compio_driver/sys/op/managed/
fallback.rs

1use std::io;
2
3use compio_buf::{IntoInner, IoBuf, IoBufMut, SetLenExt};
4use rustix::net::{RecvFlags, ReturnFlags};
5use socket2::SockAddr;
6
7use crate::{
8    AsFd, BufferPool, BufferRef, PollFirst,
9    op::{RecvMsg, TakeBuffer},
10    sys::op::{Read, ReadAt, Recv, RecvFrom},
11};
12
13/// Read a file at specified position into managed buffer.
14pub struct ReadManagedAt<S> {
15    pub(crate) op: ReadAt<BufferRef, S>,
16}
17
18impl<S> ReadManagedAt<S> {
19    /// Create [`ReadManagedAt`].
20    pub fn new(fd: S, offset: u64, pool: &BufferPool, len: usize) -> io::Result<Self> {
21        Ok(Self {
22            op: ReadAt::new(fd, offset, pool.pop()?.with_capacity(len)),
23        })
24    }
25}
26
27impl<S> TakeBuffer for ReadManagedAt<S> {
28    type Buffer = BufferRef;
29
30    fn take_buffer(self) -> Option<BufferRef> {
31        Some(self.op.into_inner())
32    }
33}
34
35/// Read a file into managed buffer.
36pub struct ReadManaged<S> {
37    pub(crate) op: Read<BufferRef, S>,
38}
39
40impl<S> ReadManaged<S> {
41    /// Create [`ReadManaged`].
42    pub fn new(fd: S, pool: &BufferPool, len: usize) -> io::Result<Self> {
43        Ok(Self {
44            op: Read::new(fd, pool.pop()?.with_capacity(len)),
45        })
46    }
47}
48
49impl<S> TakeBuffer for ReadManaged<S> {
50    type Buffer = BufferRef;
51
52    fn take_buffer(self) -> Option<BufferRef> {
53        Some(self.op.into_inner())
54    }
55}
56
57/// Receive data from remote into managed buffer.
58///
59/// It is only used for socket operations. If you want to read from a pipe,
60/// use [`ReadManaged`].
61pub struct RecvManaged<S> {
62    pub(crate) op: Recv<BufferRef, S>,
63}
64
65impl<S> RecvManaged<S> {
66    /// Create [`RecvManaged`].
67    pub fn new(fd: S, pool: &BufferPool, len: usize, flags: RecvFlags) -> io::Result<Self> {
68        Ok(Self {
69            op: Recv::new(fd, pool.pop()?.with_capacity(len), flags),
70        })
71    }
72}
73
74impl<S> PollFirst for RecvManaged<S> {
75    fn poll_first(&mut self) {
76        self.op.poll_first();
77    }
78}
79
80impl<S> TakeBuffer for RecvManaged<S> {
81    type Buffer = BufferRef;
82
83    fn take_buffer(self) -> Option<BufferRef> {
84        Some(self.op.into_inner())
85    }
86}
87
88/// Receive data and source address into managed buffer.
89pub struct RecvFromManaged<S: AsFd> {
90    pub(crate) op: RecvFrom<BufferRef, S>,
91}
92
93impl<S: AsFd> RecvFromManaged<S> {
94    /// Create [`RecvFromManaged`].
95    pub fn new(fd: S, pool: &BufferPool, len: usize, flags: RecvFlags) -> io::Result<Self> {
96        Ok(Self {
97            op: RecvFrom::new(fd, pool.pop()?.with_capacity(len), flags),
98        })
99    }
100}
101
102impl<S: AsFd> PollFirst for RecvFromManaged<S> {
103    fn poll_first(&mut self) {
104        self.op.poll_first();
105    }
106}
107
108impl<S: AsFd> TakeBuffer for RecvFromManaged<S> {
109    type Buffer = (BufferRef, Option<SockAddr>);
110
111    fn take_buffer(self) -> Option<Self::Buffer> {
112        Some(self.op.into_inner())
113    }
114}
115
116/// Receive data into managed buffer, and ancillary data into control buffer.
117pub struct RecvMsgManaged<C: IoBufMut, S: AsFd> {
118    pub(crate) op: RecvMsg<[BufferRef; 1], C, S>,
119}
120
121impl<C: IoBufMut, S: AsFd> RecvMsgManaged<C, S> {
122    /// Create [`RecvMsgManaged`].
123    pub fn new(
124        fd: S,
125        pool: &BufferPool,
126        len: usize,
127        control: C,
128        flags: RecvFlags,
129    ) -> io::Result<Self> {
130        Ok(Self {
131            op: RecvMsg::new(fd, [pool.pop()?.with_capacity(len)], control, flags),
132        })
133    }
134}
135
136impl<C: IoBufMut, S: AsFd> PollFirst for RecvMsgManaged<C, S> {
137    fn poll_first(&mut self) {
138        self.op.poll_first();
139    }
140}
141
142impl<C: IoBufMut, S: AsFd> TakeBuffer for RecvMsgManaged<C, S> {
143    type Buffer = ((BufferRef, C), Option<SockAddr>, usize, ReturnFlags);
144
145    fn take_buffer(self) -> Option<Self::Buffer> {
146        let (([buf], control), addr, len, flags) = self.op.into_inner();
147        Some(((buf, control), addr, len, flags))
148    }
149}
150
151/// Read a file at specified position into multiple managed buffers.
152pub type ReadMultiAt<S> = ReadManagedAt<S>;
153/// Read a file into multiple managed buffers.
154pub type ReadMulti<S> = ReadManaged<S>;
155/// Receive data from remote into multiple managed buffers.
156pub type RecvMulti<S> = RecvManaged<S>;
157
158/// Result of [`RecvFromMulti`].
159pub struct RecvFromMultiResult {
160    buffer: BufferRef,
161    addr: Option<SockAddr>,
162}
163
164impl RecvFromMultiResult {
165    #[doc(hidden)]
166    pub unsafe fn new(_: BufferRef) -> Self {
167        unreachable!("should not be called directly")
168    }
169
170    /// Get the payload data.
171    pub fn data(&self) -> &[u8] {
172        self.buffer.as_init()
173    }
174
175    /// Get the source address if applicable.
176    pub fn addr(&self) -> Option<SockAddr> {
177        self.addr.clone()
178    }
179}
180
181impl IntoInner for RecvFromMultiResult {
182    type Inner = BufferRef;
183
184    fn into_inner(self) -> Self::Inner {
185        self.buffer
186    }
187}
188
189/// Receive data and source address multi times into multiple managed buffers.
190pub struct RecvFromMulti<S: AsFd> {
191    pub(crate) op: RecvFromManaged<S>,
192    pub(crate) len: usize,
193}
194
195impl<S: AsFd> RecvFromMulti<S> {
196    /// Create [`RecvFromMulti`].
197    pub fn new(fd: S, pool: &BufferPool, flags: RecvFlags) -> io::Result<Self> {
198        Ok(Self {
199            op: RecvFromManaged::new(fd, pool, 0, flags)?,
200            len: 0,
201        })
202    }
203}
204
205impl<S: AsFd> TakeBuffer for RecvFromMulti<S> {
206    type Buffer = RecvFromMultiResult;
207
208    fn take_buffer(self) -> Option<Self::Buffer> {
209        let (mut buffer, addr) = self.op.take_buffer()?;
210        unsafe { buffer.advance_to(self.len) };
211        Some(RecvFromMultiResult { buffer, addr })
212    }
213}
214
215/// Result of [`RecvMsgMulti`].
216pub struct RecvMsgMultiResult {
217    buffer: BufferRef,
218    control: BufferRef,
219    addr: Option<SockAddr>,
220    return_flags: ReturnFlags,
221}
222
223impl RecvMsgMultiResult {
224    #[doc(hidden)]
225    pub unsafe fn new(_: BufferRef, _: usize) -> Self {
226        unreachable!("should not be called directly")
227    }
228
229    /// Get the payload data.
230    pub fn data(&self) -> &[u8] {
231        self.buffer.as_init()
232    }
233
234    /// Get the source address if applicable.
235    pub fn addr(&self) -> Option<SockAddr> {
236        self.addr.clone()
237    }
238
239    /// Get the ancillary data.
240    pub fn ancillary(&self) -> &[u8] {
241        self.control.as_init()
242    }
243
244    /// Get flags returned by `recvmsg`.
245    pub fn flags(&self) -> ReturnFlags {
246        self.return_flags
247    }
248}
249
250impl IntoInner for RecvMsgMultiResult {
251    type Inner = BufferRef;
252
253    fn into_inner(self) -> Self::Inner {
254        self.buffer
255    }
256}
257
258/// Receive data, ancillary data and source address multi times into multiple
259/// managed buffers.
260pub struct RecvMsgMulti<S: AsFd> {
261    pub(crate) op: RecvMsgManaged<BufferRef, S>,
262    pub(crate) len: usize,
263}
264
265impl<S: AsFd> RecvMsgMulti<S> {
266    /// Create [`RecvMsgMulti`].
267    pub fn new(fd: S, pool: &BufferPool, control_len: usize, flags: RecvFlags) -> io::Result<Self> {
268        Ok(Self {
269            op: RecvMsgManaged::new(fd, pool, 0, pool.pop()?.with_capacity(control_len), flags)?,
270            len: 0,
271        })
272    }
273}
274
275impl<S: AsFd> TakeBuffer for RecvMsgMulti<S> {
276    type Buffer = RecvMsgMultiResult;
277
278    fn take_buffer(self) -> Option<Self::Buffer> {
279        let ((mut buffer, mut control), addr, control_len, return_flags) = self.op.take_buffer()?;
280        unsafe { buffer.advance_to(self.len) };
281        unsafe { control.advance_to(control_len) };
282        Some(RecvMsgMultiResult {
283            buffer,
284            control,
285            addr,
286            return_flags,
287        })
288    }
289}
compio_driver/sys/op/managed/fallback.rs

compio_driver/sys/op/managed/
fallback.rs
