add w3d6 Task 3: Engine Interface and Serializable Validation

Author: Zhou-jw

mini-lsm-starter/README.md

@@ -109,3 +109,15 @@ fn main() {
 ```
 https://rustwiki.org/zh-CN/book/appendix-03-derivable-traits.html
 ```
+
+## week3 day6
+### Task 3: Engine Interface and Serializable Validation
+1. serializable validation
+go through committed txn in range (read_ts, expected_commit_ts)
+
+if read_set overlaps with write_set of those txns, validation fails
+
+if validation succeeds, commit the write_batch, insert write_set into `self.inner.mvcc().committed_txns`
+
+
+

mini-lsm-starter/src/lsm_storage.rs

@@ -591,7 +591,7 @@ impl LsmStorageInner {
     }
 
     /// Write a batch of data into the storage. Implement in week 2 day 7.
-    pub fn write_batch<T: AsRef<[u8]>>(&self, batch: &[WriteBatchRecord<T>]) -> Result<()> {
+    pub fn write_batch_inner<T: AsRef<[u8]>>(&self, batch: &[WriteBatchRecord<T>]) -> Result<u64> {
         let _lk = self.mvcc().write_lock.lock();
         let ts = self.mvcc().latest_commit_ts() + 1;
         for record in batch {
@@ -625,17 +625,54 @@ impl LsmStorageInner {
             }
         }
         self.mvcc().update_commit_ts(ts);
+        Ok(ts)
+    }
+
+    /// Write a batch of data into the storage. Implement in week 2 day 7.
+    pub fn write_batch<T: AsRef<[u8]>>(self: &Arc<Self>, batch: &[WriteBatchRecord<T>]) -> Result<()> {
+        if !self.options.serializable {
+            self.write_batch_inner(batch)?;
+        }
+        else {
+            let new_txn = self.mvcc().new_txn(self.clone(), self.options.serializable);
+            for record in batch {
+                match record {
+                    WriteBatchRecord::Put(key, value) => {
+                        new_txn.put(key.as_ref(), value.as_ref());
+                    }
+                    WriteBatchRecord::Del(key) => {
+                        new_txn.delete(key.as_ref());
+                    }
+                }
+            }
+        }
         Ok(())
     }
 
     /// Put a key-value pair into the storage by writing into the current memtable.
-    pub fn put(&self, key: &[u8], value: &[u8]) -> Result<()> {
-        self.write_batch(&[WriteBatchRecord::Put(key, value)])
+    pub fn put(self: &Arc<Self>, key: &[u8], value: &[u8]) -> Result<()> {
+        if !self.options.serializable {
+            self.write_batch_inner(&[WriteBatchRecord::Put(key, value)])?;
+        }
+        else {
+            let new_txn = self.mvcc().new_txn(self.clone(), self.options.serializable);
+            new_txn.put(key, value);
+            new_txn.commit()?;
+        }
+        Ok(())
     }
 
     /// Remove a key from the storage by writing an empty value.
-    pub fn delete(&self, key: &[u8]) -> Result<()> {
-        self.write_batch(&[WriteBatchRecord::Del(key)])
+    pub fn delete(self: &Arc<Self>, key: &[u8]) -> Result<()> {
+         if !self.options.serializable {
+            self.write_batch_inner(&[WriteBatchRecord::Del(key)])?;
+        }
+        else {
+            let new_txn = self.mvcc().new_txn(self.clone(), self.options.serializable);
+            new_txn.delete(key);
+            new_txn.commit()?;
+        }
+        Ok(())
     }
 
     pub(crate) fn path_of_sst_static(path: impl AsRef<Path>, id: usize) -> PathBuf {

mini-lsm-starter/src/mvcc.rs

@@ -64,7 +64,11 @@ impl LsmMvccInner {
             inner,
             local_storage: Arc::new(SkipMap::new()),
             committed: Arc::new(AtomicBool::new(false)),
-            key_hashes: if serializable {Some(Mutex::new((HashSet::new(), HashSet::new())))} else {None},
+            key_hashes: if serializable {
+                Some(Mutex::new((HashSet::new(), HashSet::new())))
+            } else {
+                None
+            },
         })
     }
 }

mini-lsm-starter/src/mvcc/txn.rs

@@ -7,7 +7,7 @@ use std::{
     sync::{atomic::AtomicBool, atomic::Ordering, Arc},
 };
 
-use anyhow::Result;
+use anyhow::{bail, Result};
 use bytes::Bytes;
 use crossbeam_skiplist::SkipMap;
 use ouroboros::self_referencing;
@@ -20,6 +20,8 @@ use crate::{
     mem_table::map_bound,
 };
 
+use super::CommittedTxnData;
+
 pub struct Transaction {
     pub(crate) read_ts: u64,
     pub(crate) inner: Arc<LsmStorageInner>,
@@ -37,7 +39,7 @@ impl Transaction {
 
         if self.key_hashes.is_some() {
             let mut guard = self.key_hashes.as_ref().unwrap().lock();
-            let (_, read_set)  = & mut *guard;
+            let (_, read_set) = &mut *guard;
             read_set.insert(crc32fast::hash(key));
         }
 
@@ -90,6 +92,28 @@ impl Transaction {
     }
 
     pub fn commit(&self) -> Result<()> {
+        // serializable validation
+        let mut serializable_check = false;
+        let _commit_lock = self.inner.mvcc().commit_lock.lock();
+        let expected_commit_ts = self.inner.mvcc().latest_commit_ts() + 1;
+        if self.key_hashes.is_some() {
+            let guard = self.key_hashes.as_ref().unwrap().lock();
+            let (write_set, read_set) = &*guard;
+            if !write_set.is_empty() {
+                let committed_txns = self.inner.mvcc().committed_txns.lock();
+                for (_ts, txn_data) in committed_txns.iter() {
+                    let ts_overlap = self.read_ts < txn_data.commit_ts
+                        && txn_data.commit_ts < expected_commit_ts;
+                    // commit timestamp within range (read_ts, expected_commit_ts), and has jointset
+                    if ts_overlap && !read_set.is_disjoint(&txn_data.key_hashes) {
+                        bail!("Serializable Validation fail!");
+                    }
+                }
+            }
+            serializable_check = true;
+        }
+
+        // begin commit process
         self.committed
             .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
             .expect("cannot operate on committed txn!");
@@ -107,7 +131,22 @@ impl Transaction {
             })
             .collect::<Vec<_>>();
 
-        self.inner.write_batch(&batch)
+        let ts = self.inner.write_batch_inner(&batch)?;
+        if serializable_check {
+            let mut guard = self.key_hashes.as_ref().unwrap().lock();
+            let (write_set, read_set) = &mut *guard;
+            let mut committed_txns = self.inner.mvcc().committed_txns.lock();
+            let old_tnx_data = committed_txns.insert(
+                ts,
+                CommittedTxnData {
+                    commit_ts: ts,
+                    key_hashes: std::mem::take(write_set),
+                    read_ts: self.read_ts,
+                },
+            );
+            assert!(old_tnx_data.is_none());
+        }
+        Ok(())
     }
 }