linkrs-sync 0.1.0

Synchronization primitives for Linkrs
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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

use dashmap::DashMap;
use std::time::Instant;

use super::trait_def::BatchBuffer;
use crate::types::IndexOperation;

type IndexKey = (u64, String, String);

#[derive(Debug, Default, Clone)]
pub struct BufferEntry {
    pub inserts: Vec<IndexOperation>,
    pub deletes: Vec<String>,
}

#[derive(Debug)]
pub struct OpBatchBuffer {
    buffers: DashMap<IndexKey, BufferEntry>,
    last_commit: DashMap<IndexKey, Instant>,
}

impl Default for OpBatchBuffer {
    fn default() -> Self {
        Self::new()
    }
}

impl OpBatchBuffer {
    pub fn new() -> Self {
        Self {
            buffers: DashMap::new(),
            last_commit: DashMap::new(),
        }
    }

    pub fn add_insert(&self, key: &IndexKey, operation: IndexOperation) {
        let mut entry = self.buffers.entry(key.clone()).or_default();
        entry.inserts.push(operation);
        self.last_commit
            .entry(key.clone())
            .or_insert_with(Instant::now);
    }

    pub fn add_delete(&self, key: &IndexKey, id: String) {
        let mut entry = self.buffers.entry(key.clone()).or_default();
        entry.deletes.push(id);
        self.last_commit
            .entry(key.clone())
            .or_insert_with(Instant::now);
    }

    pub fn drain_inserts(&self, key: &IndexKey) -> Vec<IndexOperation> {
        if let Some(mut entry) = self.buffers.get_mut(key) {
            std::mem::take(&mut entry.inserts)
        } else {
            Vec::new()
        }
    }

    pub fn drain_deletes(&self, key: &IndexKey) -> Vec<String> {
        if let Some(mut entry) = self.buffers.get_mut(key) {
            std::mem::take(&mut entry.deletes)
        } else {
            Vec::new()
        }
    }

    pub fn drain_all(&self, key: &IndexKey) -> BufferEntry {
        self.buffers.remove(key).map(|(_, v)| v).unwrap_or_default()
    }

    pub fn peek_entry(&self, key: &IndexKey) -> BufferEntry {
        self.buffers
            .get(key)
            .map(|entry| BufferEntry {
                inserts: entry.inserts.clone(),
                deletes: entry.deletes.clone(),
            })
            .unwrap_or_default()
    }

    pub fn re_enqueue(&self, key: &IndexKey, entry: BufferEntry) {
        if entry.inserts.is_empty() && entry.deletes.is_empty() {
            return;
        }
        let mut buffer = self.buffers.entry(key.clone()).or_default();
        buffer.inserts.extend(entry.inserts);
        buffer.deletes.extend(entry.deletes);
    }

    pub fn count(&self, key: &IndexKey) -> usize {
        self.buffers
            .get(key)
            .map(|e| e.inserts.len() + e.deletes.len())
            .unwrap_or(0)
    }

    pub fn insert_count(&self, key: &IndexKey) -> usize {
        self.buffers.get(key).map(|e| e.inserts.len()).unwrap_or(0)
    }

    pub fn delete_count(&self, key: &IndexKey) -> usize {
        self.buffers.get(key).map(|e| e.deletes.len()).unwrap_or(0)
    }

    pub fn is_timeout(&self, key: &IndexKey, timeout: std::time::Duration) -> bool {
        self.last_commit
            .get(key)
            .map(|last| last.elapsed() >= timeout)
            .unwrap_or(false)
    }

    pub fn update_commit_time(&self, key: &IndexKey) {
        self.last_commit.insert(key.clone(), Instant::now());
    }

    pub fn keys(&self) -> Vec<IndexKey> {
        self.buffers.iter().map(|e| e.key().clone()).collect()
    }

    pub fn total_count(&self) -> usize {
        self.buffers
            .iter()
            .map(|e| e.inserts.len() + e.deletes.len())
            .sum()
    }

    pub fn clear(&self) {
        self.buffers.clear();
        self.last_commit.clear();
    }

    pub fn remove(&self, key: &IndexKey) -> Option<BufferEntry> {
        self.buffers.remove(key).map(|(_, v)| v)
    }
}

impl BatchBuffer<IndexKey, IndexOperation> for OpBatchBuffer {
    fn add(&self, key: &IndexKey, value: IndexOperation) {
        self.add_insert(key, value);
    }

    fn drain(&self, key: &IndexKey) -> Vec<IndexOperation> {
        self.drain_inserts(key)
    }

    fn peek(&self, key: &IndexKey) -> Vec<IndexOperation> {
        self.peek_entry(key).inserts
    }

    fn count(&self, key: &IndexKey) -> usize {
        self.insert_count(key)
    }

    fn is_empty(&self, key: &IndexKey) -> bool {
        self.insert_count(key) == 0
    }

    fn keys(&self) -> Vec<IndexKey> {
        self.keys()
    }

    fn clear(&self) {
        self.clear();
    }

    fn total_count(&self) -> usize {
        self.total_count()
    }
}