nuclei/net/
released.rs

1//! The buffer of a zero-copy write, handed back once the kernel let go of it.
2
3use core::future::Future;
4use core::marker::PhantomData;
5use core::pin::Pin;
6use core::task::{Context, Poll};
7
8use crate::buf::owned::{IoBuf, OwnedBuf};
9
10/// The buffer of a [`TcpStream::write_zero_copy`](crate::net::TcpStream::write_zero_copy), as a
11/// future that resolves to it once the kernel let go of every byte. For a send the kernel did not
12/// copy, that is once the peer acknowledged the bytes, which may be long after the write ended.
13///
14/// It may be polled from any thread, or dropped: the buffer is then dropped once the kernel let go
15/// of it, never before. A write the kernel copied gives one that is ready at once.
16///
17/// The buffer comes back even after the runtime that sent it shut down: a keeper on the blocking
18/// pool takes over the completions of its worker's ring then, and lets go of the ring once the
19/// kernel let go of every buffer.
20pub struct Released<B = Vec<u8>> {
21    inner: Inner,
22    buf: PhantomData<fn() -> B>,
23}
24
25enum Inner {
26    Ready(Option<OwnedBuf>),
27    #[cfg(ring)]
28    Waiting(crate::sync::oneshot::Receiver<OwnedBuf>),
29}
30
31impl<B> Released<B> {
32    /// One that is ready with `buf`.
33    pub(crate) fn ready(buf: OwnedBuf) -> Self {
34        Self {
35            inner: Inner::Ready(Some(buf)),
36            buf: PhantomData,
37        }
38    }
39
40    /// One whose buffer comes through `back` once its last hold let go.
41    #[cfg(ring)]
42    pub(crate) fn waiting(back: crate::sync::oneshot::Receiver<OwnedBuf>) -> Self {
43        Self {
44            inner: Inner::Waiting(back),
45            buf: PhantomData,
46        }
47    }
48}
49
50impl<B> Future for Released<B>
51where
52    B: IoBuf,
53{
54    type Output = B;
55
56    /// The buffer once the kernel let go of it. Polled again after that, an empty buffer.
57    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<B> {
58        #[cfg(not(ring))]
59        let _ = cx;
60        let back = match &mut self.inner {
61            Inner::Ready(buf) => buf.take().unwrap_or_default(),
62            // The last hold always sends before it goes: an error is only a receiver polled
63            // again after it resolved.
64            #[cfg(ring)]
65            Inner::Waiting(back) => match Pin::new(back).poll(cx) {
66                Poll::Pending => return Poll::Pending,
67                Poll::Ready(buf) => buf.unwrap_or_default(),
68            },
69        };
70        Poll::Ready(back.into_written())
71    }
72}
73
74impl<B> core::fmt::Debug for Released<B> {
75    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
76        let ready = match &self.inner {
77            Inner::Ready(_) => true,
78            #[cfg(ring)]
79            Inner::Waiting(back) => !back.is_empty(),
80        };
81        f.debug_struct("Released").field("ready", &ready).finish()
82    }
83}
nuclei/net/released.rs

nuclei/net/
released.rs
