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 RecordWriteBytes for RecordInput {
57    fn resident_bytes(&self, history: &crate::StoreHistory) -> Result<usize> {
58        let metadata = self.borrowed().resident_bytes(history)?;
59        metadata
60            .checked_add(self.payload.capacity() - self.payload.len())
61            .ok_or(Error::InvalidInput { field: "storage admission", reason: "size overflow" })
62    }
63}
64
65impl Drop for WriteWaiter {
66    fn drop(&mut self) {
67        let counter = if self.measurement { &self.inner.measurement_waiting } else { &self.inner.control_waiting };
68        counter.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
69        self.inner.notify();
70    }
71}
72
73impl Drop for Admission {
74    fn drop(&mut self) {
75        if self.active {
76            {
77                let mut state = self.inner.state.lock().unwrap();
78                state.reserved = 0;
79                state.reserved_batches = 0;
80                state.collect();
81            }
82            self.inner.notify();
83        }
84    }
85}
86
87impl Write for Counter {
88    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
89        self.0 = self.0.checked_add(bytes.len()).ok_or_else(|| io::Error::other("encoded size overflow"))?;
90        Ok(bytes.len())
91    }
92
93    fn flush(&mut self) -> io::Result<()> {
94        Ok(())
95    }
96}
97
98/* === Public Functions === */
99
100pub fn encoded_size<T: Serialize + ?Sized>(value: &T) -> Result<usize> {
101    let mut counter = Counter(0);
102    serde_json::to_writer(&mut counter, value)
103        .map_err(|_| Error::InvalidInput { field: "storage payload", reason: "cannot measure encoded size" })?;
104    Ok(counter.0)
105}
106
107pub(crate) fn recovery_preparation_bytes(
108    payload: usize, history: &crate::StoreHistory, data_history: &crate::StoreHistory,
109) -> Result<usize> {
110    let capacity = payload
111        .checked_mul(4)
112        .and_then(|bytes| bytes.checked_add(128))
113        .ok_or(Error::InvalidInput { field: "managed recovery", reason: "size overflow" })?;
114    recovery_resident_bytes(capacity, history, data_history)
115}
116
117pub(crate) fn recovery_resident_bytes(payload: usize, history: &crate::StoreHistory, data_history: &crate::StoreHistory) -> Result<usize> {
118    [
119        payload,
120        history.store.as_str().len(),
121        history.epoch.as_str().len(),
122        data_history.store.as_str().len(),
123        data_history.epoch.as_str().len(),
124        std::mem::size_of::<RecoveryEntry>(),
125        ENTRY_OVERHEAD,
126    ]
127    .into_iter()
128    .try_fold(0usize, |bytes, next| bytes.checked_add(next))
129    .ok_or(Error::InvalidInput { field: "managed recovery", reason: "size overflow" })
130}
131
132/* === Private Functions === */
133
134/* === Tests === */
harmos_storage/manager/encoding.rs

harmos_storage/manager/
encoding.rs
