Performance bottleneck due to cache contention

crossbeam-skiplist/Cargo.toml

@@ -40,6 +40,7 @@ alloc = ["crossbeam-epoch/alloc"]
 [dependencies]
 crossbeam-epoch = { version = "0.9.17", path = "../crossbeam-epoch", default-features = false }
 crossbeam-utils = { version = "0.8.18", path = "../crossbeam-utils", default-features = false }
+rand = "0.10"
 
 [dev-dependencies]
 fastrand = "2"

crossbeam-skiplist/examples/parallel.rs

@@ -0,0 +1,107 @@
+use std::{
+    sync::{
+        Mutex,
+        atomic::{AtomicUsize, Ordering},
+    },
+    time::{Duration, Instant},
+};
+
+use crossbeam_skiplist::SkipSet;
+use rand::RngExt;
+
+fn main() {
+    let results: Vec<_> = std::env::args()
+        .skip(1)
+        .map(|nthreads| {
+            // Run 30 iterations with 1_048_576 keys and 10_000_000 ops per thread
+            let result = run(30, nthreads.parse().unwrap(), 1_048_576, 10_000_000);
+            (nthreads, result)
+        })
+        .collect();
+    println!("threads,throughput");
+    for (nthreads, throughput) in results {
+        println!("{nthreads},{throughput:.0}");
+    }
+}
+
+fn run(iters: usize, nthreads: usize, nkeys: usize, ops_per_thread: usize) -> f32 {
+    let total_runtime = Mutex::new(Duration::ZERO);
+
+    eprintln!(
+        "running {} operations per thread ({} threads, {} total operations)...",
+        ops_per_thread,
+        nthreads,
+        ops_per_thread * nthreads,
+    );
+
+    for i in 0..iters {
+        // Initialize barrier (barrier == 2 * nthreads)
+        let barrier = AtomicUsize::new(2 * nthreads);
+        let set: SkipSet<usize> = SkipSet::new();
+
+        // Fill the set to 50% by inserting all even-numbered keys
+        for x in 0..(nkeys / 2) {
+            set.insert(x * 2);
+        }
+
+        std::thread::scope(|s| {
+            for tid in 0..nthreads {
+                let barrier = &barrier;
+                let set = &set;
+                let total_runtime = &total_runtime;
+                s.spawn(move || {
+                    let mut rng = rand::rng();
+
+                    // Wait for all threads to reach here (barrier == nthreads)
+                    barrier.fetch_sub(1, Ordering::Relaxed);
+                    while barrier.load(Ordering::Relaxed) > nthreads {
+                        std::hint::spin_loop();
+                    }
+                    let start = Instant::now();
+
+                    // BEGIN benchmarked region
+                    for _ in 0..ops_per_thread {
+                        // Pick a random key and random operation with equal likelihood: insert, remove, or contains
+                        let key = rng.random_range(0..nkeys);
+                        let operation = rng.random_range(0..3);
+                        match operation {
+                            0 => {
+                                set.insert(key);
+                            }
+                            1 => {
+                                set.remove(&key);
+                            }
+                            _ => {
+                                set.contains(&key);
+                            }
+                        };
+                    }
+                    // END benchmarked region
+
+                    barrier.fetch_sub(1, Ordering::Relaxed);
+
+                    if tid == 0 {
+                        // Wait for all threads to complete (barrier == 0)
+                        while barrier.load(Ordering::Relaxed) != 0 {
+                            std::hint::spin_loop();
+                        }
+
+                        let elapsed = start.elapsed();
+                        eprintln!(
+                            "iter {:2}: {:.4?}, throughput {:.2} ops/s",
+                            i + 1,
+                            elapsed,
+                            (ops_per_thread * nthreads) as f32 / elapsed.as_secs_f32(),
+                        );
+                        *total_runtime.lock().unwrap() += elapsed;
+                    }
+                });
+            }
+        });
+    }
+
+    let average_time = total_runtime.into_inner().unwrap() / iters as u32;
+    let throughput = (ops_per_thread * nthreads) as f32 / average_time.as_secs_f32();
+    eprintln!("\naverage time: {average_time:?}, throughput {throughput} ops/s");
+    throughput
+}

crossbeam-skiplist/src/base.rs

@@ -441,11 +441,11 @@ struct Position<'a, K, V> {
 
 /// Frequently modified data associated with a skip list.
 struct HotData {
-    /// The seed for random height generation.
-    seed: AtomicUsize,
+    // /// The seed for random height generation.
+    // seed: AtomicUsize,
 
-    /// The number of entries in the skip list.
-    len: AtomicUsize,
+    // /// The number of entries in the skip list.
+    // len: AtomicUsize,
 
     /// Highest tower currently in use. This value is used as a hint for where
     /// to start lookups and never decreases.
@@ -496,8 +496,8 @@ impl<K, V, C> SkipList<K, V, C> {
             head: Head::new(),
             collector,
             hot_data: CachePadded::new(HotData {
-                seed: AtomicUsize::new(1),
-                len: AtomicUsize::new(0),
+                // seed: AtomicUsize::new(1),
+                // len: AtomicUsize::new(0),
                 max_height: AtomicUsize::new(1),
             }),
             comparator,
@@ -514,11 +514,12 @@ impl<K, V, C> SkipList<K, V, C> {
     /// If the skip list is being concurrently modified, consider the returned number just an
     /// approximation without any guarantees.
     pub fn len(&self) -> usize {
-        let len = self.hot_data.len.load(Ordering::Relaxed);
+        // let len = self.hot_data.len.load(Ordering::Relaxed);
 
-        // Due to the relaxed memory ordering, the length counter may sometimes
-        // underflow and produce a very large value. We treat such values as 0.
-        if len > isize::MAX as usize { 0 } else { len }
+        // // Due to the relaxed memory ordering, the length counter may sometimes
+        // // underflow and produce a very large value. We treat such values as 0.
+        // if len > isize::MAX as usize { 0 } else { len }
+        0
     }
 
     /// Ensures that all `Guard`s used with the skip list come from the same
@@ -710,11 +711,12 @@ where
         //
         // This particular set of operations generates 32-bit integers. See:
         // https://en.wikipedia.org/wiki/Xorshift#Example_implementation
-        let mut num = self.hot_data.seed.load(Ordering::Relaxed);
-        num ^= num << 13;
-        num ^= num >> 17;
-        num ^= num << 5;
-        self.hot_data.seed.store(num, Ordering::Relaxed);
+        // let mut num = self.hot_data.seed.load(Ordering::Relaxed);
+        // num ^= num << 13;
+        // num ^= num >> 17;
+        // num ^= num << 5;
+        // self.hot_data.seed.store(num, Ordering::Relaxed);
+        let num = rand::random::<u32>();
 
         let mut height = cmp::min(MAX_HEIGHT, num.trailing_zeros() as usize + 1);
         unsafe {
@@ -1064,8 +1066,8 @@ where
                 )
             };
 
-            // Optimistically increment `len`.
-            self.hot_data.len.fetch_add(1, Ordering::Relaxed);
+            // // Optimistically increment `len`.
+            // self.hot_data.len.fetch_add(1, Ordering::Relaxed);
 
             loop {
                 // Set the lowest successor of `n` to `search.right[0]`.
@@ -1087,7 +1089,7 @@ where
                     // This node has been abandoned
                     if let Some(r) = search.found {
                         if r.mark_tower() {
-                            self.hot_data.len.fetch_sub(1, Ordering::Relaxed);
+                            // self.hot_data.len.fetch_sub(1, Ordering::Relaxed);
                         }
                     }
                     break;
@@ -1115,7 +1117,7 @@ where
                         if let Some(e) = RefEntry::try_acquire(self, r) {
                             // Destroy the new node.
                             Node::finalize(node.as_raw() as *mut Node<K, V>);
-                            self.hot_data.len.fetch_sub(1, Ordering::Relaxed);
+                            // self.hot_data.len.fetch_sub(1, Ordering::Relaxed);
 
                             return e;
                         }
@@ -1296,7 +1298,7 @@ where
                 // Try removing the node by marking its tower.
                 if n.mark_tower() {
                     // Success! Decrement `len`.
-                    self.hot_data.len.fetch_sub(1, Ordering::Relaxed);
+                    // self.hot_data.len.fetch_sub(1, Ordering::Relaxed);
 
                     // Unlink the node at each level of the skip list. We could do this by simply
                     // repeating the search, but it's usually faster to unlink it manually using
@@ -1396,7 +1398,7 @@ where
                     // Try removing the current entry.
                     if e.node.mark_tower() {
                         // Success! Decrement `len`.
-                        self.hot_data.len.fetch_sub(1, Ordering::Relaxed);
+                        // self.hot_data.len.fetch_sub(1, Ordering::Relaxed);
                     }
 
                     entry = next;
@@ -1531,7 +1533,7 @@ where
         // Try marking the tower.
         if self.node.mark_tower() {
             // Success - the entry is removed. Now decrement `len`.
-            self.parent.hot_data.len.fetch_sub(1, Ordering::Relaxed);
+            // self.parent.hot_data.len.fetch_sub(1, Ordering::Relaxed);
 
             // Search for the key to unlink the node from the skip list.
             self.parent
@@ -1697,7 +1699,7 @@ where
         // Try marking the tower.
         if self.node.mark_tower() {
             // Success - the entry is removed. Now decrement `len`.
-            self.parent.hot_data.len.fetch_sub(1, Ordering::Relaxed);
+            // self.parent.hot_data.len.fetch_sub(1, Ordering::Relaxed);
 
             // Search for the key to unlink the node from the skip list.
             self.parent