air_sys_syscall/io_uring/
multishot.rs

1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4
5//! Opérations **multishot** (Temps 3d) : une **unique soumission** produit un
6//! **flux** de complétions. Sous-module `air-sys-syscall::io_uring::multishot`.
7//! Repose sur le drapeau de complétion `CQE_F_MORE` (axe E), sur les buffers
8//! fournis du **Temps 3b** (recv/read) et les descripteurs directs du **Temps
9//! 3a** (accept direct). **Aucun register opcode** ; un opcode dédié
10//! (`READ_MULTISHOT`, 49) et des drapeaux d'op.
11//!
12//! Référence normative : `docs/specs/layer-0/io-uring-3d-multishot.md`.
13//!
14//! ## Cycle de vie (slab S1)
15//!
16//! Chaque complétion intermédiaire porte **`CQE_F_MORE`** (« d'autres suivront
17//! pour ce SQE ») ; la **terminale** ne le porte pas (erreur, pénurie de
18//! buffers, ou annulation). Le slot S1 reste **vivant** tant que `CQE_F_MORE`
19//! et n'est **libéré qu'à la terminale** — second cas (après le NOTIF zero-copy
20//! du Temps 2b) où un slot survit à sa première complétion. La **génération** du
21//! jeton filtre les CQE tardifs après annulation (Temps 1 §4.2).
22//!
23//! Une op multishot rend un [`MultishotToken`](super::MultishotToken) (distinct
24//! du `SubmissionToken` mono-coup) ; toutes ses complétions le portent
25//! ([`Completion::multishot_token`](super::Completion::multishot_token)).
26//! [`Completion::has_more`](super::Completion::has_more) distingue les
27//! intermédiaires de la terminale. Le **réarmement** (resoumission) est explicite.
28
29use super::owned::OwnedOp;
30use super::raw::{self, IoUringSqe};
31use super::{CancelTarget, IoUring, MultishotToken, ProvidedBufferRing, SubmitOptions};
32use air_sys_types::Errno;
33use air_sys_types::fd::{AsRawFd, BorrowedFd};
34use air_sys_types::io_uring::{PollEvents, TimeoutFlags, TimeoutSpec};
35use air_sys_types::net::{AcceptFlags, MessageFlags};
36use alloc::boxed::Box;
37use core::time::Duration;
38
39impl IoUring {
40    /// Réserve un slot **multishot** (S1, survit aux `CQE_F_MORE`), prépare le SQE
41    /// (rempli par `fill`), pose `user_data`/`personality`/flags `IOSQE_*`, et rend
42    /// un [`MultishotToken`]. Sans publication (comme `submit_op`). Pas de gestion
43    /// `skip_cqe` : un multishot **consomme chaque CQE** (le slot vit jusqu'à la
44    /// terminale).
45    fn submit_op_multishot(
46        &mut self,
47        payload: Option<OwnedOp>,
48        fill: impl FnOnce(&mut IoUringSqe),
49    ) -> Result<MultishotToken, Errno> {
50        if self.sq.space_left() == 0 {
51            return Err(Errno::EBUSY);
52        }
53        let token = self
54            .slab
55            .reserve_multishot(payload)
56            .map_err(|_| Errno::EBUSY)?;
57        let opts = self.pending_options;
58        self.pending_options = SubmitOptions::default();
59        let personality = self.pending_personality;
60        self.pending_personality = 0;
61        // SAFETY: place SQ vérifiée > 0 juste au-dessus ; le slot (zéro-initialisé
62        // par `prepare`) est rempli intégralement ci-dessous avant toute publication.
63        let sqe_ptr = unsafe { self.sq.prepare() }.expect("place SQ vérifiée");
64        // SAFETY: `sqe_ptr` pointe le slot SQE, exclusivement détenu jusqu'à la
65        // publication ; le `&mut` ne vit que cette portée.
66        unsafe {
67            let sqe = &mut *sqe_ptr;
68            fill(sqe);
69            sqe.user_data = token.to_user_data();
70            sqe.personality = personality;
71            sqe.flags = opts.iosqe_flags();
72        }
73        Ok(MultishotToken::new(token.slot(), token.generation()))
74    }
75
76    // ── Accept multishot (§2) ─────────────────────────────────────────────
77
78    /// `ACCEPT` (13) + `IORING_ACCEPT_MULTISHOT` : **un seul SQE accepte en
79    /// continu** — chaque connexion produit une complétion portant le FD accepté
80    /// ([`Completion::accepted_fd`](super::Completion::accepted_fd)), `CQE_F_MORE`
81    /// maintenu jusqu'à la terminale (ex. listener fermé). `O_CLOEXEC` posé d'office.
82    ///
83    /// # Errors
84    ///
85    /// [`Errno::EBUSY`] si la SQ ou le slab sont pleins.
86    pub fn submit_accept_multishot(
87        &mut self,
88        listener: BorrowedFd<'_>,
89        flags: AcceptFlags,
90    ) -> Result<MultishotToken, Errno> {
91        let fd = listener.as_raw_fd();
92        let accept_flags = (flags | AcceptFlags::CLOEXEC).bits().cast_unsigned();
93        self.submit_op_multishot(None, |sqe| {
94            sqe.opcode = raw::IORING_OP_ACCEPT;
95            sqe.fd = fd;
96            sqe.op_flags = accept_flags;
97            sqe.ioprio |= raw::IORING_ACCEPT_MULTISHOT;
98        })
99    }
100
101    /// Variante **descripteurs directs** : chaque connexion atterrit dans un slot
102    /// **auto-alloué** de la `FixedFdTable` (Temps 3a, `FixedSlotTarget::Alloc`).
103    /// Le slot choisi est rendu dans `cqe->res`
104    /// ([`Completion::allocated_slot`](super::Completion::allocated_slot)). Pour
105    /// les serveurs à fort taux de connexions (pas de FD ordinaire).
106    ///
107    /// Le kernel **rejette** `SOCK_CLOEXEC` sur un descripteur direct : non posé
108    /// (le `O_CLOEXEC` est appliqué à `fixed_fd_install`).
109    ///
110    /// # Errors
111    ///
112    /// Voir [`IoUring::submit_accept_multishot`].
113    pub fn submit_accept_multishot_direct(
114        &mut self,
115        listener: BorrowedFd<'_>,
116        flags: AcceptFlags,
117    ) -> Result<MultishotToken, Errno> {
118        let fd = listener.as_raw_fd();
119        let accept_flags = flags.bits().cast_unsigned();
120        self.submit_op_multishot(None, |sqe| {
121            sqe.opcode = raw::IORING_OP_ACCEPT;
122            sqe.fd = fd;
123            sqe.op_flags = accept_flags;
124            sqe.ioprio |= raw::IORING_ACCEPT_MULTISHOT;
125            // Auto-allocation dans la table de FD fixes (chaque connexion → slot).
126            sqe.splice_fd_in_or_file_index = raw::IORING_FILE_INDEX_ALLOC;
127        })
128    }
129
130    // ── Recv / Read multishot (§3, buffers fournis 3b) ────────────────────
131
132    /// `RECV` (27) + `IORING_RECV_MULTISHOT` sur les buffers de `group`
133    /// (`IOSQE_BUFFER_SELECT`) : chaque arrivée de données prend un buffer du
134    /// groupe et produit une complétion. Complétion via
135    /// [`Completion::into_provided_buffer`](super::Completion::into_provided_buffer).
136    ///
137    /// **Pénurie de buffers** : si le groupe est vide à l'arrivée de données, la
138    /// complétion porte **`-ENOBUFS` et TERMINE** le multishot (pas de `F_MORE`) —
139    /// à distinguer d'une erreur réseau. L'application réapprovisionne puis
140    /// **resoumet**.
141    ///
142    /// # Errors
143    ///
144    /// [`Errno::EBUSY`] si la SQ/le slab sont pleins.
145    pub fn submit_receive_multishot(
146        &mut self,
147        sock: BorrowedFd<'_>,
148        group: &ProvidedBufferRing,
149        flags: MessageFlags,
150    ) -> Result<MultishotToken, Errno> {
151        let fd = sock.as_raw_fd();
152        let bgid = group.group_id();
153        let msg_flags = flags.bits().cast_unsigned();
154        self.pending_options = self.pending_options.buffer_select();
155        self.submit_op_multishot(None, |sqe| {
156            sqe.opcode = raw::IORING_OP_RECV;
157            sqe.fd = fd;
158            sqe.op_flags = msg_flags;
159            sqe.buf_index_or_group = bgid;
160            sqe.ioprio |= raw::IORING_RECV_MULTISHOT;
161        })
162    }
163
164    /// `READ_MULTISHOT` (49) sur les buffers de `group` : flux de lectures, chaque
165    /// arrivée prenant un buffer du groupe. Même sémantique de pénurie `-ENOBUFS`
166    /// terminante que [`IoUring::submit_receive_multishot`]. L'opcode dédié implique
167    /// le multishot (pas de drapeau d'op séparé).
168    ///
169    /// # Errors
170    ///
171    /// Voir [`IoUring::submit_receive_multishot`].
172    pub fn submit_read_multishot(
173        &mut self,
174        fd: BorrowedFd<'_>,
175        group: &ProvidedBufferRing,
176        offset: Option<u64>,
177    ) -> Result<MultishotToken, Errno> {
178        let fd = fd.as_raw_fd();
179        let bgid = group.group_id();
180        let off = offset.unwrap_or(u64::MAX);
181        self.pending_options = self.pending_options.buffer_select();
182        self.submit_op_multishot(None, |sqe| {
183            sqe.opcode = raw::IORING_OP_READ_MULTISHOT;
184            sqe.fd = fd;
185            sqe.off_or_addr2 = off;
186            sqe.buf_index_or_group = bgid;
187        })
188    }
189
190    // ── Poll multishot (§4) ───────────────────────────────────────────────
191
192    /// `POLL_ADD` (6) + `IORING_POLL_ADD_MULTI` : chaque transition d'état du FD
193    /// vers les `events` surveillés produit une complétion
194    /// ([`Completion::into_poll_result`](super::Completion::into_poll_result)),
195    /// `F_MORE` maintenu. **Edge-triggered** (la surface validée ne porte pas
196    /// d'option level-triggered).
197    ///
198    /// # Errors
199    ///
200    /// [`Errno::EBUSY`] si la SQ/le slab sont pleins.
201    pub fn submit_poll_multishot(
202        &mut self,
203        fd: BorrowedFd<'_>,
204        events: PollEvents,
205    ) -> Result<MultishotToken, Errno> {
206        let fd = fd.as_raw_fd();
207        let poll_events = events.bits();
208        self.submit_op_multishot(None, |sqe| {
209            sqe.opcode = raw::IORING_OP_POLL_ADD;
210            sqe.fd = fd;
211            sqe.op_flags = poll_events; // poll32_events
212            sqe.len = raw::IORING_POLL_ADD_MULTI; // drapeaux poll (multishot)
213        })
214    }
215
216    // ── Timeout multishot (§5) ────────────────────────────────────────────
217
218    /// `TIMEOUT` (11) + `IORING_TIMEOUT_MULTISHOT` : émet une complétion à
219    /// **intervalle régulier** (timer répétitif) jusqu'à annulation — battement
220    /// périodique dans le reactor sans resoumettre à chaque tick. Le
221    /// `__kernel_timespec` est gardé en vie dans le slot (vivant jusqu'à la
222    /// terminale).
223    ///
224    /// # Errors
225    ///
226    /// [`Errno::EBUSY`] si la SQ/le slab sont pleins.
227    pub fn submit_timeout_multishot(
228        &mut self,
229        interval: Duration,
230        flags: TimeoutFlags,
231    ) -> Result<MultishotToken, Errno> {
232        let ts = Box::new(super::async_ops::timespec_of(TimeoutSpec::after(interval)));
233        let ts_ptr = core::ptr::from_ref(&*ts) as u64;
234        let op_flags = flags.bits() | raw::IORING_TIMEOUT_MULTISHOT;
235        self.submit_op_multishot(Some(OwnedOp::Timeout(ts)), |sqe| {
236            sqe.opcode = raw::IORING_OP_TIMEOUT;
237            sqe.fd = -1;
238            sqe.addr_or_splice_off_in = ts_ptr;
239            sqe.len = 1;
240            sqe.op_flags = op_flags;
241        })
242    }
243
244    // ── Annulation (§6) ───────────────────────────────────────────────────
245
246    /// Annule un multishot en vol, en ciblant son `token`. La complétion terminale
247    /// (sans `F_MORE`, souvent `-ECANCELED`) **libère le slot** ; les CQE tardifs
248    /// éventuels sont filtrés par la **génération** (S1).
249    ///
250    /// Utilise l'annulation **synchrone** par jeton (`IORING_REGISTER_SYNC_CANCEL`,
251    /// `CancelTarget::Token`) — cohérente avec un retour `Result<()>` (pas de CQE
252    /// d'annulation supplémentaire à drainer ; la terminale du multishot suffit).
253    /// `submit_cancel` (Temps 2c) reste disponible pour les annulations groupées.
254    ///
255    /// # Errors
256    ///
257    /// [`Errno::ENOENT`] si le multishot s'est déjà terminé (rien à annuler) ;
258    /// autres erreurs de `io_uring_register`.
259    pub fn cancel_multishot(&mut self, token: MultishotToken) -> Result<(), Errno> {
260        self.sync_cancel(CancelTarget::Token(token.as_submission()))
261            .map(|_count| ())
262    }
263}
264
265// ───────────────────────────────────────────────────────────────────────────
266// Tests
267// ───────────────────────────────────────────────────────────────────────────
268//
269// Intégration kernel (multishot non modélisé par Miri) → `#[cfg_attr(miri, ignore)]`.
270// Le cycle de vie S1 multishot (slot vivant tant que F_MORE, libéré à la
271// terminale, génération anti-CQE-tardif) est prouvé **sous Miri** par les tests
272// purs du slab (`slab::proptests::multishot_slot_survives_until_final_…`).
273
274#[cfg(test)]
275mod tests {
276    use super::*;
277    use crate::io_uring::{
278        Completion, FixedFdTable, IoUring, ProvidedBufferRing, ProvidedBufferRingOptions,
279    };
280    use crate::test_support::{sfd, std_file};
281    use air_sys_types::fd::AsFd;
282    use air_sys_types::io_uring::PollEvents;
283    use air_sys_types::ipc::PipeFlags;
284    use core::num::{NonZeroU16, NonZeroU32};
285    use std::io::Write;
286
287    fn ring8() -> IoUring {
288        IoUring::new(NonZeroU32::new(8).expect("8 ≠ 0")).expect("io_uring_setup 6.12")
289    }
290
291    fn drain_n(ring: &mut IoUring, n: u32) -> Vec<Completion> {
292        ring.submit_and_wait(n).expect("submit_and_wait");
293        let mut out = Vec::new();
294        while out.len() < usize::try_from(n).expect("n ⊆ usize") {
295            out.push(ring.wait_completion().expect("wait_completion"));
296        }
297        out
298    }
299
300    // ── Accept multishot ──────────────────────────────────────────────────
301
302    #[test]
303    #[cfg_attr(miri, ignore = "io_uring multishot non supporté par Miri")]
304    fn accept_multishot_streams_connections_then_cancel() {
305        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
306        let addr = listener.local_addr().expect("addr");
307        // Deux connexions en attente dans le backlog.
308        let _c1 = std::net::TcpStream::connect(addr).expect("connect 1");
309        let _c2 = std::net::TcpStream::connect(addr).expect("connect 2");
310        std::thread::sleep(std::time::Duration::from_millis(20));
311        let mut ring = ring8();
312        let token = ring
313            .submit_accept_multishot(sfd(&listener), AcceptFlags::empty())
314            .expect("accept_multishot");
315        // Deux complétions intermédiaires (F_MORE) portant chacune un FD accepté.
316        let completions = drain_n(&mut ring, 2);
317        for c in completions {
318            assert_eq!(c.multishot_token(), Some(token), "même jeton multishot");
319            assert!(c.has_more(), "F_MORE : flux maintenu");
320            let _fd = c.accepted_fd().expect("FD accepté valide");
321        }
322        assert_eq!(ring.in_flight(), 1, "slot multishot vivant tant que F_MORE");
323        // Annulation → complétion terminale (-ECANCELED), libère le slot.
324        ring.cancel_multishot(token).expect("cancel_multishot");
325        let terminal = ring.wait_completion().expect("terminale");
326        assert_eq!(terminal.multishot_token(), Some(token));
327        assert!(!terminal.has_more(), "terminale : pas de F_MORE");
328        assert_eq!(terminal.into_result(), Err(Errno::ECANCELED));
329        assert_eq!(ring.in_flight(), 0, "slot libéré à la terminale");
330    }
331
332    #[test]
333    #[cfg_attr(miri, ignore = "io_uring multishot non supporté par Miri")]
334    fn accept_multishot_direct_lands_in_fixed_slots() {
335        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
336        let addr = listener.local_addr().expect("addr");
337        let _c1 = std::net::TcpStream::connect(addr).expect("connect");
338        std::thread::sleep(std::time::Duration::from_millis(20));
339        let mut ring = ring8();
340        let table = FixedFdTable::register(&mut ring, NonZeroU32::new(4).expect("4"))
341            .expect("register table");
342        let token = ring
343            .submit_accept_multishot_direct(sfd(&listener), AcceptFlags::empty())
344            .expect("accept_multishot_direct");
345        let completions = drain_n(&mut ring, 1);
346        let c = completions.into_iter().next().expect("une complétion");
347        assert_eq!(c.multishot_token(), Some(token));
348        assert!(c.has_more());
349        let slot = c.allocated_slot().expect("slot auto-alloué");
350        assert!(slot < 4, "connexion dans un slot fixe");
351        ring.cancel_multishot(token).expect("cancel");
352        let terminal = ring.wait_completion().expect("terminale");
353        assert!(!terminal.has_more());
354        table.unregister(&mut ring).expect("unregister");
355    }
356
357    // ── Recv multishot (buffers fournis) + pénurie -ENOBUFS ───────────────
358
359    #[test]
360    #[cfg_attr(miri, ignore = "io_uring multishot non supporté par Miri")]
361    fn recv_multishot_streams_buffers() {
362        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
363        let addr = listener.local_addr().expect("addr");
364        let mut client = std::net::TcpStream::connect(addr).expect("connect");
365        let (server, _) = listener.accept().expect("accept");
366        client.write_all(b"hello").expect("write");
367        client.flush().expect("flush");
368        std::thread::sleep(std::time::Duration::from_millis(20));
369        let mut ring = ring8();
370        let mut group = ProvidedBufferRing::register(
371            &mut ring,
372            1,
373            NonZeroU16::new(4).expect("4"),
374            NonZeroU32::new(64).expect("64"),
375            ProvidedBufferRingOptions::default(),
376        )
377        .expect("register group");
378        let token = ring
379            .submit_receive_multishot(sfd(&server), &group, MessageFlags::empty())
380            .expect("recv_multishot");
381        let completions = drain_n(&mut ring, 1);
382        let c = completions.into_iter().next().expect("une complétion");
383        assert_eq!(c.multishot_token(), Some(token));
384        assert!(c.has_more(), "flux maintenu (groupe non vide)");
385        {
386            let buf = c.into_provided_buffer(&mut group).expect("buffer choisi");
387            assert_eq!(buf.data(), b"hello");
388        } // drop → réappro
389        ring.cancel_multishot(token).expect("cancel");
390        let terminal = ring.wait_completion().expect("terminale");
391        assert!(!terminal.has_more());
392        group.unregister(&mut ring).expect("unregister");
393    }
394
395    #[test]
396    #[cfg_attr(miri, ignore = "io_uring multishot non supporté par Miri")]
397    fn recv_multishot_enobufs_terminates_then_resubmit() {
398        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
399        let addr = listener.local_addr().expect("addr");
400        let mut client = std::net::TcpStream::connect(addr).expect("connect");
401        let (server, _) = listener.accept().expect("accept");
402        // 8 octets pour 1 buffer de 4 ⇒ le 2e segment manque de buffer ⇒ -ENOBUFS.
403        client.write_all(b"AAAABBBB").expect("write 8");
404        client.flush().expect("flush");
405        std::thread::sleep(std::time::Duration::from_millis(20));
406        let mut ring = ring8();
407        let mut group = ProvidedBufferRing::register(
408            &mut ring,
409            2,
410            NonZeroU16::new(1).expect("1"),
411            NonZeroU32::new(4).expect("4"),
412            ProvidedBufferRingOptions::default(),
413        )
414        .expect("register count=1");
415        let token = ring
416            .submit_receive_multishot(sfd(&server), &group, MessageFlags::empty())
417            .expect("recv_multishot");
418        ring.submit_and_wait(1).expect("submit");
419        // 1ère complétion : buffer rempli (F_MORE) ; on la traite et on rend le buffer.
420        let first = ring.wait_completion().expect("1ère");
421        assert_eq!(first.multishot_token(), Some(token));
422        assert!(first.has_more());
423        {
424            let buf = first.into_provided_buffer(&mut group).expect("buffer");
425            assert_eq!(buf.data(), b"AAAA");
426        } // drop → réappro (buffer 0 rendu)
427        // 2ᵉ complétion : pénurie ⇒ -ENOBUFS **terminant** le multishot (pas de F_MORE).
428        let second = ring.wait_completion().expect("2ᵉ");
429        assert_eq!(second.multishot_token(), Some(token));
430        assert!(!second.has_more(), "ENOBUFS termine le multishot");
431        assert!(
432            second.into_provided_buffer(&mut group).is_none(),
433            "pas de buffer"
434        );
435        assert_eq!(ring.in_flight(), 0, "slot libéré à la terminale ENOBUFS");
436        // Réapprovisionné : resoumission récupère les octets restants.
437        let token2 = ring
438            .submit_receive_multishot(sfd(&server), &group, MessageFlags::empty())
439            .expect("re-recv_multishot");
440        let again = drain_n(&mut ring, 1)
441            .into_iter()
442            .next()
443            .expect("complétion");
444        assert_eq!(again.multishot_token(), Some(token2));
445        let buf = again.into_provided_buffer(&mut group).expect("buffer 2");
446        assert_eq!(buf.data(), b"BBBB");
447        drop(buf);
448        ring.cancel_multishot(token2).expect("cancel");
449        let _ = ring.wait_completion().expect("terminale");
450        group.unregister(&mut ring).expect("unregister");
451    }
452
453    // ── Poll multishot ────────────────────────────────────────────────────
454
455    #[test]
456    #[cfg_attr(miri, ignore = "io_uring multishot non supporté par Miri")]
457    fn poll_multishot_signals_then_cancel() {
458        let (reader, writer) = crate::ipc::pipe2(PipeFlags::empty()).expect("pipe2");
459        let mut w = std_file(writer);
460        w.write_all(b"x").expect("write pipe");
461        w.flush().expect("flush");
462        let mut ring = ring8();
463        let token = ring
464            .submit_poll_multishot(reader.as_fd(), PollEvents::IN)
465            .expect("poll_multishot");
466        let c = drain_n(&mut ring, 1)
467            .into_iter()
468            .next()
469            .expect("complétion");
470        assert_eq!(c.multishot_token(), Some(token));
471        assert!(c.has_more(), "poll multishot maintient F_MORE");
472        let events = c.into_poll_result().expect("événements");
473        assert!(events.contains(PollEvents::IN), "POLLIN signalé");
474        ring.cancel_multishot(token).expect("cancel");
475        let terminal = ring.wait_completion().expect("terminale");
476        assert!(!terminal.has_more());
477        assert_eq!(ring.in_flight(), 0);
478    }
479
480    // ── Timeout multishot ─────────────────────────────────────────────────
481
482    #[test]
483    #[cfg_attr(miri, ignore = "io_uring multishot non supporté par Miri")]
484    fn timeout_multishot_ticks_then_cancel() {
485        use air_sys_types::io_uring::TimeoutFlags;
486        let mut ring = ring8();
487        let token = ring
488            .submit_timeout_multishot(std::time::Duration::from_millis(20), TimeoutFlags::empty())
489            .expect("timeout_multishot");
490        // Deux ticks (chacun F_MORE).
491        for _ in 0..2 {
492            let tick = drain_n(&mut ring, 1).into_iter().next().expect("tick");
493            assert_eq!(tick.multishot_token(), Some(token));
494            assert!(tick.has_more(), "tick maintient F_MORE");
495        }
496        assert_eq!(ring.in_flight(), 1, "slot vivant entre les ticks");
497        ring.cancel_multishot(token).expect("cancel");
498        let terminal = ring.wait_completion().expect("terminale");
499        assert!(!terminal.has_more());
500        assert_eq!(ring.in_flight(), 0);
501    }
502
503    // ── Mono-coup : pas de jeton multishot ────────────────────────────────
504
505    #[test]
506    #[cfg_attr(miri, ignore = "io_uring non supporté par Miri")]
507    fn mono_shot_completion_has_no_multishot_token() {
508        let mut ring = ring8();
509        let _tok = ring.submit_nop().expect("nop");
510        let c = drain_n(&mut ring, 1)
511            .into_iter()
512            .next()
513            .expect("complétion");
514        assert_eq!(
515            c.multishot_token(),
516            None,
517            "op mono-coup : pas de jeton multishot"
518        );
519        assert!(!c.has_more());
520    }
521
522    // ── Read multishot (buffers fournis, sur pipe) ────────────────────────
523
524    #[test]
525    #[cfg_attr(miri, ignore = "io_uring multishot non supporté par Miri")]
526    fn read_multishot_streams_buffers_from_pipe() {
527        let (reader, writer) = crate::ipc::pipe2(PipeFlags::empty()).expect("pipe2");
528        // Le writer reste **ouvert** (pas d'EOF) ⇒ le read multishot reste armé
529        // après avoir consommé les données disponibles.
530        let mut w = std_file(writer);
531        w.write_all(b"piped").expect("write pipe");
532        w.flush().expect("flush");
533        let mut ring = ring8();
534        let mut group = ProvidedBufferRing::register(
535            &mut ring,
536            6,
537            NonZeroU16::new(4).expect("4"),
538            NonZeroU32::new(64).expect("64"),
539            ProvidedBufferRingOptions::default(),
540        )
541        .expect("register group");
542        let token = ring
543            .submit_read_multishot(reader.as_fd(), &group, None)
544            .expect("read_multishot");
545        let c = drain_n(&mut ring, 1)
546            .into_iter()
547            .next()
548            .expect("complétion");
549        assert_eq!(c.multishot_token(), Some(token));
550        assert!(c.has_more(), "armé : writer encore ouvert");
551        let buf = c.into_provided_buffer(&mut group).expect("buffer choisi");
552        assert_eq!(buf.data(), b"piped");
553        drop(buf);
554        ring.cancel_multishot(token).expect("cancel");
555        let terminal = ring.wait_completion().expect("terminale");
556        assert!(!terminal.has_more());
557        group.unregister(&mut ring).expect("unregister");
558        drop(w);
559    }
560
561    // ── Back-pressure : SQ pleine ⇒ EBUSY (aucun slot réservé) ────────────
562
563    #[test]
564    #[cfg_attr(miri, ignore = "io_uring multishot non supporté par Miri")]
565    fn multishot_on_full_sq_returns_ebusy() {
566        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
567        let mut ring = ring8();
568        // Remplit la SQ (8 entrées) sans publier.
569        for _ in 0..8 {
570            ring.submit_nop().expect("nop");
571        }
572        // 9e soumission multishot : SQ pleine ⇒ EBUSY **avant** réservation de slot.
573        assert_eq!(
574            ring.submit_accept_multishot(sfd(&listener), AcceptFlags::empty()),
575            Err(Errno::EBUSY)
576        );
577        // Les 8 nops se vident normalement.
578        let _ = drain_n(&mut ring, 8);
579        assert_eq!(ring.in_flight(), 0);
580    }
581}
air_sys_syscall/io_uring/multishot.rs

air_sys_syscall/io_uring/
multishot.rs
