chore(phantun) bump to v0.3.0

Bumps [actions/checkout](https://github.com/actions/checkout) from 2 to 3.
- [Release notes](https://github.com/actions/checkout/releases)
- [Changelog](https://github.com/actions/checkout/blob/main/CHANGELOG.md)
- [Commits](https://github.com/actions/checkout/compare/v2...v3)

---
updated-dependencies:
- dependency-name: actions/checkout
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>

.github/workflows/release.yml

@@ -31,7 +31,7 @@ jobs:
           - mipsel-unknown-linux-musl
 
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - uses: actions-rs/toolchain@v1
         with:
           toolchain: stable

.github/workflows/rust.yml

@@ -11,7 +11,7 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v3
     - uses: actions-rs/toolchain@v1
       with:
         toolchain: stable

LICENSE-APACHE

@@ -186,7 +186,7 @@ APPENDIX: How to apply the Apache License to your work.
    same "printed page" as the copyright notice for easier
    identification within third-party archives.
 
-Copyright 2014-2021 The Rust Project Developers
+Copyright 2021-2022 Datong Sun (dndx@idndx.com)
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

LICENSE-MIT

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2014-2021 The Rust Project Developers
+Copyright (c) 2021-2022 Datong Sun (dndx@idndx.com)
 
 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated

README.md

@@ -24,6 +24,7 @@ Table of Contents
 * [MTU overhead](#mtu-overhead)
     * [MTU calculation for WireGuard](#mtu-calculation-for-wireguard)
 * [Version compatibility](#version-compatibility)
+* [Documentations](#documentations)
 * [Performance](#performance)
 * [Future plans](#future-plans)
 * [Compariation to udp2raw](#compariation-to-udp2raw)
@@ -31,16 +32,16 @@ Table of Contents
 
 # Latest release
 
-[v0.2.2](https://github.com/dndx/phantun/releases/tag/v0.2.2)
+[v0.2.5](https://github.com/dndx/phantun/releases/tag/v0.2.5)
 
 # Overview
 
-Phanton is a project that obfuscated UDP packets into TCP connections. It aims to
+Phantun is a project that obfuscated UDP packets into TCP connections. It aims to
 achieve maximum performance with minimum processing and encapsulation overhead.
 
 It is commonly used in environments where UDP is blocked/throttled but TCP is allowed through.
 
-Phanton simply converts a stream of UDP packets into obfuscated TCP stream packets. The TCP stack
+Phantun simply converts a stream of UDP packets into obfuscated TCP stream packets. The TCP stack
 used by Phantun is designed to pass through most L3/L4 stateful/stateless firewalls/NAT
 devices. It will **not** be able to pass through L7 proxies.
 However, the advantage of this approach is that none of the common UDP over TCP performance killer
@@ -71,7 +72,7 @@ NIC address and Phantun's TUN interface address.
 You may customize the name of Tun interface created by Phantun and the assigned addresses. Please
 run the executable with `-h` options to see how to change them.
 
-Another way to help understand this network topology:
+Another way to help understand this network topology (please see the diagram above for an illustration of this topology):
 
 Phantun Client is like a machine with private IP address (`192.168.200.2`) behind a router.
 In order for it to reach the Internet, you will need to SNAT the private IP address before it's traffic
@@ -134,7 +135,7 @@ iptables -t nat -A POSTROUTING -o eth0 -j MASQUERADE
 Server needs to DNAT the TCP listening port to Phantun's TUN interface address.
 
 Note: change `eth0` to whatever actual physical interface name is and `4567` to
-actual TCP port number used by Phanton server
+actual TCP port number used by Phantun server
 
 [Back to TOC](#table-of-contents)
 
@@ -176,6 +177,8 @@ sudo setcap cap_net_admin=+pe phantun_client
 
 **Note:** Run Phantun executable with `-h` option to see full detailed options.
 
+[Back to TOC](#table-of-contents)
+
 ### Server
 
 Note: `4567` is the TCP port Phantun should listen on and must corresponds to the DNAT
@@ -185,6 +188,12 @@ rule specified above. `127.0.0.1:1234` is the UDP Server to connect to for new c
 RUST_LOG=info /usr/local/bin/phantun_server --local 4567 --remote 127.0.0.1:1234
 ```
 
+Or use host name with `--remote`:
+
+```
+RUST_LOG=info /usr/local/bin/phantun_server --local 4567 --remote example.com:1234
+```
+
 [Back to TOC](#table-of-contents)
 
 ### Client
@@ -196,6 +205,12 @@ the Phantun Server to connect.
 RUST_LOG=info /usr/local/bin/phantun_client --local 127.0.0.1:1234 --remote 10.0.0.1:4567
 ```
 
+Or use host name with `--remote`:
+
+```
+RUST_LOG=info /usr/local/bin/phantun_client --local 127.0.0.1:1234 --remote example.com:4567
+```
+
 [Back to TOC](#table-of-contents)
 
 # MTU overhead
@@ -240,6 +255,13 @@ of Server/Client of Phantun on both ends to ensure maximum compatibility.
 
 [Back to TOC](#table-of-contents)
 
+# Documentations
+
+For users who wish to use `fake-tcp` library inside their own project, refer to the documentations for the library at:
+[https://docs.rs/fake-tcp](https://docs.rs/fake-tcp).
+
+[Back to TOC](#table-of-contents)
+
 # Performance
 
 Performance was tested on AWS t3.xlarge instance with 4 vCPUs and 5 Gb/s NIC. WireGuard was used
@@ -264,7 +286,7 @@ for tunneling TCP/UDP traffic between two test instances and MTU has been tuned
 # Compariation to udp2raw
 [udp2raw](https://github.com/wangyu-/udp2raw-tunnel) is another popular project by [@wangyu-](https://github.com/wangyu-)
 that is very similar to what Phantun can do. In fact I took inspirations of Phantun from udp2raw. The biggest reason for
-developing Phanton is because of lack of performance when running udp2raw (especially on multi-core systems such as Raspberry Pi).
+developing Phantun is because of lack of performance when running udp2raw (especially on multi-core systems such as Raspberry Pi).
 However, the goal is never to be as feature complete as udp2raw and only support the most common use cases. Most notably, UDP over ICMP
 and UDP over UDP mode are not supported and there is no anti-replay nor encryption support. The benefit of this is much better
 performance overall and less MTU overhead because lack of additional headers inside the TCP payload.
@@ -288,7 +310,7 @@ Here is a quick overview of comparison between those two to help you choose:
 
 # License
 
-Copyright 2021 Datong Sun <dndx@idndx.com>
+Copyright 2021-2022 Datong Sun (dndx@idndx.com)
 
 Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
 [https://www.apache.org/licenses/LICENSE-2.0](https://www.apache.org/licenses/LICENSE-2.0)> or the MIT license

fake-tcp/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fake-tcp"
-version = "0.2.0"
+version = "0.2.4"
 edition = "2021"
 authors = ["Datong Sun <dndx@idndx.com>"]
 license = "MIT OR Apache-2.0"
@@ -16,9 +16,9 @@ benchmark = []
 
 [dependencies]
 bytes = "1"
-pnet = "0.28.0"
-tokio = { version = "1.12.0", features = ["full"] }
-rand = { version = "0.8.4", features = ["small_rng"] }
+pnet = "0.29"
+tokio = { version = "1.14", features = ["full"] }
+rand = { version = "0.8", features = ["small_rng"] }
 log = "0.4"
-internet-checksum = "0.2.0"
-dndx-fork-tokio-tun = "0.4.0"
+internet-checksum = "0.2"
+tokio-tun = "0.5"

fake-tcp/LICENSE-APACHE

@@ -186,7 +186,7 @@ APPENDIX: How to apply the Apache License to your work.
    same "printed page" as the copyright notice for easier
    identification within third-party archives.
 
-Copyright 2014-2021 The Rust Project Developers
+Copyright 2021-2022 Datong Sun (dndx@idndx.com)
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

fake-tcp/LICENSE-MIT

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2014-2021 The Rust Project Developers
+Copyright (c) 2021-2022 Datong Sun (dndx@idndx.com)
 
 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated

fake-tcp/src/lib.rs

@@ -1,7 +1,46 @@
+//! A minimum, userspace TCP based datagram stack
+//!
+//! # Overview
+//!
+//! `fake-tcp` is a reusable library that implements a minimum TCP stack in
+//! user space using the Tun interface. It allows programs to send datagrams
+//! as if they are part of a TCP connection. `fake-tcp` has been tested to
+//! be able to pass through a variety of NAT and stateful firewalls while
+//! fully preserves certain desirable behavior such as out of order delivery
+//! and no congestion/flow controls.
+//!
+//! # Core Concepts
+//!
+//! The core of the `fake-tcp` crate compose of two structures. [`Stack`] and
+//! [`Socket`].
+//!
+//! ## [`Stack`]
+//!
+//! [`Stack`] represents a virtual TCP stack that operates at
+//! Layer 3. It is responsible for:
+//!
+//! * TCP active and passive open and handshake
+//! * `RST` handling
+//! * Interact with the Tun interface at Layer 3
+//! * Distribute incoming datagrams to corresponding [`Socket`]
+//!
+//! ## [`Socket`]
+//!
+//! [`Socket`] represents a TCP connection. It registers the identifying
+//! tuple `(src_ip, src_port, dest_ip, dest_port)` inside the [`Stack`] so
+//! so that incoming packets can be distributed to the right [`Socket`] with
+//! using a channel. It is also what the client should use for
+//! sending/receiving datagrams.
+//!
+//! # Examples
+//!
+//! Please see [`client.rs`](https://github.com/dndx/phantun/blob/main/phantun/src/bin/client.rs)
+//! and [`server.rs`](https://github.com/dndx/phantun/blob/main/phantun/src/bin/server.rs) files
+//! from the `phantun` crate for how to use this library in client/server mode, respectively.
+
 #![cfg_attr(feature = "benchmark", feature(test))]
 
 pub mod packet;
-extern crate dndx_fork_tokio_tun as tokio_tun;
 
 use bytes::{Bytes, BytesMut};
 use log::{error, info, trace, warn};
@@ -24,7 +63,7 @@ const RETRIES: usize = 6;
 const MPSC_BUFFER_LEN: usize = 512;
 
 #[derive(Hash, Eq, PartialEq, Clone, Debug)]
-pub struct AddrTuple {
+struct AddrTuple {
     local_addr: SocketAddrV4,
     remote_addr: SocketAddrV4,
 }
@@ -57,7 +96,6 @@ pub enum State {
     SynSent,
     SynReceived,
     Established,
-    Closed,
 }
 
 pub struct Socket {
@@ -71,6 +109,13 @@ pub struct Socket {
     state: State,
 }
 
+/// A socket that represents a unique TCP connection between a server and client.
+///
+/// The `Socket` object itself satisfies `Sync` and `Send`, which means it can
+/// be safely called within an async future.
+///
+/// To close a TCP connection that is no longer needed, simply drop this object
+/// out of scope.
 impl Socket {
     fn new(
         shared: Arc<Shared>,
@@ -108,6 +153,13 @@ impl Socket {
         )
     }
 
+    /// Sends a datagram to the other end.
+    ///
+    /// This method takes `&self`, and it can be called safely by multiple threads
+    /// at the same time.
+    ///
+    /// A return of `None` means the Tun socket returned an error
+    /// and this socket must be closed.
     pub async fn send(&self, payload: &[u8]) -> Option<()> {
         match self.state {
             State::Established => {
@@ -116,16 +168,21 @@ impl Socket {
 
                 tokio::select! {
                     res = self.tun.send(&buf) => {
-                        res.unwrap();
-                        Some(())
+                        res.ok().and(Some(()))
                     },
                 }
             }
-            State::Closed => None,
             _ => unreachable!(),
         }
     }
 
+    /// Attempt to receive a datagram from the other end.
+    ///
+    /// This method takes `&self`, and it can be called safely by multiple threads
+    /// at the same time.
+    ///
+    /// A return of `None` means the TCP connection is broken
+    /// and this socket must be closed.
     pub async fn recv(&self, buf: &mut [u8]) -> Option<usize> {
         match self.state {
             State::Established => {
@@ -148,7 +205,6 @@ impl Socket {
                     Some(payload.len())
                 })
             }
-            State::Closed => None,
             _ => unreachable!(),
         }
     }
@@ -251,9 +307,8 @@ impl Socket {
 }
 
 impl Drop for Socket {
+    /// Drop the socket and close the TCP connection
     fn drop(&mut self) {
-        self.state = State::Closed;
-
         let tuple = AddrTuple::new(self.local_addr, self.remote_addr);
         // dissociates ourself from the dispatch map
         assert!(self.shared.tuples.write().unwrap().remove(&tuple).is_some());
@@ -270,6 +325,7 @@ impl Drop for Socket {
 }
 
 impl fmt::Display for Socket {
+    /// User-friendly string representation of the socket
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(
             f,
@@ -279,7 +335,12 @@ impl fmt::Display for Socket {
     }
 }
 
+/// A userspace TCP state machine
 impl Stack {
+    /// Create a new stack, `tun` is an array of [`Tun`](tokio_tun::Tun).
+    /// When more than one [`Tun`](tokio_tun::Tun) object is passed in, same amount
+    /// of reader will be spawned later. This allows user to utilize the performance
+    /// benefit of Multiqueue Tun support on machines with SMP.
     pub fn new(tun: Vec<Tun>) -> Stack {
         let tun: Vec<Arc<Tun>> = tun.into_iter().map(Arc::new).collect();
         let (ready_tx, ready_rx) = mpsc::channel(MPSC_BUFFER_LEN);
@@ -308,14 +369,18 @@ impl Stack {
         }
     }
 
+    /// Listens for incoming connections on the given `port`.
     pub fn listen(&mut self, port: u16) {
         assert!(self.shared.listening.write().unwrap().insert(port));
     }
 
+    /// Accepts an incoming connection.
     pub async fn accept(&mut self) -> Socket {
         self.ready.recv().await.unwrap()
     }
 
+    /// Connects to the remote end. `None` returned means
+    /// the connection attempt failed.
     pub async fn connect(&mut self, addr: SocketAddrV4) -> Option<Socket> {
         let mut rng = SmallRng::from_entropy();
         let local_port: u16 = rng.gen_range(1024..65535);

phantun/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "phantun"
-version = "0.2.3"
+version = "0.3.0"
 edition = "2021"
 authors = ["Datong Sun <dndx@idndx.com>"]
 license = "MIT OR Apache-2.0"
@@ -11,11 +11,12 @@ Transforms UDP stream into (fake) TCP streams that can go through
 Layer 3 & Layer 4 (NAPT) firewalls/NATs.
 """
 [dependencies]
-clap = "2.33.3"
-socket2 = { version = "0.4.2", features = ["all"] }
-fake-tcp = "0.2.0"
-tokio = { version = "1.12.0", features = ["full"] }
+clap = { version = "3.0", features = ["cargo"] }
+socket2 = { version = "0.4", features = ["all"] }
+fake-tcp = { path = "../fake-tcp", version = "0.2" }
+tokio = { version = "1.14", features = ["full"] }
+tokio-util = "0.7"
 log = "0.4"
-pretty_env_logger = "0.4.0"
-dndx-fork-tokio-tun = "0.4.0"
-num_cpus = "1.13.0"
+pretty_env_logger = "0.4"
+tokio-tun = "0.5"
+num_cpus = "1.13"

phantun/LICENSE-APACHE

@@ -186,7 +186,7 @@ APPENDIX: How to apply the Apache License to your work.
    same "printed page" as the copyright notice for easier
    identification within third-party archives.
 
-Copyright 2014-2021 The Rust Project Developers
+Copyright 2021-2022 Datong Sun (dndx@idndx.com)
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

phantun/LICENSE-MIT

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2014-2021 The Rust Project Developers
+Copyright (c) 2021-2022 Datong Sun (dndx@idndx.com)
 
 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated

phantun/src/bin/client.rs

@@ -1,51 +1,28 @@
-extern crate dndx_fork_tokio_tun as tokio_tun;
-
-use clap::{crate_version, App, Arg};
+use clap::{crate_version, Arg, Command};
 use fake_tcp::packet::MAX_PACKET_LEN;
 use fake_tcp::{Socket, Stack};
 use log::{debug, error, info};
+use phantun::utils::new_udp_reuseport;
 use std::collections::HashMap;
-use std::convert::TryInto;
 use std::net::{Ipv4Addr, SocketAddr};
 use std::sync::Arc;
-use std::time::Duration;
-use tokio::net::UdpSocket;
-use tokio::sync::RwLock;
+use tokio::sync::{Notify, RwLock};
 use tokio::time;
 use tokio_tun::TunBuilder;
+use tokio_util::sync::CancellationToken;
 
-const UDP_TTL: Duration = Duration::from_secs(180);
-
-fn new_udp_reuseport(addr: SocketAddr) -> UdpSocket {
-    let udp_sock = socket2::Socket::new(
-        if addr.is_ipv4() {
-            socket2::Domain::IPV4
-        } else {
-            socket2::Domain::IPV6
-        },
-        socket2::Type::DGRAM,
-        None,
-    )
-    .unwrap();
-    udp_sock.set_reuse_port(true).unwrap();
-    // from tokio-rs/mio/blob/master/src/sys/unix/net.rs
-    udp_sock.set_cloexec(true).unwrap();
-    udp_sock.set_nonblocking(true).unwrap();
-    udp_sock.bind(&socket2::SockAddr::from(addr)).unwrap();
-    let udp_sock: std::net::UdpSocket = udp_sock.into();
-    udp_sock.try_into().unwrap()
-}
+use phantun::UDP_TTL;
 
 #[tokio::main]
 async fn main() {
     pretty_env_logger::init();
 
-    let matches = App::new("Phantun Client")
+    let matches = Command::new("Phantun Client")
         .version(crate_version!())
         .author("Datong Sun (github.com/dndx)")
         .arg(
-            Arg::with_name("local")
-                .short("l")
+            Arg::new("local")
+                .short('l')
                 .long("local")
                 .required(true)
                 .value_name("IP:PORT")
@@ -53,8 +30,8 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("remote")
-                .short("r")
+            Arg::new("remote")
+                .short('r')
                 .long("remote")
                 .required(true)
                 .value_name("IP or HOST NAME:PORT")
@@ -62,7 +39,7 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("tun")
+            Arg::new("tun")
                 .long("tun")
                 .required(false)
                 .value_name("tunX")
@@ -71,7 +48,7 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("tun_local")
+            Arg::new("tun_local")
                 .long("tun-local")
                 .required(false)
                 .value_name("IP")
@@ -80,7 +57,7 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("tun_peer")
+            Arg::new("tun_peer")
                 .long("tun-peer")
                 .required(false)
                 .value_name("IP")
@@ -101,12 +78,12 @@ async fn main() {
     let remote_addr = tokio::net::lookup_host(matches.value_of("remote").unwrap())
         .await
         .expect("bad remote address or host")
-        .next()
-        .expect("unable to resolve remote host name");
+        .find(|addr| addr.is_ipv4())
+        .expect("unable to resolve remote host name or no valid A record was returned");
     let remote_addr = if let SocketAddr::V4(addr) = remote_addr {
         addr
     } else {
-        panic!("only IPv4 remote address is supported");
+        unreachable!();
     };
     info!("Remote address is: {}", remote_addr);
 
@@ -121,6 +98,8 @@ async fn main() {
         .parse()
         .expect("bad peer address for Tun interface");
 
+    let num_cpus = num_cpus::get();
+
     let tun = TunBuilder::new()
         .name(matches.value_of("tun").unwrap()) // if name is empty, then it is set by kernel.
         .tap(false) // false (default): TUN, true: TAP.
@@ -128,7 +107,7 @@ async fn main() {
         .up() // or set it up manually using `sudo ip link set <tun-name> up`.
         .address(tun_local)
         .destination(tun_peer)
-        .try_build_mq(num_cpus::get())
+        .try_build_mq(num_cpus)
         .unwrap();
 
     info!("Created TUN device {}", tun[0].name());
@@ -170,52 +149,85 @@ async fn main() {
                     assert!(connections.write().await.insert(addr, sock.clone()).is_none());
                     debug!("inserted fake TCP socket into connection table");
 
-                    let connections = connections.clone();
-
                     // spawn "fastpath" UDP socket and task, this will offload main task
                     // from forwarding UDP packets
-                    tokio::spawn(async move {
-                        let mut buf_udp = [0u8; MAX_PACKET_LEN];
-                        let mut buf_tcp = [0u8; MAX_PACKET_LEN];
-                        let udp_sock = new_udp_reuseport(local_addr);
-                        udp_sock.connect(addr).await.unwrap();
 
+                    let packet_received = Arc::new(Notify::new());
+                    let quit = CancellationToken::new();
+
+                    for i in 0..num_cpus {
+                        let sock = sock.clone();
+                        let quit = quit.child_token();
+                        let packet_received = packet_received.clone();
+
+                        tokio::spawn(async move {
+                            let mut buf_udp = [0u8; MAX_PACKET_LEN];
+                            let mut buf_tcp = [0u8; MAX_PACKET_LEN];
+                            let udp_sock = new_udp_reuseport(local_addr);
+                            udp_sock.connect(addr).await.unwrap();
+
+                            loop {
+                                tokio::select! {
+                                    Ok(size) = udp_sock.recv(&mut buf_udp) => {
+                                        if sock.send(&buf_udp[..size]).await.is_none() {
+                                            debug!("removed fake TCP socket from connections table");
+                                            quit.cancel();
+                                            return;
+                                        }
+
+                                        packet_received.notify_one();
+                                    },
+                                    res = sock.recv(&mut buf_tcp) => {
+                                        match res {
+                                            Some(size) => {
+                                                if size > 0 {
+                                                    if let Err(e) = udp_sock.send(&buf_tcp[..size]).await {
+                                                        error!("Unable to send UDP packet to {}: {}, closing connection", e, addr);
+                                                        quit.cancel();
+                                                        return;
+                                                    }
+                                                }
+                                            },
+                                            None => {
+                                                debug!("removed fake TCP socket from connections table");
+                                                quit.cancel();
+                                                return;
+                                            },
+                                        }
+
+                                        packet_received.notify_one();
+                                    },
+                                    _ = quit.cancelled() => {
+                                        debug!("worker {} terminated", i);
+                                        return;
+                                    },
+                                };
+                            }
+                        });
+                    }
+
+                    let connections = connections.clone();
+                    tokio::spawn(async move {
                         loop {
                             let read_timeout = time::sleep(UDP_TTL);
+                            let packet_received_fut = packet_received.notified();
 
                             tokio::select! {
-                                Ok(size) = udp_sock.recv(&mut buf_udp) => {
-                                    if sock.send(&buf_udp[..size]).await.is_none() {
-                                        connections.write().await.remove(&addr);
-                                        debug!("removed fake TCP socket from connections table");
-                                        return;
-                                    }
-                                },
-                                res = sock.recv(&mut buf_tcp) => {
-                                    match res {
-                                        Some(size) => {
-                                            if size > 0 {
-                                                if let Err(e) = udp_sock.send(&buf_tcp[..size]).await {
-                                                    connections.write().await.remove(&addr);
-                                                    error!("Unable to send UDP packet to {}: {}, closing connection", e, addr);
-                                                    return;
-                                                }
-                                            }
-                                        },
-                                        None => {
-                                            connections.write().await.remove(&addr);
-                                            debug!("removed fake TCP socket from connections table");
-                                            return;
-                                        },
-                                    }
-                                },
                                 _ = read_timeout => {
                                     info!("No traffic seen in the last {:?}, closing connection", UDP_TTL);
                                     connections.write().await.remove(&addr);
                                     debug!("removed fake TCP socket from connections table");
+
+                                    quit.cancel();
                                     return;
-                                }
-                            };
+                                },
+                                _ = quit.cancelled() => {
+                                    connections.write().await.remove(&addr);
+                                    debug!("removed fake TCP socket from connections table");
+                                    return;
+                                },
+                                _ = packet_received_fut => {},
+                            }
                         }
                     });
                 },

phantun/src/bin/server.rs

@@ -1,25 +1,28 @@
-extern crate dndx_fork_tokio_tun as tokio_tun;
-
-use clap::{crate_version, App, Arg};
+use clap::{crate_version, Arg, Command};
 use fake_tcp::packet::MAX_PACKET_LEN;
 use fake_tcp::Stack;
-use log::{error, info};
+use log::{debug, error, info};
+use phantun::utils::new_udp_reuseport;
 use std::net::Ipv4Addr;
+use std::sync::Arc;
 use tokio::net::UdpSocket;
-use tokio::time::{self, Duration};
+use tokio::sync::Notify;
+use tokio::time;
 use tokio_tun::TunBuilder;
-const UDP_TTL: Duration = Duration::from_secs(180);
+use tokio_util::sync::CancellationToken;
+
+use phantun::UDP_TTL;
 
 #[tokio::main]
 async fn main() {
     pretty_env_logger::init();
 
-    let matches = App::new("Phantun Server")
+    let matches = Command::new("Phantun Server")
         .version(crate_version!())
         .author("Datong Sun (github.com/dndx)")
         .arg(
-            Arg::with_name("local")
-                .short("l")
+            Arg::new("local")
+                .short('l')
                 .long("local")
                 .required(true)
                 .value_name("PORT")
@@ -27,8 +30,8 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("remote")
-                .short("r")
+            Arg::new("remote")
+                .short('r')
                 .long("remote")
                 .required(true)
                 .value_name("IP or HOST NAME:PORT")
@@ -36,7 +39,7 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("tun")
+            Arg::new("tun")
                 .long("tun")
                 .required(false)
                 .value_name("tunX")
@@ -45,7 +48,7 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("tun_local")
+            Arg::new("tun_local")
                 .long("tun-local")
                 .required(false)
                 .value_name("IP")
@@ -54,7 +57,7 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("tun_peer")
+            Arg::new("tun_peer")
                 .long("tun-peer")
                 .required(false)
                 .value_name("IP")
@@ -90,6 +93,8 @@ async fn main() {
         .parse()
         .expect("bad peer address for Tun interface");
 
+    let num_cpus = num_cpus::get();
+
     let tun = TunBuilder::new()
         .name(matches.value_of("tun").unwrap()) // if name is empty, then it is set by kernel.
         .tap(false) // false (default): TUN, true: TAP.
@@ -97,7 +102,7 @@ async fn main() {
         .up() // or set it up manually using `sudo ip link set <tun-name> up`.
         .address(tun_local)
         .destination(tun_peer)
-        .try_build_mq(num_cpus::get())
+        .try_build_mq(num_cpus)
         .unwrap();
 
     info!("Created TUN device {}", tun[0].name());
@@ -112,46 +117,82 @@ async fn main() {
         let mut buf_tcp = [0u8; MAX_PACKET_LEN];
 
         loop {
-            let sock = stack.accept().await;
+            let sock = Arc::new(stack.accept().await);
             info!("New connection: {}", sock);
 
-            tokio::spawn(async move {
-                let udp_sock = UdpSocket::bind(if remote_addr.is_ipv4() {
-                    "0.0.0.0:0"
-                } else {
-                    "[::]:0"
-                })
-                .await
-                .unwrap();
-                udp_sock.connect(remote_addr).await.unwrap();
+            let packet_received = Arc::new(Notify::new());
+            let quit = CancellationToken::new();
+            let udp_sock = UdpSocket::bind(if remote_addr.is_ipv4() {
+                "0.0.0.0:0"
+            } else {
+                "[::]:0"
+            })
+            .await
+            .unwrap();
+            let local_addr = udp_sock.local_addr().unwrap();
+            drop(udp_sock);
 
+            for i in 0..num_cpus {
+                let sock = sock.clone();
+                let quit = quit.child_token();
+                let packet_received = packet_received.clone();
+                let udp_sock = new_udp_reuseport(local_addr);
+
+                tokio::spawn(async move {
+                    udp_sock.connect(remote_addr).await.unwrap();
+
+                    loop {
+                        tokio::select! {
+                            Ok(size) = udp_sock.recv(&mut buf_udp) => {
+                                if sock.send(&buf_udp[..size]).await.is_none() {
+                                    quit.cancel();
+                                    return;
+                                }
+
+                                packet_received.notify_one();
+                            },
+                            res = sock.recv(&mut buf_tcp) => {
+                                match res {
+                                    Some(size) => {
+                                        if size > 0 {
+                                            if let Err(e) = udp_sock.send(&buf_tcp[..size]).await {
+                                                error!("Unable to send UDP packet to {}: {}, closing connection", e, remote_addr);
+                                                quit.cancel();
+                                                return;
+                                            }
+                                        }
+                                    },
+                                    None => {
+                                        quit.cancel();
+                                        return;
+                                    },
+                                }
+
+                                packet_received.notify_one();
+                            },
+                            _ = quit.cancelled() => {
+                                debug!("worker {} terminated", i);
+                                return;
+                            },
+                        };
+                    }
+                });
+            }
+
+            tokio::spawn(async move {
                 loop {
                     let read_timeout = time::sleep(UDP_TTL);
+                    let packet_received_fut = packet_received.notified();
 
                     tokio::select! {
-                        Ok(size) = udp_sock.recv(&mut buf_udp) => {
-                            if sock.send(&buf_udp[..size]).await.is_none() {
-                                return;
-                            }
-                        },
-                        res = sock.recv(&mut buf_tcp) => {
-                            match res {
-                                Some(size) => {
-                                    if size > 0 {
-                                        if let Err(e) = udp_sock.send(&buf_tcp[..size]).await {
-                                            error!("Unable to send UDP packet to {}: {}, closing connection", e, remote_addr);
-                                            return;
-                                        }
-                                    }
-                                },
-                                None => { return; },
-                            }
-                        },
                         _ = read_timeout => {
                             info!("No traffic seen in the last {:?}, closing connection", UDP_TTL);
+
+                            quit.cancel();
                             return;
-                        }
-                    };
+                        },
+                        _ = packet_received_fut => {},
+                    }
                 }
             });
         }

phantun/src/lib.rs

@@ -0,0 +1,5 @@
+use std::time::Duration;
+
+pub mod utils;
+
+pub const UDP_TTL: Duration = Duration::from_secs(180);

phantun/src/utils.rs

@@ -0,0 +1,22 @@
+use std::net::SocketAddr;
+use tokio::net::UdpSocket;
+
+pub fn new_udp_reuseport(local_addr: SocketAddr) -> UdpSocket {
+    let udp_sock = socket2::Socket::new(
+        if local_addr.is_ipv4() {
+            socket2::Domain::IPV4
+        } else {
+            socket2::Domain::IPV6
+        },
+        socket2::Type::DGRAM,
+        None,
+    )
+    .unwrap();
+    udp_sock.set_reuse_port(true).unwrap();
+    // from tokio-rs/mio/blob/master/src/sys/unix/net.rs
+    udp_sock.set_cloexec(true).unwrap();
+    udp_sock.set_nonblocking(true).unwrap();
+    udp_sock.bind(&socket2::SockAddr::from(local_addr)).unwrap();
+    let udp_sock: std::net::UdpSocket = udp_sock.into();
+    udp_sock.try_into().unwrap()
+}