1+ use crate :: io:: blocking:: Buf ;
2+ use crate :: io:: uring:: utils:: { ArcFd , UringFd } ;
13use crate :: runtime:: driver:: op:: { CancelData , Cancellable , Completable , CqeResult , Op } ;
24
35use io_uring:: { opcode, types} ;
6+ use std:: fmt;
47use std:: io:: { self , Error } ;
5- use std:: os:: fd:: { AsRawFd , OwnedFd } ;
8+ use std:: os:: fd:: OwnedFd ;
69
7- #[ derive( Debug ) ]
8- pub ( crate ) struct Read {
9- fd : OwnedFd ,
10- buf : Vec < u8 > ,
10+ /// Trait for buffers that can be used with io-uring read operations.
11+ pub ( crate ) trait ReadBuffer : Send + ' static {
12+ /// Prepare the buffer for a read operation.
13+ /// Returns a pointer and length for the io-uring SQE.
14+ fn uring_read_prepare ( & mut self , max_len : usize ) -> ( * mut u8 , u32 ) ;
15+
16+ /// Complete a read of `n` bytes.
17+ ///
18+ /// # Safety
19+ ///
20+ /// The caller must ensure the kernel wrote exactly `n` bytes
21+ /// into the buffer at the pointer returned by `uring_read_prepare`.
22+ unsafe fn uring_read_complete ( & mut self , n : u32 ) ;
1123}
1224
13- impl Completable for Read {
14- type Output = ( io:: Result < u32 > , OwnedFd , Vec < u8 > ) ;
25+ impl ReadBuffer for Vec < u8 > {
26+ fn uring_read_prepare ( & mut self , max_len : usize ) -> ( * mut u8 , u32 ) {
27+ assert ! ( self . spare_capacity_mut( ) . len( ) >= max_len) ;
28+ let ptr = self . spare_capacity_mut ( ) . as_mut_ptr ( ) . cast ( ) ;
29+ ( ptr, max_len as u32 )
30+ }
31+
32+ unsafe fn uring_read_complete ( & mut self , n : u32 ) {
33+ // SAFETY: the kernel wrote `n` bytes into spare capacity starting
34+ // at the old self.len(), so self.len() + n bytes are now initialized.
35+ unsafe { self . set_len ( self . len ( ) + n as usize ) } ;
36+ }
37+ }
38+
39+ impl ReadBuffer for Buf {
40+ fn uring_read_prepare ( & mut self , max_len : usize ) -> ( * mut u8 , u32 ) {
41+ self . prepare_uring_read ( max_len)
42+ }
43+
44+ unsafe fn uring_read_complete ( & mut self , n : u32 ) {
45+ // SAFETY: caller guarantees kernel wrote exactly n bytes.
46+ unsafe { self . complete_uring_read ( n as usize ) } ;
47+ }
48+ }
49+
50+ pub ( crate ) struct Read < B , F = ArcFd > {
51+ fd : F ,
52+ buf : B ,
53+ }
54+
55+ impl < B : fmt:: Debug , F > fmt:: Debug for Read < B , F > {
56+ fn fmt ( & self , f : & mut fmt:: Formatter < ' _ > ) -> fmt:: Result {
57+ f. debug_struct ( "Read" )
58+ . field ( "buf" , & self . buf )
59+ . finish_non_exhaustive ( )
60+ }
61+ }
62+
63+ impl < B : ReadBuffer , F : UringFd > Completable for Read < B , F > {
64+ type Output = ( io:: Result < u32 > , F , B ) ;
1565
1666 fn complete ( self , cqe : CqeResult ) -> Self :: Output {
1767 let mut buf = self . buf ;
18-
1968 if let Ok ( len) = cqe. result {
20- let new_len = buf. len ( ) + len as usize ;
21- // SAFETY: Kernel read len bytes
22- unsafe { buf. set_len ( new_len) } ;
69+ // SAFETY: kernel wrote exactly `len` bytes into the prepared buffer.
70+ unsafe { buf. uring_read_complete ( len) } ;
2371 }
24-
2572 ( cqe. result , self . fd , buf)
2673 }
2774
@@ -30,32 +77,38 @@ impl Completable for Read {
3077 }
3178}
3279
33- impl Cancellable for Read {
80+ impl Cancellable for Read < Vec < u8 > , OwnedFd > {
3481 fn cancel ( self ) -> CancelData {
35- CancelData :: Read ( self )
82+ CancelData :: ReadVec ( self )
3683 }
3784}
3885
39- impl Op < Read > {
40- // Submit a request to read a FD at given length and offset into a
41- // dynamic buffer with uninitialized memory. The read happens on uninitialized
42- // buffer and no overwriting happens.
86+ impl Cancellable for Read < Buf , ArcFd > {
87+ fn cancel ( self ) -> CancelData {
88+ CancelData :: ReadBuf ( self )
89+ }
90+ }
4391
44- // SAFETY: The `len` of the amount to be read and the buffer that is passed
45- // should have capacity > len.
46- //
47- // If `len` read is higher than vector capacity then setting its length by
48- // the caller in terms of size_read can be unsound.
49- pub ( crate ) fn read ( fd : OwnedFd , mut buf : Vec < u8 > , len : u32 , offset : u64 ) -> Self {
50- // don't overwrite on already written part
51- assert ! ( buf. spare_capacity_mut( ) . len( ) >= len as usize ) ;
52- let buf_mut_ptr = buf. spare_capacity_mut ( ) . as_mut_ptr ( ) . cast ( ) ;
92+ impl < B , F > Op < Read < B , F > >
93+ where
94+ B : ReadBuffer + fmt:: Debug ,
95+ F : UringFd ,
96+ Read < B , F > : Cancellable ,
97+ {
98+ /// Submit a read operation via io-uring.
99+ ///
100+ /// `max_len` is the maximum number of bytes to read.
101+ /// `offset` is the file offset; use `u64::MAX` for the current cursor.
102+ pub ( crate ) fn read_at ( fd : F , mut buf : B , max_len : usize , offset : u64 ) -> Self {
103+ let ( ptr, len) = buf. uring_read_prepare ( max_len) ;
53104
54- let read_op = opcode:: Read :: new ( types:: Fd ( fd . as_raw_fd ( ) ) , buf_mut_ptr , len)
105+ let sqe = opcode:: Read :: new ( types:: Fd ( UringFd :: as_raw_fd ( & fd ) ) , ptr , len)
55106 . offset ( offset)
56107 . build ( ) ;
57108
58- // SAFETY: Parameters are valid for the entire duration of the operation
59- unsafe { Op :: new ( read_op, Read { fd, buf } ) }
109+ // SAFETY: `fd` and `buf`, which owns the heap buffer, are moved into `Read`,
110+ // which is held by the `Op` for the entire duration of the io-uring operation.
111+ // The buffer pointer remains valid because Vec heap data doesn't move.
112+ unsafe { Op :: new ( sqe, Read { fd, buf } ) }
60113 }
61114}
0 commit comments