using io uring

tcp/src/bin/server.rs

@@ -1,11 +1,9 @@
 // src/main.rs
 
+use io_uring::{opcode, types, IoUring};
+
 use nix::sys::socket;
-use nix::poll;
-use nix::sys::socket::MsgFlags;
-use std::os::fd::FromRawFd;
 use std::os::fd::AsRawFd;
-use std::os::fd::AsFd;
 
 fn main() {
     let tcp_soc = socket::socket(
@@ -27,43 +25,61 @@ fn main() {
     ).unwrap();
     println!("Listening: {:?}", listening);
 
-    // waiting in here until socket becomes readable
+    // ASYNC
-    let pollfd = poll::PollFd::new(tcp_soc.as_fd(), poll::PollFlags::POLLIN);
+    let mut ring = IoUring::new(8).unwrap();
-    let mut fds = [pollfd];
+    let (submitter, mut sq, mut cq) = ring.split();
+    let read_e = opcode::Accept::new(types::Fd(rawfd), std::ptr::null_mut(), std::ptr::null_mut())
+        .build()
+        .user_data(0);
+
+    unsafe {
+        sq.push(&read_e).unwrap();
+    }
+    sq.sync();
+    submitter.submit().unwrap();
+    std::thread::sleep(std::time::Duration::from_millis(100));
     let accepted;
     loop {
-        poll::poll(&mut fds, poll::PollTimeout::NONE).unwrap();
+        cq.sync();
-        let revents = fds[0].revents();  // what happened?
+        if let Some(cqe) = cq.next() {
-
+            accepted = cqe.result();
-        if revents.unwrap().contains(poll::PollFlags::POLLIN) {
-            accepted = socket::accept(rawfd).unwrap();
             println!("Accepted: {:?}", accepted);
             break;
         }
+        std::thread::sleep(std::time::Duration::from_millis(10));
     }
 
-    let client_fd = unsafe { std::os::fd::OwnedFd::from_raw_fd(accepted) };
+    let mut buf = [0u8; 7];
-    let pollfd2 = poll::PollFd::new(client_fd.as_fd(), poll::PollFlags::POLLIN);
-    let mut fds2 = [pollfd2];
-
     loop {
-        poll::poll(&mut fds2, poll::PollTimeout::NONE).unwrap();
+        let recv_e = opcode::Recv::new(types::Fd(accepted), buf.as_mut_ptr(), buf.len() as u32)
-        let mut buf = [0u8; 7];
+            .build()
-        let received = socket::recv(
+            .user_data(1);
-            accepted,
-            &mut buf,
-            MsgFlags::empty()
-        ).unwrap();
-        if received != 0 {
-            // println!("Recv: {}", received);
-            // println!("Bytes: {:?}", buf);
-            // println!("Recv msg: {:?}", std::str::from_utf8(&buf).unwrap_or(""));
-            socket::send(
-                accepted,
-                &buf,
-                MsgFlags::empty()
-            );
-        }
-    }
 
+        unsafe {
+            sq.push(&recv_e).unwrap();
+        }
+
+        sq.sync();
+        submitter.submit().unwrap();
+        loop {
+            cq.sync();
+            if let Some(cqe) = cq.next() {
+                if cqe.user_data() == 1 {
+                    break;
+                }
+            }
+            std::thread::sleep(std::time::Duration::from_millis(10));
+        }
+
+        let send_e = opcode::Send::new(types::Fd(accepted), buf.as_ptr(), buf.len() as u32)
+            .build()
+            .user_data(2);
+
+        unsafe {
+            sq.push(&send_e).unwrap();
+        }
+
+        sq.sync();
+        submitter.submit().unwrap();
+    }
 }