harmos_runtime/storage/
manager.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14pub use harmos_storage::ErasedCatalog;
15pub(crate) use harmos_storage::encoded_size;
16
17/* === Imports === */
18
19// Standard Library Imports
20use std::sync::{Arc, Mutex};
21
22// Third Party Imports
23
24// Local Imports
25use super::{StorageBackends, StorageSettings};
26use crate::operations::{OperationManager, resident::Lease};
27use crate::services::{Fault, Once, Resident, Service, ServiceDefinition};
28use crate::supervision::executor;
29use crate::{Application, Error, Result};
30
31/* === Constants === */
32
33/* === Macros === */
34
35/* === Type Aliases === */
36
37/* === Structs === */
38
39pub struct StorageManager<C = ()> {
40    engine: harmos_storage::StorageManager<C>,
41}
42
43struct Worker(harmos_storage::Worker);
44
45struct Owner {
46    leases: Mutex<Vec<Lease>>,
47}
48
49/* === Enums === */
50
51/* === Traits === */
52
53/* === Implementations === */
54
55impl<C> Clone for StorageManager<C> {
56    fn clone(&self) -> Self {
57        Self { engine: self.engine.clone() }
58    }
59}
60
61impl<C> std::ops::Deref for StorageManager<C> {
62    type Target = harmos_storage::StorageManager<C>;
63    fn deref(&self) -> &Self::Target {
64        &self.engine
65    }
66}
67
68impl<C: super::ObjectCatalog> StorageManager<C> {
69    pub fn new(backends: StorageBackends, settings: StorageSettings) -> Result<Self> {
70        let execution = harmos_storage::execution::Execution::new(executor::now, |duration| Box::pin(executor::sleep(duration)));
71        Ok(Self { engine: harmos_storage::StorageManager::with_execution(backends, settings, execution)? })
72    }
73}
74
75impl<C> StorageManager<C> {
76    pub(crate) fn claim_owner(&self, owner: Arc<()>) -> Result<()> {
77        self.engine.claim_owner(owner).map_err(Into::into)
78    }
79    pub(crate) async fn acquire(&self) -> Result<()> {
80        self.engine.acquire().await.map_err(Into::into)
81    }
82    pub async fn close(&self) -> Result<()> {
83        self.engine.close().await.map_err(Into::into)
84    }
85
86    pub(crate) fn erased(&self) -> StorageManager<ErasedCatalog> {
87        StorageManager { engine: self.engine.erased() }
88    }
89    pub(crate) fn same_instance<D>(&self, other: &StorageManager<D>) -> bool {
90        self.engine.same_instance(&other.engine)
91    }
92    pub async fn start<A: Application>(&self, runtime: &crate::Runtime<A>) -> Result<()> {
93        self.start_on::<A>(&runtime.operations).await
94    }
95    pub(crate) async fn start_on<A: Application>(&self, operations: &OperationManager) -> Result<()> {
96        self.engine
97            .start_with(operations.resident_identity(), |worker| {
98                let lease = operations
99                    .start_service::<A>(Once::new(ServiceDefinition::new("harmos-storage"), Worker(worker)), true, false)
100                    .map_err(|error| {
101                        harmos_storage::Error::backend("start storage worker", harmos_storage::BackendFailure::Permanent, error)
102                    })?;
103                Ok(Box::new(Owner { leases: Mutex::new(vec![lease]) }))
104            })
105            .await
106            .map_err(Into::into)
107    }
108}
109
110impl<A: Application> Resident<A> for Worker {
111    type Error = Error;
112    async fn cleanup(&mut self) -> Result<()> {
113        self.0.cleanup().await.map_err(Into::into)
114    }
115    async fn run(&mut self, mut context: Service<A>) -> std::result::Result<(), Fault<Error>> {
116        context.ready();
117        context.until_stopping(self.0.run()).await.unwrap_or(Ok(())).map_err(|error| Fault::Environmental(error.into()))
118    }
119}
120
121impl harmos_storage::WorkerOwner for Owner {
122    fn stop(&self) -> harmos_storage::StorageFuture<'_, ()> {
123        Box::pin(async move {
124            let receipts: Vec<_> = self.leases.lock().unwrap().iter_mut().map(Lease::retire).collect();
125            let mut failure = None;
126            for receipt in receipts {
127                if let Err(error) = receipt.await {
128                    failure.get_or_insert_with(|| {
129                        harmos_storage::Error::backend("stop storage worker", harmos_storage::BackendFailure::Permanent, error)
130                    });
131                }
132            }
133            failure.map_or(Ok(()), Err)
134        })
135    }
136}
137
138/* === Public Functions === */
139
140/* === Private Functions === */
141
142/* === Tests === */
harmos_runtime/storage/manager.rs

harmos_runtime/storage/
manager.rs
