Use RangeSet instead of tracking every complete job (#1663)

Right now, we store each completed `JobId` and search linearly through
the list when checking for dependencies. This means that a flush, which
is dependent on all previous jobs, takes `O(N^2)` time to check its
dependencies.

This PR switches to a
[`RangeSet`](https://docs.rs/rangemap/latest/rangemap/set/struct.RangeSet.html),
which provides `O(log N)` lookup. In practice, we expect it to be
`O(1)`: jobs are completed in order, so the `RangeSet` will only contain
a single contiguous range.

I did a quick audit of the `rangemap` crate, and it seems reasonable:

- No dependencies
- `proptest` based testing, Github CI
- Consistent docstrings

There are two small behavior changes:

- `completed()` now allocates, expanding the `RangeSet` into a
  `Vec<JobId>` (instead of a `&[JobId]`). It's only used in unit tests and
  our debug work-printing function, so I think that's fine.
- The output of `completed()` is now sorted. This required changing one
  test assertion.

Author: mkeeter

Cargo.lock

@@ -78,6 +78,7 @@ rayon = "1.10.0"
 rand = { version = "0.8.5", features = ["min_const_gen", "small_rng"] }
 rand_chacha = "0.3.1"
 reedline = "0.38.0"
+rangemap = "1.5.1"
 reqwest = { version = "0.12", features = ["default", "blocking", "json", "stream"] }
 ringbuffer = "0.15.0"
 rusqlite = { version = "0.32" }

Cargo.toml

@@ -31,6 +31,7 @@ opentelemetry.workspace = true
 oximeter-producer.workspace = true
 oximeter.workspace = true
 rand.workspace = true
+rangemap.workspace = true
 rayon.workspace = true
 repair-client.workspace = true
 reqwest.workspace = true

downstairs/Cargo.toml

@@ -1,15 +1,19 @@
 use crate::JobId;
+use rangemap::RangeSet;
 
 /// Stores a set of complete jobs
 #[derive(Debug)]
 pub struct CompletedJobs {
-    completed: Vec<JobId>,
+    completed: RangeSet<JobId>,
 }
 
 impl CompletedJobs {
     pub fn new(last_flush: Option<JobId>) -> Self {
         Self {
-            completed: last_flush.into_iter().collect(),
+            completed: last_flush
+                .into_iter()
+                .map(|id| id..JobId(id.0 + 1))
+                .collect(),
         }
     }
 
@@ -20,7 +24,7 @@ impl CompletedJobs {
 
     /// Records a new complete job in the list
     pub fn push(&mut self, id: JobId) {
-        self.completed.push(id);
+        self.completed.insert(id..JobId(id.0 + 1));
     }
 
     /// Resets the data structure, given a new barrier operation
@@ -29,20 +33,20 @@ impl CompletedJobs {
     /// oldest complete job.
     pub fn reset(&mut self, id: JobId) {
         self.completed.clear();
-        self.completed.push(id);
+        self.completed.insert(id..JobId(id.0 + 1));
     }
 
     /// Checks whether the given job is complete
     ///
     /// A job is complete if it is listed in the set of complete jobs.
     pub fn is_complete(&self, id: JobId) -> bool {
-        // We deliberately reverse the `completed` list because new jobs are at
-        // the back and are more likely to be what we care about
-        self.completed.iter().rev().any(|j| *j == id)
+        self.completed.contains(&id)
     }
 
     /// Returns the list of completed jobs
-    pub fn completed(&self) -> &[JobId] {
-        &self.completed
+    pub fn completed(&self) -> impl Iterator<Item = JobId> + use<'_> {
+        self.completed
+            .iter()
+            .flat_map(|r| (r.start.0..r.end.0).map(JobId))
     }
 }

downstairs/src/complete_jobs.rs

@@ -3288,8 +3288,8 @@ impl Work {
         }
     }
 
-    fn completed(&self) -> &[JobId] {
-        self.completed.completed()
+    fn completed(&self) -> Vec<JobId> {
+        self.completed.completed().collect()
     }
 
     /// Pushes a new job to the back of the queue
@@ -4947,7 +4947,7 @@ mod test {
         test_do_work(&mut work, next_jobs);
         assert_eq!(
             work.completed(),
-            vec![JobId(1000), JobId(2000), JobId(1001), JobId(2001)]
+            vec![JobId(1000), JobId(1001), JobId(2000), JobId(2001)]
         );
 
         let next_jobs = test_push_next_jobs(&mut work);