tempest_rt/fs/
sync_file.rs

1//! Async file sync (fsync) operation.
2
3use std::{
4    io,
5    marker::PhantomData,
6    mem::replace,
7    pin::Pin,
8    task::{Context, Poll},
9};
10
11use tempest_io::{Io, OpHandle};
12
13use crate::context::{current_io, get_op_handle};
14
15enum SyncFileState<I: Io> {
16    NeedsSubmit { fd: I::Fd, handle: Option<OpHandle> },
17    InFlight { handle: OpHandle },
18    Done,
19}
20
21/// Future that fsyncs a file descriptor, ensuring data is flushed to storage.
22#[must_use = "futures do nothing unless awaited"]
23pub struct SyncFile<I: Io> {
24    state: SyncFileState<I>,
25    // NB: we use I as the return here, so that SyncFile stays Unpin
26    _marker: PhantomData<fn() -> I>,
27}
28
29/// Issues an fsync on `fd`, resolving when the kernel confirms the flush.
30pub fn sync_file<I: Io>(fd: I::Fd) -> SyncFile<I> {
31    SyncFile {
32        state: SyncFileState::NeedsSubmit { fd, handle: None },
33        _marker: PhantomData,
34    }
35}
36
37// NB: required because of Fd, which is Copy
38impl<I: Io> Unpin for SyncFile<I> {}
39
40impl<I: Io> Future for SyncFile<I> {
41    type Output = io::Result<()>;
42
43    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
44        let this = self.get_mut();
45
46        match replace(&mut this.state, SyncFileState::Done) {
47            SyncFileState::NeedsSubmit { fd, handle } => {
48                let handle = handle.unwrap_or_else(|| unsafe { get_op_handle(cx) });
49
50                // SAFETY: we do not hold on to io outside of this function
51                let io = unsafe { current_io::<I>() };
52                match io.fsync(fd, handle) {
53                    Ok(()) => {
54                        this.state = SyncFileState::InFlight { handle };
55                        Poll::Pending
56                    }
57                    Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
58                        // io SQ is full, retry next tick
59                        this.state = SyncFileState::NeedsSubmit {
60                            fd,
61                            handle: Some(handle),
62                        };
63                        cx.waker().wake_by_ref();
64                        Poll::Pending
65                    }
66                    Err(e) => Poll::Ready(Err(e)),
67                }
68            }
69            SyncFileState::InFlight { handle } => {
70                // SAFETY: we do not hold on to io outside of this function
71                let io = unsafe { current_io::<I>() };
72                match io.get_cqe(handle).transpose()? {
73                    Some(_) => Poll::Ready(Ok(())),
74                    None => Poll::Pending,
75                }
76            }
77            SyncFileState::Done => panic!("polled after completion"),
78        }
79    }
80}
tempest_rt/fs/sync_file.rs

tempest_rt/fs/
sync_file.rs
