acyclic-fs 0.1.5

Rust SDK for versioned workspaces and embedded native mounts
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

//! Shared synchronous/asynchronous driver for resumable point frontiers.

use crate::async_storage::AsyncObjectStore;
use crate::cancellation::CancellationToken;
use crate::performance::{WorkBudget, WorkCounters};
use crate::storage::{ObjectFailure, ObjectId, ObjectRead, ObjectReceipt, ObjectStore};

#[derive(Clone, Copy)]
pub(crate) struct ReadRequest {
    pub(crate) page: ObjectId,
    pub(crate) maximum_bytes: u64,
    pub(crate) remaining: WorkBudget,
    pub(crate) prospective: WorkCounters,
}

pub(crate) trait Machine {
    type Output;
    type Failure;

    fn complete(&mut self) -> Result<Option<Self::Output>, Self::Failure>;
    fn prepare_read(&mut self) -> Result<ReadRequest, Self::Failure>;
    fn accept(
        &mut self,
        prospective: WorkCounters,
        receipt: &ObjectReceipt<ObjectRead>,
    ) -> Result<Option<Self::Output>, Self::Failure>;
    fn storage_failure(&self, prospective: WorkCounters, failure: ObjectFailure) -> Self::Failure;
    fn cancelled(&self) -> Self::Failure;
}

pub(crate) fn drive_sync<S, M>(store: &S, machine: &mut M) -> Result<M::Output, M::Failure>
where
    S: ObjectStore,
    M: Machine,
{
    loop {
        if let Some(output) = machine.complete()? {
            return Ok(output);
        }
        let request = machine.prepare_read()?;
        let receipt = ObjectStore::read(
            store,
            request.page,
            request.maximum_bytes,
            request.remaining,
        )
        .map_err(|failure| machine.storage_failure(request.prospective, failure))?;
        if let Some(output) = machine.accept(request.prospective, &receipt)? {
            return Ok(output);
        }
    }
}

pub(crate) async fn drive_async<S, M>(
    store: &S,
    machine: &mut M,
    cancellation: &CancellationToken,
) -> Result<M::Output, M::Failure>
where
    S: AsyncObjectStore,
    M: Machine,
{
    loop {
        if cancellation.is_cancelled() {
            return Err(machine.cancelled());
        }
        if let Some(output) = machine.complete()? {
            return Ok(output);
        }
        let request = machine.prepare_read()?;
        let receipt = AsyncObjectStore::read(
            store,
            request.page,
            request.maximum_bytes,
            request.remaining,
            cancellation,
        )
        .await
        .map_err(|failure| machine.storage_failure(request.prospective, failure))?;
        if let Some(output) = machine.accept(request.prospective, &receipt)? {
            return Ok(output);
        }
    }
}