Removed unnecessary Runtime Handle in servers entry function

BREAKING: run_local, run_server, run_manager have removed Handle parameter

src/bin/local.rs

@@ -294,10 +294,9 @@ fn main() {
         builder.threaded_scheduler();
     }
     let mut runtime = builder.enable_all().build().expect("create tokio Runtime");
-    let rt_handle = runtime.handle().clone();
     runtime.block_on(async move {
         let abort_signal = monitor::create_signal_monitor();
-        let server = run_local(config, rt_handle);
+        let server = run_local(config);
 
         tokio::pin!(abort_signal);
         tokio::pin!(server);

src/bin/manager.rs

@@ -150,10 +150,9 @@ fn main() {
         builder.threaded_scheduler();
     }
     let mut runtime = builder.enable_all().build().expect("create tokio Runtime");
-    let rt_handle = runtime.handle().clone();
     runtime.block_on(async move {
         let abort_signal = monitor::create_signal_monitor();
-        let server = run_manager(config, rt_handle);
+        let server = run_manager(config);
 
         tokio::pin!(abort_signal);
         tokio::pin!(server);

src/bin/server.rs

@@ -174,10 +174,9 @@ fn main() {
         builder.threaded_scheduler();
     }
     let mut runtime = builder.enable_all().build().expect("create tokio Runtime");
-    let rt_handle = runtime.handle().clone();
     runtime.block_on(async move {
         let abort_signal = monitor::create_signal_monitor();
-        let server = run_server(config, rt_handle);
+        let server = run_server(config);
 
         tokio::pin!(abort_signal);
         tokio::pin!(server);

src/context.rs

@@ -17,18 +17,17 @@ use log::warn;
 #[cfg(feature = "local-dns-relay")]
 use lru_time_cache::LruCache;
 use spin::Mutex;
-use tokio::runtime::Handle;
 #[cfg(feature = "trust-dns")]
 use trust_dns_resolver::TokioAsyncResolver;
 
 #[cfg(feature = "trust-dns")]
 use crate::relay::dns_resolver::create_resolver;
-#[cfg(feature = "local-flow-stat")]
-use crate::relay::flow::ServerFlowStatistic;
-#[cfg(feature = "local-dns-relay")]
-use crate::relay::dnsrelay::upstream::LocalUpstream;
 #[cfg(not(feature = "local-dns-relay"))]
 use crate::relay::dns_resolver::resolve;
+#[cfg(feature = "local-dns-relay")]
+use crate::relay::dnsrelay::upstream::LocalUpstream;
+#[cfg(feature = "local-flow-stat")]
+use crate::relay::flow::ServerFlowStatistic;
 use crate::{
     acl::AccessControl,
     config::{Config, ConfigType, ServerConfig},
@@ -130,9 +129,9 @@ pub struct ServerState {
 #[cfg(feature = "trust-dns")]
 impl ServerState {
     /// Create a global shared server state
-    pub async fn new_shared(config: &Config, rt: Handle) -> SharedServerState {
+    pub async fn new_shared(config: &Config) -> SharedServerState {
         let state = ServerState {
-            dns_resolver: match create_resolver(config.get_dns_config(), config.timeout, config.ipv6_first, rt).await {
+            dns_resolver: match create_resolver(config.get_dns_config(), config.timeout, config.ipv6_first).await {
                 Ok(resolver) => Some(resolver),
                 Err(..) => None,
             },
@@ -335,11 +334,12 @@ impl Context {
     /// Add a record to the reverse lookup cache
     #[cfg(feature = "local-dns-relay")]
     pub fn add_to_reverse_lookup_cache(&self, addr: &IpAddr, forward: bool) {
-        let is_exception = forward != match self.acl() {
-            // Proxy everything by default
-            None => true,
-            Some(ref a) => a.check_ip_in_proxy_list(addr)
-        };
+        let is_exception = forward
+            != match self.acl() {
+                // Proxy everything by default
+                None => true,
+                Some(ref a) => a.check_ip_in_proxy_list(addr),
+            };
         let mut reverse_lookup_cache = self.reverse_lookup_cache.lock();
         match reverse_lookup_cache.get_mut(addr) {
             Some(value) => {

src/lib.rs

@@ -55,10 +55,9 @@
 //! use shadowsocks::{run_local, Config, ConfigType};
 //!
 //! let mut rt = Runtime::new().expect("Failed to create runtime");
-//! let rt_handle = rt.handle().clone();
 //!
 //! let config = Config::load_from_file("shadowsocks.json", ConfigType::Socks5Local).unwrap();
-//! rt.block_on(run_local(config, rt_handle));
+//! rt.block_on(run_local(config));
 //! ```
 //!
 //! That's all! And let me show you how to run a proxy server
@@ -68,10 +67,9 @@
 //! use shadowsocks::{run_server, Config, ConfigType};
 //!
 //! let mut rt = Runtime::new().expect("Failed to create runtime");
-//! let rt_handle = rt.handle().clone();
 //!
 //! let config = Config::load_from_file("shadowsocks.json", ConfigType::Server).unwrap();
-//! rt.block_on(run_server(config, rt_handle));
+//! rt.block_on(run_server(config));
 //! ```
 
 #![crate_type = "lib"]
@@ -80,8 +78,6 @@
 
 use std::io;
 
-use tokio::runtime::Handle;
-
 /// ShadowSocks version
 pub const VERSION: &str = env!("CARGO_PKG_VERSION");
 
@@ -105,10 +101,10 @@ pub mod relay;
 /// Start a ShadowSocks' server
 ///
 /// For `config.config_type` in `Socks5Local`, `HttpLocal` and `TunnelLocal`, server will run in Local mode.
-pub async fn run(config: Config, rt: Handle) -> io::Result<()> {
+pub async fn run(config: Config) -> io::Result<()> {
     if config.config_type.is_local() {
-        run_local(config, rt).await
+        run_local(config).await
     } else {
-        run_server(config, rt).await
+        run_server(config).await
     }
 }

src/relay/dns_resolver/trust_dns_resolver.rs

@@ -7,7 +7,6 @@ use std::{
 };
 
 use log::{error, trace};
-use tokio::{self, runtime::Handle};
 use trust_dns_resolver::{
     config::{LookupIpStrategy, ResolverConfig, ResolverOpts},
     TokioAsyncResolver,
@@ -21,7 +20,6 @@ pub async fn create_resolver(
     dns: Option<ResolverConfig>,
     timeout: Option<Duration>,
     ipv6_first: bool,
-    rt: Handle,
 ) -> io::Result<TokioAsyncResolver> {
     let mut resolver_opts = ResolverOpts::default();
     if let Some(d) = timeout {
@@ -32,76 +30,63 @@ pub async fn create_resolver(
         resolver_opts.ip_strategy = LookupIpStrategy::Ipv6thenIpv4;
     }
 
-    {
-        // To make this independent, if targeting macOS, BSD, Linux, or Windows, we can use the system's configuration:
-        #[cfg(any(unix, windows))]
-        {
-            if let Some(conf) = dns {
-                trace!(
-                    "initializing DNS resolver with config {:?} opts {:?}",
-                    conf,
-                    resolver_opts
-                );
-                TokioAsyncResolver::new(conf, resolver_opts, rt)
-            } else {
-                use trust_dns_resolver::system_conf::read_system_conf;
-                // use the system resolver configuration
-                let (config, mut opts) = match read_system_conf() {
-                    Ok(o) => o,
-                    Err(err) => {
-                        error!("failed to initialize DNS resolver with system-config, error: {}", err);
-
-                        // From::from is required because on error type is different on Windows
-                        #[allow(clippy::identity_conversion)]
-                        return Err(From::from(err));
-                    }
-                };
-
-                // NOTE: timeout will be set by config (for example, /etc/resolv.conf on UNIX-like system)
-                //
-                // Only ip_strategy should be changed
-                if ipv6_first {
-                    opts.ip_strategy = LookupIpStrategy::Ipv6thenIpv4;
-                }
-
-                trace!(
-                    "initializing DNS resolver with system-config {:?} opts {:?}",
-                    config,
-                    opts
-                );
-
-                TokioAsyncResolver::new(config, opts, rt)
-            }
+    // Customized dns resolution
+    match dns {
+        Some(conf) => {
+            trace!(
+                "initializing DNS resolver with config {:?} opts {:?}",
+                conf,
+                resolver_opts
+            );
+            TokioAsyncResolver::tokio(conf, resolver_opts).await
         }
 
-        // For other operating systems, we can use one of the preconfigured definitions
-        #[cfg(not(any(unix, windows)))]
-        {
-            if let Some(conf) = dns {
-                trace!(
-                    "initializing DNS resolver with config {:?} opts {:?}",
-                    conf,
-                    resolver_opts
-                );
-                TokioAsyncResolver::new(conf, resolver_opts, rt)
-            } else {
-                // Get a new resolver with the google nameservers as the upstream recursive resolvers
-                trace!(
-                    "initializing DNS resolver with google-config {:?} opts {:?}",
-                    ResolverConfig::google(),
-                    resolver_opts
-                );
-                TokioAsyncResolver::new(ResolverConfig::google(), resolver_opts, rt)
+        // To make this independent, if targeting macOS, BSD, Linux, or Windows, we can use the system's configuration
+        #[cfg(any(unix, windows))]
+        None => {
+            use tokio::runtime::Handle;
+            use trust_dns_resolver::system_conf::read_system_conf;
+
+            // use the system resolver configuration
+            let (config, mut opts) = match read_system_conf() {
+                Ok(o) => o,
+                Err(err) => {
+                    error!("failed to initialize DNS resolver with system-config, error: {}", err);
+
+                    // From::from is required because on error type is different on Windows
+                    #[allow(clippy::identity_conversion)]
+                    return Err(From::from(err));
+                }
+            };
+
+            // NOTE: timeout will be set by config (for example, /etc/resolv.conf on UNIX-like system)
+            //
+            // Only ip_strategy should be changed
+            if ipv6_first {
+                opts.ip_strategy = LookupIpStrategy::Ipv6thenIpv4;
             }
+
+            trace!(
+                "initializing DNS resolver with system-config {:?} opts {:?}",
+                config,
+                opts
+            );
+
+            TokioAsyncResolver::new(config, opts, Handle::current()).await
+        }
+
+        #[cfg(not(any(unix, windows)))]
+        None => {
+            // Get a new resolver with the google nameservers as the upstream recursive resolvers
+            trace!(
+                "initializing DNS resolver with google-config {:?} opts {:?}",
+                ResolverConfig::google(),
+                resolver_opts
+            );
+            TokioAsyncResolver::tokio(ResolverConfig::google(), resolver_opts).await
         }
     }
-    .await
-    .map_err(|err| {
-        Error::new(
-            ErrorKind::Other,
-            format!("failed to create trust-dns DNS resolver, error: {}", err),
-        )
-    })
+    .map_err(From::from)
 }
 
 /// Perform a DNS resolution

src/relay/local.rs

@@ -4,7 +4,6 @@ use std::io::{self, ErrorKind};
 
 use futures::{future::select_all, FutureExt};
 use log::{debug, error, trace, warn};
-use tokio::runtime::Handle;
 
 #[cfg(feature = "local-flow-stat")]
 use crate::context::SharedContext;
@@ -16,7 +15,7 @@ use crate::{
 };
 
 /// Relay server running under local environment.
-pub async fn run(mut config: Config, rt: Handle) -> io::Result<()> {
+pub async fn run(mut config: Config) -> io::Result<()> {
     trace!("initializing local server with {:?}", config);
 
     assert!(config.config_type.is_local());
@@ -43,7 +42,7 @@ pub async fn run(mut config: Config, rt: Handle) -> io::Result<()> {
     let mode = config.mode;
 
     // Create a context containing a DNS resolver and server running state flag.
-    let state = ServerState::new_shared(&config, rt).await;
+    let state = ServerState::new_shared(&config).await;
 
     let mut vf = Vec::new();
 

src/relay/manager.rs

@@ -16,7 +16,7 @@ use futures::future;
 use log::{debug, error, trace, warn};
 #[cfg(unix)]
 use tokio::net::UnixDatagram;
-use tokio::{self, net::UdpSocket, runtime::Handle, sync::oneshot};
+use tokio::{self, net::UdpSocket, sync::oneshot};
 
 use crate::{
     config::{Config, ConfigType, ManagerAddr, Mode, ServerAddr, ServerConfig},
@@ -533,7 +533,7 @@ impl ManagerService {
 }
 
 /// Server manager for supporting [Manage Multiple Users](https://github.com/shadowsocks/shadowsocks/wiki/Manage-Multiple-Users) APIs
-pub async fn run(config: Config, rt: Handle) -> io::Result<()> {
+pub async fn run(config: Config) -> io::Result<()> {
     assert!(config.config_type.is_manager());
 
     if let Some(nofile) = config.nofile {
@@ -555,7 +555,7 @@ pub async fn run(config: Config, rt: Handle) -> io::Result<()> {
     }
 
     // Create a context containing a DNS resolver and server running state flag.
-    let state = ServerState::new_shared(&config, rt.clone()).await;
+    let state = ServerState::new_shared(&config).await;
     let context = Context::new_shared(config, state.clone());
 
     let bind_addr = match context.config().manager_addr {

src/relay/server.rs

@@ -7,7 +7,7 @@ use std::{
 
 use futures::future::{select_all, FutureExt};
 use log::{debug, error, trace, warn};
-use tokio::{runtime::Handle, time};
+use tokio::time;
 
 use crate::{
     config::Config,
@@ -24,9 +24,9 @@ use crate::{
 
 /// Runs Relay server on server side.
 #[inline]
-pub async fn run(config: Config, rt: Handle) -> io::Result<()> {
+pub async fn run(config: Config) -> io::Result<()> {
     // Create a context containing a DNS resolver and server running state flag.
-    let server_state = ServerState::new_shared(&config, rt).await;
+    let server_state = ServerState::new_shared(&config).await;
 
     // Create statistics for multiple servers
     //

tests/socks5.rs

@@ -2,7 +2,6 @@ use std::net::{SocketAddr, ToSocketAddrs};
 
 use tokio::{
     prelude::*,
-    runtime::{Builder, Handle},
     time::{self, Duration},
 };
 
@@ -51,19 +50,19 @@ impl Socks5TestServer {
         &self.local_addr
     }
 
-    pub async fn run(&self, rt_handle: Handle) {
+    pub async fn run(&self) {
         let svr_cfg = self.svr_config.clone();
-        tokio::spawn(run_server(svr_cfg, rt_handle.clone()));
+        tokio::spawn(run_server(svr_cfg));
 
         let client_cfg = self.cli_config.clone();
-        tokio::spawn(run_local(client_cfg, rt_handle));
+        tokio::spawn(run_local(client_cfg));
 
         time::delay_for(Duration::from_secs(1)).await;
     }
 }
 
-#[test]
-fn socks5_relay_stream() {
+#[tokio::test]
+async fn socks5_relay_stream() {
     let _ = env_logger::try_init();
 
     const SERVER_ADDR: &str = "127.0.0.1:8100";
@@ -72,33 +71,28 @@ fn socks5_relay_stream() {
     const PASSWORD: &str = "test-password";
     const METHOD: CipherType = CipherType::Aes128Gcm;
 
-    let mut rt = Builder::new().basic_scheduler().enable_all().build().unwrap();
-    let rt_handle = rt.handle().clone();
+    let svr = Socks5TestServer::new(SERVER_ADDR, LOCAL_ADDR, PASSWORD, METHOD, false);
+    svr.run().await;
 
-    rt.block_on(async move {
-        let svr = Socks5TestServer::new(SERVER_ADDR, LOCAL_ADDR, PASSWORD, METHOD, false);
-        svr.run(rt_handle).await;
+    let mut c = Socks5Client::connect(
+        Address::DomainNameAddress("www.example.com".to_owned(), 80),
+        svr.client_addr(),
+    )
+    .await
+    .unwrap();
 
-        let mut c = Socks5Client::connect(
-            Address::DomainNameAddress("www.example.com".to_owned(), 80),
-            svr.client_addr(),
-        )
-        .await
-        .unwrap();
+    let req = b"GET / HTTP/1.0\r\nHost: www.example.com\r\nAccept: */*\r\n\r\n";
+    c.write_all(req).await.unwrap();
+    c.flush().await.unwrap();
 
-        let req = b"GET / HTTP/1.0\r\nHost: www.example.com\r\nAccept: */*\r\n\r\n";
-        c.write_all(req).await.unwrap();
-        c.flush().await.unwrap();
+    let mut buf = Vec::new();
+    c.read_to_end(&mut buf).await.unwrap();
 
-        let mut buf = Vec::new();
-        c.read_to_end(&mut buf).await.unwrap();
-
-        println!("Got reply from server: {}", String::from_utf8(buf).unwrap());
-    });
+    println!("Got reply from server: {}", String::from_utf8(buf).unwrap());
 }
 
-#[test]
-fn socks5_relay_aead() {
+#[tokio::test]
+async fn socks5_relay_aead() {
     let _ = env_logger::try_init();
 
     const SERVER_ADDR: &str = "127.0.0.1:8110";
@@ -107,27 +101,22 @@ fn socks5_relay_aead() {
     const PASSWORD: &str = "test-password";
     const METHOD: CipherType = CipherType::Aes256Gcm;
 
-    let mut rt = Builder::new().basic_scheduler().enable_all().build().unwrap();
-    let rt_handle = rt.handle().clone();
+    let svr = Socks5TestServer::new(SERVER_ADDR, LOCAL_ADDR, PASSWORD, METHOD, false);
+    svr.run().await;
 
-    rt.block_on(async move {
-        let svr = Socks5TestServer::new(SERVER_ADDR, LOCAL_ADDR, PASSWORD, METHOD, false);
-        svr.run(rt_handle).await;
+    let mut c = Socks5Client::connect(
+        Address::DomainNameAddress("www.example.com".to_owned(), 80),
+        svr.client_addr(),
+    )
+    .await
+    .unwrap();
 
-        let mut c = Socks5Client::connect(
-            Address::DomainNameAddress("www.example.com".to_owned(), 80),
-            svr.client_addr(),
-        )
-        .await
-        .unwrap();
+    let req = b"GET / HTTP/1.0\r\nHost: www.example.com\r\nAccept: */*\r\n\r\n";
+    c.write_all(req).await.unwrap();
+    c.flush().await.unwrap();
 
-        let req = b"GET / HTTP/1.0\r\nHost: www.example.com\r\nAccept: */*\r\n\r\n";
-        c.write_all(req).await.unwrap();
-        c.flush().await.unwrap();
+    let mut buf = Vec::new();
+    c.read_to_end(&mut buf).await.unwrap();
 
-        let mut buf = Vec::new();
-        c.read_to_end(&mut buf).await.unwrap();
-
-        println!("Got reply from server: {}", String::from_utf8(buf).unwrap());
-    });
+    println!("Got reply from server: {}", String::from_utf8(buf).unwrap());
 }

tests/tunnel.rs

@@ -2,7 +2,6 @@ use tokio::{
     self,
     net::{TcpStream, UdpSocket},
     prelude::*,
-    runtime::Builder,
     time::{self, Duration},
 };
 
@@ -13,8 +12,8 @@ use shadowsocks::{
     run_server,
 };
 
-#[test]
-fn tcp_tunnel() {
+#[tokio::test]
+async fn tcp_tunnel() {
     let _ = env_logger::try_init();
 
     let mut local_config = Config::load_from_str(
@@ -43,31 +42,26 @@ fn tcp_tunnel() {
     )
     .unwrap();
 
-    let mut rt = Builder::new().basic_scheduler().enable_all().build().unwrap();
-    let rt_handle = rt.handle().clone();
+    tokio::spawn(run_local(local_config));
+    tokio::spawn(run_server(server_config));
 
-    rt.block_on(async move {
-        tokio::spawn(run_local(local_config, rt_handle.clone()));
-        tokio::spawn(run_server(server_config, rt_handle));
+    time::delay_for(Duration::from_secs(1)).await;
 
-        time::delay_for(Duration::from_secs(1)).await;
+    // Connect it directly, because it is now established a TCP tunnel with www.example.com
+    let mut stream = TcpStream::connect("127.0.0.1:9110").await.unwrap();
 
-        // Connect it directly, because it is now established a TCP tunnel with www.example.com
-        let mut stream = TcpStream::connect("127.0.0.1:9110").await.unwrap();
+    let req = b"GET / HTTP/1.0\r\nHost: www.example.com\r\nAccept: */*\r\n\r\n";
+    stream.write_all(req).await.unwrap();
+    stream.flush().await.unwrap();
 
-        let req = b"GET / HTTP/1.0\r\nHost: www.example.com\r\nAccept: */*\r\n\r\n";
-        stream.write_all(req).await.unwrap();
-        stream.flush().await.unwrap();
+    let mut buf = Vec::new();
+    stream.read_to_end(&mut buf).await.unwrap();
 
-        let mut buf = Vec::new();
-        stream.read_to_end(&mut buf).await.unwrap();
-
-        println!("Got reply from server: {}", String::from_utf8(buf).unwrap());
-    });
+    println!("Got reply from server: {}", String::from_utf8(buf).unwrap());
 }
 
-#[test]
-fn udp_tunnel() {
+#[tokio::test]
+async fn udp_tunnel() {
     let _ = env_logger::try_init();
 
     let mut local_config = Config::load_from_str(
@@ -98,33 +92,28 @@ fn udp_tunnel() {
     )
     .unwrap();
 
-    let mut rt = Builder::new().basic_scheduler().enable_all().build().unwrap();
-    let rt_handle = rt.handle().clone();
+    tokio::spawn(run_local(local_config));
+    tokio::spawn(run_server(server_config));
 
-    rt.block_on(async move {
-        tokio::spawn(run_local(local_config, rt_handle.clone()));
-        tokio::spawn(run_server(server_config, rt_handle));
-
-        // Start a UDP echo server
-        tokio::spawn(async {
-            let mut socket = UdpSocket::bind("127.0.0.1:9230").await.unwrap();
-
-            let mut buf = vec![0u8; 65536];
-            let (n, src) = socket.recv_from(&mut buf).await.unwrap();
-
-            println!("UDP Echo server received packet, size: {}, src: {}", n, src);
-
-            socket.send_to(&buf[..n], src).await.unwrap();
-        });
-
-        time::delay_for(Duration::from_secs(1)).await;
-
-        let mut socket = UdpSocket::bind("0.0.0.0:0").await.unwrap();
-        socket.send_to(b"HELLO WORLD", "127.0.0.1:9210").await.unwrap();
+    // Start a UDP echo server
+    tokio::spawn(async {
+        let mut socket = UdpSocket::bind("127.0.0.1:9230").await.unwrap();
 
         let mut buf = vec![0u8; 65536];
-        let n = socket.recv(&mut buf).await.unwrap();
+        let (n, src) = socket.recv_from(&mut buf).await.unwrap();
 
-        println!("Got reply from server: {}", ::std::str::from_utf8(&buf[..n]).unwrap());
+        println!("UDP Echo server received packet, size: {}, src: {}", n, src);
+
+        socket.send_to(&buf[..n], src).await.unwrap();
     });
+
+    time::delay_for(Duration::from_secs(1)).await;
+
+    let mut socket = UdpSocket::bind("0.0.0.0:0").await.unwrap();
+    socket.send_to(b"HELLO WORLD", "127.0.0.1:9210").await.unwrap();
+
+    let mut buf = vec![0u8; 65536];
+    let n = socket.recv(&mut buf).await.unwrap();
+
+    println!("Got reply from server: {}", ::std::str::from_utf8(&buf[..n]).unwrap());
 }

tests/udp.rs

@@ -9,7 +9,6 @@ use bytes::{BufMut, BytesMut};
 use log::debug;
 use tokio::{
     prelude::*,
-    runtime::{Builder, Handle},
     time::{self, Duration},
 };
 
@@ -60,12 +59,12 @@ fn get_client_addr() -> SocketAddr {
     LOCAL_ADDR.parse().unwrap()
 }
 
-fn start_server(handle: Handle) {
-    tokio::spawn(run_server(get_svr_config(), handle));
+fn start_server() {
+    tokio::spawn(run_server(get_svr_config()));
 }
 
-fn start_local(handle: Handle) {
-    tokio::spawn(run_local(get_cli_config(), handle));
+fn start_local() {
+    tokio::spawn(run_local(get_cli_config()));
 }
 
 fn start_udp_echo_server() {
@@ -102,56 +101,51 @@ fn start_udp_request_holder(addr: Address) {
     });
 }
 
-#[test]
-fn udp_relay() {
+#[tokio::test]
+async fn udp_relay() {
     use tokio::net::UdpSocket;
 
     let _ = env_logger::try_init();
 
     let remote_addr = Address::SocketAddress(UDP_ECHO_SERVER_ADDR.parse().unwrap());
 
-    let mut rt = Builder::new().basic_scheduler().enable_all().build().unwrap();
-    let rt_handle = rt.handle().clone();
+    start_server();
+    start_local();
 
-    rt.block_on(async move {
-        start_server(rt_handle.clone());
-        start_local(rt_handle);
+    start_udp_echo_server();
 
-        start_udp_echo_server();
+    // Wait until all server starts
+    time::delay_for(Duration::from_secs(1)).await;
 
-        // Wait until all server starts
-        time::delay_for(Duration::from_secs(1)).await;
+    start_udp_request_holder(remote_addr.clone());
 
-        start_udp_request_holder(remote_addr.clone());
+    let mut l = UdpSocket::bind(UDP_LOCAL_ADDR).await.unwrap();
 
-        let mut l = UdpSocket::bind(UDP_LOCAL_ADDR).await.unwrap();
+    let header = UdpAssociateHeader::new(0, remote_addr);
+    let mut buf = BytesMut::with_capacity(header.serialized_len());
+    header.write_to_buf(&mut buf);
 
-        let header = UdpAssociateHeader::new(0, remote_addr);
-        let mut buf = BytesMut::with_capacity(header.serialized_len());
-        header.write_to_buf(&mut buf);
+    let payload = b"HEllo WORld";
 
-        let payload = b"HEllo WORld";
+    buf.reserve(payload.len());
+    buf.put_slice(payload);
 
-        buf.reserve(payload.len());
-        buf.put_slice(payload);
+    let local_addr = LOCAL_ADDR.parse::<SocketAddr>().unwrap();
+    l.send_to(&buf[..], &local_addr).await.unwrap();
 
-        let local_addr = LOCAL_ADDR.parse::<SocketAddr>().unwrap();
-        l.send_to(&buf[..], &local_addr).await.unwrap();
+    let mut buf = vec![0u8; 65536];
+    let (amt, _) = time::timeout(Duration::from_secs(5), l.recv_from(&mut buf))
+        .await
+        .unwrap()
+        .unwrap();
+    println!("Received buf size={} {:?}", amt, &buf[..amt]);
 
-        let mut buf = vec![0u8; 65536];
-        let (amt, _) = time::timeout(Duration::from_secs(5), l.recv_from(&mut buf))
-            .await
-            .unwrap()
-            .unwrap();
-        println!("Received buf size={} {:?}", amt, &buf[..amt]);
+    let mut cur = Cursor::new(buf[..amt].to_vec());
+    let header = UdpAssociateHeader::read_from(&mut cur).await.unwrap();
+    println!("{:?}", header);
+    let header_len = cur.position() as usize;
+    let buf = cur.into_inner();
+    let buf = &buf[header_len..];
 
-        let mut cur = Cursor::new(buf[..amt].to_vec());
-        let header = UdpAssociateHeader::read_from(&mut cur).await.unwrap();
-        println!("{:?}", header);
-        let header_len = cur.position() as usize;
-        let buf = cur.into_inner();
-        let buf = &buf[header_len..];
-
-        assert_eq!(buf, payload);
-    });
+    assert_eq!(buf, payload);
 }