Merge pull request #202 from JakeRoggenbuck/abort-commit-record-level

record lock approach

Author: JakeRoggenbuck

benches/merge_benchmarks.rs

@@ -1,8 +1,8 @@
 use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
 use redoxql::pagerange::PageRange;
-use redoxql::record::{Record, RecordAddress};
+use redoxql::record::{Record, RecordAddress, RecordLock};
 use redoxql::table::PageDirectory;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, RwLock};
 use std::time::Duration;
 
 fn setup_benchmark_data(
@@ -28,6 +28,7 @@ fn setup_benchmark_data(
             let record = Record {
                 rid,
                 addresses: Arc::new(Mutex::new(addresses)),
+                lock: Arc::new(RwLock::new(RecordLock::default())),
             };
 
             pd.directory.insert(rid, record);
@@ -45,6 +46,7 @@ fn setup_benchmark_data(
             let record = Record {
                 rid,
                 addresses: Arc::new(Mutex::new(addresses)),
+                lock: Arc::new(RwLock::new(RecordLock::default())),
             };
 
             pd.directory.insert(rid, record);

src/container.rs

@@ -1,8 +1,10 @@
+use crate::record::RecordLock;
+
 use super::filewriter::{build_binary_writer, Writer};
 use super::page::PhysicalPage;
 use super::record::{Record, RecordAddress};
 use serde::{Deserialize, Serialize};
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, RwLock};
 
 #[derive(Clone, Default, Deserialize, Serialize, Debug)]
 pub struct BaseContainerMetadata {
@@ -191,6 +193,7 @@ impl BaseContainer {
         Record {
             rid,
             addresses: addresses.clone(),
+            lock: Arc::new(RwLock::new(RecordLock::default())),
         }
     }
 
@@ -483,6 +486,7 @@ impl TailContainer {
         Record {
             rid,
             addresses: addresses.clone(),
+            lock: Arc::new(RwLock::new(RecordLock::default())),
         }
     }
 

src/database.rs

@@ -21,6 +21,26 @@ pub struct RDatabaseMetadata {
     db_filepath: Option<String>,
 }
 
+// Define lock types for 2PL
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum LockType {
+    Shared,    // Read lock (S)
+    Exclusive, // Write lock (X)
+}
+
+impl std::fmt::Display for LockType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            LockType::Shared => write!(f, "Shared"),
+            LockType::Exclusive => write!(f, "Exclusive"),
+        }
+    }
+}
+
+// A unique identifier for a record: (RTableHandle, record_id, lock_type)
+pub type RecordId = (RTableHandle, i64, LockType);
+pub type RecordId2 = (String, i64, LockType);
+
 #[pyclass]
 pub struct RDatabase {
     /// This is where we keep all of the tables
@@ -157,14 +177,6 @@ impl RDatabase {
             index_guard.set_owner(Arc::downgrade(&arc_table));
         }
 
-        // PREVIOUS IMPLEMENTATION
-        // let i = self.tables.len();
-
-        // Map a name of a table to it's index on the self.tables field
-        // self.tables_hashmap.insert(name, i);
-
-        // self.tables.push(t);
-
         // Push t into the tables vector so its address becomes stable.
         self.tables.push(arc_table.clone());
         let i = self.tables.len() - 1;

src/pagerange.rs

@@ -3,12 +3,12 @@ use super::container::{
 };
 use super::filewriter::{build_binary_writer, Writer};
 use super::record::Record;
-use crate::record::RecordAddress;
+use crate::record::{RecordAddress, RecordLock};
 use crate::table::PageDirectory;
 use log::info;
 use serde::{Deserialize, Serialize};
 use std::collections::HashSet;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, RwLock};
 use std::thread;
 
 #[derive(Clone, Default, Deserialize, Serialize, Debug)]
@@ -136,6 +136,7 @@ impl PageRange {
                     let new_record = Record {
                         rid: base_rid,
                         addresses: Arc::new(Mutex::new(Vec::new())),
+                        lock: Arc::new(RwLock::new(RecordLock::default())),
                     };
 
                     {
@@ -345,6 +346,7 @@ impl PageRange {
                     let new_record = Record {
                         rid: base_rid,
                         addresses: Arc::new(Mutex::new(Vec::new())),
+                        lock: Arc::new(RwLock::new(RecordLock::default())),
                     };
 
                     {

src/record.rs

@@ -3,7 +3,7 @@ use crate::container::{ReservedColumns, NUM_RESERVED_COLUMNS};
 use pyo3::prelude::*;
 use rustc_hash::FxHashMap;
 use serde::{Deserialize, Serialize};
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, RwLock};
 
 type RedoxQLHashMap<K, V> = FxHashMap<K, V>;
 
@@ -89,6 +89,7 @@ impl RecordMetadata {
         Record {
             rid: self.rid,
             addresses: Arc::new(Mutex::new(rec_addrs)),
+            lock: Arc::new(RwLock::new(RecordLock::default())),
         }
     }
 }
@@ -102,6 +103,14 @@ pub struct Record {
     /// The Record keeps a Vector of the RecordAddress, which allow us to actually call
     /// RecordAddress.page.read() to get the value stored at the page using the offset
     pub addresses: Arc<Mutex<Vec<RecordAddress>>>,
+
+    pub lock: Arc<RwLock<RecordLock>>,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct RecordLock {
+    pub reader_count: i64,
+    pub writer_count: i64,
 }
 
 impl Record {
@@ -121,6 +130,48 @@ impl Record {
 
         return rm;
     }
+
+    pub fn attempt_obtain_read(&self) -> bool {
+        let mut a = self.lock.write().unwrap();
+
+        if a.writer_count != 0 {
+            return false;
+        }
+
+        a.reader_count += 1;
+
+        true
+    }
+
+    pub fn attempt_obtain_write(&self) -> bool {
+        let mut a = self.lock.write().unwrap();
+
+        if a.writer_count != 0 || a.reader_count != 0 {
+            return false;
+        }
+
+        a.writer_count += 1;
+
+        true
+    }
+
+    pub fn release_read_lock(&self) {
+        let mut a = self.lock.write().unwrap();
+        a.reader_count -= 1;
+
+        if a.reader_count < 0 {
+            panic!("write_count less than 0, rid {}", self.rid)
+        }
+    }
+
+    pub fn release_write_lock(&self) {
+        let mut a = self.lock.write().unwrap();
+        a.writer_count -= 1;
+
+        if a.writer_count < 0 {
+            panic!("write_count less than 0, rid {}", self.rid)
+        }
+    }
 }
 
 #[pymethods]