feat: recv/recvmsg multishot support for io_uring driver

If you have time for review and interested in the changes I can also split this PR into 4 separate ones, i kept it this way to better demonstrate the scope of changes. Otherwise would appreciate any comments on the implementation. - [io_uring multishot recv](https://man7.org/linux/man-pages/man3/io_uring_prep_recv_multishot.3.html) - [io_uring multishot recvmsg](https://man7.org/linux/man-pages/man3/io_uring_prep_recvmsg_multishot.3.html) - [io_uring register buf_ring](https://man7.org/linux/man-pages/man3/io_uring_register_buf_ring.3.html) ## 1. feat: add multishot lifecycle support Extends the lifecycle state machine to handle multishot operations. Unlike single-shot operations that produce one completion per submission, multishot operations yield multiple completions until explicitly terminated via the IORING_CQE_F_MORE flag. Example: single-shot recv ```rust let (result, buf) = socket.recv(buf).await; // ├─ submit() → Lifecycle::Submitted, returns Op<Recv> // ├─ .await → poll() sees Submitted, stores waker → Lifecycle::Waiting(waker) // │ returns Poll::Pending, task suspends // │ // │ kernel receives packet, posts CQE // │ // ├─ complete()→ Lifecycle::Completed(result, flags), waker.wake() // ├─ .await → poll() sees Completed, removes from slab, returns Poll::Ready // └─ done ``` Example: multishot recv ```rust let ring = UserRingBuf::new(4, 1500, 0)?; // 4 buffers, 1500 bytes each let mut stream = socket.recv_multishot(ring)?; // Lifecycle::Multishot { queue: [], terminated: false } // ↳ submits RecvMulti SQE to io_uring sender.send_to(b"hello", addr).await; // kernel receives, posts CQE with MORE=1 // ↳ complete(): queue.push(cqe), waker.wake() let buf = stream.next().await; // poll_multishot(): queue.pop() → Ready(cqe) assert_eq!(&*buf, b"hello"); // ↳ buf holds buffer #0 drop(buf); // ↳ buffer #0 returned to ring sender.send_to(b"world", addr).await; // kernel posts CQE with MORE=1 let buf = stream.next().await; // poll_multishot() → Ready(cqe) assert_eq!(&*buf, b"world"); drop(buf); drop(stream); // user drops while kernel still working // ↳ drop_op(): Lifecycle → Ignored(boxed_data) // ↳ slab entry kept alive // once kernel finishes and posts cqe // ↳ complete(): if MORE=0, self.remove() // ↳ slab entry finally removed ``` ## 2. feat: add ringbuf implementation Implements io_uring provided [buffer rings](https://man7.org/linux/man-pages/man3/io_uring_register_buf_ring.3.html). The kernel picks buffers directly from a shared ring. Buffer State Example: Ring entries (slots) and buffer memory are decoupled. When returning a buffer, we write its address to slot[tail & 3] (next available slot), not its "original" slot. ```rust let ring = UserRingBuf::new(4, 1500, 0)?; let mut stream = socket.recv_multishot(ring)?; // INITIAL tail=4 // ring slots: [slot0→buf0, slot1→buf1, slot2→buf2, slot3→buf3] // user holds: (none) sender.send_to(b"aaa", addr).await; let a = stream.next().await?; // kernel picks slot0→buf0, gives to user // STEP 1 tail=4 // ring slots: [slot0:empty, slot1→buf1, slot2→buf2, slot3→buf3] // user holds: a=buf0 sender.send_to(b"bbb", addr).await; let b = stream.next().await?; // kernel picks slot1→buf1, gives to user // STEP 2 tail=4 // ring slots: [slot0:empty, slot1:empty, slot2→buf2, slot3→buf3] // user holds: a=buf0, b=buf1 sender.send_to(b"ccc", addr).await; let c = stream.next().await?; // kernel picks slot2→buf2 drop(c); // return_buffer(buf2): // slot[4 & 3] = slot0 → buf2, tail=5 // STEP 3 tail=5 // ring slots: [slot0→buf2, slot1:empty, slot2:empty, slot3→buf3] // ↑ buf2 now in slot0 (was slot2) // user holds: a=buf0, b=buf1 sender.send_to(b"ddd", addr).await; let d = stream.next().await?; // kernel picks slot3→buf3 drop(d); // return_buffer(buf3): // slot[5 & 3] = slot1 → buf3, tail=6 // STEP 4 tail=6 // ring slots: [slot0→buf2, slot1→buf3, slot2:empty, slot3:empty] // ↑ buf3 now in slot1 (was slot3) // user holds: a=buf0, b=buf1 // // kernel cycles buf2↔buf3 via slots 0,1 // buf0,buf1 memory untouched - not in any slot drop(b); // slot[6 & 3] = slot2 → buf1, tail=7 drop(a); // slot[7 & 3] = slot3 → buf0, tail=8 // FINAL tail=8 // ring slots: [slot0→buf2, slot1→buf3, slot2→buf1, slot3→buf0] // user holds: (none) ``` For UserRecvMsgRingBuf, each buffer holds header + address + payload. Parser type parameter determines sockaddr size at compile time. ```bash ┌────────────────┬─────────────────────┬──────────────────────┐ │ RecvMsgOut Hdr │ sockaddr_in[6] │ Payload │ │ (16 bytes) │ (16-28 bytes) │ (variable) │ ├────────────────┴─────────────────────┼──────────────────────┤ │ parsed by parse_recvmsg() │ returned as RawBuffer│ └──────────────────────────────────────┴──────────────────────┘ ``` ## 3. feat: add multishot op Generic `MultishotOp<T>` wrapper for io_uring multishot operations. io_uring multishot operations: - recv multishot - with provided buffers (implemented) - recvmsg multishot - with provided buffers (implemented) - accept multishot - IORING_ACCEPT_MULTISHOT - poll multishot - IORING_POLL_ADD_MULTI - timeout multishot - IORING_TIMEOUT_MULTISHOT - read multishot - IORING_OP_READ_MULTISHOT To add new multishot ops: implement OpAble with uring_op() returning the multishot SQE, wrap in `MultishotOp<T>`. ## 4. feat: add recv/recvmsg multishot with tests Uses [recv_multishot](https://man7.org/linux/man-pages/man3/io_uring_prep_recv_multishot.3.html) and [recvmsg_multishot](https://man7.org/linux/man-pages/man3/io_uring_prep_recvmsg_multishot.3.html). ```rust pub struct RecvMultishot<R: RingBuf> { ring: R, // owns the buffer ring op: MultishotOp<RecvMultishotOp>, // owns the io_uring operation cancellation_guard: Option<AssociateGuard>, } pub struct RecvMultishotStream<'a, R: RingBuf> { ring: &'a R, // borrows ring op: &'a mut MultishotOp<RecvMultishotOp>, // borrows op mutably cancellation_guard: &'a Option<AssociateGuard>, } ``` This separation enforces a single mutable poller (stream holds `&mut op`), while allowing multiple buffers to be held simultaneously without borrow conflicts (buffers only need `&ring`). ```rust let mut multishot = socket.recv_multishot(ring)?; let mut stream = multishot.stream(); let buf1 = stream.next().await?; // Buffer<'a> let buf2 = stream.next().await?; // Buffer<'a>, buf1 still valid let buf3 = stream.next().await?; // all three coexist process(&buf1, &buf2, &buf3); drop(buf3); drop(buf2); drop(buf1); // returned to ring in any order ``` Ring is reclaimable only after the multishot operation is fully terminated: 1. Kernel posts final CQE with IORING_CQE_F_MORE=0 (no more completions coming) 2. User drains all queued completions until stream.next() returns None Termination happens naturally on error (e.g. ENOBUFS) or can be triggered via cancellation: ```rust canceller.cancel(); // sends AsyncCancel SQE to io_uring // kernel posts final CQE with MORE=0 while stream.next().await.is_some() {} // drain queue until None // None means: terminated AND queue empty drop(stream); // release borrow let ring = multishot.try_into_ring()?; // Ok only if terminated, Err(self) otherwise ring.unregister()?; // safe now: no in-flight buffers ```