Merge pull request #365 from tox-rs/timeouts

Refactor timeouts

Author: Roman Proskuryakov

src/toxcore/dht/lan_discovery.rs

@@ -10,7 +10,7 @@ use futures::sync::mpsc;
 use get_if_addrs;
 use get_if_addrs::IfAddr;
 use tokio::timer::Interval;
-use tokio::timer::timeout::Error as TimeoutError;
+use tokio::util::FutureExt;
 
 use crate::toxcore::crypto_core::*;
 use crate::toxcore::io_tokio::*;
@@ -48,9 +48,6 @@ pub const END_PORT: u16 = 33546;
 /// Interval in seconds between `LanDiscovery` packet sending.
 pub const LAN_DISCOVERY_INTERVAL: u64 = 10;
 
-/// Timeout in seconds for packet sending
-pub const LAN_DISCOVERY_SEND_TIMEOUT: u64 = 1;
-
 /// Shorthand for the transmit half of the message channel.
 type Tx = mpsc::Sender<(Packet, SocketAddr)>;
 
@@ -130,7 +127,7 @@ impl LanDiscoverySender {
     }
 
     /// Send `LanDiscovery` packets.
-    fn send(&mut self) -> impl Future<Item=(), Error=TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
+    fn send(&mut self) -> impl Future<Item=(), Error=mpsc::SendError<(Packet, SocketAddr)>> + Send {
         let addrs = self.get_broadcast_socket_addrs();
         let lan_packet = Packet::LanDiscovery(LanDiscovery {
             pk: self.dht_pk,
@@ -140,7 +137,7 @@ impl LanDiscoverySender {
             addrs.into_iter().map(move |addr| (lan_packet.clone(), addr))
         );
 
-        send_all_to_bounded(&self.tx, stream, Duration::from_secs(LAN_DISCOVERY_SEND_TIMEOUT))
+        send_all_to(&self.tx, stream)
     }
 
     /// Run LAN discovery periodically. Result future will never be completed
@@ -152,7 +149,7 @@ impl LanDiscoverySender {
             .map_err(|e| e.context(LanDiscoveryErrorKind::Wakeup).into())
             .for_each(move |_instant| {
                 trace!("LAN discovery sender wake up");
-                self.send().then(|r| {
+                self.send().timeout(interval).then(|r| {
                     if let Err(e) = r {
                         warn!("Failed to send LAN discovery packets: {}", e);
                         if let Some(e) = e.into_inner() {

src/toxcore/dht/server/mod.rs

@@ -12,7 +12,7 @@ use futures::future::{Either, join_all};
 use futures::sync::mpsc;
 use parking_lot::RwLock;
 use tokio::timer::Interval;
-use tokio::timer::timeout::Error as TimeoutError;
+use tokio::util::FutureExt;
 
 use std::collections::HashMap;
 use std::net::SocketAddr;
@@ -70,8 +70,6 @@ pub const FAKE_FRIENDS_NUMBER: usize = 2;
 /// Maximum number of entry in Lru cache for precomputed keys.
 pub const PRECOMPUTED_LRU_CACHE_SIZE: usize = KBUCKET_DEFAULT_SIZE as usize * KBUCKET_MAX_ENTRIES as usize + // For KTree.
     KBUCKET_DEFAULT_SIZE as usize * (2 + 10); // For friend's close_nodes of 2 fake friends + 10 friends reserved
-/// Timeout in seconds for packet sending
-pub const DHT_SEND_TIMEOUT: u64 = 1;
 /// How often DHT main loop should be called.
 const MAIN_LOOP_INTERVAL: u64 = 1;
 
@@ -382,15 +380,22 @@ impl Server {
             .map_err(|e| e.context(RunErrorKind::Wakeup).into())
             .for_each(move |_instant| {
                 trace!("Bootstrap wake up");
-                self.send_bootstrap_requests()
-                    .map_err(|e| e.context(RunErrorKind::SendTo).into())
+                self.send_bootstrap_requests().timeout(interval).then(|r| {
+                    if let Err(e) = r {
+                        warn!("Failed to send initial bootstrap packets: {}", e);
+                        if let Some(e) = e.into_inner() {
+                            return future::err(e.context(RunErrorKind::SendTo).into())
+                        }
+                    }
+                    future::ok(())
+                })
             })
     }
 
     /// Check if all nodes in Ktree are discarded (including the case when
     /// it's empty) and if so then send `NodesRequest` packet to nodes from
     /// initial bootstrap list and from Ktree.
-    fn send_bootstrap_requests(&self) -> impl Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
+    fn send_bootstrap_requests(&self) -> impl Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send {
         let mut request_queue = self.request_queue.write();
         let close_nodes = self.close_nodes.read();
 
@@ -417,9 +422,12 @@ impl Server {
             .map_err(|e| e.context(RunErrorKind::Wakeup).into())
             .for_each(move |_instant| {
                 trace!("DHT server wake up");
-                self.dht_main_loop().then(|res| {
+                self.dht_main_loop().timeout(interval).then(|res| {
                     if let Err(e) = res {
                         warn!("Failed to send DHT periodical packets: {}", e);
+                        if let Some(e) = e.into_inner() {
+                            return future::err(e.context(RunErrorKind::SendTo).into())
+                        }
                     }
                     future::ok(())
                 })
@@ -455,7 +463,7 @@ impl Server {
     }
 
     /// Send `PingRequest` packets to nodes from `nodes_to_ping` list.
-    fn send_pings(&self) -> impl Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
+    fn send_pings(&self) -> impl Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send {
         let nodes_to_ping = mem::replace(
             &mut *self.nodes_to_ping.write(),
             Kbucket::<PackedNode>::new(MAX_TO_PING)
@@ -478,7 +486,7 @@ impl Server {
     /// a friend and we don't know it's address then this method will send
     /// `PingRequest` immediately instead of adding to a `nodes_to_ping`
     /// list.
-    fn ping_add(&self, node: &PackedNode) -> impl Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
+    fn ping_add(&self, node: &PackedNode) -> impl Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send {
         let close_nodes = self.close_nodes.read();
 
         if !close_nodes.can_add(&node) {
@@ -502,7 +510,7 @@ impl Server {
     /// necessary to check whether node is alive before adding it to close
     /// nodes lists.
     fn ping_nodes_to_bootstrap(&self, request_queue: &mut RequestQueue<PublicKey>, nodes_to_bootstrap: &mut Kbucket<PackedNode>, pk: PublicKey)
-        -> impl Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
+        -> impl Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send {
         let capacity = nodes_to_bootstrap.capacity() as u8;
         let nodes_to_bootstrap = mem::replace(nodes_to_bootstrap, Kbucket::new(capacity));
 
@@ -516,7 +524,7 @@ impl Server {
     /// Iterate over nodes from close nodes list and send `NodesRequest` packets
     /// to them if necessary.
     fn ping_close_nodes<'a, T>(&self, request_queue: &mut RequestQueue<PublicKey>, nodes: T, pk: PublicKey)
-        -> Box<dyn Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send>
+        -> Box<dyn Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send>
         where T: Iterator<Item = &'a mut DhtNode> // if change to impl Future the result will be dependent on nodes lifetime
     {
         let futures = nodes
@@ -539,7 +547,7 @@ impl Server {
     /// it was sent less than `NODES_REQ_INTERVAL`. This function should be
     /// called every second.
     fn send_nodes_req_random<'a, T>(&self, request_queue: &mut RequestQueue<PublicKey>, nodes: T, pk: PublicKey)
-        -> Box<dyn Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send>
+        -> Box<dyn Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send>
         where T: Iterator<Item = &'a DhtNode> // if change to impl Future the result will be dependent on nodes lifetime
     {
         let good_nodes = nodes
@@ -572,7 +580,7 @@ impl Server {
 
     /// Send `PingRequest` packet to the node.
     fn send_ping_req(&self, node: &PackedNode, request_queue: &mut RequestQueue<PublicKey>)
-        -> impl Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
+        -> impl Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send {
         let payload = PingRequestPayload {
             id: request_queue.new_ping_id(node.pk),
         };
@@ -586,7 +594,7 @@ impl Server {
 
     /// Send `NodesRequest` packet to the node.
     fn send_nodes_req(&self, node: &PackedNode, request_queue: &mut RequestQueue<PublicKey>, search_pk: PublicKey)
-        -> impl Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
+        -> impl Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send {
         // Check if packet is going to be sent to ourselves.
         if self.pk == node.pk {
             trace!("Attempt to send NodesRequest to ourselves.");
@@ -607,7 +615,7 @@ impl Server {
 
     /// Send `NatPingRequest` packet to all friends and try to punch holes.
     fn send_nat_ping_req(&self, request_queue: &mut RequestQueue<PublicKey>, friends: &mut HashMap<PublicKey, DhtFriend>)
-        -> impl Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
+        -> impl Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send {
         let futures = friends.values_mut()
             .filter(|friend| !friend.is_addr_known())
             .map(|friend| {
@@ -645,7 +653,7 @@ impl Server {
 
     /// Try to punch holes to specified friend.
     fn punch_holes(&self, request_queue: &mut RequestQueue<PublicKey>, friend: &mut DhtFriend, returned_addrs: &[SocketAddr])
-        -> impl Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
+        -> impl Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send {
         let punch_addrs = friend.hole_punch.next_punch_addrs(returned_addrs);
 
         let packets = punch_addrs.into_iter().map(|addr| {
@@ -661,13 +669,13 @@ impl Server {
             (packet, addr)
         }).collect::<Vec<_>>();
 
-        send_all_to_bounded(&self.tx, stream::iter_ok(packets), Duration::from_secs(DHT_SEND_TIMEOUT))
+        send_all_to(&self.tx, stream::iter_ok(packets))
     }
 
     /// Send `NatPingRequest` packet to all close nodes of friend in the hope
     /// that they will redirect it to this friend.
     fn send_nat_ping_req_inner(&self, friend: &DhtFriend, nat_ping_req_packet: DhtRequest)
-        -> impl Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
+        -> impl Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send {
         let packet = Packet::DhtRequest(nat_ping_req_packet);
         let futures = friend.close_nodes.nodes
             .iter()
@@ -713,8 +721,8 @@ impl Server {
 
     /// Send UDP packet to specified address.
     fn send_to(&self, addr: SocketAddr, packet: Packet)
-        -> impl Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
-        send_to_bounded(&self.tx, (packet, addr), Duration::from_secs(DHT_SEND_TIMEOUT))
+        -> impl Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send {
+        send_to(&self.tx, (packet, addr))
     }
 
     /// Handle received `PingRequest` packet and response with `PingResponse`
@@ -1335,7 +1343,7 @@ impl Server {
     /// Handle `OnionRequest` from TCP relay and send `OnionRequest1` packet
     /// to the next node in the onion path.
     pub fn handle_tcp_onion_request(&self, packet: OnionRequest, addr: SocketAddr)
-        -> impl Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
+        -> impl Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send {
         let onion_symmetric_key = self.onion_symmetric_key.read();
 
         let onion_return = OnionReturn::new(

src/toxcore/io_tokio.rs

@@ -2,11 +2,8 @@
 
 use std::fmt::Debug;
 use std::io::{Error as IoError};
-use std::time::Duration;
 
 use futures::{Future, Sink, Stream};
-use tokio::util::FutureExt;
-use tokio::timer::timeout::Error as TimeoutError;
 
 /// A convenience typedef around a `Future` whose error component is `io::Error`
 pub type IoFuture<T> = Box<Future<Item = T, Error = IoError> + Send>;
@@ -24,14 +21,6 @@ pub fn send_to<T: Send + 'static, Tx, E: Debug>(tx: &Tx, v: T) -> impl Future<It
         .map(|_tx| ()) // ignore tx because it was cloned
 }
 
-/// Send item to a sink using reference with timeout
-pub fn send_to_bounded<T: Send + 'static, Tx, E: Debug>(tx: &Tx, v: T, timeout: Duration) -> impl Future<Item=(), Error=TimeoutError<E>> + Send
-    where Tx: Sink<SinkItem = T, SinkError = E> + Send + Clone + 'static
-{
-    send_to(tx, v)
-        .timeout(timeout)
-}
-
 /// Send item to a sink using reference
 pub fn send_all_to<T: Send + 'static, S, Tx, E: Debug>(tx: &Tx, s: S) -> impl Future<Item=(), Error=E> + Send
     where S: Stream<Item = T, Error = E> + Send + 'static,
@@ -43,11 +32,3 @@ pub fn send_all_to<T: Send + 'static, S, Tx, E: Debug>(tx: &Tx, s: S) -> impl Fu
         .map(|_tx| ()) // ignore tx because it was cloned
 }
 
-/// Send item to a sink using reference with timeout
-pub fn send_all_to_bounded<T: Send + 'static, S, Tx, E: Debug>(tx: &Tx, s: S, timeout: Duration) -> impl Future<Item=(), Error=TimeoutError<E>> + Send
-    where S: Stream<Item = T, Error = E> + Send + 'static,
-          Tx: Sink<SinkItem = T, SinkError = E> + Send + Clone + 'static
-{
-    send_all_to(tx, s)
-        .timeout(timeout)
-}

src/toxcore/net_crypto/mod.rs

@@ -31,7 +31,7 @@ use futures::future::Either;
 use futures::sync::mpsc;
 use parking_lot::RwLock;
 use tokio::timer::Interval;
-use tokio::timer::timeout::Error as TimeoutError;
+use tokio::util::FutureExt;
 
 use crate::toxcore::binary_io::*;
 use crate::toxcore::crypto_core::*;
@@ -68,9 +68,6 @@ const PACKET_ID_LOSSY_RANGE_START: u8 = 192;
 /// `PACKET_ID_LOSSY_RANGE_END` are considered lossy packets.
 const PACKET_ID_LOSSY_RANGE_END: u8 = 254;
 
-/// Timeout for packet sending
-const NET_CRYPTO_SEND_TIMEOUT: Duration = Duration::from_millis(50);
-
 /// Shorthand for the transmit half of the message channel for sending DHT
 /// packets.
 type UdpTx = mpsc::Sender<(Packet, SocketAddr)>;
@@ -292,8 +289,8 @@ impl NetCrypto {
     }
 
     /// Send `Packet` packet to UDP socket
-    fn send_to_udp(&self, addr: SocketAddr, packet: Packet) -> impl Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
-        send_to_bounded(&self.udp_tx, (packet, addr), NET_CRYPTO_SEND_TIMEOUT)
+    fn send_to_udp(&self, addr: SocketAddr, packet: Packet) -> impl Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send {
+        send_to(&self.udp_tx, (packet, addr))
     }
 
     /// Get long term `PublicKey` of the peer by its UDP address
@@ -747,7 +744,7 @@ impl NetCrypto {
 
     /// Send packet to crypto connection choosing TCP or UDP protocol
     fn send_packet(&self, packet: Packet, connection: &mut CryptoConnection)
-        -> impl Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
+        -> impl Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send {
         // TODO: can backpressure be used instead of congestion control? It
         // seems it's possible to implement wrapper for bounded sender with
         // priority queue and just send packets there
@@ -778,7 +775,7 @@ impl NetCrypto {
     /// Send `CookieRequest` or `CryptoHandshake` packet if needed depending on
     /// connection status and update sent counter
     fn send_status_packet(&self, connection: &mut CryptoConnection)
-        -> impl Future<Item = (), Error = TimeoutError<mpsc::SendError<(Packet, SocketAddr)>>> + Send {
+        -> impl Future<Item = (), Error = mpsc::SendError<(Packet, SocketAddr)>> + Send {
         match connection.packet_to_send() {
             Some(packet) => Either::A(self.send_packet(packet, connection)),
             None => Either::B(future::ok(())),
@@ -900,8 +897,15 @@ impl NetCrypto {
 
         wakeups
             .map_err(|e| e.context(RunErrorKind::Wakeup).into())
-            .for_each(move |_instant| self.main_loop()
-                .map_err(|e| e.context(RunErrorKind::SendData).into())
+            .for_each(move |_instant| self.main_loop().timeout(PACKET_COUNTER_AVERAGE_INTERVAL).then(|r| {
+                    if let Err(e) = r {
+                        warn!("Failed to send net crypto packets: {}", e);
+                        if let Some(e) = e.into_inner() {
+                            return future::err(e.context(RunErrorKind::SendData).into())
+                        }
+                    }
+                    future::ok(())
+                })
             )
     }
 }

src/toxcore/tcp/server/client.rs

@@ -16,6 +16,7 @@ use std::time::{Instant, Duration};
 
 use futures::Future;
 use futures::sync::mpsc;
+use tokio::util::FutureExt;
 
 /// Interval in seconds for sending TCP PingRequest
 pub const TCP_PING_FREQUENCY: u64 = 30;
@@ -128,7 +129,7 @@ impl Client {
     /** Send a packet. This method does not ignore IO error
     */
     fn send(&self, packet: Packet) -> impl Future<Item = (), Error = Error> + Send {
-        send_to_bounded(&self.tx, packet, Duration::from_secs(TCP_SEND_TIMEOUT)).map_err(|e|
+        send_to(&self.tx, packet).timeout(Duration::from_secs(TCP_SEND_TIMEOUT)).map_err(|e|
             Error::new(ErrorKind::Other,
                 format!("Failed to send packet: {:?}", e)
         ))