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

compio_fs/
file.rs
