chiralmap 0.1.2

Left-Right map using IndexMap
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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

//! Inner type used to access the memory writer / map

pub(crate) mod single_map;

#[cfg(feature = "metrics")]
use metrics::{Counter, Gauge, counter, gauge};

use left_right::{ReadHandle, WriteHandle};
use single_map::SingleMap;
use std::{
    hash::Hash,
    sync::mpsc::{Receiver, RecvTimeoutError},
    time::{Duration, Instant},
};
use tracing::{debug, error};

use super::{ChiralMapConfig, FlushableOp, Operation};

/// This inner struct is maintained unique referenced and cannot be cloned,
/// because it holds on the chiral writer.
/// Here is also where the write operations get forwarded.
pub(crate) struct ChiralMapInner<K, V>
where
    K: Hash + Eq + Clone,
    V: Clone,
{
    writer: WriteHandle<SingleMap<K, V>, Operation<K, V>>,
    reader: ReadHandle<SingleMap<K, V>>,
    rx: Receiver<FlushableOp<K, V>>,
    opcount: usize,
    auto_flush_count: usize,
    auto_flush_seconds: u64,

    #[cfg(feature = "metrics")]
    met_flushes: Counter,
    #[cfg(feature = "metrics")]
    met_opslog: Gauge,
}

impl<K, V> ChiralMapInner<K, V>
where
    K: Hash + Eq + Clone,
    V: Clone,
{
    #[allow(clippy::type_complexity)]
    pub(crate) fn new(config: &ChiralMapConfig, rx: Receiver<FlushableOp<K, V>>) -> Self {
        let inner_map = SingleMap::<K, V>::new(config);
        let (mut writer, reader) = left_right::new_from_empty(inner_map);
        // This will trigger the first sync and swap the two maps around twice,
        // entrering the situation when (even if empty) they are rotated with
        // the readers too, and the Absorb::sync_with would be already called.
        writer.publish();
        writer.publish();

        #[cfg(feature = "metrics")]
        let met_flushes = counter!("chiral_map_flushes", "id" => config.id.clone());
        #[cfg(feature = "metrics")]
        let met_opslog = gauge!("chiral_map_queued_ops", "id" => config.id.clone());

        Self {
            writer,
            reader,
            rx,
            opcount: 0,
            auto_flush_count: config.auto_flush_count,
            auto_flush_seconds: config.auto_flush_seconds,

            #[cfg(feature = "metrics")]
            met_flushes,
            #[cfg(feature = "metrics")]
            met_opslog,
        }
    }

    pub(crate) fn op(&mut self, op: Operation<K, V>) {
        self.writer.append(op);
        self.opcount += 1;

        #[cfg(feature = "metrics")]
        self.met_opslog.set(self.opcount as f64);

        if self.opcount >= self.auto_flush_count {
            debug!("Auto flush triggered by opscount");
            self.flush();
        }
    }

    pub(crate) fn flush(&mut self) {
        debug!("Flushing");
        self.writer.flush();
        self.opcount = 0;

        #[cfg(feature = "metrics")]
        self.met_opslog.set(self.opcount as f64);
        #[cfg(feature = "metrics")]
        self.met_flushes.increment(1);
    }

    pub(crate) fn reader(&self) -> ReadHandle<SingleMap<K, V>> {
        self.reader.clone()
    }

    pub(crate) fn receiver_loop(mut self)
    where
        K: Hash + Eq + Clone,
        V: Clone,
    {
        let dur = Duration::from_secs(self.auto_flush_seconds);
        let mut interval = Instant::now();
        loop {
            match self
                .rx
                .recv_timeout(Duration::from_secs(self.auto_flush_seconds))
            {
                Ok(op) => match op {
                    FlushableOp::Flush(rec) => {
                        self.flush();
                        if let Some(rec) = rec {
                            let _ = rec.send(());
                        }
                    }
                    FlushableOp::Op(op) => self.op(op),
                },
                Err(e) => match e {
                    RecvTimeoutError::Disconnected => {
                        error!("ChiralMap receiver disconnected");
                        return;
                    }
                    RecvTimeoutError::Timeout => {
                        self.flush();
                        debug!("Auto flush triggered by time(out)");
                    }
                },
            }
            if interval.elapsed() > dur {
                self.flush();
                debug!("Auto flush triggered by time");
                interval = Instant::now();
            }
        }
    }
}