compio_fs/
file.rs

1use std::{future::Future, io, mem::ManuallyDrop, path::Path};
2
3use compio_buf::{BufResult, IntoInner, IoBuf, IoBufMut};
4#[cfg(unix)]
5use compio_driver::op::FileStat;
6use compio_driver::{
7    BufferRef, ResultTakeBuffer, ToSharedFd, impl_raw_fd,
8    op::{BufResultExt, CloseFile, ReadAt, ReadManagedAt, Sync, WriteAt},
9};
10use compio_io::{AsyncReadAt, AsyncReadManagedAt, AsyncWriteAt, util::Splittable};
11use compio_runtime::{Runtime, fd::AsyncFd};
12#[cfg(all(unix, not(solarish)))]
13use {
14    compio_buf::{IoVectoredBuf, IoVectoredBufMut},
15    compio_driver::op::{ReadVectoredAt, WriteVectoredAt},
16};
17
18use crate::{Metadata, OpenOptions, Permissions};
19
20/// A reference to an open file on the filesystem.
21///
22/// An instance of a `File` can be read and/or written depending on what options
23/// it was opened with. The `File` type provides **positional** read and write
24/// operations. The file does not maintain an internal cursor. The caller is
25/// required to specify an offset when issuing an operation.
26///
27///
28/// If you'd like to use methods from [`AsyncRead`](`compio_io::AsyncRead`) or
29/// [`AsyncWrite`](`compio_io::AsyncWrite`) traits, you can wrap `File` with
30/// [`std::io::Cursor`].
31///
32/// # Examples
33/// ```no_run
34/// // Most users depend on the `compio` crate rather than on the sub-crates
35/// // directly. In that case, import these as `compio::fs::File`,
36/// // `compio::buf::BufResult` and `compio::io::AsyncReadExt` instead.
37/// use std::io::Cursor;
38///
39/// use compio_buf::BufResult;
40/// use compio_fs::File;
41/// use compio_io::AsyncReadExt;
42///
43/// # compio_runtime::Runtime::new().unwrap().block_on(async {
44/// let file = File::open("foo.txt").await?;
45/// let mut cursor = Cursor::new(file);
46///
47/// let int = cursor.read_u32().await?;
48/// let float = cursor.read_f32().await?;
49///
50/// let string = String::new();
51/// let BufResult(result, string) = cursor.read_to_string(string).await;
52///
53/// let buf = vec![0; 1024];
54/// let BufResult(result, buf) = cursor.read_exact(buf).await;
55/// # std::io::Result::Ok(()) });
56/// ```
57#[derive(Debug, Clone)]
58pub struct File {
59    pub(crate) inner: AsyncFd<std::fs::File>,
60}
61
62impl File {
63    pub(crate) fn from_std(file: std::fs::File) -> io::Result<Self> {
64        Ok(Self {
65            inner: AsyncFd::new(file)?,
66        })
67    }
68
69    /// Attempts to open a file in read-only mode.
70    ///
71    /// See the [`OpenOptions::open`] method for more details.
72    pub async fn open(path: impl AsRef<Path>) -> io::Result<Self> {
73        OpenOptions::new().read(true).open(path).await
74    }
75
76    /// Opens a file in write-only mode.
77    ///
78    /// This function will create a file if it does not exist,
79    /// and will truncate it if it does.
80    ///
81    /// See the [`OpenOptions::open`] function for more details.
82    pub async fn create(path: impl AsRef<Path>) -> io::Result<Self> {
83        OpenOptions::new()
84            .create(true)
85            .write(true)
86            .truncate(true)
87            .open(path)
88            .await
89    }
90
91    /// Close the file. If the returned future is dropped before polling, the
92    /// file won't be closed.
93    ///
94    /// As [`File`] is clonable, users can call `close` on a clone, but the
95    /// future will never complete until all clones are dropped. Some
96    /// operations may keep a strong reference to the file, so the future
97    /// may never complete if there are pending operations.
98    ///
99    /// It's OK to drop the [`File`] directly without calling `close`, but the
100    /// file may not be closed immediately.
101    pub fn close(self) -> impl Future<Output = io::Result<()>> {
102        // Make sure that fd won't be dropped after `close` called.
103        // Users may call this method and drop the future immediately. In that
104        // way `close` should be cancelled.
105        let this = ManuallyDrop::new(self);
106        async move {
107            let fd = ManuallyDrop::into_inner(this)
108                .inner
109                .into_inner()
110                .take()
111                .await;
112            if let Some(fd) = fd {
113                let op = CloseFile::new(fd.into());
114                compio_runtime::submit(op).await.0?;
115            }
116            Ok(())
117        }
118    }
119
120    /// Queries metadata about the underlying file.
121    #[cfg(windows)]
122    pub async fn metadata(&self) -> io::Result<Metadata> {
123        crate::spawn_blocking_with(self.to_shared_fd(), |file| {
124            file.metadata().map(Metadata::from_std)
125        })
126        .await
127    }
128
129    #[cfg(windows)]
130    /// Truncates or extends the underlying file, updating the size of this file
131    /// to become `size`.
132    pub async fn set_len(&self, size: u64) -> io::Result<()> {
133        crate::spawn_blocking_with(self.to_shared_fd(), move |file| file.set_len(size)).await
134    }
135
136    #[cfg(unix)]
137    /// Truncates or extends the underlying file, updating the size of this file
138    /// to become `size`.
139    ///
140    /// NOTE: On Linux kernel <= 6.9 or when io uring is disabled, the operation
141    /// will be offloaded to the separate blocking thread
142    pub async fn set_len(&self, size: u64) -> io::Result<()> {
143        use compio_driver::op::TruncateFile;
144
145        let op = TruncateFile::new(self.to_shared_fd(), size);
146        compio_runtime::submit(op).await.0.map(|_| ())
147    }
148
149    /// Queries metadata about the underlying file.
150    #[cfg(unix)]
151    pub async fn metadata(&self) -> io::Result<Metadata> {
152        let op = FileStat::new(self.to_shared_fd());
153        let BufResult(res, op) = compio_runtime::submit(op).await;
154        res.map(|_| Metadata::from_attr(op.into_inner()))
155    }
156
157    /// Reads an extended attribute from this open file, with `fgetxattr`
158    /// semantics.
159    ///
160    /// The file descriptor identifies the object; no pathname is resolved
161    /// again. In particular, this does not read attributes from a symbolic
162    /// link itself.
163    ///
164    /// The value is written from the start of `buffer`, using its full
165    /// capacity. On success, the result is the value's length and the
166    /// buffer's initialized length advances to at least that length. With
167    /// zero capacity, only the required length is returned and the buffer
168    /// remains empty. The attribute may change between a sizing query and a
169    /// subsequent read.
170    ///
171    /// A missing attribute returns `ENODATA`; insufficient nonzero capacity
172    /// returns `ERANGE`. Other OS errors are preserved. A name containing a NUL
173    /// byte returns [`io::ErrorKind::InvalidInput`]. Errors return the original
174    /// buffer without advancing its initialized length.
175    ///
176    /// The submitted operation owns the name and buffer and holds a reference
177    /// to the file descriptor until completion, even if this future is
178    /// dropped. Cancellation does not return the buffer to the caller.
179    #[cfg(any(target_os = "linux", target_os = "android"))]
180    pub async fn get_xattr<T: IoBufMut>(
181        &self,
182        name: impl AsRef<std::ffi::OsStr>,
183        buffer: T,
184    ) -> BufResult<usize, T> {
185        use std::{ffi::CString, os::unix::ffi::OsStrExt};
186
187        use compio_buf::{IoBufMutExt, buf_try};
188        use compio_driver::op::FGetXattr;
189
190        let (name, mut buffer) = buf_try!(
191            CString::new(name.as_ref().as_bytes()).map_err(io::Error::from),
192            buffer
193        );
194        let query_size = buffer.buf_capacity() == 0;
195        let op = FGetXattr::new(self.to_shared_fd(), name, buffer);
196        let res = compio_runtime::submit(op).await.into_inner();
197        if query_size {
198            res
199        } else {
200            // SAFETY: A successful nonzero-capacity fgetxattr initializes
201            // exactly the returned byte count, bounded by the
202            // supplied buffer capacity. Size-only queries are
203            // excluded because they do not write bytes.
204            unsafe { res.map_advanced() }
205        }
206    }
207
208    /// Changes the permissions on the underlying file.
209    #[cfg(windows)]
210    pub async fn set_permissions(&self, perm: Permissions) -> io::Result<()> {
211        crate::spawn_blocking_with(self.to_shared_fd(), move |file| {
212            if let Some(p) = perm.0.original {
213                file.set_permissions(p)
214            } else {
215                let mut p = file.metadata()?.permissions();
216                p.set_readonly(perm.readonly());
217                file.set_permissions(p)
218            }
219        })
220        .await
221    }
222
223    /// Changes the permissions on the underlying file.
224    #[cfg(unix)]
225    pub async fn set_permissions(&self, perm: Permissions) -> io::Result<()> {
226        crate::spawn_blocking_with(self.to_shared_fd(), |file| file.set_permissions(perm.0)).await
227    }
228
229    async fn sync_impl(&self, datasync: bool) -> io::Result<()> {
230        let op = Sync::new(self.to_shared_fd(), datasync);
231        compio_runtime::submit(op).await.0?;
232        Ok(())
233    }
234
235    /// Attempts to sync all OS-internal metadata to disk.
236    ///
237    /// This function will attempt to ensure that all in-memory data reaches the
238    /// filesystem before returning.
239    pub async fn sync_all(&self) -> io::Result<()> {
240        self.sync_impl(false).await
241    }
242
243    /// This function is similar to [`sync_all`], except that it might not
244    /// synchronize file metadata to the filesystem.
245    ///
246    /// This is intended for use cases that must synchronize content, but don't
247    /// need the metadata on disk. The goal of this method is to reduce disk
248    /// operations.
249    ///
250    /// Note that some platforms may simply implement this in terms of
251    /// [`sync_all`].
252    ///
253    /// [`sync_all`]: File::sync_all
254    pub async fn sync_data(&self) -> io::Result<()> {
255        self.sync_impl(true).await
256    }
257}
258
259impl AsyncReadAt for File {
260    async fn read_at<T: IoBufMut>(&self, buffer: T, pos: u64) -> BufResult<usize, T> {
261        let fd = self.inner.to_shared_fd();
262        let op = ReadAt::new(fd, pos, buffer);
263        let res = compio_runtime::submit(op).await.into_inner();
264        unsafe { res.map_advanced() }
265    }
266
267    #[cfg(all(unix, not(solarish)))]
268    async fn read_vectored_at<T: IoVectoredBufMut>(
269        &self,
270        buffer: T,
271        pos: u64,
272    ) -> BufResult<usize, T> {
273        use compio_driver::op::VecBufResultExt;
274
275        let fd = self.inner.to_shared_fd();
276        let op = ReadVectoredAt::new(fd, pos, buffer);
277        let res = compio_runtime::submit(op).await.into_inner();
278        unsafe { res.map_vec_advanced() }
279    }
280}
281
282impl AsyncReadManagedAt for File {
283    type Buffer = BufferRef;
284
285    async fn read_managed_at(&self, len: usize, pos: u64) -> io::Result<Option<Self::Buffer>> {
286        let fd = self.inner.to_shared_fd();
287        let res = Runtime::with_current(|rt| {
288            let buffer_pool = rt.buffer_pool()?;
289            let op = ReadManagedAt::new(fd, pos, &buffer_pool, len)?;
290            io::Result::Ok(rt.submit(op))
291        })?
292        .await;
293        unsafe { res.take_buffer() }
294    }
295}
296
297impl AsyncWriteAt for File {
298    #[inline]
299    async fn write_at<T: IoBuf>(&mut self, buf: T, pos: u64) -> BufResult<usize, T> {
300        (&*self).write_at(buf, pos).await
301    }
302
303    #[cfg(all(unix, not(solarish)))]
304    #[inline]
305    async fn write_vectored_at<T: IoVectoredBuf>(
306        &mut self,
307        buf: T,
308        pos: u64,
309    ) -> BufResult<usize, T> {
310        (&*self).write_vectored_at(buf, pos).await
311    }
312}
313
314impl AsyncWriteAt for &File {
315    async fn write_at<T: IoBuf>(&mut self, buffer: T, pos: u64) -> BufResult<usize, T> {
316        let fd = self.inner.to_shared_fd();
317        let op = WriteAt::new(fd, pos, buffer);
318        compio_runtime::submit(op).await.into_inner()
319    }
320
321    #[cfg(all(unix, not(solarish)))]
322    async fn write_vectored_at<T: IoVectoredBuf>(
323        &mut self,
324        buffer: T,
325        pos: u64,
326    ) -> BufResult<usize, T> {
327        let fd = self.inner.to_shared_fd();
328        let op = WriteVectoredAt::new(fd, pos, buffer);
329        compio_runtime::submit(op).await.into_inner()
330    }
331}
332
333impl Splittable for File {
334    type ReadHalf = File;
335    type WriteHalf = File;
336
337    fn split(self) -> (Self::ReadHalf, Self::WriteHalf) {
338        (self.clone(), self)
339    }
340}
341
342impl Splittable for &File {
343    type ReadHalf = File;
344    type WriteHalf = File;
345
346    fn split(self) -> (Self::ReadHalf, Self::WriteHalf) {
347        (self.clone(), self.clone())
348    }
349}
350
351impl Splittable for &mut File {
352    type ReadHalf = File;
353    type WriteHalf = File;
354
355    fn split(self) -> (Self::ReadHalf, Self::WriteHalf) {
356        (self.clone(), self.clone())
357    }
358}
359
360impl_raw_fd!(File, std::fs::File, inner, file);
compio_fs/file.rs

compio_fs/
file.rs
