mountos_native/
ops.rs

1//! Clone and tree operations. `clone_entry` is one round trip. `clone_dir` and `remove_dir`
2//! return an `Operation` at once. The server reports its end as notices on the connection that
3//! started it, and the reader routes them here by op id.
4
5use crate::client::{Client, Slot};
6use crate::error::{Error, Result};
7use crate::events::lock;
8use crate::wire::{self, method};
9use std::io;
10use std::sync::{Arc, Condvar, Mutex};
11use std::time::{Duration, Instant};
12
13/// Most failures an `Operation` keeps. The server sends at most this many per operation.
14/// `OpCounts::failed` can be larger.
15pub const MAX_OP_FAILURES: usize = 64;
16
17/// Extra properties `clone_entry` and `clone_dir` copy from the source. New kinds can be added.
18#[derive(Clone, Copy, Debug, PartialEq, Eq)]
19#[non_exhaustive]
20pub enum Preserve {
21  /// Keep the source times. Without it the new entry gets the current time.
22  Times,
23  /// Copy extended attributes.
24  Xattr,
25  /// Copy access control lists.
26  Acl,
27}
28
29impl Preserve {
30  fn to_wire(self) -> wire::PreserveKind {
31    match self {
32      Preserve::Times => wire::PreserveKind::Times,
33      Preserve::Xattr => wire::PreserveKind::Xattr,
34      Preserve::Acl => wire::PreserveKind::Acl,
35    }
36  }
37}
38
39/// Settings for `clone_entry` and `clone_dir`. Start from `CloneOptions::default()` and chain the
40/// `with_` methods.
41#[derive(Clone, Debug, Default, PartialEq, Eq)]
42#[non_exhaustive]
43pub struct CloneOptions {
44  /// Replace the content of a destination that differs. Without it such a destination fails
45  /// with `EEXIST`.
46  pub overwrite: bool,
47  /// Properties to copy besides data, mode and owner.
48  pub preserve: Vec<Preserve>,
49}
50
51impl CloneOptions {
52  /// Sets `overwrite`.
53  pub fn with_overwrite(mut self, overwrite: bool) -> CloneOptions {
54    self.overwrite = overwrite;
55    self
56  }
57
58  /// Sets `preserve`.
59  pub fn with_preserve(mut self, preserve: Vec<Preserve>) -> CloneOptions {
60    self.preserve = preserve;
61    self
62  }
63
64  fn to_wire(&self) -> wire::CloneOptions {
65    wire::CloneOptions {
66      overwrite: self.overwrite,
67      preserve: self.preserve.iter().map(|p| p.to_wire() as i32).collect(),
68    }
69  }
70}
71
72/// What `clone_entry` did with the destination. New outcomes can be added.
73#[derive(Clone, Copy, Debug, PartialEq, Eq)]
74#[non_exhaustive]
75pub enum CloneOutcome {
76  /// The destination did not exist and was created.
77  Cloned,
78  /// The destination held different content and was replaced.
79  Overwritten,
80  /// The destination already held the same content. Nothing changed.
81  Skipped,
82}
83
84/// Totals an operation reports when it ends.
85#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
86#[non_exhaustive]
87pub struct OpCounts {
88  pub cloned: u64,
89  pub overwritten: u64,
90  pub skipped: u64,
91  /// Can exceed the number of failures kept.
92  pub failed: u64,
93  pub removed: u64,
94  /// Entries kept because the mount protects them.
95  pub protected: u64,
96}
97
98impl OpCounts {
99  pub(crate) fn from_notice(m: &wire::Notice) -> OpCounts {
100    OpCounts {
101      cloned: m.cloned,
102      overwritten: m.overwritten,
103      skipped: m.skipped,
104      failed: m.failed,
105      removed: m.removed,
106      protected: m.protected,
107    }
108  }
109}
110
111/// One entry that failed inside an operation.
112#[derive(Clone, Debug, PartialEq, Eq)]
113#[non_exhaustive]
114pub struct OpFailure {
115  /// Relative to the mount root. Empty when the name is not valid UTF-8.
116  pub path: String,
117  pub code: String,
118  pub message: String,
119}
120
121impl OpFailure {
122  pub(crate) fn new(path: String, code: String, message: String) -> OpFailure {
123    OpFailure { path, code, message }
124  }
125}
126
127/// How an operation ended.
128#[derive(Clone, Debug, PartialEq, Eq)]
129#[non_exhaustive]
130pub struct OpResult {
131  pub counts: OpCounts,
132  /// At most `MAX_OP_FAILURES`, in the order the server reported them.
133  pub failures: Vec<OpFailure>,
134}
135
136struct Inner {
137  failures: Vec<OpFailure>,
138  /// Set once: the totals, or why they cannot be known.
139  end: Option<std::result::Result<OpCounts, Error>>,
140}
141
142/// The shared record of one operation. The reader thread fills it, `Operation::wait` reads it.
143pub(crate) struct OpState {
144  inner: Mutex<Inner>,
145  cv: Condvar,
146}
147
148impl OpState {
149  pub(crate) fn new() -> Arc<OpState> {
150    Arc::new(OpState {
151      inner: Mutex::new(Inner {
152        failures: Vec::new(),
153        end: None,
154      }),
155      cv: Condvar::new(),
156    })
157  }
158
159  /// Keeps `f` while the operation runs and fewer than `MAX_OP_FAILURES` are kept.
160  pub(crate) fn add_failure(&self, f: OpFailure) {
161    let mut g = lock(&self.inner);
162    if g.end.is_none() && g.failures.len() < MAX_OP_FAILURES {
163      g.failures.push(f);
164    }
165  }
166
167  /// Ends the operation with its totals. The first ending wins.
168  pub(crate) fn finish(&self, counts: OpCounts) {
169    self.settle(Ok(counts));
170  }
171
172  /// Ends the operation because its end cannot be known. The first ending wins.
173  pub(crate) fn fail(&self, e: Error) {
174    self.settle(Err(e));
175  }
176
177  fn settle(&self, end: std::result::Result<OpCounts, Error>) {
178    let mut g = lock(&self.inner);
179    if g.end.is_none() {
180      g.end = Some(end);
181      drop(g);
182      self.cv.notify_all();
183    }
184  }
185
186  fn wait(&self, timeout: Duration) -> Result<OpResult> {
187    let deadline = Instant::now().checked_add(timeout);
188    let mut g = lock(&self.inner);
189    loop {
190      match &g.end {
191        Some(Ok(counts)) => {
192          return Ok(OpResult {
193            counts: *counts,
194            failures: g.failures.clone(),
195          })
196        }
197        Some(Err(e)) => return Err(e.dup()),
198        None => {}
199      }
200      let left = match deadline {
201        Some(d) => d.saturating_duration_since(Instant::now()),
202        None => Duration::from_secs(3600),
203      };
204      if left.is_zero() {
205        return Err(Error::Io(io::Error::new(
206          io::ErrorKind::TimedOut,
207          "operation wait timed out",
208        )));
209      }
210      g = self.cv.wait_timeout(g, left).unwrap_or_else(|e| e.into_inner()).0;
211    }
212  }
213}
214
215/// A running or ended `clone_dir` or `remove_dir`. It lives on the connection that started it.
216pub struct Operation {
217  id: u64,
218  slot: Arc<Slot>,
219  epoch: u64,
220  state: Arc<OpState>,
221}
222
223impl Operation {
224  /// The server's operation id, which is the `ref_id` of its events.
225  pub fn id(&self) -> u64 {
226    self.id
227  }
228
229  /// Waits up to `timeout` for the end of the operation. `Ok` means the server reported the end,
230  /// and the result has the totals and the failures. `Error::OperationStateUnknown` means notices
231  /// were lost. An `Error::Io` of kind `ConnectionAborted` means the connection dropped or the
232  /// client closed. A timeout is an `Error::Io` of kind `TimedOut`. Calling again is allowed.
233  pub fn wait(&self, timeout: Duration) -> Result<OpResult> {
234    self.state.wait(timeout)
235  }
236
237  /// Asks the server to stop the operation. It still ends with an `OpDone` event, so call `wait`
238  /// for the partial totals. An operation that already ended fails with `ENOENT`.
239  pub fn cancel(&self) -> Result<()> {
240    let req = wire::OpCancelRequest { op_id: self.id };
241    self
242      .slot
243      .run(Some(self.epoch), |c| {
244        c.call_msg::<wire::Empty>(method::OP_CANCEL, &req, &[])
245      })
246      .map(drop)
247  }
248}
249
250impl Client {
251  /// Copies one entry that is not a directory (a file, a symlink or a special file) and returns
252  /// what happened to the destination. Needs the `clone` capability. Without it the call fails
253  /// with `UNSUPPORTED_OPERATION`.
254  pub fn clone_entry(&self, from: &str, to: &str, opts: &CloneOptions) -> Result<CloneOutcome> {
255    let req = wire::CloneRequest {
256      from: from.into(),
257      to: to.into(),
258      options: Some(opts.to_wire()),
259    };
260    let r: wire::CloneResponse = self.unary(method::CLONE, &req)?;
261    match wire::CloneOutcome::try_from(r.outcome) {
262      Ok(wire::CloneOutcome::Cloned) => Ok(CloneOutcome::Cloned),
263      Ok(wire::CloneOutcome::Overwritten) => Ok(CloneOutcome::Overwritten),
264      Ok(wire::CloneOutcome::Skipped) => Ok(CloneOutcome::Skipped),
265      Err(_) => Err(Error::Protocol(format!("unknown clone outcome {}", r.outcome))),
266    }
267  }
268
269  /// Starts copying the tree at `from` to `to` in the background and returns at once. Needs the
270  /// `ops` capability. Fails with `EAGAIN` when the server runs too many operations.
271  pub fn clone_dir(&self, from: &str, to: &str, opts: &CloneOptions) -> Result<Operation> {
272    let req = wire::CloneDirRequest {
273      from: from.into(),
274      to: to.into(),
275      options: Some(opts.to_wire()),
276    };
277    self.start_op(|c, st| c.call_op::<wire::CloneDirResponse>(method::CLONE_DIR, &req, st, |r| r.op_id))
278  }
279
280  /// Starts removing `path` and everything below it in the background and returns at once.
281  /// Unlike `remove_all` it keeps going after a failed entry and reports each failure. Needs the
282  /// `ops` capability. Fails with `EAGAIN` when the server runs too many operations.
283  pub fn remove_dir(&self, path: &str) -> Result<Operation> {
284    let req = wire::RemoveDirRequest { path: path.into() };
285    self.start_op(|c, st| c.call_op::<wire::RemoveDirResponse>(method::REMOVE_DIR, &req, st, |r| r.op_id))
286  }
287
288  fn start_op(&self, f: impl FnOnce(&crate::client::Sock, Arc<OpState>) -> Result<u64>) -> Result<Operation> {
289    let state = OpState::new();
290    let st = state.clone();
291    let (id, slot, epoch) = self.run_any(move |c| f(c, st))?;
292    Ok(Operation {
293      id,
294      slot: slot.clone(),
295      epoch,
296      state,
297    })
298  }
299}
mountos_native/ops.rs

mountos_native/
ops.rs
