Allow for compression of the oplog in case of mergeable operations.

src/lib.rs

@@ -189,6 +189,17 @@ pub use crate::read::{ReadGuard, ReadHandle, ReadHandleFactory};
 
 pub mod aliasing;
 
+/// The result of calling [`Absorb::try_compress`](Absorb::try_compress).
+#[derive(Debug)]
+pub enum TryCompressResult<O> {
+    /// Returned when [`try_compress`](Absorb::try_compress) was successful.
+    Compressed,
+    /// Returned when [`try_compress`](Absorb::try_compress) failed because `prev` and `next` are independent of each other, contains `next`.
+    Independent(O),
+    /// Returned when [`try_compress`](Absorb::try_compress) failed because `prev` must precede `next`, contains `next`.
+    Dependent(O),
+}
+
 /// Types that can incorporate operations of type `O`.
 ///
 /// This trait allows `left-right` to keep the two copies of the underlying data structure (see the
@@ -261,6 +272,32 @@ pub trait Absorb<O> {
     /// subtly affect results like the `RandomState` of a `HashMap` which can change iteration
     /// order.
     fn sync_with(&mut self, first: &Self);
+
+    /// Range at which [`WriteHandle`] tries to compress the oplog, reset each time a compression succeeds.
+    ///
+    /// Can be used to avoid having insertion into the oplog be O(oplog.len * ops.len) if it is filled with mainly independent ops.
+    ///
+    /// Defaults to `0`, which disables compression and allows the usage of an efficient fallback.
+    const MAX_COMPRESS_RANGE: usize = 0;
+
+    /// Try to compress two ops into a single op to optimize the oplog.
+    ///
+    /// `prev` is the target of the compression and temporarily removed from the oplog, `next` is the op to be inserted.
+    ///
+    /// A return value of [`TryCompressResult::Compressed`] means the ops were successfully compressed,
+    /// [`TryCompressResult::Independent`] that while the ops can't be compressed,
+    /// `next` can safely precede `prev`, which therefore can be simply skipped over,
+    /// and [`TryCompressResult::Dependent`] that they can not be compressed, and `prev` must precede `next`.
+    ///
+    /// Defaults to [`TryCompressResult::Dependent`], which sub-optimally disables compression.
+    /// Setting [`Self::MAX_COMPRESS_RANGE`](Absorb::MAX_COMPRESS_RANGE) to `0` is vastly more efficient for that.
+    fn try_compress(prev: &mut O, next: O) -> TryCompressResult<O> {
+        // yes, unnecessary, but: makes it so that prev is not an unused variable
+        // and really matches the mental model of 'all ops are dependent'.
+        match prev {
+            _ => TryCompressResult::Dependent(next),
+        }
+    }
 }
 
 /// Construct a new write and read handle pair from an empty data structure.

src/utilities.rs

@@ -8,13 +8,63 @@ impl Absorb<CounterAddOp> for i32 {
         *self += operation.0;
     }
 
-    fn absorb_second(&mut self, operation: CounterAddOp, _: &Self) {
-        *self += operation.0;
+    fn sync_with(&mut self, first: &Self) {
+        *self = *first
     }
+}
+
+#[cfg(test)]
+#[derive(Debug, Eq, PartialEq)]
+pub enum CompressibleCounterOp<const MAX_COMPRESS_RANGE: usize> {
+    Set(i32),
+    Add(i32),
+    Sub(i32),
+}
 
-    fn drop_first(self: Box<Self>) {}
+#[cfg(test)]
+impl<const MAX_COMPRESS_RANGE: usize> Absorb<CompressibleCounterOp<MAX_COMPRESS_RANGE>> for i32 {
+    fn absorb_first(
+        &mut self,
+        operation: &mut CompressibleCounterOp<MAX_COMPRESS_RANGE>,
+        _: &Self,
+    ) {
+        match operation {
+            CompressibleCounterOp::Set(v) => *self = *v,
+            CompressibleCounterOp::Add(v) => *self += *v,
+            CompressibleCounterOp::Sub(v) => *self -= *v,
+        }
+    }
 
     fn sync_with(&mut self, first: &Self) {
         *self = *first
     }
+
+    const MAX_COMPRESS_RANGE: usize = MAX_COMPRESS_RANGE;
+
+    fn try_compress(
+        prev: &mut CompressibleCounterOp<MAX_COMPRESS_RANGE>,
+        next: CompressibleCounterOp<MAX_COMPRESS_RANGE>,
+    ) -> TryCompressResult<CompressibleCounterOp<MAX_COMPRESS_RANGE>> {
+        match (prev, next) {
+            (CompressibleCounterOp::Add(prev), CompressibleCounterOp::Add(next)) => {
+                *prev += next;
+                TryCompressResult::Compressed
+            }
+            (CompressibleCounterOp::Sub(prev), CompressibleCounterOp::Sub(next)) => {
+                *prev += next;
+                TryCompressResult::Compressed
+            }
+            (CompressibleCounterOp::Add(_), CompressibleCounterOp::Sub(next)) => {
+                TryCompressResult::Independent(CompressibleCounterOp::Sub(next))
+            }
+            (CompressibleCounterOp::Sub(_), CompressibleCounterOp::Add(next)) => {
+                TryCompressResult::Independent(CompressibleCounterOp::Add(next))
+            }
+            (CompressibleCounterOp::Set(_), next) => TryCompressResult::Dependent(next),
+            (prev, CompressibleCounterOp::Set(next)) => {
+                *prev = CompressibleCounterOp::Set(next);
+                TryCompressResult::Compressed
+            }
+        }
+    }
 }