refactor(tcp_server): release state lock before sending packets

Currently we send tcp packets while holding the state lock. At the
same time we use bounded queues for sending of packets. This means
that a single stuck client can block the whole server until the
timeout happens. This change makes the server release the state lock
before sending any packet.

Author: kurnevsky

tox_core/src/relay/server/client.rs

@@ -3,26 +3,20 @@
 
 use tox_crypto::*;
 use tox_packet::relay::*;
-use tox_packet::relay::connection_id::ConnectionId;
 use crate::relay::links::Links;
-use tox_packet::onion::InnerOnionResponse;
 use crate::time::*;
 use crate::utils::*;
 
-use std::io::{Error, ErrorKind};
 use std::net::IpAddr;
 use std::time::{Instant, Duration};
 
 use futures::channel::mpsc;
-use futures::SinkExt;
 use rand::thread_rng;
 
 /// Interval of time for sending TCP PingRequest
 pub const TCP_PING_FREQUENCY: Duration = Duration::from_secs(30);
 /// Interval of time for waiting response of PingRequest sent
 pub const TCP_PING_TIMEOUT: Duration = Duration::from_secs(10);
-/// Interval of time for packet sending
-pub const TCP_SEND_TIMEOUT: Duration = Duration::from_secs(1);
 
 /** Structure that represents how Server keeps connected clients. A write-only socket with
 human interface. A client cannot send a message directly to another client, whereas server can.
@@ -125,93 +119,16 @@ impl Client {
         &mut self.links
     }
 
-    /** Send a packet. This method does not ignore IO error
-    */
-    async fn send(&self, packet: Packet) -> Result<(), Error> {
-        let mut tx = self.tx.clone();
-
-        let timeout = tokio::time::timeout(
-            TCP_SEND_TIMEOUT,
-            tx.send(packet)
-        );
-
-        match timeout.await {
-            Err(e) => Err(Error::new(
-                ErrorKind::Other,
-                format!("Failed to send packet: {:?}", e)
-            )),
-            Ok(Err(e)) => Err(Error::new(
-                ErrorKind::Other,
-                format!("Failed to send packet: {:?}", e)
-            )),
-            Ok(_) => Ok(())
-        }
-    }
-    /** Send a packet. This method ignores IO error
-    */
-    async fn send_ignore_error(&self, packet: Packet) {
-        // ignore if somehow failed to send it
-        self.send(packet).await.ok();
-    }
-    /** Construct RouteResponse and send it to Client
-    */
-    pub async fn send_route_response(&self, pk: PublicKey, connection_id: ConnectionId) -> Result<(), Error> {
-        self.send(
-            Packet::RouteResponse(RouteResponse { connection_id, pk })
-        ).await
-    }
-    /** Construct ConnectNotification and send it to Client ignoring IO error
-    */
-    pub async fn send_connect_notification(&self, connection_id: ConnectionId) {
-        self.send_ignore_error(
-            Packet::ConnectNotification(ConnectNotification { connection_id })
-        ).await;
-    }
-    /** Construct DisconnectNotification and send it to Client ignoring IO error
-    */
-    pub async fn send_disconnect_notification(&self, connection_id: ConnectionId) {
-        self.send_ignore_error(
-            Packet::DisconnectNotification(DisconnectNotification { connection_id })
-        ).await;
+    pub fn tx(&self) -> mpsc::Sender<Packet> {
+        self.tx.clone()
     }
-    /** Construct PongResponse and send it to Client
-    */
-    pub async fn send_pong_response(&self, ping_id: u64) -> Result<(), Error> {
-        self.send(
-            Packet::PongResponse(PongResponse { ping_id })
-        ).await
-    }
-    /** Construct OobReceive and send it to Client ignoring IO error
-    */
-    pub async fn send_oob(&self, sender_pk: PublicKey, data: Vec<u8>) {
-        self.send_ignore_error(
-            Packet::OobReceive(OobReceive { sender_pk, data })
-        ).await;
-    }
-    /** Construct OnionResponse and send it to Client
-    */
-    pub async fn send_onion_response(&self, payload: InnerOnionResponse) -> Result<(), Error> {
-        self.send(
-            Packet::OnionResponse(OnionResponse { payload })
-        ).await
-    }
-    /** Construct Data and send it to Client
-    */
-    pub async fn send_data(&self, connection_id: ConnectionId, data: DataPayload) -> Result<(), Error> {
-        self.send(
-            Packet::Data(Data { connection_id, data })
-        ).await
-    }
-    /** Construct PingRequest and send it to Client
-    */
-    pub async fn send_ping_request(&mut self) -> Result<(), Error> {
+
+    pub fn new_ping_id(&mut self) -> u64 {
         let ping_id = gen_ping_id(&mut thread_rng());
 
         self.last_pinged = Instant::now();
         self.ping_id = ping_id;
 
-        self.send(
-            Packet::PingRequest(PingRequest { ping_id })
-        ).await
+        ping_id
     }
 }