Introduced the concept of the deadline stopwatch and deadline based locking strategies.

DevP2P uses now this strategy to access data from the transaction queue, with a deadline of 200ms for retrieving transactions (supplier) and 1000ms for planning transaction retrieval.
DMDcoin#236

Author: SurfingNerd

Cargo.lock

@@ -24,6 +24,7 @@ use std::{
         atomic::{self, AtomicUsize},
         Arc,
     },
+    time::Duration,
 };
 
 use self::scoring::ScoringEvent;
@@ -680,12 +681,23 @@ impl TransactionQueue {
     ///
     /// Given transaction hash looks up that transaction in the pool
     /// and returns a shared pointer to it or `None` if it's not present, or a readlock could not get acquired.
-    pub fn find_if_readable(&self, hash: &H256) -> Option<Arc<pool::VerifiedTransaction>> {
+    pub fn find_if_readable(
+        &self,
+        hash: &H256,
+        max_lock_duration: &Duration,
+    ) -> Option<Arc<pool::VerifiedTransaction>> {
+        let splitted_duration = max_lock_duration.div_f32(3.0);
         self.cached_enforced_pending
-            .try_read()?
+            .try_read_for(splitted_duration.clone())?
             .find(hash)
-            .or(self.cached_non_enforced_pending.try_read()?.find(hash))
-            .or(self.pool.try_read()?.find(hash))
+            .or(self
+                .cached_non_enforced_pending
+                .try_read_for(splitted_duration.clone())?
+                .find(hash))
+            .or(self
+                .pool
+                .try_read_for(splitted_duration.clone())?
+                .find(hash))
     }
 
     /// Remove a set of transactions from the pool.

crates/concensus/miner/src/pool/queue.rs

@@ -2426,8 +2426,14 @@ impl BlockChainClient for Client {
         self.importer.miner.transaction(&hash)
     }
 
-    fn transaction_if_readable(&self, hash: &H256) -> Option<Arc<VerifiedTransaction>> {
-        self.importer.miner.transaction_if_readable(&hash)
+    fn transaction_if_readable(
+        &self,
+        hash: &H256,
+        max_lock_duration: &Duration,
+    ) -> Option<Arc<VerifiedTransaction>> {
+        self.importer
+            .miner
+            .transaction_if_readable(&hash, max_lock_duration)
     }
 
     fn uncle(&self, id: UncleId) -> Option<encoded::Header> {

crates/ethcore/src/client/client.rs

@@ -23,6 +23,7 @@ use std::{
         Arc,
         atomic::{AtomicBool, AtomicUsize, Ordering as AtomicOrder},
     },
+    time::Duration,
 };
 
 use crate::{
@@ -1160,8 +1161,12 @@ impl BlockChainClient for TestBlockChainClient {
         self.miner.transaction(tx_hash)
     }
 
-    fn transaction_if_readable(&self, hash: &H256) -> Option<Arc<VerifiedTransaction>> {
-        self.miner.transaction_if_readable(hash)
+    fn transaction_if_readable(
+        &self,
+        hash: &H256,
+        max_lock_duration: &Duration,
+    ) -> Option<Arc<VerifiedTransaction>> {
+        self.miner.transaction_if_readable(hash, max_lock_duration)
     }
 
     /// Returns the devp2p network endpoint IP and Port information that is used to communicate with other peers.

crates/ethcore/src/client/test_client.rs

@@ -20,6 +20,7 @@ use std::{
     collections::{BTreeMap, BTreeSet},
     net::SocketAddr,
     sync::Arc,
+    time::Duration,
 };
 
 use crate::{
@@ -425,7 +426,11 @@ pub trait BlockChainClient:
 
     /// see queued_transactions(&self).
     /// Get pool transaction with a given hash, but returns NONE fast, if if cannot acquire a readlock fast.
-    fn transaction_if_readable(&self, hash: &H256) -> Option<Arc<VerifiedTransaction>>;
+    fn transaction_if_readable(
+        &self,
+        hash: &H256,
+        max_lock_duration: &Duration,
+    ) -> Option<Arc<VerifiedTransaction>>;
 
     /// Sorted list of transaction gas prices from at least last sample_size blocks.
     fn gas_price_corpus(&self, sample_size: usize) -> ::stats::Corpus<U256> {

crates/ethcore/src/client/traits.rs

@@ -1468,8 +1468,13 @@ impl miner::MinerService for Miner {
         return result;
     }
 
-    fn transaction_if_readable(&self, hash: &H256) -> Option<Arc<VerifiedTransaction>> {
-        self.transaction_queue.find_if_readable(hash)
+    fn transaction_if_readable(
+        &self,
+        hash: &H256,
+        max_lock_duration: &Duration,
+    ) -> Option<Arc<VerifiedTransaction>> {
+        self.transaction_queue
+            .find_if_readable(hash, max_lock_duration)
     }
 
     fn remove_transaction(&self, hash: &H256) -> Option<Arc<VerifiedTransaction>> {

crates/ethcore/src/miner/miner.rs

@@ -35,6 +35,7 @@ pub use ethcore_miner::{
 use std::{
     collections::{BTreeMap, BTreeSet},
     sync::Arc,
+    time::Duration,
 };
 
 use crate::types::{
@@ -201,7 +202,11 @@ pub trait MinerService: Send + Sync {
 
     /// Query transaction from the pool given it's hash, without blocking.
     /// Might return "None" in cases when the lock could not get acquired.
-    fn transaction_if_readable(&self, hash: &H256) -> Option<Arc<VerifiedTransaction>>;
+    fn transaction_if_readable(
+        &self,
+        hash: &H256,
+        max_lock_duration: &Duration,
+    ) -> Option<Arc<VerifiedTransaction>>;
 
     /// Returns next valid nonce for given address.
     ///

crates/ethcore/src/miner/mod.rs

@@ -38,6 +38,7 @@ rand_xorshift = "0.2.0"
 rlp = { version = "0.4.6" }
 trace-time = "0.1"
 triehash-ethereum = {version = "0.2", path = "../../util/triehash-ethereum" }
+time-utils = { path = "../../util/time-utils" }
 stats = { path = "../../util/stats" }
 crossbeam-channel = "0.5.2"
 

crates/ethcore/sync/Cargo.toml

@@ -32,7 +32,11 @@ use ethereum_types::{H256, H512, U256};
 use hash::keccak;
 use network::{PeerId, client_version::ClientVersion};
 use rlp::Rlp;
-use std::{cmp, mem, time::Instant};
+use std::{
+    cmp, mem,
+    time::{Duration, Instant},
+};
+use time_utils::DeadlineStopwatch;
 
 use super::{
     request_id::strip_request_id,
@@ -858,6 +862,11 @@ impl SyncHandler {
         peer_id: PeerId,
         tx_rlp: &Rlp,
     ) -> Result<(), DownloaderImportError> {
+        // those P2P operations must not take forever, a better , configurable but balanced timeout managment would be nice to have.
+        let max_duration = Duration::from_millis(500);
+
+        let deadline = DeadlineStopwatch::new(max_duration);
+
         for item in tx_rlp {
             let hash = item
                 .as_val::<H256>()
@@ -868,7 +877,17 @@ impl SyncHandler {
 
             // if we cant read the pool here, we are asuming we dont know the transaction yet.
             // in the worst case we are refetching a transaction that we already have.
-            if io.chain().transaction_if_readable(&hash).is_none() {
+
+            if deadline.is_expired() {
+                // we did run out of time to finish this opation, but thats Ok.
+                return Ok(());
+            }
+
+            if io
+                .chain()
+                .transaction_if_readable(&hash, &deadline.time_left())
+                .is_none()
+            {
                 sync.peers
                     .get_mut(&peer_id)
                     .map(|peer| peer.unfetched_pooled_transactions.insert(hash));

crates/ethcore/sync/src/chain/handler.rs

@@ -18,6 +18,7 @@ use bytes::Bytes;
 
 #[cfg(not(test))]
 use devp2p::PAYLOAD_SOFT_LIMIT;
+use time_utils::DeadlineStopwatch;
 #[cfg(test)]
 pub const PAYLOAD_SOFT_LIMIT: usize = 100_000;
 
@@ -27,7 +28,7 @@ use ethereum_types::{H256, H512};
 use network::{self, PeerId};
 use parking_lot::RwLock;
 use rlp::{Rlp, RlpStream};
-use std::cmp;
+use std::{cmp, time::Duration};
 
 use crate::sync_io::SyncIo;
 
@@ -316,14 +317,23 @@ impl SyncSupplier {
         rlp.begin_unbounded_list();
         let mut not_found = 0;
         let mut parse_errors = 0;
+
+        let deadline = DeadlineStopwatch::new(Duration::from_millis(200));
         for v in r {
             if let Ok(hash) = v.as_val::<H256>() {
                 // io.chain().transaction(hash)
 
+                if !deadline.should_continue() {
+                    break;
+                }
+
                 // we do not lock here, if we cannot access the memory at this point in time,
                 // we will just skip this transaction, otherwise the other peer might wait to long, resulting in a timeout.
                 // also this solved a potential deadlock situation:
-                if let Some(tx) = io.chain().transaction_if_readable(&hash) {
+                if let Some(tx) = io
+                    .chain()
+                    .transaction_if_readable(&hash, &deadline.time_left())
+                {
                     tx.signed().rlp_append(&mut rlp);
                     added += 1;
                     if rlp.len() > PAYLOAD_SOFT_LIMIT {