@@ -1,8 +1,8 @@
|
||||
use crate::lsm_storage::MiniLsm;
|
||||
use crate::lsm_storage::{LsmStorageInner, MiniLsm};
|
||||
|
||||
impl MiniLsm {
|
||||
impl LsmStorageInner {
|
||||
pub fn dump_structure(&self) {
|
||||
let snapshot = self.inner.state.read();
|
||||
let snapshot = self.state.read();
|
||||
if !snapshot.l0_sstables.is_empty() {
|
||||
println!(
|
||||
"L0 ({}): {:?}",
|
||||
@@ -15,3 +15,9 @@ impl MiniLsm {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl MiniLsm {
|
||||
pub fn dump_structure(&self) {
|
||||
self.inner.dump_structure()
|
||||
}
|
||||
}
|
||||
|
@@ -3,8 +3,6 @@
|
||||
|
||||
use anyhow::Result;
|
||||
|
||||
use crate::key::KeySlice;
|
||||
|
||||
use super::StorageIterator;
|
||||
|
||||
/// Merges two iterators of different types into one. If the two iterators have the same key, only
|
||||
@@ -16,8 +14,8 @@ pub struct TwoMergeIterator<A: StorageIterator, B: StorageIterator> {
|
||||
}
|
||||
|
||||
impl<
|
||||
A: 'static + for<'a> StorageIterator<KeyType<'a> = KeySlice<'a>>,
|
||||
B: 'static + for<'a> StorageIterator<KeyType<'a> = KeySlice<'a>>,
|
||||
A: 'static + StorageIterator,
|
||||
B: 'static + for<'a> StorageIterator<KeyType<'a> = A::KeyType<'a>>,
|
||||
> TwoMergeIterator<A, B>
|
||||
{
|
||||
pub fn create(a: A, b: B) -> Result<Self> {
|
||||
@@ -26,13 +24,13 @@ impl<
|
||||
}
|
||||
|
||||
impl<
|
||||
A: 'static + for<'a> StorageIterator<KeyType<'a> = KeySlice<'a>>,
|
||||
B: 'static + for<'a> StorageIterator<KeyType<'a> = KeySlice<'a>>,
|
||||
A: 'static + StorageIterator,
|
||||
B: 'static + for<'a> StorageIterator<KeyType<'a> = A::KeyType<'a>>,
|
||||
> StorageIterator for TwoMergeIterator<A, B>
|
||||
{
|
||||
type KeyType<'a> = KeySlice<'a>;
|
||||
type KeyType<'a> = A::KeyType<'a>;
|
||||
|
||||
fn key(&self) -> KeySlice {
|
||||
fn key(&self) -> Self::KeyType<'_> {
|
||||
unimplemented!()
|
||||
}
|
||||
|
||||
|
@@ -7,6 +7,7 @@ pub mod lsm_iterator;
|
||||
pub mod lsm_storage;
|
||||
pub mod manifest;
|
||||
pub mod mem_table;
|
||||
pub mod mvcc;
|
||||
pub mod table;
|
||||
pub mod wal;
|
||||
|
||||
|
@@ -18,6 +18,7 @@ use crate::compact::{
|
||||
use crate::lsm_iterator::{FusedIterator, LsmIterator};
|
||||
use crate::manifest::Manifest;
|
||||
use crate::mem_table::MemTable;
|
||||
use crate::mvcc::LsmMvccInner;
|
||||
use crate::table::SsTable;
|
||||
|
||||
pub type BlockCache = moka::sync::Cache<(usize, usize), Arc<Block>>;
|
||||
@@ -122,6 +123,7 @@ pub(crate) struct LsmStorageInner {
|
||||
pub(crate) options: Arc<LsmStorageOptions>,
|
||||
pub(crate) compaction_controller: CompactionController,
|
||||
pub(crate) manifest: Option<Manifest>,
|
||||
pub(crate) mvcc: Option<LsmMvccInner>,
|
||||
}
|
||||
|
||||
/// A thin wrapper for `LsmStorageInner` and the user interface for MiniLSM.
|
||||
@@ -249,6 +251,7 @@ impl LsmStorageInner {
|
||||
compaction_controller,
|
||||
manifest: None,
|
||||
options: options.into(),
|
||||
mvcc: None,
|
||||
};
|
||||
|
||||
Ok(storage)
|
||||
|
58
mini-lsm-starter/src/mvcc.rs
Normal file
58
mini-lsm-starter/src/mvcc.rs
Normal file
@@ -0,0 +1,58 @@
|
||||
#![allow(unused_variables)] // TODO(you): remove this lint after implementing this mod
|
||||
#![allow(dead_code)] // TODO(you): remove this lint after implementing this mod
|
||||
|
||||
pub mod txn;
|
||||
mod watermark;
|
||||
|
||||
use std::{
|
||||
collections::{BTreeMap, HashSet},
|
||||
sync::Arc,
|
||||
};
|
||||
|
||||
use parking_lot::Mutex;
|
||||
|
||||
use crate::lsm_storage::LsmStorageInner;
|
||||
|
||||
use self::{txn::Transaction, watermark::Watermark};
|
||||
|
||||
pub(crate) struct CommittedTxnData {
|
||||
pub(crate) key_hashes: HashSet<u32>,
|
||||
#[allow(dead_code)]
|
||||
pub(crate) read_ts: u64,
|
||||
#[allow(dead_code)]
|
||||
pub(crate) commit_ts: u64,
|
||||
}
|
||||
|
||||
pub(crate) struct LsmMvccInner {
|
||||
pub(crate) write_lock: Mutex<()>,
|
||||
pub(crate) ts: Arc<Mutex<(u64, Watermark)>>,
|
||||
pub(crate) committed_txns: Arc<Mutex<BTreeMap<u64, CommittedTxnData>>>,
|
||||
}
|
||||
|
||||
impl LsmMvccInner {
|
||||
pub fn new(initial_ts: u64) -> Self {
|
||||
Self {
|
||||
write_lock: Mutex::new(()),
|
||||
ts: Arc::new(Mutex::new((initial_ts, Watermark::new()))),
|
||||
committed_txns: Arc::new(Mutex::new(BTreeMap::new())),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn latest_commit_ts(&self) -> u64 {
|
||||
self.ts.lock().0
|
||||
}
|
||||
|
||||
pub fn update_commit_ts(&self, ts: u64) {
|
||||
self.ts.lock().0 = ts;
|
||||
}
|
||||
|
||||
/// All ts (strictly) below this ts can be garbage collected.
|
||||
pub fn watermark(&self) -> u64 {
|
||||
let ts = self.ts.lock();
|
||||
ts.1.watermark().unwrap_or(ts.0)
|
||||
}
|
||||
|
||||
pub fn new_txn(&self, inner: Arc<LsmStorageInner>, serializable: bool) -> Arc<Transaction> {
|
||||
unimplemented!()
|
||||
}
|
||||
}
|
128
mini-lsm-starter/src/mvcc/txn.rs
Normal file
128
mini-lsm-starter/src/mvcc/txn.rs
Normal file
@@ -0,0 +1,128 @@
|
||||
#![allow(unused_variables)] // TODO(you): remove this lint after implementing this mod
|
||||
#![allow(dead_code)] // TODO(you): remove this lint after implementing this mod
|
||||
|
||||
use std::{
|
||||
collections::HashSet,
|
||||
ops::Bound,
|
||||
sync::{atomic::AtomicBool, Arc},
|
||||
};
|
||||
|
||||
use anyhow::Result;
|
||||
use bytes::Bytes;
|
||||
use crossbeam_skiplist::SkipMap;
|
||||
use ouroboros::self_referencing;
|
||||
use parking_lot::Mutex;
|
||||
|
||||
use crate::{
|
||||
iterators::{two_merge_iterator::TwoMergeIterator, StorageIterator},
|
||||
lsm_iterator::{FusedIterator, LsmIterator},
|
||||
lsm_storage::LsmStorageInner,
|
||||
};
|
||||
|
||||
pub struct Transaction {
|
||||
pub(crate) read_ts: u64,
|
||||
pub(crate) inner: Arc<LsmStorageInner>,
|
||||
pub(crate) local_storage: Arc<SkipMap<Bytes, Bytes>>,
|
||||
pub(crate) committed: Arc<AtomicBool>,
|
||||
/// Write set and read set
|
||||
pub(crate) key_hashes: Option<Mutex<(HashSet<u32>, HashSet<u32>)>>,
|
||||
}
|
||||
|
||||
impl Transaction {
|
||||
pub fn get(&self, key: &[u8]) -> Result<Option<Bytes>> {
|
||||
unimplemented!()
|
||||
}
|
||||
|
||||
pub fn scan(self: &Arc<Self>, lower: Bound<&[u8]>, upper: Bound<&[u8]>) -> Result<TxnIterator> {
|
||||
unimplemented!()
|
||||
}
|
||||
|
||||
pub fn put(&self, key: &[u8], value: &[u8]) {
|
||||
unimplemented!()
|
||||
}
|
||||
|
||||
pub fn delete(&self, key: &[u8]) {
|
||||
unimplemented!()
|
||||
}
|
||||
|
||||
pub fn commit(&self) -> Result<()> {
|
||||
unimplemented!()
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Transaction {
|
||||
fn drop(&mut self) {}
|
||||
}
|
||||
|
||||
type SkipMapRangeIter<'a> =
|
||||
crossbeam_skiplist::map::Range<'a, Bytes, (Bound<Bytes>, Bound<Bytes>), Bytes, Bytes>;
|
||||
|
||||
#[self_referencing]
|
||||
pub struct TxnLocalIterator {
|
||||
/// Stores a reference to the skipmap.
|
||||
map: Arc<SkipMap<Bytes, Bytes>>,
|
||||
/// Stores a skipmap iterator that refers to the lifetime of `MemTableIterator` itself.
|
||||
#[borrows(map)]
|
||||
#[not_covariant]
|
||||
iter: SkipMapRangeIter<'this>,
|
||||
/// Stores the current key-value pair.
|
||||
item: (Bytes, Bytes),
|
||||
}
|
||||
|
||||
impl StorageIterator for TxnLocalIterator {
|
||||
type KeyType<'a> = &'a [u8];
|
||||
|
||||
fn value(&self) -> &[u8] {
|
||||
unimplemented!()
|
||||
}
|
||||
|
||||
fn key(&self) -> &[u8] {
|
||||
unimplemented!()
|
||||
}
|
||||
|
||||
fn is_valid(&self) -> bool {
|
||||
unimplemented!()
|
||||
}
|
||||
|
||||
fn next(&mut self) -> Result<()> {
|
||||
unimplemented!()
|
||||
}
|
||||
}
|
||||
|
||||
pub struct TxnIterator {
|
||||
_txn: Arc<Transaction>,
|
||||
iter: TwoMergeIterator<TxnLocalIterator, FusedIterator<LsmIterator>>,
|
||||
}
|
||||
|
||||
impl TxnIterator {
|
||||
pub fn create(
|
||||
txn: Arc<Transaction>,
|
||||
iter: TwoMergeIterator<TxnLocalIterator, FusedIterator<LsmIterator>>,
|
||||
) -> Result<Self> {
|
||||
unimplemented!()
|
||||
}
|
||||
}
|
||||
|
||||
impl StorageIterator for TxnIterator {
|
||||
type KeyType<'a> = &'a [u8] where Self: 'a;
|
||||
|
||||
fn value(&self) -> &[u8] {
|
||||
self.iter.value()
|
||||
}
|
||||
|
||||
fn key(&self) -> Self::KeyType<'_> {
|
||||
self.iter.key()
|
||||
}
|
||||
|
||||
fn is_valid(&self) -> bool {
|
||||
self.iter.is_valid()
|
||||
}
|
||||
|
||||
fn next(&mut self) -> Result<()> {
|
||||
unimplemented!()
|
||||
}
|
||||
|
||||
fn num_active_iterators(&self) -> usize {
|
||||
self.iter.num_active_iterators()
|
||||
}
|
||||
}
|
24
mini-lsm-starter/src/mvcc/watermark.rs
Normal file
24
mini-lsm-starter/src/mvcc/watermark.rs
Normal file
@@ -0,0 +1,24 @@
|
||||
#![allow(unused_variables)] // TODO(you): remove this lint after implementing this mod
|
||||
#![allow(dead_code)] // TODO(you): remove this lint after implementing this mod
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
pub struct Watermark {
|
||||
readers: BTreeMap<u64, usize>,
|
||||
}
|
||||
|
||||
impl Watermark {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
readers: BTreeMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn add_reader(&mut self, ts: u64) {}
|
||||
|
||||
pub fn remove_reader(&mut self, ts: u64) {}
|
||||
|
||||
pub fn watermark(&self) -> Option<u64> {
|
||||
Some(0)
|
||||
}
|
||||
}
|
Reference in New Issue
Block a user