feat: Support Bind and Listen opcodes (#325)

* feat: Support Bind and Listen opcodes

* Create a separate test for bind/listen

Author: pchaseh

io-uring-test/src/main.rs

@@ -140,6 +140,7 @@ fn test<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
 
     tests::net::test_tcp_shutdown(&mut ring, &test)?;
     tests::net::test_socket(&mut ring, &test)?;
+    tests::net::test_socket_bind_listen(&mut ring, &test)?;
     tests::net::test_udp_recvmsg_multishot(&mut ring, &test)?;
     tests::net::test_udp_recvmsg_multishot_trunc(&mut ring, &test)?;
     tests::net::test_udp_send_with_dest(&mut ring, &test)?;

io-uring-test/src/tests/net.rs

@@ -1590,6 +1590,143 @@ pub fn test_socket<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
     Ok(())
 }
 
+pub fn test_socket_bind_listen<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
+    ring: &mut IoUring<S, C>,
+    test: &Test,
+) -> anyhow::Result<()> {
+    use socket2::{Domain, Protocol, Socket, Type};
+
+    require!(
+        test;
+        test.probe.is_supported(opcode::Socket::CODE);
+        test.probe.is_supported(opcode::Bind::CODE);
+        test.probe.is_supported(opcode::Listen::CODE);
+    );
+
+    println!("test socket_bind_listen");
+
+    // Open a TCP socket through old-style `socket(2)` syscall.
+    // This is used both as a kernel sanity check, and for comparing the returned io-uring FD.
+    let plain_socket = Socket::new(Domain::IPV4, Type::STREAM, Some(Protocol::TCP)).unwrap();
+    let plain_fd = plain_socket.as_raw_fd();
+
+    let socket_fd_op = opcode::Socket::new(
+        Domain::IPV4.into(),
+        Type::STREAM.into(),
+        Protocol::TCP.into(),
+    );
+    unsafe {
+        ring.submission()
+            .push(&socket_fd_op.build().user_data(42).into())
+            .expect("queue is full");
+    }
+    ring.submit_and_wait(1)?;
+
+    let cqes: Vec<cqueue::Entry> = ring.completion().map(Into::into).collect();
+    assert_eq!(cqes.len(), 1);
+    assert_eq!(cqes[0].user_data(), 42);
+    assert!(cqes[0].result() >= 0);
+    let io_uring_socket = unsafe { Socket::from_raw_fd(cqes[0].result()) };
+    assert!(io_uring_socket.as_raw_fd() != plain_fd);
+    assert_eq!(cqes[0].flags(), 0);
+
+    // Try to bind.
+    {
+        let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap();
+        let server_addr: socket2::SockAddr = server_addr.into();
+        let op = io_uring::opcode::Bind::new(
+            io_uring::types::Fd(io_uring_socket.as_raw_fd()),
+            server_addr.as_ptr() as *const _,
+            server_addr.len(),
+        )
+        .build()
+        .user_data(2345);
+        unsafe {
+            ring.submission().push(&op.into()).expect("queue is full");
+        }
+        ring.submit_and_wait(1)?;
+        let cqes: Vec<cqueue::Entry> = ring.completion().map(Into::into).collect();
+        assert_eq!(cqes.len(), 1);
+        assert_eq!(cqes[0].user_data(), 2345);
+        assert_eq!(cqes[0].result(), 0);
+        assert_eq!(cqes[0].flags(), 0);
+
+        assert_eq!(
+            io_uring_socket
+                .local_addr()
+                .expect("no local addr")
+                .as_socket_ipv4()
+                .expect("no IPv4 address")
+                .ip(),
+            server_addr.as_socket_ipv4().unwrap().ip()
+        );
+    }
+
+    // Try to listen.
+    {
+        let op =
+            io_uring::opcode::Listen::new(io_uring::types::Fd(io_uring_socket.as_raw_fd()), 128)
+                .build()
+                .user_data(3456);
+        unsafe {
+            ring.submission().push(&op.into()).expect("queue is full");
+        }
+        ring.submit_and_wait(1)?;
+        let cqes: Vec<cqueue::Entry> = ring.completion().map(Into::into).collect();
+        assert_eq!(cqes.len(), 1);
+        assert_eq!(cqes[0].user_data(), 3456);
+        assert_eq!(cqes[0].result(), 0);
+        assert_eq!(cqes[0].flags(), 0);
+
+        // Ensure the socket is actually in the listening state.
+        _ = TcpStream::connect(
+            io_uring_socket
+                .local_addr()
+                .unwrap()
+                .as_socket_ipv4()
+                .unwrap(),
+        )?;
+    }
+
+    // Close both sockets, to avoid leaking FDs.
+    drop(plain_socket);
+    drop(io_uring_socket);
+
+    // Cleanup all fixed files (if any), then reserve slot 0.
+    let _ = ring.submitter().unregister_files();
+    ring.submitter().register_files_sparse(1).unwrap();
+
+    let fixed_socket_op = opcode::Socket::new(
+        Domain::IPV4.into(),
+        Type::DGRAM.into(),
+        Protocol::UDP.into(),
+    );
+    let dest_slot = types::DestinationSlot::try_from_slot_target(0).unwrap();
+    unsafe {
+        ring.submission()
+            .push(
+                &fixed_socket_op
+                    .file_index(Some(dest_slot))
+                    .build()
+                    .user_data(55)
+                    .into(),
+            )
+            .expect("queue is full");
+    }
+    ring.submit_and_wait(1)?;
+
+    let cqes: Vec<cqueue::Entry> = ring.completion().map(Into::into).collect();
+    assert_eq!(cqes.len(), 1);
+    assert_eq!(cqes[0].user_data(), 55);
+    assert_eq!(cqes[0].result(), 0);
+    assert_eq!(cqes[0].flags(), 0);
+
+    // If the fixed-socket operation worked properly, this must not fail.
+    ring.submitter().unregister_files().unwrap();
+
+    Ok(())
+}
+
 pub fn test_udp_recvmsg_multishot<S: squeue::EntryMarker, C: cqueue::EntryMarker>(
     ring: &mut IoUring<S, C>,
     test: &Test,

src/opcode.rs

@@ -2062,3 +2062,49 @@ opcode! {
         Entry(sqe)
     }
 }
+
+// === 6.11 ===
+
+opcode! {
+    /// Bind a socket, equivalent to `bind(2)`.
+    pub struct Bind {
+        fd: { impl sealed::UseFixed },
+        addr: { *const libc::sockaddr },
+        addrlen: { libc::socklen_t }
+        ;;
+    }
+
+    pub const CODE = sys::IORING_OP_BIND;
+
+    pub fn build(self) -> Entry {
+        let Bind { fd, addr, addrlen } = self;
+
+        let mut sqe = sqe_zeroed();
+        sqe.opcode = Self::CODE;
+        assign_fd!(sqe.fd = fd);
+        sqe.__bindgen_anon_2.addr = addr as _;
+        sqe.__bindgen_anon_1.off = addrlen as _;
+        Entry(sqe)
+    }
+}
+
+opcode! {
+    /// Listen on a socket, equivalent to `listen(2)`.
+    pub struct Listen {
+        fd: { impl sealed::UseFixed },
+        backlog: { i32 },
+        ;;
+    }
+
+    pub const CODE = sys::IORING_OP_LISTEN;
+
+    pub fn build(self) -> Entry {
+        let Listen { fd, backlog } = self;
+
+        let mut sqe = sqe_zeroed();
+        sqe.opcode = Self::CODE;
+        assign_fd!(sqe.fd = fd);
+        sqe.len = backlog as _;
+        Entry(sqe)
+    }
+}