Deuxfleurs
/
aerogramme
15
0
Fork 1
Code Issues 33 Pull Requests 1 Projects 11 Releases 4 Wiki Activity

implemented an in memory storage

This commit is contained in:
Quentin 2023-12-19 19:02:22 +01:00
parent 3d41f40dc8
commit c75f2d91ff
Signed by: quentin
GPG Key ID: E9602264D639FF68
5 changed files with 201 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/mail/incoming.rs
View File

@ -386,7 +386,7 @@ async fn k2v_lock_loop_internal(
_ => None,
};
if let Some(ct) = release {
match storage.row_rm(&storage::Selector::Single(&ct)).await {
match storage.row_rm_single(&ct).await {
Err(e) => warn!("Unable to release lock {:?}: {}", ct, e),
Ok(_) => (),
};

7
src/mail/mailbox.rs
View File

@ -361,7 +361,12 @@ impl MailboxInternal {
async {
// Delete mail meta from K2V
let sk = ident.to_string();
self.storage.row_rm(&Selector::Single(&RowRef::new(&self.mail_path, &sk))).await?;
let res = self.storage
.row_fetch(&storage::Selector::Single(&RowRef::new(&self.mail_path, &sk)))
.await?;
if let Some(row_val) = res.into_iter().next() {
self.storage.row_rm_single(&row_val.row_ref).await?;
}
Ok::<_, anyhow::Error>(())
}
)?;

8
src/storage/garage.rs
View File

@ -56,14 +56,18 @@ impl IStore for GarageStore {
unimplemented!();
}
async fn row_rm_single(&self, entry: &RowRef) -> Result<(), StorageError> {
unimplemented!();
}
async fn blob_fetch(&self, blob_ref: &BlobRef) -> Result<BlobVal, StorageError> {
unimplemented!();
}
async fn blob_insert(&self, blob_val: &BlobVal) -> Result<BlobVal, StorageError> {
async fn blob_insert(&self, blob_val: &BlobVal) -> Result<(), StorageError> {
unimplemented!();
}
async fn blob_copy(&self, src: &BlobRef, dst: &BlobRef) -> Result<BlobVal, StorageError> {
async fn blob_copy(&self, src: &BlobRef, dst: &BlobRef) -> Result<(), StorageError> {
unimplemented!();
}

240
src/storage/in_memory.rs
View File

@ -1,19 +1,67 @@
use crate::storage::*;
use std::collections::{HashMap, BTreeMap};
use std::ops::Bound::{Included, Unbounded, Excluded};
use std::ops::Bound::{Included, Unbounded, Excluded, self};
use std::sync::{Arc, RwLock};
use tokio::sync::Notify;
/// This implementation is very inneficient, and not completely correct
/// Indeed, when the connector is dropped, the memory is freed.
/// It means that when a user disconnects, its data are lost.
/// It's intended only for basic debugging, do not use it for advanced tests...
pub type ArcRow = Arc<RwLock<HashMap<String, BTreeMap<String, Vec<u8>>>>>;
pub type ArcBlob = Arc<RwLock<HashMap<String, Vec<u8>>>>;
#[derive(Debug, Clone)]
enum InternalData {
Tombstone,
Value(Vec<u8>),
}
impl InternalData {
fn to_alternative(&self) -> Alternative {
match self {
Self::Tombstone => Alternative::Tombstone,
Self::Value(x) => Alternative::Value(x.clone()),
}
}
}
#[derive(Debug, Default)]
struct InternalRowVal {
data: Vec<InternalData>,
version: u64,
change: Arc<Notify>,
}
impl InternalRowVal {
fn concurrent_values(&self) -> Vec<Alternative> {
self.data.iter().map(InternalData::to_alternative).collect()
}
fn to_row_val(&self, row_ref: RowRef) -> RowVal {
RowVal{
row_ref: row_ref.with_causality(self.version.to_string()),
value: self.concurrent_values(),
}
}
}
#[derive(Debug, Default, Clone)]
struct InternalBlobVal {
data: Vec<u8>,
metadata: HashMap<String, String>,
}
impl InternalBlobVal {
fn to_blob_val(&self, bref: &BlobRef) -> BlobVal {
BlobVal {
blob_ref: bref.clone(),
meta: self.metadata.clone(),
value: self.data.clone(),
}
}
}
type ArcRow = Arc<RwLock<HashMap<String, BTreeMap<String, InternalRowVal>>>>;
type ArcBlob = Arc<RwLock<BTreeMap<String, InternalBlobVal>>>;
#[derive(Clone, Debug)]
pub struct MemBuilder {
user: String,
unicity: Vec<u8>,
row: ArcRow,
blob: ArcBlob,
@ -25,10 +73,9 @@ impl MemBuilder {
unicity.extend_from_slice(file!().as_bytes());
unicity.extend_from_slice(user.as_bytes());
Arc::new(Self {
user: user.to_string(),
unicity,
row: Arc::new(RwLock::new(HashMap::new())),
blob: Arc::new(RwLock::new(HashMap::new())),
blob: Arc::new(RwLock::new(BTreeMap::new())),
})
}
}
@ -51,105 +98,180 @@ pub struct MemStore {
blob: ArcBlob,
}
impl MemStore {
fn inner_fetch(&self, row_ref: &RowRef) -> Result<Vec<u8>, StorageError> {
Ok(self.row
.read()
.or(Err(StorageError::Internal))?
.get(&row_ref.uid.shard)
.ok_or(StorageError::NotFound)?
.get(&row_ref.uid.sort)
.ok_or(StorageError::NotFound)?
.clone())
fn prefix_last_bound(prefix: &str) -> Bound<String> {
let mut sort_end = prefix.to_string();
match sort_end.pop() {
None => Unbounded,
Some(ch) => {
let nc = char::from_u32(ch as u32 + 1).unwrap();
sort_end.push(nc);
Excluded(sort_end)
}
}
}
#[async_trait]
impl IStore for MemStore {
async fn row_fetch<'a>(&self, select: &Selector<'a>) -> Result<Vec<RowVal>, StorageError> {
let store = self.row.read().or(Err(StorageError::Internal))?;
match select {
Selector::Range { shard, sort_begin, sort_end } => {
Ok(self.row
.read()
.or(Err(StorageError::Internal))?
Ok(store
.get(*shard)
.ok_or(StorageError::NotFound)?
.range((Included(sort_begin.to_string()), Excluded(sort_end.to_string())))
.map(|(k, v)| RowVal {
row_ref: RowRef { uid: RowUid { shard: shard.to_string(), sort: k.to_string() }, causality: Some("c".to_string()) },
value: vec![Alternative::Value(v.clone())],
})
.map(|(k, v)| v.to_row_val(RowRef::new(shard, k)))
.collect::<Vec<_>>())
},
Selector::List(rlist) => {
let mut acc = vec![];
for row_ref in rlist {
let bytes = self.inner_fetch(row_ref)?;
let row_val = RowVal {
row_ref: row_ref.clone(),
value: vec![Alternative::Value(bytes)]
};
acc.push(row_val);
let intval = store
.get(&row_ref.uid.shard)
.ok_or(StorageError::NotFound)?
.get(&row_ref.uid.sort)
.ok_or(StorageError::NotFound)?;
acc.push(intval.to_row_val(row_ref.clone()));
}
Ok(acc)
},
Selector::Prefix { shard, sort_prefix } => {
let mut sort_end = sort_prefix.to_string();
let last_bound = match sort_end.pop() {
None => Unbounded,
Some(ch) => {
let nc = char::from_u32(ch as u32 + 1).unwrap();
sort_end.push(nc);
Excluded(sort_end)
}
};
let last_bound = prefix_last_bound(sort_prefix);
Ok(self.row
.read()
.or(Err(StorageError::Internal))?
Ok(store
.get(*shard)
.ok_or(StorageError::NotFound)?
.range((Included(sort_prefix.to_string()), last_bound))
.map(|(k, v)| RowVal {
row_ref: RowRef { uid: RowUid { shard: shard.to_string(), sort: k.to_string() }, causality: Some("c".to_string()) },
value: vec![Alternative::Value(v.clone())],
})
.map(|(k, v)| v.to_row_val(RowRef::new(shard, k)))
.collect::<Vec<_>>())
},
Selector::Single(row_ref) => {
let bytes = self.inner_fetch(row_ref)?;
Ok(vec![RowVal{ row_ref: (*row_ref).clone(), value: vec![Alternative::Value(bytes)]}])
let intval = store
.get(&row_ref.uid.shard)
.ok_or(StorageError::NotFound)?
.get(&row_ref.uid.sort)
.ok_or(StorageError::NotFound)?;
Ok(vec![intval.to_row_val((*row_ref).clone())])
}
}
}
async fn row_rm_single(&self, entry: &RowRef) -> Result<(), StorageError> {
let mut store = self.row.write().or(Err(StorageError::Internal))?;
let shard = &entry.uid.shard;
let sort = &entry.uid.sort;
let cauz = match entry.causality.as_ref().map(|v| v.parse::<u64>()) {
Some(Ok(v)) => v,
_ => 0,
};
let bt = store.entry(shard.to_string()).or_default();
let intval = bt.entry(sort.to_string()).or_default();
if cauz == intval.version {
intval.data.clear();
}
intval.data.push(InternalData::Tombstone);
intval.version += 1;
intval.change.notify_waiters();
Ok(())
}
async fn row_rm<'a>(&self, select: &Selector<'a>) -> Result<(), StorageError> {
unimplemented!();
//@FIXME not efficient at all...
let values = self.row_fetch(select).await?;
for v in values.into_iter() {
self.row_rm_single(&v.row_ref).await?;
}
Ok(())
}
async fn row_insert(&self, values: Vec<RowVal>) -> Result<(), StorageError> {
unimplemented!();
let mut store = self.row.write().or(Err(StorageError::Internal))?;
for v in values.into_iter() {
let shard = v.row_ref.uid.shard;
let sort = v.row_ref.uid.sort;
let val = match v.value.into_iter().next() {
Some(Alternative::Value(x)) => x,
_ => vec![],
};
let cauz = match v.row_ref.causality.map(|v| v.parse::<u64>()) {
Some(Ok(v)) => v,
_ => 0,
};
let bt = store.entry(shard).or_default();
let intval = bt.entry(sort).or_default();
if cauz == intval.version {
intval.data.clear();
}
intval.data.push(InternalData::Value(val));
intval.version += 1;
intval.change.notify_waiters();
}
Ok(())
}
async fn row_poll(&self, value: &RowRef) -> Result<RowVal, StorageError> {
unimplemented!();
let shard = &value.uid.shard;
let sort = &value.uid.sort;
let cauz = match value.causality.as_ref().map(|v| v.parse::<u64>()) {
Some(Ok(v)) => v,
_ => 0,
};
let notify_me = {
let store = self.row.read().or(Err(StorageError::Internal))?;
let intval = store
.get(shard)
.ok_or(StorageError::NotFound)?
.get(sort)
.ok_or(StorageError::NotFound)?;
if intval.version != cauz {
return Ok(intval.to_row_val(value.clone()));
}
intval.change.clone()
};
notify_me.notified().await;
let res = self.row_fetch(&Selector::Single(value)).await?;
res.into_iter().next().ok_or(StorageError::NotFound)
}
async fn blob_fetch(&self, blob_ref: &BlobRef) -> Result<BlobVal, StorageError> {
unimplemented!();
let store = self.blob.read().or(Err(StorageError::Internal))?;
store.get(&blob_ref.0).ok_or(StorageError::NotFound).map(|v| v.to_blob_val(blob_ref))
}
async fn blob_insert(&self, blob_val: &BlobVal) -> Result<BlobVal, StorageError> {
unimplemented!();
async fn blob_insert(&self, blob_val: &BlobVal) -> Result<(), StorageError> {
let mut store = self.blob.write().or(Err(StorageError::Internal))?;
let entry = store.entry(blob_val.blob_ref.0.clone()).or_default();
entry.data = blob_val.value.clone();
entry.metadata = blob_val.meta.clone();
Ok(())
}
async fn blob_copy(&self, src: &BlobRef, dst: &BlobRef) -> Result<BlobVal, StorageError> {
unimplemented!();
async fn blob_copy(&self, src: &BlobRef, dst: &BlobRef) -> Result<(), StorageError> {
let mut store = self.blob.write().or(Err(StorageError::Internal))?;
let blob_src = store.entry(src.0.clone()).or_default().clone();
store.insert(dst.0.clone(), blob_src);
Ok(())
}
async fn blob_list(&self, prefix: &str) -> Result<Vec<BlobRef>, StorageError> {
unimplemented!();
let store = self.blob.read().or(Err(StorageError::Internal))?;
let last_bound = prefix_last_bound(prefix);
let blist = store.range((Included(prefix.to_string()), last_bound)).map(|(k, _)| BlobRef(k.to_string())).collect::<Vec<_>>();
Ok(blist)
}
async fn blob_rm(&self, blob_ref: &BlobRef) -> Result<(), StorageError> {
unimplemented!();
let mut store = self.blob.write().or(Err(StorageError::Internal))?;
store.remove(&blob_ref.0);
Ok(())
}
}

9
src/storage/mod.rs
View File

@ -61,6 +61,10 @@ impl RowRef {
causality: None,
}
}
pub fn with_causality(mut self, causality: String) -> Self {
self.causality = Some(causality);
self
}
}
#[derive(Debug, Clone)]
@ -118,12 +122,13 @@ pub enum Selector<'a> {
pub trait IStore {
async fn row_fetch<'a>(&self, select: &Selector<'a>) -> Result<Vec<RowVal>, StorageError>;
async fn row_rm<'a>(&self, select: &Selector<'a>) -> Result<(), StorageError>;
async fn row_rm_single(&self, entry: &RowRef) -> Result<(), StorageError>;
async fn row_insert(&self, values: Vec<RowVal>) -> Result<(), StorageError>;
async fn row_poll(&self, value: &RowRef) -> Result<RowVal, StorageError>;
async fn blob_fetch(&self, blob_ref: &BlobRef) -> Result<BlobVal, StorageError>;
async fn blob_insert(&self, blob_val: &BlobVal) -> Result<BlobVal, StorageError>;
async fn blob_copy(&self, src: &BlobRef, dst: &BlobRef) -> Result<BlobVal, StorageError>;
async fn blob_insert(&self, blob_val: &BlobVal) -> Result<(), StorageError>;
async fn blob_copy(&self, src: &BlobRef, dst: &BlobRef) -> Result<(), StorageError>;
async fn blob_list(&self, prefix: &str) -> Result<Vec<BlobRef>, StorageError>;
async fn blob_rm(&self, blob_ref: &BlobRef) -> Result<(), StorageError>;
}