harmos_storage/manager/
encoding.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17
18// Third Party Imports
19
20// Local Imports
21use super::*;
22
23/* === Constants === */
24
25/* === Macros === */
26
27/* === Type Aliases === */
28
29/* === Structs === */
30
31/* === Enums === */
32
33/* === Traits === */
34
35/* === Implementations === */
36
37impl RecordWriteBytes for RecordWrite<'_> {
38    fn resident_bytes(&self, history: &crate::StoreHistory) -> Result<usize> {
39        [
40            self.payload.len(),
41            self.schema.id.as_str().len(),
42            self.scope.map_or(0, |value| value.as_str().len()),
43            self.item.as_str().len(),
44            self.source.map_or(0, |value| value.as_str().len()),
45            history.store.as_str().len(),
46            history.epoch.as_str().len(),
47            std::mem::size_of::<StoredRecord>(),
48            ENTRY_OVERHEAD,
49        ]
50        .into_iter()
51        .try_fold(0usize, |bytes, next| bytes.checked_add(next))
52        .ok_or(Error::InvalidInput { field: "storage admission", reason: "size overflow" })
53    }
54}
55
56impl Drop for WriteWaiter {
57    fn drop(&mut self) {
58        let counter = if self.measurement { &self.inner.measurement_waiting } else { &self.inner.control_waiting };
59        counter.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
60        self.inner.notify();
61    }
62}
63
64impl Drop for Admission {
65    fn drop(&mut self) {
66        if self.active {
67            {
68                let mut state = self.inner.state.lock().unwrap();
69                state.reserved = 0;
70                state.reserved_batches = 0;
71                state.collect();
72            }
73            self.inner.notify();
74        }
75    }
76}
77
78impl Write for Counter {
79    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
80        self.0 = self.0.checked_add(bytes.len()).ok_or_else(|| io::Error::other("encoded size overflow"))?;
81        Ok(bytes.len())
82    }
83
84    fn flush(&mut self) -> io::Result<()> {
85        Ok(())
86    }
87}
88
89/* === Public Functions === */
90
91pub fn encoded_size<T: Serialize + ?Sized>(value: &T) -> Result<usize> {
92    let mut counter = Counter(0);
93    serde_json::to_writer(&mut counter, value)
94        .map_err(|_| Error::InvalidInput { field: "storage payload", reason: "cannot measure encoded size" })?;
95    Ok(counter.0)
96}
97
98pub(crate) fn recovery_preparation_bytes(
99    payload: usize, history: &crate::StoreHistory, data_history: &crate::StoreHistory,
100) -> Result<usize> {
101    let capacity = payload
102        .checked_mul(4)
103        .and_then(|bytes| bytes.checked_add(128))
104        .ok_or(Error::InvalidInput { field: "managed recovery", reason: "size overflow" })?;
105    recovery_resident_bytes(capacity, history, data_history)
106}
107
108pub(crate) fn recovery_resident_bytes(payload: usize, history: &crate::StoreHistory, data_history: &crate::StoreHistory) -> Result<usize> {
109    [
110        payload,
111        history.store.as_str().len(),
112        history.epoch.as_str().len(),
113        data_history.store.as_str().len(),
114        data_history.epoch.as_str().len(),
115        std::mem::size_of::<RecoveryEntry>(),
116        ENTRY_OVERHEAD,
117    ]
118    .into_iter()
119    .try_fold(0usize, |bytes, next| bytes.checked_add(next))
120    .ok_or(Error::InvalidInput { field: "managed recovery", reason: "size overflow" })
121}
122
123/* === Private Functions === */
124
125/* === Tests === */
harmos_storage/manager/encoding.rs

harmos_storage/manager/
encoding.rs
