Refactor to support other OSes

This moves unix specific calls out of the main server and client
functions.  It does this by introducing several new types: PipeListener, PipeConnection,  and ClientConnection.  These types are contain the unix specific functionality to communitcate with Unix Domain sockets and are hidden behind a conditional compilation flag.

Signed-off-by: James Sturtevant <[email protected]>

Author: jsturtevant

src/sync/channel.rs

@@ -12,18 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use nix::sys::socket::*;
-use std::os::unix::io::RawFd;
 
 use crate::error::{get_rpc_status, sock_error_msg, Error, Result};
+use crate::sync::sys::{PipeConnection};
 use crate::proto::{Code, MessageHeader, MESSAGE_HEADER_LENGTH, MESSAGE_LENGTH_MAX};
 
-fn retryable(e: nix::Error) -> bool {
-    use ::nix::Error;
-    e == Error::EINTR || e == Error::EAGAIN
-}
-
-fn read_count(fd: RawFd, count: usize) -> Result<Vec<u8>> {
+fn read_count (fd: &PipeConnection, count: usize) -> Result<Vec<u8>> {
     let mut v: Vec<u8> = vec![0; count];
     let mut len = 0;
 
@@ -32,19 +26,14 @@ fn read_count(fd: RawFd, count: usize) -> Result<Vec<u8>> {
     }
 
     loop {
-        match recv(fd, &mut v[len..], MsgFlags::empty()) {
+        match  fd.read(&mut v[len..]) {
             Ok(l) => {
                 len += l;
                 // when socket peer closed, it would return 0.
                 if len == count || l == 0 {
                     break;
                 }
             }
-
-            Err(e) if retryable(e) => {
-                // Should retry
-            }
-
             Err(e) => {
                 return Err(Error::Socket(e.to_string()));
             }
@@ -54,26 +43,21 @@ fn read_count(fd: RawFd, count: usize) -> Result<Vec<u8>> {
     Ok(v[0..len].to_vec())
 }
 
-fn write_count(fd: RawFd, buf: &[u8], count: usize) -> Result<usize> {
+fn write_count(fd: &PipeConnection, buf: &[u8], count: usize) -> Result<usize> {
     let mut len = 0;
 
     if count == 0 {
         return Ok(0);
     }
 
     loop {
-        match send(fd, &buf[len..], MsgFlags::empty()) {
+        match  fd.write(&buf[len..]){
             Ok(l) => {
                 len += l;
                 if len == count {
                     break;
                 }
             }
-
-            Err(e) if retryable(e) => {
-                // Should retry
-            }
-
             Err(e) => {
                 return Err(Error::Socket(e.to_string()));
             }
@@ -83,7 +67,7 @@ fn write_count(fd: RawFd, buf: &[u8], count: usize) -> Result<usize> {
     Ok(len)
 }
 
-fn read_message_header(fd: RawFd) -> Result<MessageHeader> {
+fn read_message_header(fd: &PipeConnection) -> Result<MessageHeader> {
     let buf = read_count(fd, MESSAGE_HEADER_LENGTH)?;
     let size = buf.len();
     if size != MESSAGE_HEADER_LENGTH {
@@ -98,7 +82,7 @@ fn read_message_header(fd: RawFd) -> Result<MessageHeader> {
     Ok(mh)
 }
 
-pub fn read_message(fd: RawFd) -> Result<(MessageHeader, Vec<u8>)> {
+pub fn read_message(fd: &PipeConnection) -> Result<(MessageHeader, Vec<u8>)> {
     let mh = read_message_header(fd)?;
     trace!("Got Message header {:?}", mh);
 
@@ -125,7 +109,7 @@ pub fn read_message(fd: RawFd) -> Result<(MessageHeader, Vec<u8>)> {
     Ok((mh, buf))
 }
 
-fn write_message_header(fd: RawFd, mh: MessageHeader) -> Result<()> {
+fn write_message_header(fd: &PipeConnection, mh: MessageHeader) -> Result<()> {
     let buf: Vec<u8> = mh.into();
 
     let size = write_count(fd, &buf, MESSAGE_HEADER_LENGTH)?;
@@ -139,7 +123,7 @@ fn write_message_header(fd: RawFd, mh: MessageHeader) -> Result<()> {
     Ok(())
 }
 
-pub fn write_message(fd: RawFd, mh: MessageHeader, buf: Vec<u8>) -> Result<()> {
+pub fn write_message(fd: &PipeConnection, mh: MessageHeader, buf: Vec<u8>) -> Result<()> {
     write_message_header(fd, mh)?;
 
     let size = write_count(fd, &buf, buf.len())?;

src/sync/client.rs

@@ -14,18 +14,15 @@
 
 //! Sync client of ttrpc.
 
-use nix::sys::socket::*;
-use nix::unistd::close;
+
 use std::collections::HashMap;
 use std::os::unix::io::RawFd;
 use std::sync::mpsc;
 use std::sync::{Arc, Mutex};
-use std::{io, thread};
+use std::{thread};
 
-#[cfg(target_os = "macos")]
-use crate::common::set_fd_close_exec;
-use crate::common::{client_connect, SOCK_CLOEXEC};
 use crate::error::{Error, Result};
+use crate::sync::sys::{ClientConnection};
 use crate::proto::{Code, Codec, MessageHeader, Request, Response, MESSAGE_TYPE_RESPONSE};
 use crate::sync::channel::{read_message, write_message};
 use std::time::Duration;
@@ -36,38 +33,36 @@ type Receiver = mpsc::Receiver<(Vec<u8>, mpsc::SyncSender<Result<Vec<u8>>>)>;
 /// A ttrpc Client (sync).
 #[derive(Clone)]
 pub struct Client {
-    _fd: RawFd,
+    _fd: Arc<ClientConnection>,
     sender_tx: Sender,
-    _client_close: Arc<ClientClose>,
 }
 
 impl Client {
     pub fn connect(sockaddr: &str) -> Result<Client> {
-        let fd = unsafe { client_connect(sockaddr)? };
-        Ok(Self::new(fd))
+        let conn = ClientConnection::client_connect(sockaddr)?;
+        
+        Ok(Self::new_client(conn))
     }
 
     /// Initialize a new [`Client`] from raw file descriptor.
     pub fn new(fd: RawFd) -> Client {
-        let (sender_tx, rx): (Sender, Receiver) = mpsc::channel();
+        let conn = ClientConnection::new(fd);
 
-        let (recver_fd, close_fd) =
-            socketpair(AddressFamily::Unix, SockType::Stream, None, SOCK_CLOEXEC).unwrap();
+        Self::new_client(conn)
+    }
 
-        // MacOS doesn't support descriptor creation with SOCK_CLOEXEC automically,
-        // so there is a chance of leak if fork + exec happens in between of these calls.
-        #[cfg(target_os = "macos")]
-        {
-            set_fd_close_exec(recver_fd).unwrap();
-            set_fd_close_exec(close_fd).unwrap();
-        }
+    fn new_client(pipe_client: ClientConnection) -> Client {
+        let client = Arc::new(pipe_client);
+        
 
-        let client_close = Arc::new(ClientClose { fd, close_fd });
+        let (sender_tx, rx): (Sender, Receiver) = mpsc::channel();
 
+        
         let recver_map_orig = Arc::new(Mutex::new(HashMap::new()));
 
         //Sender
         let recver_map = recver_map_orig.clone();
+        let sender_client = client.clone();
         thread::spawn(move || {
             let mut stream_id: u32 = 1;
             for (buf, recver_tx) in rx.iter() {
@@ -80,7 +75,8 @@ impl Client {
                 }
                 let mut mh = MessageHeader::new_request(0, buf.len() as u32);
                 mh.set_stream_id(current_stream_id);
-                if let Err(e) = write_message(fd, mh, buf) {
+                let c = sender_client.get_pipe_connection();
+                if let Err(e) = write_message(&c, mh, buf) {
                     //Remove current_stream_id and recver_tx to recver_map
                     {
                         let mut map = recver_map.lock().unwrap();
@@ -95,53 +91,28 @@ impl Client {
         });
 
         //Recver
+        let reciever_client = client.clone();
         thread::spawn(move || {
-            let mut pollers = vec![
-                libc::pollfd {
-                    fd: recver_fd,
-                    events: libc::POLLIN,
-                    revents: 0,
-                },
-                libc::pollfd {
-                    fd,
-                    events: libc::POLLIN,
-                    revents: 0,
-                },
-            ];
+          
 
             loop {
-                let returned = unsafe {
-                    let pollers: &mut [libc::pollfd] = &mut pollers;
-                    libc::poll(
-                        pollers as *mut _ as *mut libc::pollfd,
-                        pollers.len() as _,
-                        -1,
-                    )
-                };
-
-                if returned == -1 {
-                    let err = io::Error::last_os_error();
-                    if err.raw_os_error() == Some(libc::EINTR) {
+                
+                match reciever_client.ready() {
+                    Ok(None) => {
                         continue;
                     }
-
-                    error!("fatal error in process reaper:{}", err);
-                    break;
-                } else if returned < 1 {
-                    continue;
-                }
-
-                if pollers[0].revents != 0 {
-                    break;
-                }
-
-                if pollers[pollers.len() - 1].revents == 0 {
-                    continue;
+                    Ok(_) => {}
+                    Err(e) => {
+                        error!("pipeConnection ready error {:?}", e);
+                        break;
+                    }
                 }
-
                 let mh;
                 let buf;
-                match read_message(fd) {
+
+                let pipe_connection = reciever_client.get_pipe_connection();
+
+                match read_message(&pipe_connection) {
                     Ok((x, y)) => {
                         mh = x;
                         buf = y;
@@ -190,20 +161,18 @@ impl Client {
                 map.remove(&mh.stream_id);
             }
 
-            let _ = close(recver_fd).map_err(|e| {
+            let _ = reciever_client.close_receiver().map_err(|e| {
                 warn!(
-                    "failed to close recver_fd: {} with error: {:?}",
-                    recver_fd, e
+                    "failed to close with error: {:?}", e
                 )
             });
 
             trace!("Recver quit");
         });
 
         Client {
-            _fd: fd,
+            _fd: client,
             sender_tx,
-            _client_close: client_close,
         }
     }
     pub fn request(&self, req: Request) -> Result<Response> {
@@ -239,15 +208,9 @@ impl Client {
     }
 }
 
-struct ClientClose {
-    fd: RawFd,
-    close_fd: RawFd,
-}
-
-impl Drop for ClientClose {
+impl Drop for ClientConnection {
     fn drop(&mut self) {
-        close(self.close_fd).unwrap();
-        close(self.fd).unwrap();
-        trace!("All client is droped");
+        self.close().unwrap();
+        trace!("All client is dropped");
     }
 }

src/sync/mod.rs

@@ -8,6 +8,7 @@
 mod channel;
 mod client;
 mod server;
+mod sys;
 
 #[macro_use]
 mod utils;