liz/veilid
1
0
Fork 0
Code Pull Requests Packages Releases Activity

storage work

Browse Source
This commit is contained in:
John Smith
2023-04-06 20:21:45 -04:00
parent 7eded89b11
commit e46d64f648
7 changed files with 326 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
veilid-core/src/storage_manager/mod.rs
View File

@@ -1,6 +1,6 @@
mod record_store;
mod value_record;
mod record_store_limits;
mod value_record;
use record_store::*;
use record_store_limits::*;
use value_record::*;
@@ -58,7 +58,6 @@ impl StorageManager {
fn local_limits_from_config(config: VeilidConfig) -> RecordStoreLimits {
RecordStoreLimits {
record_cache_size: todo!(),
subkey_cache_size: todo!(),
max_records: None,
max_subkey_cache_memory_mb: Some(xxx),
@@ -68,11 +67,10 @@ impl StorageManager {
fn remote_limits_from_config(config: VeilidConfig) -> RecordStoreLimits {
RecordStoreLimits {
record_cache_size: todo!(),
subkey_cache_size: todo!(),
max_records: Some(xxx),
max_subkey_cache_memory_mb: Some(xxx),
max_disk_space_mb: Some(xxx)
max_disk_space_mb: Some(xxx),
}
}
@@ -84,7 +82,6 @@ impl StorageManager {
block_store: BlockStore,
rpc_processor: RPCProcessor,
) -> StorageManager {
StorageManager {
unlocked_inner: Arc::new(Self::new_unlocked_inner(
config,
@@ -94,7 +91,7 @@ impl StorageManager {
block_store,
rpc_processor,
)),
inner: Arc::new(Mutex::new(Self::new_inner()))
inner: Arc::new(Mutex::new(Self::new_inner())),
}
}
@@ -105,8 +102,16 @@ impl StorageManager {
let local_limits = Self::local_limits_from_config(config.clone());
let remote_limits = Self::remote_limits_from_config(config.clone());
inner.local_record_store = Some(RecordStore::new(self.unlocked_inner.table_store.clone(), "local", local_limits));
inner.remote_record_store = Some(RecordStore::new(self.unlocked_inner.table_store.clone(), "remote", remote_limits));
inner.local_record_store = Some(RecordStore::new(
self.unlocked_inner.table_store.clone(),
"local",
local_limits,
));
inner.remote_record_store = Some(RecordStore::new(
self.unlocked_inner.table_store.clone(),
"remote",
remote_limits,
));
Ok(())
}
@@ -120,15 +125,17 @@ impl StorageManager {
debug!("finished storage manager shutdown");
}
async fn add_value_record(&self, key: TypedKey, record: ValueRecord) -> EyreResult<()> {
async fn new_local_record(&self, key: TypedKey, record: ValueRecord) -> EyreResult<()> {
// add value record to record store
let mut inner = self.inner.lock();
inner.record_store.
let Some(local_record_store) = inner.local_record_store else {
apibail_generic!("not initialized");
};
local_record_store.new_record(key, record)
}
/// Creates a new DHT value with a specified crypto kind and schema
/// Returns the newly allocated DHT Key if successful.
pub async fn create_value(
pub async fn create_record(
&self,
kind: CryptoKind,
schema: &DHTSchema,
@@ -144,60 +151,50 @@ impl StorageManager {
let key = TypedKey::new(kind, keypair.key);
let secret = keypair.secret;
// Add value record
let record = ValueRecord::new(Some(secret), schema, safety_selection);
self.add_value_record(key, record)
// Add new local value record
let cur_ts = get_aligned_timestamp();
let record = ValueRecord::new(cur_ts, Some(secret), schema, safety_selection);
self.new_local_record(key, record)
.await
.map_err(VeilidAPIError::internal)?;
Ok(key)
}
/// Opens a DHT value at a specific key. Associates an owner secret if one is provided.
/// Returns the DHT key descriptor for the opened key if successful
/// Value may only be opened or created once. To re-open with a different routing context, first close the value.
pub async fn open_value(
pub async fn open_record(
key: TypedKey,
secret: Option<SecretKey>,
safety_selection: SafetySelection,
) -> Result<DHTDescriptor, VeilidAPIError> {
) -> Result<DHTRecordDescriptor, VeilidAPIError> {
unimplemented!();
}
/// Closes a DHT value at a specific key that was opened with create_value or open_value.
/// Closing a value allows you to re-open it with a different routing context
pub async fn close_value(key: TypedKey) -> Result<(), VeilidAPIError> {
pub async fn close_record(key: TypedKey) -> Result<(), VeilidAPIError> {
unimplemented!();
}
pub async fn delete_value(key: TypedKey) -> Result<(), VeilidAPIError> {
unimplemented!();
}
/// Gets the latest value of a subkey from the network
/// Returns the possibly-updated value data of the subkey
pub async fn get_value(
&self,
key: TypedKey,
subkey: ValueSubkey,
force_refresh: bool,
) -> Result<ValueData, VeilidAPIError> {
unimplemented!();
}
/// Pushes a changed subkey value to the network
/// Returns None if the value was successfully put
/// Returns Some(newer_value) if the value put was older than the one available on the network
pub async fn set_value(
&self,
key: TypedKey,
subkey: ValueSubkey,
value_data: ValueData,
) -> Result<Option<ValueData>, VeilidAPIError> {
unimplemented!();
}
/// Watches changes to an opened or created value
/// Changes to subkeys within the subkey range are returned via a ValueChanged callback
/// If the subkey range is empty, all subkey changes are considered
/// Expiration can be infinite to keep the watch for the maximum amount of time
/// Return value upon success is the amount of time allowed for the watch
pub async fn set_value(
&self,
key: TypedKey,
subkey: ValueSubkey,
data: Vec<u8>,
) -> Result<Option<ValueData>, VeilidAPIError> {
unimplemented!();
}
pub async fn watch_value(
&self,
key: TypedKey,
@@ -208,8 +205,6 @@ impl StorageManager {
unimplemented!();
}
/// Cancels a watch early
/// This is a convenience function that cancels watching all subkeys in a range
pub async fn cancel_watch_value(
&self,
key: TypedKey,

294
veilid-core/src/storage_manager/record_store.rs
View File

@@ -1,9 +1,15 @@
/// RecordStore
/// Keeps an LRU cache of dht keys and their associated subkey valuedata.
/// Instances of this store are used for 'local' (persistent) and 'remote' (ephemeral) dht key storage.
/// This store does not perform any validation on the schema, and all ValueRecordData passed in must have been previously validated.
/// Uses an in-memory store for the records, backed by the TableStore. Subkey data is LRU cached and rotated out by a limits policy,
/// and backed to the TableStore for persistence.
use super::*;
use hashlink::LruCache;
pub type RecordIndex = u32;
#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
struct RecordIndexKey {
pub key: TypedKey,
}
@@ -30,7 +36,7 @@ impl TryFrom<&[u8]> for RecordIndexKey {
}
}
#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
struct SubkeyCacheKey {
pub key: TypedKey,
pub subkey: ValueSubkey,
@@ -71,6 +77,9 @@ pub struct RecordStore {
subkey_table: Option<TableDB>,
record_index: LruCache<RecordIndexKey, ValueRecord>,
subkey_cache: LruCache<SubkeyCacheKey, ValueRecordData>,
dead_records: Vec<(RecordIndexKey, ValueRecord)>,
changed_records: HashSet<(RecordIndexKey, Timestamp)>,
}
impl RecordStore {
@@ -84,6 +93,8 @@ impl RecordStore {
subkey_table: None,
record_index: LruCache::new(limits.max_records.unwrap_or(usize::MAX)),
subkey_cache: LruCache::new(subkey_cache_size),
dead_records: Vec::new(),
changed_records: HashSet::new(),
}
}
@@ -118,145 +129,163 @@ impl RecordStore {
})
}
// Delete dead keys
if !dead_records.empty() {
let rt_xact = record_table.transact();
let st_xact = subkey_table.transact();
for (k, v) in dead_records {
// Delete record
rt_xact.delete(0, &k.bytes());
// Delete subkeys
let subkey_count = v.subkey_count();
for sk in 0..subkey_count {
let sck = SubkeyCacheKey {
key: k.key,
subkey: sk,
};
st_xact.delete(0, &sck.bytes())?;
}
}
rt_xact.commit().await?;
st_xact.commit().await?;
}
self.record_table = Some(record_table);
self.subkey_table = Some(record_table);
Ok(())
}
fix up new record
fn add_dead_record(&mut self, key: RecordIndexKey, record: ValueRecord) {
self.dead_records.push((key, record));
}
pub fn new_record(&mut self, key: TypedKey, record: ValueRecord) -> EyreResult<()> {
fn mark_record_changed(&mut self, key: RecordIndexKey) {
let cur_ts = get_aligned_timestamp();
self.changed_records.insert((key, cur_ts));
}
async fn purge_dead_records(&mut self) {
// Delete dead keys
if self.dead_records.empty() {
return;
}
let rt_xact = record_table.transact();
let st_xact = subkey_table.transact();
let mut dead_records = mem::take(&mut self.dead_records);
for (k, v) in dead_records {
// Delete record
rt_xact.delete(0, &k.bytes());
// Delete subkeys
let subkey_count = v.subkey_count();
for sk in 0..subkey_count {
// From table
let sck = SubkeyCacheKey {
key: k.key,
subkey: sk,
};
st_xact.delete(0, &sck.bytes())?;
// From cache
self.subkey_cache.remove(&sck);
}
}
if let Err(e) = rt_xact.commit().await {
log_stor!(error "failed to commit record table transaction: {}", e);
}
if let Err(e) = st_xact.commit().await {
log_stor!(error "failed to commit subkey table transaction: {}", e);
}
}
async fn flush_records(&mut self) {
// touch records
if self.changed_records.empty() {
return;
}
let rt_xact = record_table.transact();
let st_xact = subkey_table.transact();
let mut changed_records = mem::take(&mut self.changed_records);
for (rik, ts) in changed_records {
// Flush record and changed subkeys
}
if let Err(e) = rt_xact.commit().await {
log_stor!(error "failed to commit record table transaction: {}", e);
}
if let Err(e) = st_xact.commit().await {
log_stor!(error "failed to commit subkey table transaction: {}", e);
}
}
pub async fn tick(&mut self, last_ts: Timestamp, cur_ts: Timestamp) {
self.flush_records().await;
self.purge_dead_records().await;
}
pub fn new_record(&mut self, key: TypedKey, record: ValueRecord) -> Result<(), VeilidAPIError> {
if self.with_record(key, |_| {})?.is_some() {
bail!("record already exists");
apibail_generic!("record already exists");
}
// Get record table
let Some(record_table) = self.record_table.clone() else {
bail!("record store not initialized");
apibail_internal!("record store not initialized");
};
// Save to record table
record_table.store_rkyv(0, &key, &r).await?;
record_table
.store_rkyv(0, &key, &r)
.await
.map_err(VeilidAPIError::internal)?;
// Cache it
self.record_cache.insert(key, value, |_| {});
self.record_cache.insert(key, value, |k, v| {
self.add_dead_record(k, v);
});
Ok(())
}
pub fn with_record<R, F>(&mut self, key: TypedKey, f: F) -> EyreResult<Option<R>>
pub fn with_record<R, F>(&mut self, key: TypedKey, f: F) -> Option<R>
where
F: FnOnce(&mut RecordStore, TypedKey, &ValueRecord) -> R,
F: FnOnce(&ValueRecord) -> R,
{
// Get record table
let Some(record_table) = self.record_table.clone() else {
bail!("record store not initialized");
};
// If record exists in cache, use that
// Get record from index
let rck = RecordIndexKey { key };
if let Some(r) = self.record_cache.get(&rck) {
if let Some(r) = self.record_index.get_mut(&rck) {
// Touch
r.touch(get_aligned_timestamp());
self.mark_record_changed(&rck);
// Callback
return Ok(Some(f(self, key, r)));
return Some(f(key, r));
}
// If not in cache, try to pull from table store
let k = rck.bytes();
if let Some(r) = record_table.load_rkyv(0, &k)? {
// Callback
let out = f(self, key, &r);
// Add to cache, do nothing with lru out
self.record_cache.insert(rck, r, |_| {});
return Ok(Some(out));
};
return Ok(None);
None
}
pub fn with_record_mut<R, F>(&mut self, key: TypedKey, f: F) -> EyreResult<Option<R>>
where
F: FnOnce(&mut RecordStore, TypedKey, &mut ValueRecord) -> R,
{
// Get record table
let Some(record_table) = self.record_table.clone() else {
bail!("record store not initialized");
};
// If record exists in cache, use that
let rck = RecordIndexKey { key };
if let Some(r) = self.record_cache.get_mut(&rck) {
// Callback
return Ok(Some(f(self, key, r)));
}
// If not in cache, try to pull from table store
let k = rck.bytes();
if let Some(r) = record_table.load_rkyv(0, &k)? {
// Callback
let out = f(self, key, &mut r);
// Save changes back to record table
record_table.store_rkyv(0, &k, &r).await?;
// Add to cache, do nothing with lru out
self.record_cache.insert(rck, r, |_| {});
return Ok(Some(out));
};
Ok(None)
}
pub fn with_subkey<R, F>(
pub fn get_subkey<R, F>(
&mut self,
key: TypedKey,
subkey: ValueSubkey,
f: F,
) -> EyreResult<Option<R>>
where
F: FnOnce(&mut RecordStore, TypedKey, ValueSubkey, &ValueRecordData) -> R,
{
) -> Result<Option<ValueRecordData>, VeilidAPIError> {
// record from index
let rck = RecordIndexKey { key };
let Some(r) = self.record_index.get_mut(&rck) else {
apibail_invalid_argument!("no record at this key", "key", key);
};
// Touch
r.touch(get_aligned_timestamp());
self.mark_record_changed(&rck);
// Check if the subkey is in range
if subkey >= r.subkey_count() {
apibail_invalid_argument!("subkey out of range", "subkey", subkey);
}
// Get subkey table
let Some(subkey_table) = self.subkey_table.clone() else {
bail!("record store not initialized");
apibail_internal!("record store not initialized");
};
// If subkey exists in subkey cache, use that
let skck = SubkeyCacheKey { key, subkey };
if let Some(rd) = self.subkey_cache.get(&skck) {
// Callback
return Ok(Some(f(self, key, subkey, rd)));
if let Some(rd) = self.subkey_cache.get_mut(&skck) {
let out = rd.clone();
return Ok(Some(out));
}
// If not in cache, try to pull from table store
let k = skck.bytes();
if let Some(rd) = subkey_table.load_rkyv(0, &k)? {
// Callback
let out = f(self, key, subkey, &rd);
if let Some(rd) = subkey_table
.load_rkyv::<ValueRecordData>(0, &k)
.map_err(VeilidAPIError::internal)?
{
let out = rd.clone();
// Add to cache, do nothing with lru out
self.subkey_cache.insert(skck, r, |_| {});
self.subkey_cache.insert(skck, rd, |_| {});
return Ok(Some(out));
};
@@ -264,41 +293,52 @@ impl RecordStore {
return Ok(None);
}
pub fn with_subkey_mut<R, F>(
pub fn set_subkey<R, F>(
&mut self,
key: TypedKey,
subkey: ValueSubkey,
f: F,
) -> EyreResult<Option<R>>
where
F: FnOnce(&mut RecordStore, TypedKey, ValueSubkey, &mut ValueRecord) -> R,
{
// Get record table
let Some(subkey_table) = self.subkey_table.clone() else {
bail!("record store not initialized");
data: ValueRecordData,
) -> Result<(), VeilidAPIError> {
// Get record from index
let rck = RecordIndexKey { key };
let Some(r) = self.record_index.get_mut(&rck) else {
apibail_invalid_argument!("no record at this key", "key", key);
};
// If subkey exists in cache, use that
let skck = SubkeyCacheKey { key, subkey };
if let Some(rd) = self.subkey_cache.get_mut(&skck) {
// Callback
return Ok(Some(f(self, key, subkey, rd)));
// Touch
r.touch(get_aligned_timestamp());
self.mark_record_changed(&rck);
// Check if the subkey is in range
if subkey >= r.subkey_count() {
apibail_invalid_argument!("subkey out of range", "subkey", subkey);
}
// If not in cache, try to pull from table store
let k = skck.bytes();
if let Some(rd) = subkey_table.load_rkyv(0, &k)? {
// Callback
let out = f(self, key, subkey, &mut rd);
// Save changes back to record table
subkey_table.store_rkyv(0, &k, &rd).await?;
// Add to cache, do nothing with lru out
self.subkey_cache.insert(key, r, |_| {});
return Ok(Some(out));
// Get subkey table
let Some(subkey_table) = self.subkey_table.clone() else {
apibail_internal!("record store not initialized");
};
Ok(None)
// Write to subkey cache
let skck = SubkeyCacheKey { key, subkey };
if let Some(rd) = self.subkey_cache.insert(skck, data, |_, _| {}) {
return Ok(Some(out));
}
xxx do we flush this now or queue it?
// Write subkey
// let k = skck.bytes();
// if let Some(rd) = subkey_table.load_rkyv::<ValueRecordData>(0, &k)? {
// let out = rd.data.clone();
// // Add to cache, do nothing with lru out
// self.subkey_cache.insert(skck, rd, |_| {});
// return Ok(Some(out));
// };
return Ok(None);
}
}

4
veilid-core/src/storage_manager/value_record.rs
View File

@@ -14,8 +14,8 @@ use super::*;
)]
#[archive_attr(repr(C), derive(CheckBytes))]
pub struct ValueRecordData {
data: ValueData,
signature: Signature,
pub data: ValueData,
pub signature: Signature,
}
#[derive(