refactor send syscall

loongs-zhang · loongs-zhang · commit b742c4c05193 · 2024-07-06T10:50:20.000+08:00
diff --git a/examples/src/lib.rs b/examples/src/lib.rs
@@ -1,39 +1,127 @@
 use open_coroutine::task;
-use std::io::{BufRead, BufReader, Read, Write};
-use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpListener, TcpStream};
+use std::io::{BufRead, BufReader, ErrorKind, IoSlice, IoSliceMut, Read, Write};
+use std::net::{IpAddr, Ipv4Addr, Shutdown, SocketAddr, TcpListener, TcpStream, ToSocketAddrs};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Condvar, Mutex};
 use std::time::Duration;
 
-fn crate_co(input: i32) {
+fn crate_task(input: i32) {
     _ = task!(
         |_, param| {
-            println!("[coroutine] launched param:{}", param);
+            println!("[coroutine{}] launched", param);
         },
         input,
     );
 }
 
+pub fn start_server<A: ToSocketAddrs>(
+    addr: A,
+    server_finished: Arc<(Mutex<bool>, Condvar)>,
+) -> std::io::Result<()> {
+    let listener = TcpListener::bind(addr)?;
+    for stream in listener.incoming() {
+        let mut stream = stream?;
+        let mut buffer1 = [0; 256];
+        for _ in 0..3 {
+            assert_eq!(12, stream.read(&mut buffer1)?);
+            println!("Server Received: {}", String::from_utf8_lossy(&buffer1));
+            assert_eq!(256, stream.write(&buffer1)?);
+            println!("Server Send");
+        }
+        let mut buffer2 = [0; 256];
+        for _ in 0..3 {
+            let mut buffers = [IoSliceMut::new(&mut buffer1), IoSliceMut::new(&mut buffer2)];
+            assert_eq!(26, stream.read_vectored(&mut buffers)?);
+            println!(
+                "Server Received Multiple: {}{}",
+                String::from_utf8_lossy(&buffer1),
+                String::from_utf8_lossy(&buffer2)
+            );
+            let responses = [IoSlice::new(&buffer1), IoSlice::new(&buffer2)];
+            assert_eq!(512, stream.write_vectored(&responses)?);
+            println!("Server Send Multiple");
+        }
+        println!("Server Shutdown Write");
+        stream.shutdown(Shutdown::Write).map(|()| {
+            println!("Server Closed Connection");
+        })?;
+        let (lock, cvar) = &*server_finished;
+        let mut pending = lock.lock().unwrap();
+        *pending = false;
+        cvar.notify_one();
+    }
+    Ok(())
+}
+
+pub fn start_client<A: ToSocketAddrs>(addr: A) -> std::io::Result<()> {
+    let mut stream = connect_timeout(addr, Duration::from_secs(3))?;
+    let mut buffer1 = [0; 256];
+    for i in 0..3 {
+        assert_eq!(12, stream.write(format!("RequestPart{i}").as_ref())?);
+        println!("Client Send");
+        assert_eq!(256, stream.read(&mut buffer1)?);
+        println!("Client Received: {}", String::from_utf8_lossy(&buffer1));
+    }
+    let mut buffer2 = [0; 256];
+    for i in 0..3 {
+        let request1 = format!("RequestPart{i}1");
+        let request2 = format!("RequestPart{i}2");
+        let requests = [
+            IoSlice::new(request1.as_ref()),
+            IoSlice::new(request2.as_ref()),
+        ];
+        assert_eq!(26, stream.write_vectored(&requests)?);
+        println!("Client Send Multiple");
+        let mut buffers = [IoSliceMut::new(&mut buffer1), IoSliceMut::new(&mut buffer2)];
+        assert_eq!(512, stream.read_vectored(&mut buffers)?);
+        println!(
+            "Client Received Multiple: {}{}",
+            String::from_utf8_lossy(&buffer1),
+            String::from_utf8_lossy(&buffer2)
+        );
+    }
+    println!("Client Shutdown Write");
+    stream.shutdown(Shutdown::Write).map(|()| {
+        println!("Client Closed");
+    })
+}
+
+fn connect_timeout<A: ToSocketAddrs>(addr: A, timeout: Duration) -> std::io::Result<TcpStream> {
+    let mut last_err = None;
+    for addr in addr.to_socket_addrs()? {
+        match TcpStream::connect_timeout(&addr, timeout) {
+            Ok(l) => return Ok(l),
+            Err(e) => last_err = Some(e),
+        }
+    }
+    Err(last_err.unwrap_or_else(|| {
+        std::io::Error::new(
+            ErrorKind::InvalidInput,
+            "could not resolve to any addresses",
+        )
+    }))
+}
+
 pub fn crate_server(
     port: u16,
     server_started: Arc<AtomicBool>,
     server_finished: Arc<(Mutex<bool>, Condvar)>,
 ) {
     //invoke by libc::listen
-    crate_co(1);
+    crate_task(1);
     let mut data: [u8; 512] = unsafe { std::mem::zeroed() };
     data[511] = b'\n';
     let listener = TcpListener::bind(format!("127.0.0.1:{port}"))
         .unwrap_or_else(|_| panic!("bind to 127.0.0.1:{port} failed !"));
     server_started.store(true, Ordering::Release);
     //invoke by libc::accept
-    crate_co(2);
+    crate_task(2);
     if let Some(stream) = listener.incoming().next() {
         let mut stream = stream.expect("accept new connection failed !");
         let mut buffer: [u8; 512] = [0; 512];
         loop {
             //invoke by libc::recv
-            crate_co(6);
+            crate_task(6);
             //从流里面读内容，读到buffer中
             let bytes_read = stream.read(&mut buffer).expect("server read failed !");
             if bytes_read == 0 {
@@ -48,13 +136,13 @@ pub fn crate_server(
                 *pending = false;
                 cvar.notify_one();
                 println!("server closed");
-                crate_co(8);
+                crate_task(8);
                 return;
             }
             assert_eq!(512, bytes_read);
             assert_eq!(data, buffer);
             //invoke by libc::send
-            crate_co(7);
+            crate_task(7);
             //回写
             assert_eq!(
                 bytes_read,
@@ -74,7 +162,7 @@ pub fn crate_client(port: u16, server_started: Arc<AtomicBool>) {
     //等服务端起来
     while !server_started.load(Ordering::Acquire) {}
     //invoke by libc::connect
-    crate_co(3);
+    crate_task(3);
     let socket = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), port);
     let mut stream = TcpStream::connect_timeout(&socket, Duration::from_secs(3))
         .unwrap_or_else(|_| panic!("connect to 127.0.0.1:{port} failed !"));
@@ -83,13 +171,13 @@ pub fn crate_client(port: u16, server_started: Arc<AtomicBool>) {
     let mut buffer: Vec<u8> = Vec::with_capacity(512);
     for _ in 0..3 {
         //invoke by libc::send
-        crate_co(4);
+        crate_task(4);
         //写入stream流，如果写入失败，提示"写入失败"
         assert_eq!(512, stream.write(&data).expect("Failed to write!"));
         print!("Client Send: {}", String::from_utf8_lossy(&data[..]));
 
         //invoke by libc::recv
-        crate_co(5);
+        crate_task(5);
         let mut reader = BufReader::new(&stream);
         //一直读到换行为止（b'\n'中的b表示字节），读到buffer里面
         assert_eq!(
@@ -105,7 +193,7 @@ pub fn crate_client(port: u16, server_started: Arc<AtomicBool>) {
     //发送终止符
     assert_eq!(1, stream.write(&[b'e']).expect("Failed to write!"));
     println!("client closed");
-    crate_co(8);
+    crate_task(8);
 }
 
 pub fn crate_co_server(
@@ -114,22 +202,22 @@ pub fn crate_co_server(
     server_finished: Arc<(Mutex<bool>, Condvar)>,
 ) {
     //invoke by libc::listen
-    crate_co(11);
+    crate_task(11);
     let mut data: [u8; 512] = unsafe { std::mem::zeroed() };
     data[511] = b'\n';
     let listener = TcpListener::bind(format!("127.0.0.1:{port}"))
         .unwrap_or_else(|_| panic!("bind to 127.0.0.1:{port} failed !"));
     server_started.store(true, Ordering::Release);
     //invoke by libc::accept
-    crate_co(12);
+    crate_task(12);
     for stream in listener.incoming() {
         _ = task!(
             |_, input| {
                 let mut stream = input.expect("accept new connection failed !");
                 let mut buffer: [u8; 512] = [0; 512];
                 loop {
                     //invoke by libc::recv
-                    crate_co(16);
+                    crate_task(16);
                     //从流里面读内容，读到buffer中
                     let bytes_read = stream
                         .read(&mut buffer)
@@ -149,13 +237,13 @@ pub fn crate_co_server(
                         *pending = false;
                         cvar.notify_one();
                         println!("coroutine server closed");
-                        crate_co(18);
+                        crate_task(18);
                         return Some(Box::leak(Box::new(stream)));
                     }
                     assert_eq!(512, bytes_read);
                     assert_eq!(data, buffer);
                     //invoke by libc::send
-                    crate_co(17);
+                    crate_task(17);
                     //回写
                     assert_eq!(
                         bytes_read,
@@ -180,7 +268,7 @@ pub fn crate_co_client(port: u16, server_started: Arc<AtomicBool>) {
     _ = task!(
         |_, input| {
             //invoke by libc::connect
-            crate_co(13);
+            crate_task(13);
             let socket = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), input);
             let mut stream = TcpStream::connect_timeout(&socket, Duration::from_secs(3))
                 .unwrap_or_else(|_| panic!("connect to 127.0.0.1:{input} failed !"));
@@ -189,7 +277,7 @@ pub fn crate_co_client(port: u16, server_started: Arc<AtomicBool>) {
             let mut buffer: Vec<u8> = Vec::with_capacity(512);
             for _ in 0..3 {
                 //invoke by libc::send
-                crate_co(14);
+                crate_task(14);
                 //写入stream流，如果写入失败，提示"写入失败"
                 assert_eq!(512, stream.write(&data).expect("Failed to write!"));
                 print!(
@@ -198,7 +286,7 @@ pub fn crate_co_client(port: u16, server_started: Arc<AtomicBool>) {
                 );
 
                 //invoke by libc::recv
-                crate_co(15);
+                crate_task(15);
                 let mut reader = BufReader::new(&stream);
                 //一直读到换行为止（b'\n'中的b表示字节），读到buffer里面
                 assert_eq!(
@@ -217,7 +305,7 @@ pub fn crate_co_client(port: u16, server_started: Arc<AtomicBool>) {
             //发送终止符
             assert_eq!(1, stream.write(&[b'e']).expect("Failed to write!"));
             println!("coroutine client closed");
-            crate_co(18);
+            crate_task(18);
         },
         port,
     );
diff --git a/open-coroutine-core/src/monitor/mod.rs b/open-coroutine-core/src/monitor/mod.rs
@@ -166,10 +166,7 @@ impl Monitor {
 
     #[allow(dead_code)]
     pub fn stop() {
-        assert_eq!(
-            MonitorState::Running,
-            Self::get_instance().state.replace(MonitorState::Stopping)
-        );
+        Self::get_instance().state.set(MonitorState::Stopping);
         cfg_if::cfg_if! {
             if #[cfg(feature = "net")] {
                 let pair = EventLoops::new_condition();
diff --git a/open-coroutine-core/src/syscall/facade.rs b/open-coroutine-core/src/syscall/facade.rs
@@ -131,17 +131,6 @@ pub extern "C" fn recvmsg(
 
 /// write
 
-#[must_use]
-pub extern "C" fn send(
-    fn_ptr: Option<&extern "C" fn(c_int, *const c_void, size_t, c_int) -> ssize_t>,
-    socket: c_int,
-    buf: *const c_void,
-    len: size_t,
-    flags: c_int,
-) -> ssize_t {
-    CHAIN.send(fn_ptr, socket, buf, len, flags)
-}
-
 #[must_use]
 pub extern "C" fn sendto(
     fn_ptr: Option<
diff --git a/open-coroutine-core/src/syscall/io_uring.rs b/open-coroutine-core/src/syscall/io_uring.rs
@@ -136,17 +136,6 @@ impl<I: UnixSyscall> UnixSyscall for IoUringLinuxSyscall<I> {
         impl_io_uring!(self, recvmsg, fn_ptr, fd, msg, flags)
     }
 
-    extern "C" fn send(
-        &self,
-        fn_ptr: Option<&extern "C" fn(c_int, *const c_void, size_t, c_int) -> ssize_t>,
-        socket: c_int,
-        buf: *const c_void,
-        len: size_t,
-        flags: c_int,
-    ) -> ssize_t {
-        impl_io_uring!(self, send, fn_ptr, socket, buf, len, flags)
-    }
-
     extern "C" fn sendto(
         &self,
         fn_ptr: Option<
diff --git a/open-coroutine-core/src/syscall/mod.rs b/open-coroutine-core/src/syscall/mod.rs
@@ -127,15 +127,6 @@ pub trait UnixSyscall {
 
     /// write
 
-    extern "C" fn send(
-        &self,
-        fn_ptr: Option<&extern "C" fn(c_int, *const c_void, size_t, c_int) -> ssize_t>,
-        fd: c_int,
-        buf: *const c_void,
-        len: size_t,
-        flags: c_int,
-    ) -> ssize_t;
-
     extern "C" fn sendto(
         &self,
         fn_ptr: Option<
diff --git a/open-coroutine-core/src/syscall/nio.rs b/open-coroutine-core/src/syscall/nio.rs
@@ -527,17 +527,6 @@ impl<I: UnixSyscall> UnixSyscall for NioLinuxSyscall<I> {
         r
     }
 
-    extern "C" fn send(
-        &self,
-        fn_ptr: Option<&extern "C" fn(c_int, *const c_void, size_t, c_int) -> ssize_t>,
-        socket: c_int,
-        buf: *const c_void,
-        len: size_t,
-        flags: c_int,
-    ) -> ssize_t {
-        impl_expected_write_hook!(self.inner, send, fn_ptr, socket, buf, len, flags)
-    }
-
     extern "C" fn sendto(
         &self,
         fn_ptr: Option<
diff --git a/open-coroutine-core/src/syscall/raw.rs b/open-coroutine-core/src/syscall/raw.rs
@@ -163,21 +163,6 @@ impl UnixSyscall for RawLinuxSyscall {
 
     /// write
 
-    extern "C" fn send(
-        &self,
-        fn_ptr: Option<&extern "C" fn(c_int, *const c_void, size_t, c_int) -> ssize_t>,
-        socket: c_int,
-        buf: *const c_void,
-        len: size_t,
-        flags: c_int,
-    ) -> ssize_t {
-        if let Some(f) = fn_ptr {
-            (f)(socket, buf, len, flags)
-        } else {
-            unsafe { libc::send(socket, buf, len, flags) }
-        }
-    }
-
     extern "C" fn sendto(
         &self,
         fn_ptr: Option<
diff --git a/open-coroutine-core/src/syscall/state.rs b/open-coroutine-core/src/syscall/state.rs
@@ -146,17 +146,6 @@ impl<I: UnixSyscall> UnixSyscall for StateLinuxSyscall<I> {
         syscall_state!(self, recvmsg, fn_ptr, fd, msg, flags)
     }
 
-    extern "C" fn send(
-        &self,
-        fn_ptr: Option<&extern "C" fn(c_int, *const c_void, size_t, c_int) -> ssize_t>,
-        socket: c_int,
-        buf: *const c_void,
-        len: size_t,
-        flags: c_int,
-    ) -> ssize_t {
-        syscall_state!(self, send, fn_ptr, socket, buf, len, flags)
-    }
-
     extern "C" fn sendto(
         &self,
         fn_ptr: Option<
diff --git a/open-coroutine-core/src/syscall/unix/mod.rs b/open-coroutine-core/src/syscall/unix/mod.rs
diff --git a/open-coroutine-core/src/syscall/unix/send.rs b/open-coroutine-core/src/syscall/unix/send.rs
