sharedstate 0.5.0

Sync heavily read state across many servers
Documentation 
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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119

use std::sync::Arc;

use hotread::{HotRead, HotReadHandle, HotReadState};

use crate::state::determinstic_state::DeterministicState;

pub struct SharedState<D: DeterministicState> {
    inner: Arc<StateInner<D>>,
}

pub struct SharedStateReader<D: DeterministicState> {
    inner: Arc<StateInner<D>>,
}

impl<D: DeterministicState> Clone for SharedStateReader<D> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
        }
    }
}

pub struct SharedStateHandle<D: DeterministicState> {
    handle: HotReadHandle<HotState<D>>,
}

impl<D: DeterministicState> SharedState<D> {
    pub fn new(state: D) -> Self {
        SharedState {
            inner: Arc::new(StateInner {
                hot: HotRead::new(HotState { state }),
            }),
        }
    }

    pub fn create_reader(&self) -> SharedStateReader<D> {
        SharedStateReader {
            inner: self.inner.clone(),
        }
    }

    pub fn create_handle(&self) -> SharedStateHandle<D> {
        SharedStateHandle {
            handle: self.inner.hot.create_handle(),
        }
    }

    pub fn queue_updates<I>(&mut self, updates: I)
    where
        I: IntoIterator<Item = (u64, D::AuthorityAction)>,
    {
        self.inner.hot.queue_updates(
            updates
                .into_iter()
                .map(|(seq, action)| HotSharedStateUpdate::Authority { seq, action }),
        );
    }

    pub fn reset(&mut self, state: D) {
        self.inner.hot.queue_update(HotSharedStateUpdate::Reset(state));
    }

    pub fn maintain_state(&self) {
        self.inner.hot.maintain();
    }
}

impl<D: DeterministicState> SharedStateReader<D> {
    pub fn create_handle(&self) -> SharedStateHandle<D> {
        SharedStateHandle {
            handle: self.inner.hot.create_handle(),
        }
    }
}

impl<D: DeterministicState> SharedStateHandle<D> {
    pub fn read(&mut self) -> &D {
        &self.handle.current().state
    }

    pub fn quiescent(&mut self) {
        self.handle.quiescent();
    }
}

struct StateInner<D: DeterministicState> {
    hot: HotRead<HotState<D>>,
}

#[derive(Clone)]
struct HotState<D: DeterministicState> {
    state: D,
}

#[derive(Clone)]
enum HotSharedStateUpdate<D: DeterministicState> {
    Authority { seq: u64, action: D::AuthorityAction },
    Reset(D),
}

impl<D: DeterministicState> HotReadState for HotState<D> {
    type Action = HotSharedStateUpdate<D>;

    fn apply_update(&mut self, update: &Self::Action) {
        match update {
            HotSharedStateUpdate::Authority { seq, action } => {
                if self.state.accept_seq() != *seq {
                    panic!("Received authority ")
                }
                assert_eq!(self.state.accept_seq(), *seq);
                self.state.update(action);
                assert_eq!(self.state.accept_seq(), seq + 1);
            }
            HotSharedStateUpdate::Reset(state) => {
                self.state = state.clone();
            }
        }
    }
}