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 encoded_bytes(&self, history: &crate::StoreHistory) -> Result<usize> {
39        let value = StoredRecord {
40            position: StorageCursor {
41                history: history.clone(),
42                sequence: u64::MAX,
43            },
44            schema: self.schema.clone(),
45            scope: self.scope.cloned(),
46            item: self.item.clone(),
47            source: self.source.cloned(),
48            source_time: self.source_time,
49            payload: self.payload,
50        };
51        encoded_size(&value)
52    }
53}
54
55impl RecordWriteBytes for RecordInput {
56    fn encoded_bytes(&self, history: &crate::StoreHistory) -> Result<usize> {
57        self.borrowed().encoded_bytes(history)
58    }
59}
60
61impl Drop for WriteWaiter {
62    fn drop(&mut self) {
63        let counter = if self.measurement { &self.inner.measurement_waiting } else { &self.inner.control_waiting };
64        counter.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
65        self.inner.notify();
66    }
67}
68
69impl Drop for Admission {
70    fn drop(&mut self) {
71        if self.active {
72            {
73                let mut state = self.inner.state.lock().unwrap();
74                state.reserved = 0;
75                state.reserved_batches = 0;
76                state.collect();
77            }
78            self.inner.notify();
79        }
80    }
81}
82
83impl Write for Counter {
84    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
85        self.0 = self.0.checked_add(bytes.len()).ok_or_else(|| io::Error::other("encoded size overflow"))?;
86        Ok(bytes.len())
87    }
88
89    fn flush(&mut self) -> io::Result<()> {
90        Ok(())
91    }
92}
93
94/* === Public Functions === */
95
96pub fn encoded_size<T: Serialize + ?Sized>(value: &T) -> Result<usize> {
97    let mut counter = Counter(0);
98    serde_json::to_writer(&mut counter, value).map_err(|_| Error::InvalidInput {
99        field: "storage payload",
100        reason: "cannot measure encoded size",
101    })?;
102    Ok(counter.0)
103}
104
105pub(crate) fn recovery_encoded_bound(payload: usize, history: &crate::StoreHistory, data_history: &crate::StoreHistory) -> Result<usize> {
106    let entry = RecoveryEntry {
107        required_data: StorageCursor {
108            history: data_history.clone(),
109            sequence: u64::MAX,
110        },
111        position: RecoveryPosition {
112            history: history.clone(),
113            sequence: u64::MAX,
114        },
115        payload: Vec::new(),
116    };
117    let encoded = payload
118        .checked_div(3)
119        .and_then(|groups| groups.checked_mul(4))
120        .and_then(|bytes| {
121            bytes.checked_add(match payload % 3 {
122                0 => 0,
123                1 => 2,
124                _ => 3,
125            })
126        })
127        .ok_or(Error::InvalidInput {
128            field: "managed recovery",
129            reason: "encoded size overflow",
130        })?;
131    let snapshot = RecoverySnapshot {
132        required_data: entry.required_data.clone(),
133        through: entry.position.clone(),
134        payload: Vec::new(),
135    };
136    let header = encoded_size(&entry)?.max(encoded_size(&snapshot)?);
137    header.checked_add(encoded).ok_or(Error::InvalidInput {
138        field: "managed recovery",
139        reason: "encoded size overflow",
140    })
141}
142
143/* === Private Functions === */
144
145/* === Tests === */
harmos_storage/manager/encoding.rs

harmos_storage/manager/
encoding.rs
