Merge branch 'master' into v0.5-alpha

CHANGELOG.md

@@ -4,10 +4,20 @@
 
 ## Improvements
 
-- Add support for WebSockets in RPC.
-
 # 0.5.0
 
+## Improvements
+
+- Add fields in Receipt: gas_fee, gas_sponsored, storage_sponsored. Accumulate gas_used in Receipt, not gas_charged.
+
+- Delay block requests when we cannot process them to avoid wasting network bandwidth.
+
+- Set block gas limit for Genesis block to 30_000_000.
+
+- Define gas_used to be transaction gas limit for NotEnoughCash, the same as all other exceptions.
+
+- Add support for WebSockets in RPC.
+
 ## Bug Fixes
 
 - Fix UnexpectedResponse from honest peers that causes peer demotion.
@@ -20,7 +30,17 @@
 
 - Fix a race condition that may cause optimistic execution to panic.
 
-- Delay block requests when we cannot process them to avoid wasting network bandwidth.
+- Fill in correct block gas limit value for mining.
+
+- Fix definitions and logics in transaction early execution error checking.
+
+- Use block_count - 1 in target difficulty calculation because it's the unbiased estimation of exponential distribution parameter (past mining power).
+
+## Improvements
+
+- Add cfx_getConfirmationRiskByHash RPC to get confirmation risk by block hash.
+
+- Add getTransactionsFromPool debug RPC to collect transactions in pool.
 
 # 0.4.0
 
@@ -34,23 +54,11 @@
 
 - Add missing transaction verifications for invalid block.
 
-- Fill in correct block gas limit value for mining.
-
-- Set block gas limit for Genesis block to 30_000_000.
-
-- Fix definitions and logics in transaction early execution error checking.
-
-- Use block_count - 1 in target difficulty calculation because it's the unbiased estimation of exponential distribution parameter (past mining power).
-
-- Add fields in Receipt: gas_fee, gas_sponsored, storage_sponsored. Accumulate gas_used in Receipt, not gas_charged.
-
-- Define gas_used to be transaction gas limit for NotEnoughCash, the same as all other exceptions.
-
 ## Improvements
 
 - Improve the transaction address check at RPC
 
-- Chagne the test net PoW to use double keccak
+- Change the test net PoW to use double keccak
 
 ## EVM Updates
 

client/Cargo.toml

@@ -4,6 +4,8 @@ version = "0.6.0"
 edition = "2018"
 
 [dependencies]
+bigdecimal = "0.1.0"
+num-bigint = "0.2.6"
 mio = "0.6.8"
 clap = "2"
 term_size = "0.3"

client/src/rpc/impls/cfx.rs

@@ -896,6 +896,7 @@ impl Cfx for CfxHandler {
                 -> JsonRpcResult<RpcBlock>;
             fn block_by_hash(&self, hash: RpcH256, include_txs: bool)
                 -> JsonRpcResult<Option<RpcBlock>>;
+            fn confirmation_risk_by_hash(&self, block_hash: RpcH256) -> JsonRpcResult<Option<RpcU256>>;
             fn blocks_by_epoch(&self, num: EpochNumber) -> JsonRpcResult<Vec<RpcH256>>;
             fn skipped_blocks_by_epoch(&self, num: EpochNumber) -> JsonRpcResult<Vec<RpcH256>>;
             fn epoch_number(&self, epoch_num: Option<EpochNumber>) -> JsonRpcResult<RpcU256>;
@@ -1014,6 +1015,7 @@ impl LocalRpc for LocalRpcImpl {
             fn tx_inspect(&self, hash: RpcH256) -> JsonRpcResult<BTreeMap<String, String>>;
             fn txpool_content(&self) -> JsonRpcResult<
                 BTreeMap<String, BTreeMap<String, BTreeMap<usize, Vec<RpcTransaction>>>>>;
+            fn txs_from_pool(&self) -> JsonRpcResult<Vec<RpcTransaction>>;
             fn txpool_inspect(&self) -> JsonRpcResult<
                 BTreeMap<String, BTreeMap<String, BTreeMap<usize, Vec<String>>>>>;
             fn txpool_status(&self) -> JsonRpcResult<BTreeMap<String, usize>>;

client/src/rpc/impls/common.rs

@@ -2,7 +2,9 @@
 // Conflux is free software and distributed under GNU General Public License.
 // See http://www.gnu.org/licenses/
 
+use bigdecimal::BigDecimal;
 use jsonrpc_core::{Error as RpcError, Result as RpcResult, Value as RpcValue};
+use num_bigint::{BigInt, ToBigInt};
 use parking_lot::{Condvar, Mutex};
 use std::{
     collections::{BTreeMap, HashSet},
@@ -11,7 +13,7 @@ use std::{
     time::Duration,
 };
 
-use cfx_types::{Address, H256, U128};
+use cfx_types::{Address, H256, U128, U256};
 use cfxcore::{
     BlockDataManager, ConsensusGraph, ConsensusGraphTrait, PeerInfo,
     SharedConsensusGraph, SharedTransactionPool,
@@ -154,6 +156,36 @@ impl RpcImpl {
         }
     }
 
+    pub fn confirmation_risk_by_hash(
+        &self, block_hash: RpcH256,
+    ) -> RpcResult<Option<RpcU256>> {
+        let consensus_graph = self
+            .consensus
+            .as_any()
+            .downcast_ref::<ConsensusGraph>()
+            .expect("downcast should succeed");
+        let inner = &*consensus_graph.inner.read();
+        let result = consensus_graph
+            .confirmation_meter
+            .confirmation_risk_by_hash(inner, block_hash.into());
+        if result.is_none() {
+            Ok(None)
+        } else {
+            let risk: BigDecimal = result.unwrap().into();
+            let scale = BigInt::parse_bytes(b"FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF", 16).expect("failed to unwrap U256::max into bigInt");
+
+            //TODO: there's a precision problem here, it should be fine under a
+            // (2^256 - 1) scale
+            let scaled_risk: BigInt = (risk * scale)
+                .to_bigint()
+                .expect("failed to convert scaled risk to bigInt");
+            let (sign, big_endian_bytes) = scaled_risk.to_bytes_be();
+            assert_ne!(sign, num_bigint::Sign::Minus);
+            let rpc_result = U256::from(big_endian_bytes.as_slice());
+            Ok(Some(rpc_result.into()))
+        }
+    }
+
     pub fn block_by_hash(
         &self, hash: RpcH256, include_txs: bool,
     ) -> RpcResult<Option<RpcBlock>> {
@@ -527,6 +559,19 @@ impl RpcImpl {
         Ok(ret)
     }
 
+    pub fn txs_from_pool(&self) -> RpcResult<Vec<RpcTransaction>> {
+        let (ready_txs, deferred_txs) = self.tx_pool.content();
+        let converter = |tx: &Arc<SignedTransaction>| -> RpcTransaction {
+            RpcTransaction::from_signed(&tx, None)
+        };
+        let result = ready_txs
+            .iter()
+            .map(converter)
+            .chain(deferred_txs.iter().map(converter))
+            .collect();
+        return Ok(result);
+    }
+
     pub fn txpool_content(
         &self,
     ) -> RpcResult<

client/src/rpc/impls/light.rs

@@ -467,6 +467,7 @@ impl Cfx for CfxHandler {
             fn gas_price(&self) -> RpcResult<RpcU256>;
             fn next_nonce(&self, address: RpcH160, num: Option<BlockHashOrEpochNumber>) -> RpcResult<RpcU256>;
             fn skipped_blocks_by_epoch(&self, num: EpochNumber) -> RpcResult<Vec<RpcH256>>;
+            fn confirmation_risk_by_hash(&self, block_hash: RpcH256) -> RpcResult<Option<RpcU256>>;
         }
 
         to self.rpc_impl {
@@ -562,6 +563,7 @@ impl LocalRpc for DebugRpcImpl {
             fn net_throttling(&self) -> RpcResult<throttling::Service>;
             fn tx_inspect(&self, hash: RpcH256) -> RpcResult<BTreeMap<String, String>>;
             fn txpool_content(&self) -> RpcResult<BTreeMap<String, BTreeMap<String, BTreeMap<usize, Vec<RpcTransaction>>>>>;
+            fn txs_from_pool(&self) -> RpcResult<Vec<RpcTransaction>>;
             fn txpool_inspect(&self) -> RpcResult<BTreeMap<String, BTreeMap<String, BTreeMap<usize, Vec<String>>>>>;
             fn txpool_status(&self) -> RpcResult<BTreeMap<String, usize>>;
             fn accounts(&self) -> RpcResult<Vec<RpcH160>>;

client/src/rpc/traits/cfx.rs

@@ -191,6 +191,11 @@ pub trait Cfx {
         &self, epoch_number: Option<EpochNumber>,
     ) -> JsonRpcResult<RpcU256>;
 
+    #[rpc(name = "cfx_getConfirmationRiskByHash")]
+    fn confirmation_risk_by_hash(
+        &self, block_hash: RpcH256,
+    ) -> JsonRpcResult<Option<RpcU256>>;
+
     //        /// Returns transaction at given block hash and index.
     //        #[rpc(name = "cfx_getTransactionByBlockHashAndIndex")]
     //        fn transaction_by_block_hash_and_index(&self, RpcH256, Index) ->

client/src/rpc/traits/debug.rs

@@ -43,6 +43,9 @@ pub trait LocalRpc {
         >,
     >;
 
+    #[rpc(name = "getTransactionsFromPool")]
+    fn txs_from_pool(&self) -> JsonRpcResult<Vec<RpcTransaction>>;
+
     #[rpc(name = "clear_tx_pool")]
     fn clear_tx_pool(&self) -> JsonRpcResult<()>;
 

core/src/consensus/consensus_inner/confirmation_meter.rs

@@ -153,7 +153,7 @@ impl ConfirmationMeter {
             }
             Some(max_risk)
         } else {
-            None
+            Some(0.9)
         }
     }
 

core/src/sync/message/message.rs

@@ -278,6 +278,9 @@ pub fn handle_rlp_message(
         msgid::STATE_SYNC_CANDIDATE_RESPONSE => {
             handle_message::<StateSyncCandidateResponse>(ctx, rlp)?;
         }
+        msgid::THROTTLED => {
+            handle_message::<Throttled>(ctx, rlp)?;
+        }
         _ => return Ok(false),
     }
 

network/src/node_database.rs

@@ -391,6 +391,10 @@ impl NodeDatabase {
                         if let Some(removed_node) =
                             self.untrusted_nodes.remove_with_id(id)
                         {
+                            self.promote_with_untrusted(
+                                id,
+                                removed_node.endpoint.address.ip(),
+                            );
                             // IP address not changed and always allow to add.
                             self.trusted_node_tag_index.add_node(&removed_node);
                             self.trusted_nodes.add_node(

run/throttling.toml

@@ -9,21 +9,21 @@
 
 [sync_protocol]
 Status="90,90,1,30,0" # heartbeat interval 30s
-NewBlockHashes="200,200,20,1,10"
-Transactions="50,50,10,1,5"
-GetBlockHeaders="200,200,20,1,10"
-NewBlock="20,20,4,1,2"
-GetBlocks="50,50,10,1,5"
-GetCompactBlocks="200,200,20,1,10"
-GetBlockTxn="200,200,20,1,10"
-DynamicCapabilityChange="20,20,5,1,3"
-TransactionDigests="50,50,10,1,5"
-GetTransactions="50,50,10,1,5"
-GetTransactionsFromTxHashes="50,50,10,1,5"
-GetBlockHashesByEpoch="50,50,10,1,5"
-SnapshotManifestRequest="50,50,10,1,5"
-SnapshotChunkRequest="50,50,10,1,5"
-Throttled="100,100,10,1,5"
+NewBlockHashes="200,200,20,1,100"
+Transactions="50,50,10,1,50"
+GetBlockHeaders="200,200,100,1,100"
+NewBlock="20,20,4,1,20"
+GetBlocks="200,200,100,1,50"
+GetCompactBlocks="200,200,20,1,100"
+GetBlockTxn="200,200,20,1,100"
+DynamicCapabilityChange="20,20,5,1,30"
+TransactionDigests="50,50,10,1,50"
+GetTransactions="50,50,10,1,50"
+GetTransactionsFromTxHashes="50,50,10,1,50"
+GetBlockHashesByEpoch="50,50,10,1,50"
+SnapshotManifestRequest="50,50,10,1,50"
+SnapshotChunkRequest="50,50,10,1,50"
+Throttled="100,100,10,1,50"
 
 # Suggest to limit the IP address to access the RPC as well.
 [rpc]
@@ -43,3 +43,7 @@ cfx_getTransactionByHash="50,50,10,1,5"
 cfx_estimateGas="10,10,2,1,1"
 cfx_getBlocksByEpoch="20,20,5,1,5"
 cfx_getTransactionReceipt="50,50,10,1,5"
+
+[rpc_local]
+
+[light_protocol]

tests/network_tests/p2p_throttling_test.py

@@ -0,0 +1,82 @@
+#!/usr/bin/env python3
+import os
+import sys
+
+sys.path.insert(1, os.path.join(sys.path[0], '..'))
+
+from test_framework.block_gen_thread import BlockGenThread
+from eth_utils import decode_hex
+from rlp.sedes import Binary, BigEndianInt
+from conflux import utils
+from conflux.rpc import RpcClient
+from conflux.utils import encode_hex, bytes_to_int, priv_to_addr, parse_as_int
+from test_framework.blocktools import create_block
+from test_framework.test_framework import ConfluxTestFramework
+from test_framework.mininode import *
+from test_framework.util import *
+
+
+class P2PThrottlingTest(ConfluxTestFramework):
+    def set_test_params(self):
+        self.num_nodes = 2
+        self.conf_parameters["generate_tx"] = "true"
+        self.conf_parameters["generate_tx_period_us"] = "50000"
+
+        # Token bucket: <max_tokens>,<init_tokens>,<recharge_rate>,<default_cost>,<max_throttled_tolerates>
+        # These parameters are set to ensure that throttling will be triggered,
+        # and will not exceed max_throttled_tolerates.
+        self.throttling_array = [20, 10, 10, 1, 50]
+        throttling_setting = ",".join([str(i) for i in self.throttling_array])
+        throttling_file = "throttling.toml"
+        self.conf_parameters["throttling_conf"] = f"\"{throttling_file}\""
+        # Use heartbeat to trigger block sync from terminals.
+        self.conf_parameters["heartbeat_period_interval_ms"] = "1000"
+        self.extra_conf_files = {
+            f"{throttling_file}":
+                f"""\
+                [rpc]
+                [rpc_local]
+                [sync_protocol]
+                NewBlockHashes="{throttling_setting}"
+                Transactions="{throttling_setting}"
+                GetBlockHeaders="{throttling_setting}"
+                NewBlock="{throttling_setting}"
+                GetBlocks="{throttling_setting}"
+                GetCompactBlocks="{throttling_setting}"
+                GetBlockTxn="{throttling_setting}"
+                DynamicCapabilityChange="{throttling_setting}"
+                TransactionDigests="{throttling_setting}"
+                GetTransactions="{throttling_setting}"
+                GetTransactionsFromTxHashes="{throttling_setting}"
+                GetBlockHashesByEpoch="{throttling_setting}"
+                SnapshotManifestRequest="{throttling_setting}"
+                SnapshotChunkRequest="{throttling_setting}"
+                Throttled="{throttling_setting}"
+                [light_protocol]\
+                """
+        }
+
+    def setup_network(self):
+        self.setup_nodes()
+        connect_sample_nodes(self.nodes, self.log)
+        sync_blocks(self.nodes)
+
+    def run_test(self):
+        n_blocks = 200
+        rate = self.throttling_array[2]
+        init_token = self.throttling_array[0]
+        max_throttled_tolerates = self.throttling_array[4]
+        start = time.time()
+        # Generate blocks with about twice the throttling rate.
+        for _ in range(int(n_blocks/rate)):
+            self.nodes[0].generate_empty_blocks(rate)
+            time.sleep(0.5)
+        self.log.info("Wait for blocks to be synced")
+        sync_blocks(self.nodes, timeout=120)
+        elapsed = time.time() - start
+        assert elapsed > (n_blocks - init_token - max_throttled_tolerates) /rate
+        self.log.info(f"Pass with {elapsed} seconds")
+
+
+if __name__ == "__main__":
+    P2PThrottlingTest().main()