Merge pull request #1133 from spacejam/tyler_avoid_sorting_batch_pegs

0.34.2 - avoid sorting batch pegs

CHANGELOG.md

@@ -1,3 +1,11 @@
+# 0.34.2
+
+## Improvements
+
+* #1133 transactions and writebatch performance has been
+  significantly improved by removing a bottleneck in
+  the atomic batch stability tracking code.
+
 # 0.34.1
 
 ## New Features

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "sled"
-version = "0.34.1"
+version = "0.34.2"
 authors = ["Tyler Neely <t@jujit.su>"]
 description = "a modern embedded database"
 license = "MIT/Apache-2.0"

src/pagecache/iobuf.rs

@@ -192,14 +192,18 @@ impl IoBuf {
     ) -> std::result::Result<Header, Header> {
         debug_delay();
         let res = self.header.compare_and_swap(old, new, SeqCst);
-        if res == old { Ok(new) } else { Err(res) }
+        if res == old {
+            Ok(new)
+        } else {
+            Err(res)
+        }
     }
 }
 
 #[derive(Debug)]
 pub(crate) struct StabilityIntervals {
     fsynced_ranges: Vec<(Lsn, Lsn)>,
-    batches: Vec<(Lsn, Lsn)>,
+    batches: BTreeMap<Lsn, Lsn>,
     stable_lsn: Lsn,
 }
 
@@ -208,15 +212,13 @@ impl StabilityIntervals {
         StabilityIntervals {
             stable_lsn: lsn,
             fsynced_ranges: vec![],
-            batches: vec![],
+            batches: BTreeMap::default(),
         }
     }
 
     pub(crate) fn mark_batch(&mut self, interval: (Lsn, Lsn)) {
         assert!(interval.0 > self.stable_lsn);
-        self.batches.push(interval);
-        // reverse sort
-        self.batches.sort_unstable_by(|a, b| b.cmp(a));
+        self.batches.insert(interval.0, interval.1);
     }
 
     fn mark_fsync(&mut self, interval: (Lsn, Lsn)) -> Option<Lsn> {
@@ -274,7 +276,9 @@ impl StabilityIntervals {
         // batch has been stabilized before any parts
         // of the batch are allowed to be reused
         // due to having marked them as stable.
-        while let Some(&(low, high)) = self.batches.last() {
+        while let Some((low, high)) =
+            self.batches.iter().map(|(l, h)| (*l, *h)).next()
+        {
             assert!(
                 low < high,
                 "expected batch low mark {} to be below high mark {}",
@@ -290,7 +294,7 @@ impl StabilityIntervals {
                     assert!(bsl < high);
                 }
                 batch_stable_lsn = Some(high);
-                self.batches.pop().unwrap();
+                self.batches.remove(&low).unwrap();
             } else {
                 if low <= self.stable_lsn {
                     // the batch has not been fully written

tests/test_tree.rs

@@ -92,7 +92,7 @@ fn test_varied_compression_ratios() {
 fn concurrent_tree_pops() -> sled::Result<()> {
     use std::thread;
 
-    let db = sled::open("db")?;
+    let db = sled::Config::new().temporary(true).open()?;
 
     // Insert values 0..5
     for x in 0u32..5 {