ruda-runtime 0.1.31

Ruda portable host runtime.
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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

use super::cache::{Record, split_scoped_key};
use ruda_core::{
    bytes::Bytes,
    environment::{self, Namespace, persistence::{Storage, Insertion, Origin, storage}},
};
use std::{boxed::Box, io, string::ToString, sync::Mutex, vec::Vec};

struct BoundStorage {
    generation: u32,
    storage: Box<dyn Storage>,
}

/// The legacy bounded payload, stored in a versioned, bundle-exportable environment table.
pub(super) struct EnvironmentCache {
    namespace: Namespace,
    capacity: usize,
    state: Mutex<Option<BoundStorage>>,
}

impl EnvironmentCache {
    pub(super) fn new(capacity: usize) -> Self {
        Self {
            namespace: Namespace::scoped("stack-autotune", "records-v1"),
            capacity,
            state: Mutex::new(None),
        }
    }

    pub(super) fn table(&self) -> &str { self.namespace.as_str() }

    fn access<T>(
        &self, generation: u32, action: impl FnOnce(&dyn Storage) -> io::Result<T>,
    ) -> io::Result<Option<T>> {
        let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
        if generation != environment::generation() { return Ok(None); }
        if state.as_ref().is_none_or(|state| state.generation != generation) {
            let storage = storage::open(self.namespace.as_str());
            // A switch during open must not label the new table as belonging to an old tune.
            if generation != environment::generation() { return Ok(None); }
            *state = Some(BoundStorage { generation, storage });
        }
        let bound = state.as_ref().expect("opened environment table");
        // The handle remains bound to that table even if another thread switches now.
        action(bound.storage.as_ref()).map(Some)
    }

    pub(super) fn load(&self, scoped: &str) -> io::Result<Option<Record>> {
        let (generation, key) = split_scoped_key(scoped)
            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "unscoped environment tuning key"))?;
        self.access(generation, |storage| {
            let Some(bytes) = storage.get(key.as_bytes()) else { return Ok(None); };
            let mut record = match Record::decode(&bytes) {
                Ok(record) => record,
                Err(error) => { storage.purge_key(key.as_bytes()); return Err(error); }
            };
            if record.key != key { storage.purge_key(key.as_bytes()); return Ok(None); }
            record.key = scoped.to_string();
            Ok(Some(record))
        }).map(Option::flatten)
    }

    pub(super) fn save(&self, record: &Record) -> io::Result<bool> {
        let (generation, key) = split_scoped_key(&record.key)
            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "unscoped environment tuning key"))?;
        let mut durable = record.clone();
        durable.key = key.to_string();
        let bytes = durable.encode()?;
        self.access(generation, |storage| {
            let accepted = match storage.insert(key.as_bytes(), Bytes::from_bytes_vec(bytes.clone()), Origin::Local) {
                Insertion::Stored => true,
                Insertion::Conflict(previous) => &previous[..] == bytes.as_slice(),
                Insertion::Failed(reason) => return Err(io::Error::other(reason)),
            };
            if accepted { self.prune(storage); }
            Ok(accepted)
        }).map(|accepted| accepted.unwrap_or(false))
    }

    pub(super) fn remove(&self, scoped: &str) -> io::Result<()> {
        let (generation, key) = split_scoped_key(scoped)
            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "unscoped environment tuning key"))?;
        self.access(generation, |storage| { storage.purge_key(key.as_bytes()); Ok(()) })?;
        Ok(())
    }

    fn prune(&self, storage: &dyn Storage) {
        let mut entries = Vec::new();
        storage.scan(&mut |key, value| {
            let created = Record::decode(&value).ok()
                .filter(|record| record.key.as_bytes() == key)
                .map(|record| record.created);
            entries.push((created, key.to_vec()));
        });
        entries.sort_unstable();
        let extra = entries.len().saturating_sub(self.capacity);
        // Never touch storage from inside its scan visitor: tables may share a lock.
        for (_, key) in entries.into_iter().take(extra) { storage.purge_key(&key); }
    }
}