Merge pull request #8 from Dounutsubtly/dev

Fix incorrect UDP packet truncation. Add TCP_NODELAY.

tuic-server/README.md

@@ -100,8 +100,9 @@ replace <tag> with [current version tag](https://github.com/Itsusinn/tuic/pkgs/c
     "max_idle_time": "10s",
 
     // Optional. Maximum packet size the server can receive from outbound UDP sockets, in bytes
+    // ** Warning: This option is deprecated. **
     // Default: 1500
-    "max_external_packet_size": 1500,
+    // "max_external_packet_size": 1500,
 
     // Optional. Maximum number of bytes to transmit to a peer without acknowledgment
     // Should be set to at least the expected connection latency multiplied by the maximum desired throughput

tuic-server/src/config.rs

@@ -82,9 +82,8 @@ pub struct Config {
     )]
     pub max_idle_time: Duration,
 
-    #[serde(default = "default::max_external_packet_size")]
-    pub max_external_packet_size: usize,
-
+    // #[serde(default = "default::max_external_packet_size")]
+    // pub max_external_packet_size: usize,
     #[serde(default = "default::send_window")]
     pub send_window: u64,
 
@@ -187,9 +186,9 @@ mod default {
         Duration::from_secs(10)
     }
 
-    pub fn max_external_packet_size() -> usize {
-        1500
-    }
+    // pub fn max_external_packet_size() -> usize {
+    //     1500
+    // }
 
     pub fn send_window() -> u64 {
         8 * 1024 * 1024 * 2

tuic-server/src/connection/handle_task.rs

@@ -46,6 +46,7 @@ impl Connection {
                     for addr in addrs {
                         match TcpStream::connect(addr).await {
                             Ok(s) => {
+                                s.set_nodelay(true)?;
                                 stream = Some(s);
                                 break;
                             }
@@ -132,12 +133,7 @@ impl Connection {
                 None => match self.udp_sessions.write().await.entry(assoc_id) {
                     Entry::Occupied(entry) => entry.get().clone(),
                     Entry::Vacant(entry) => {
-                        let session = UdpSession::new(
-                            self.clone(),
-                            assoc_id,
-                            self.udp_relay_ipv6,
-                            self.max_external_pkt_size,
-                        )?;
+                        let session = UdpSession::new(self.clone(), assoc_id, self.udp_relay_ipv6)?;
                         entry.insert(session.clone());
                         session
                     }

tuic-server/src/connection/mod.rs

@@ -32,7 +32,6 @@ pub struct Connection {
     task_negotiation_timeout: Duration,
     udp_sessions: Arc<AsyncRwLock<HashMap<u16, UdpSession>>>,
     udp_relay_mode: Arc<AtomicCell<Option<UdpRelayMode>>>,
-    max_external_pkt_size: usize,
     remote_uni_stream_cnt: Counter,
     remote_bi_stream_cnt: Counter,
     max_concurrent_uni_streams: Arc<AtomicU32>,
@@ -48,7 +47,6 @@ impl Connection {
         zero_rtt_handshake: bool,
         auth_timeout: Duration,
         task_negotiation_timeout: Duration,
-        max_external_pkt_size: usize,
         gc_interval: Duration,
         gc_lifetime: Duration,
     ) {
@@ -69,7 +67,6 @@ impl Connection {
                 users,
                 udp_relay_ipv6,
                 task_negotiation_timeout,
-                max_external_pkt_size,
             ))
         };
 
@@ -139,7 +136,6 @@ impl Connection {
         users: Arc<HashMap<Uuid, Box<[u8]>>>,
         udp_relay_ipv6: bool,
         task_negotiation_timeout: Duration,
-        max_external_pkt_size: usize,
     ) -> Self {
         Self {
             inner: conn.clone(),
@@ -150,7 +146,6 @@ impl Connection {
             task_negotiation_timeout,
             udp_sessions: Arc::new(AsyncRwLock::new(HashMap::new())),
             udp_relay_mode: Arc::new(AtomicCell::new(None)),
-            max_external_pkt_size,
             remote_uni_stream_cnt: Counter::new(),
             remote_bi_stream_cnt: Counter::new(),
             max_concurrent_uni_streams: Arc::new(AtomicU32::new(DEFAULT_CONCURRENT_STREAMS)),

tuic-server/src/connection/udp_session.rs

@@ -4,7 +4,7 @@ use std::{
     sync::Arc,
 };
 
-use bytes::Bytes;
+use bytes::{Bytes, BytesMut};
 use socket2::{Domain, Protocol, SockAddr, Socket, Type};
 use tokio::{
     net::UdpSocket,
@@ -18,6 +18,8 @@ use tuic::Address;
 use super::Connection;
 use crate::error::Error;
 
+const MAX_UDP_PACKET_SIZE: usize = 65516;
+
 #[derive(Clone)]
 pub struct UdpSession(Arc<UdpSessionInner>);
 
@@ -26,17 +28,11 @@ struct UdpSessionInner {
     conn: Connection,
     socket_v4: UdpSocket,
     socket_v6: Option<UdpSocket>,
-    max_pkt_size: usize,
     close: AsyncRwLock<Option<Sender<()>>>,
 }
 
 impl UdpSession {
-    pub fn new(
-        conn: Connection,
-        assoc_id: u16,
-        udp_relay_ipv6: bool,
-        max_pkt_size: usize,
-    ) -> Result<Self, Error> {
+    pub fn new(conn: Connection, assoc_id: u16, udp_relay_ipv6: bool) -> Result<Self, Error> {
         let socket_v4 = {
             let socket = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP))
                 .map_err(|err| Error::Socket("failed to create UDP associate IPv4 socket", err))?;
@@ -92,7 +88,6 @@ impl UdpSession {
             assoc_id,
             socket_v4,
             socket_v6,
-            max_pkt_size,
             close: AsyncRwLock::new(Some(tx)),
         }));
 
@@ -146,23 +141,31 @@ impl UdpSession {
     }
 
     async fn recv(&self) -> Result<(Bytes, SocketAddr), IoError> {
-        async fn recv(
-            socket: &UdpSocket,
-            max_pkt_size: usize,
-        ) -> Result<(Bytes, SocketAddr), IoError> {
-            let mut buf = vec![0u8; max_pkt_size];
+        async fn recv(socket: &UdpSocket) -> Result<(Bytes, SocketAddr), IoError> {
+            let mut buf = BytesMut::with_capacity(MAX_UDP_PACKET_SIZE);
+
+            // unsafe, but it's actually safe.
+            unsafe {
+                buf.set_len(MAX_UDP_PACKET_SIZE);
+            }
+
             let (n, addr) = socket.recv_from(&mut buf).await?;
-            buf.truncate(n);
-            Ok((Bytes::from(buf), addr))
+
+            unsafe {
+                buf.set_len(n);
+            }
+
+            // BytesMut to Bytes, cheap
+            Ok((buf.freeze(), addr))
         }
 
         if let Some(socket_v6) = &self.0.socket_v6 {
             tokio::select! {
-                res = recv(&self.0.socket_v4, self.0.max_pkt_size) => res,
-                res = recv(socket_v6, self.0.max_pkt_size) => res,
+                res = recv(&self.0.socket_v4) => res,
+                res = recv(socket_v6) => res,
             }
         } else {
-            recv(&self.0.socket_v4, self.0.max_pkt_size).await
+            recv(&self.0.socket_v4).await
         }
     }
 

tuic-server/src/server.rs

@@ -32,7 +32,6 @@ pub struct Server {
     zero_rtt_handshake: bool,
     auth_timeout: Duration,
     task_negotiation_timeout: Duration,
-    max_external_pkt_size: usize,
     gc_interval: Duration,
     gc_lifetime: Duration,
 }
@@ -133,7 +132,6 @@ impl Server {
             zero_rtt_handshake: cfg.zero_rtt_handshake,
             auth_timeout: cfg.auth_timeout,
             task_negotiation_timeout: cfg.task_negotiation_timeout,
-            max_external_pkt_size: cfg.max_external_packet_size,
             gc_interval: cfg.gc_interval,
             gc_lifetime: cfg.gc_lifetime,
         })
@@ -156,7 +154,6 @@ impl Server {
                             self.zero_rtt_handshake,
                             self.auth_timeout,
                             self.task_negotiation_timeout,
-                            self.max_external_pkt_size,
                             self.gc_interval,
                             self.gc_lifetime,
                         ));