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
pub mod wstorage {
use serde::Serialize;
use serde::de::DeserializeOwned;
use serde_derive::{Serialize, Deserialize};
use tokio::sync::{oneshot::channel, mpsc};
use dashmap::DashMap;
use crate::storage::{start, Job};
#[derive(Serialize, Deserialize)]
pub enum WQuery<Document> {
Insert(Key, Document),
Remove(Key)
}
type Key = String;
pub struct WConcurrentStorage<Document> {
pub table: DashMap<String, Document>,
wal_chan: mpsc::Sender<Job>
}
impl<Document> WConcurrentStorage<Document>
where
Document: Serialize + DeserializeOwned + Clone
{
/// 1. check if exist Load entry
/// 2. check if not exist
/// -> create wal_chan for it
pub async fn open(table_name: String, total_segment_size: usize) -> Self {
let chan = start(".".to_owned(), table_name, total_segment_size);
let st= WConcurrentStorage {
table: DashMap::new(),
wal_chan: chan,
};
st.restore_storage().await;
return st
}
pub async fn insert_entry(&self, key: Key, doc: Document) -> Option<Document> {
let query = WQuery::Insert(key.clone(), doc.clone());
let _ = self.wal_chan.send(Job::WriteLog { data: bincode::serialize(&query).unwrap(), dst: None }).await;
self.table.insert(key, doc)
}
pub async fn remove_entry(&self, key: Key) {
match self.table.remove(&key) {
Some(_) => {
let query = WQuery::<Document>::Remove(key);
let _ = self.wal_chan.send(Job::WriteLog { data: bincode::serialize(&query).unwrap(), dst: None }).await;
}
None => {
}
}
}
pub async fn restore_storage(&self) {
let mut segment_index = 1;
loop {
let (sx, rx) = channel();
match self.wal_chan.send(Job::GetSegment { segment_index, dst: sx }).await {
Ok(_) => {
match rx.await {
Ok(res) => {
match res {
Ok(mut log) => {
segment_index += 1;
let iter = log.iter(..).unwrap();
for qline in iter {
let query: WQuery<Document> = bincode::deserialize(&qline.unwrap()).unwrap();
match query {
WQuery::Insert(key, doc) => {
self.table.insert(key, doc);
}
WQuery::Remove(k) => {
self.table.remove(&k);
}
}
}
}
Err(_err) => {
println!("End");
return;
}
}
}
Err(_) => {
println!("114");
return;
}
}
}
Err(err) => {
println!("ERROR: {}", err.to_string());
return;
}
}
}
}
}
}