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
pub mod rstorage {
use std::collections::HashMap;
use serde::Serialize;
use serde::de::DeserializeOwned;
use serde_derive::{Serialize, Deserialize};
use tokio::sync::{oneshot::channel, mpsc, RwLock};
use crate::storage::{start, Job};
#[derive(Serialize, Deserialize)]
pub enum RQuery<Document> {
Insert(Key, Document),
Remove(Key)
}
type Key = String;
pub struct RConcurrentStorage<Document> {
pub table: RwLock<HashMap<String, Document>>,
wal_chan: mpsc::Sender<Job>
}
impl<Document> RConcurrentStorage<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= RConcurrentStorage {
table: RwLock::new(HashMap::new()),
wal_chan: chan,
};
st.restore_storage().await;
return st
}
pub async fn insert_entry(&self, key: Key, doc: Document) -> Option<Document> {
let query = RQuery::Insert(key.clone(), doc.clone());
let _ = self.wal_chan.send(Job::WriteLog { data: bincode::serialize(&query).unwrap(), dst: None }).await;
let mut t = self.table.write().await;
t.insert(key, doc)
}
pub async fn remove_entry(&self, key: Key) {
let mut t = self.table.write().await;
match t.remove(&key) {
Some(_) => {
let query = RQuery::<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();
let mut t = self.table.write().await;
for qline in iter {
let query: RQuery<Document> = bincode::deserialize(&qline.unwrap()).unwrap();
match query {
RQuery::Insert(key, doc) => {
t.insert(key, doc);
}
RQuery::Remove(k) => {
t.remove(&k);
}
}
}
}
Err(_err) => {
println!("End");
return;
}
}
}
Err(_) => {
println!("114");
return;
}
}
}
Err(err) => {
println!("ERROR: {}", err.to_string());
return;
}
}
}
}
}
}