feat: Add TCP socket support

Add TCP support for non-windows platforms:
- Add helpers to parse TCP addresses
- Add TCP transport and helpers for the async implementation

Signed-off-by: Kostis Papazafeiropoulos <papazof@gmail.com>

Author: papazof

src/asynchronous/server.rs

@@ -101,6 +101,15 @@ impl Server {
         Ok(self.add_listener(listener))
     }
 
+    #[cfg(unix)]
+    /// # Safety
+    /// The file descriptor must represent a unix listener.
+    pub unsafe fn add_tcp_listener(self, fd: RawFd) -> Result<Server> {
+        let listener = Listener::from_raw_tcp_listener_fd(fd)
+            .map_err(err_to_others_err!(e, "from_raw_tcp_listener_fd error"))?;
+        Ok(self.add_listener(listener))
+    }
+
     #[cfg(any(target_os = "linux", target_os = "android"))]
     /// # Safety
     /// The file descriptor must represent a vsock listener.

src/asynchronous/transport/mod.rs

@@ -22,6 +22,8 @@ macro_rules! io_other {
 #[cfg(unix)]
 mod unix;
 
+mod tcp;
+
 #[cfg(any(target_os = "linux", target_os = "android"))]
 mod vsock;
 
@@ -43,6 +45,10 @@ impl Listener {
             return Self::bind_unix(addr);
         }
 
+        if let Some(addr) = addr.strip_prefix("tcp://") {
+            return Self::bind_tcp(addr);
+        }
+
         #[cfg(any(target_os = "linux", target_os = "android"))]
         if let Some(addr) = addr.strip_prefix("vsock://") {
             return Self::bind_vsock(addr);
@@ -70,6 +76,10 @@ impl Socket {
             return Self::connect_unix(addr).await;
         }
 
+        if let Some(addr) = addr.strip_prefix("tcp://") {
+            return Self::connect_tcp(addr).await;
+        }
+
         #[cfg(any(target_os = "linux", target_os = "android"))]
         if let Some(addr) = addr.strip_prefix("vsock://") {
             return Self::connect_vsock(addr).await;

src/asynchronous/transport/tcp.rs

@@ -0,0 +1,80 @@
+use std::convert::TryFrom;
+use std::io::{Error as IoError, Result as IoResult};
+use std::os::fd::{FromRawFd as _, RawFd};
+use std::net::{
+    SocketAddr, TcpListener as StdTcpListener, TcpStream as StdTcpStream,
+};
+
+use async_stream::stream;
+use tokio::net::{TcpListener, TcpStream};
+
+use super::{Listener, Socket};
+
+impl Listener {
+    pub fn bind_tcp(addr: impl AsRef<str>) -> IoResult<Self> {
+        let addr = parse_tcp_addr(addr)?;
+        let listener = StdTcpListener::bind(addr)?;
+        Self::try_from(listener)
+    }
+
+    /// # Safety
+    /// The file descriptor must represent a tcp listener.
+    pub unsafe fn from_raw_tcp_listener_fd(fd: std::os::fd::RawFd) -> IoResult<Self> {
+        let listener = unsafe { StdTcpListener::from_raw_fd(fd) };
+        Self::try_from(listener)
+    }
+}
+
+impl Socket {
+    pub async fn connect_tcp(addr: impl AsRef<str>) -> IoResult<Self> {
+        let addr = parse_tcp_addr(addr)?;
+        let socket = StdTcpStream::connect(addr)?;
+        Self::try_from(socket)
+    }
+
+    /// # Safety
+    /// The file descriptor must represent a tcp socket.
+    pub unsafe fn from_raw_tcp_socket_fd(fd: RawFd) -> IoResult<Self> {
+        let socket = unsafe { StdTcpStream::from_raw_fd(fd) };
+        Self::try_from(socket)
+    }
+}
+
+impl From<TcpListener> for Listener {
+    fn from(listener: TcpListener) -> Self {
+        Self::new(stream! {
+            loop {
+                yield listener.accept().await.map(|(socket, _)| socket);
+            }
+        })
+    }
+}
+
+impl TryFrom<StdTcpListener> for Listener {
+    type Error = IoError;
+    fn try_from(listener: StdTcpListener) -> IoResult<Self> {
+        listener.set_nonblocking(true)?;
+        Ok(Self::from(TcpListener::from_std(listener)?))
+    }
+}
+
+impl From<TcpStream> for Socket {
+    fn from(socket: TcpStream) -> Self {
+        Self::new(socket)
+    }
+}
+
+impl TryFrom<StdTcpStream> for Socket {
+    type Error = IoError;
+    fn try_from(socket: StdTcpStream) -> IoResult<Self> {
+        socket.set_nonblocking(true)?;
+        Ok(Self::from(TcpStream::from_std(socket)?))
+    }
+}
+
+fn parse_tcp_addr(addr: impl AsRef<str>) -> IoResult<SocketAddr> {
+    let addr = addr.as_ref();
+
+    addr.parse::<SocketAddr>()
+        .map_err(|e| io_other!("Failed to parse TCP address '{}': {}", addr, e))
+}

src/common.rs

@@ -8,13 +8,15 @@
 
 use nix::fcntl::{fcntl, FcntlArg, OFlag};
 use nix::sys::socket::*;
-use std::os::unix::io::RawFd;
+use std::str::FromStr;
+use std::{env, os::unix::io::RawFd};
 
 use crate::error::{Error, Result};
 
 #[derive(Debug, Clone, Copy, PartialEq)]
 pub(crate) enum Domain {
     Unix,
+    Tcp,
     #[cfg(any(target_os = "linux", target_os = "android"))]
     Vsock,
 }
@@ -39,6 +41,10 @@ fn parse_sockaddr(addr: &str) -> Result<(Domain, &str)> {
         return Ok((Domain::Vsock, addr));
     }
 
+    if let Some(addr) = addr.strip_prefix("tcp://") {
+        return Ok((Domain::Tcp, addr));
+    }
+
     Err(Error::Others(format!("Scheme {addr:?} is not supported")))
 }
 
@@ -53,6 +59,10 @@ fn parse_sockaddr(addr: &str) -> Result<(Domain, &str)> {
         return Ok((Domain::Unix, addr));
     }
 
+    if let Some(addr) = addr.strip_prefix("tcp://") {
+        return Ok((Domain::Tcp, addr));
+    }
+
     Err(Error::Others(format!("Scheme {addr:?} is not supported")))
 }
 
@@ -83,8 +93,8 @@ fn make_addr(domain: Domain, sockaddr: &str) -> Result<UnixAddr> {
                 UnixAddr::new(sockaddr).map_err(err_to_others_err!(e, ""))
             }
         }
-        Domain::Vsock => Err(Error::Others(
-            "function make_addr does not support create vsock socket".to_string(),
+        Domain::Vsock | Domain::Tcp => Err(Error::Others(
+            "function make_addr does not support create vsock/tcp socket".to_string(),
         )),
     }
 }
@@ -130,7 +140,7 @@ fn parse_vscok(addr: &str) -> Result<(u32, u32)> {
 fn make_socket(sockaddr: &str) -> Result<(RawFd, Domain, Box<dyn SockaddrLike>)> {
     let (domain, sockaddrv) = parse_sockaddr(sockaddr)?;
 
-    let get_sock_addr = |domain, sockaddr| -> Result<(RawFd, Box<dyn SockaddrLike>)> {
+    let get_unix_addr = |domain, sockaddr| -> Result<(RawFd, Box<dyn SockaddrLike>)> {
         let fd = socket(AddressFamily::Unix, SockType::Stream, SOCK_CLOEXEC, None)
             .map_err(|e| Error::Socket(e.to_string()))?;
 
@@ -141,9 +151,25 @@ fn make_socket(sockaddr: &str) -> Result<(RawFd, Domain, Box<dyn SockaddrLike>)>
         let sockaddr = make_addr(domain, sockaddr)?;
         Ok((fd, Box::new(sockaddr)))
     };
+    let get_tcp_addr = |sockaddr: &str| -> Result<(RawFd, Box<dyn SockaddrLike>)> {
+        let fd = socket(
+            AddressFamily::Inet,
+            SockType::Stream,
+            SockFlag::SOCK_CLOEXEC,
+            None,
+        )
+        .map_err(|e| Error::Socket(e.to_string()))?;
+
+        #[cfg(target_os = "macos")]
+        set_fd_close_exec(fd)?;
+        let sockaddr = SockaddrIn::from_str(sockaddr).map_err(err_to_others_err!(e, ""))?;
+
+        Ok((fd, Box::new(sockaddr)))
+    };
 
     let (fd, sockaddr): (i32, Box<dyn SockaddrLike>) = match domain {
-        Domain::Unix => get_sock_addr(domain, sockaddrv)?,
+        Domain::Unix => get_unix_addr(domain, sockaddrv)?,
+        Domain::Tcp => get_tcp_addr(sockaddrv)?,
         #[cfg(any(target_os = "linux", target_os = "android"))]
         Domain::Vsock => {
             let (cid, port) = parse_vscok(sockaddrv)?;
@@ -162,18 +188,41 @@ fn make_socket(sockaddr: &str) -> Result<(RawFd, Domain, Box<dyn SockaddrLike>)>
     Ok((fd, domain, sockaddr))
 }
 
+fn set_socket_opts(fd: RawFd, domain: Domain, is_bind: bool) -> Result<()> {
+    if domain != Domain::Tcp {
+        return Ok(());
+    }
+
+    if is_bind {
+        setsockopt(fd, sockopt::ReusePort, &true)?;
+    }
+
+    let tcp_nodelay_enabled = match env::var("TTRPC_TCP_NODELAY_ENABLED") {
+        Ok(val) if val == "1" || val.eq_ignore_ascii_case("true") => true,
+        Ok(val) if val == "0" || val.eq_ignore_ascii_case("false") => false,
+        _ => false,
+    };
+    if tcp_nodelay_enabled {
+        setsockopt(fd, sockopt::TcpNoDelay, &true)?;
+    }
+
+    Ok(())
+}
+
 pub(crate) fn do_bind(sockaddr: &str) -> Result<(RawFd, Domain)> {
     let (fd, domain, sockaddr) = make_socket(sockaddr)?;
 
+    set_socket_opts(fd, domain, true)?;
     bind(fd, sockaddr.as_ref()).map_err(err_to_others_err!(e, ""))?;
 
     Ok((fd, domain))
 }
 
 /// Creates a unix socket for client.
 pub(crate) unsafe fn client_connect(sockaddr: &str) -> Result<RawFd> {
-    let (fd, _, sockaddr) = make_socket(sockaddr)?;
+    let (fd, domain, sockaddr) = make_socket(sockaddr)?;
 
+    set_socket_opts(fd, domain, false)?;
     connect(fd, sockaddr.as_ref())?;
 
     Ok(fd)
@@ -202,6 +251,12 @@ mod tests {
                 true,
             ),
             ("abc:///run/c.sock", None, "", false),
+            (
+                "tcp://127.0.0.1:65500",
+                Some(Domain::Tcp),
+                "127.0.0.1:65500",
+                true,
+            ),
         ] {
             let (input, domain, addr, success) = (i.0, i.1, i.2, i.3);
             let r = parse_sockaddr(input);
@@ -229,6 +284,12 @@ mod tests {
             ("Vsock:///run/c.sock", None, "", false),
             ("unix://@/run/b.sock", None, "", false),
             ("abc:///run/c.sock", None, "", false),
+            (
+                "tcp://127.0.0.1:65500",
+                Some(Domain::Tcp),
+                "127.0.0.1:65500",
+                true,
+            ),
         ] {
             let (input, domain, addr, success) = (i.0, i.1, i.2, i.3);
             let r = parse_sockaddr(input);