perf: improve apply update performance

.github/workflows/y-octo-node.yml

@@ -38,7 +38,6 @@ jobs:
     runs-on: ${{ matrix.settings.host }}
     env:
       RUSTFLAGS: "-C debuginfo=1"
-    environment: development
     steps:
       - uses: actions/checkout@v3
       - name: Setup Node.js

y-octo-node/src/utils.rs

@@ -1,7 +1,5 @@
-use std::collections::HashMap;
-
 use napi::{bindgen_prelude::Either4, Env, Error, JsObject, JsUnknown, Result, Status, ValueType};
-use y_octo::{Any, Value};
+use y_octo::{AHashMap, Any, HashMapExt, Value};
 
 use super::*;
 
@@ -55,7 +53,7 @@ pub fn get_any_from_js_object(object: JsObject) -> Result<Any> {
         }
         Ok(Any::Array(array))
     } else {
-        let mut map = HashMap::new();
+        let mut map = AHashMap::new();
         let keys = object.get_property_names()?;
         if let Ok(length) = keys.get_array_length() {
             for i in 0..length {

y-octo-utils/Cargo.toml

@@ -52,6 +52,10 @@ name    = "map_ops_benchmarks"
 harness = false
 name    = "text_ops_benchmarks"
 
+[[bench]]
+harness = false
+name    = "apply_benchmarks"
+
 [[bench]]
 harness = false
 name    = "update_benchmarks"

y-octo-utils/benches/apply_benchmarks.rs

@@ -0,0 +1,35 @@
+mod utils;
+
+use std::time::Duration;
+
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
+use path_ext::PathExt;
+use utils::Files;
+
+fn apply(c: &mut Criterion) {
+    let files = Files::load();
+
+    let mut group = c.benchmark_group("apply");
+    group.measurement_time(Duration::from_secs(15));
+
+    for file in &files.files {
+        group.throughput(Throughput::Bytes(file.content.len() as u64));
+        group.bench_with_input(
+            BenchmarkId::new("apply with yrs", file.path.name_str()),
+            &file.content,
+            |b, content| {
+                b.iter(|| {
+                    use yrs::{updates::decoder::Decode, Doc, Transact, Update};
+                    let update = Update::decode_v1(content).unwrap();
+                    let doc = Doc::new();
+                    doc.transact_mut().apply_update(update);
+                });
+            },
+        );
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, apply);
+criterion_main!(benches);

y-octo/Cargo.toml

@@ -19,6 +19,7 @@ version = "0.0.1"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+ahash         = "0.8"
 bitvec        = "1.0"
 byteorder     = "1.5"
 log           = "0.4"
@@ -70,6 +71,10 @@ name    = "map_ops_benchmarks"
 harness = false
 name    = "text_ops_benchmarks"
 
+[[bench]]
+harness = false
+name    = "apply_benchmarks"
+
 [[bench]]
 harness = false
 name    = "update_benchmarks"

y-octo/benches/apply_benchmarks.rs

@@ -0,0 +1,34 @@
+mod utils;
+
+use std::time::Duration;
+
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
+use path_ext::PathExt;
+use utils::Files;
+
+fn apply(c: &mut Criterion) {
+    let files = Files::load();
+
+    let mut group = c.benchmark_group("apply");
+    group.measurement_time(Duration::from_secs(15));
+
+    for file in &files.files {
+        group.throughput(Throughput::Bytes(file.content.len() as u64));
+        group.bench_with_input(
+            BenchmarkId::new("apply with jwst", file.path.name_str()),
+            &file.content,
+            |b, content| {
+                b.iter(|| {
+                    use y_octo::*;
+                    let mut doc = Doc::new();
+                    doc.apply_update_from_binary(content.clone()).unwrap()
+                });
+            },
+        );
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, apply);
+criterion_main!(benches);

y-octo/src/doc/codec/any.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, fmt, ops::RangeInclusive};
+use std::{fmt, ops::RangeInclusive};
 
 use ordered_float::OrderedFloat;
 

y-octo/src/doc/codec/content.rs

@@ -219,7 +219,8 @@ impl Content {
         match self {
             Self::Deleted(len) => *len,
             Self::Json(strings) => strings.len() as u64,
-            Self::String(string) => string.encode_utf16().count() as u64,
+            // TODO: need a custom wrapper with length cached, this cost too much
+            Self::String(string) => string.chars().map(|c| c.len_utf16()).sum::<usize>() as u64,
             Self::Any(any) => any.len() as u64,
             Self::Binary(_) | Self::Embed(_) | Self::Format { .. } | Self::Type(_) | Self::Doc { .. } => 1,
         }

y-octo/src/doc/codec/delete_set.rs

@@ -1,5 +1,5 @@
 use std::{
-    collections::{hash_map::Entry, HashMap},
+    collections::{hash_map::Entry, VecDeque},
     ops::{Deref, DerefMut, Range},
 };
 
@@ -28,10 +28,10 @@ impl<R: CrdtReader> CrdtRead<R> for OrderRange {
         if num_of_deletes == 1 {
             Ok(OrderRange::Range(Range::<u64>::read(decoder)?))
         } else {
-            let mut deletes = Vec::with_capacity(num_of_deletes);
+            let mut deletes = VecDeque::with_capacity(num_of_deletes);
 
             for _ in 0..num_of_deletes {
-                deletes.push(Range::<u64>::read(decoder)?);
+                deletes.push_back(Range::<u64>::read(decoder)?);
             }
 
             Ok(OrderRange::Fragment(deletes))
@@ -59,10 +59,10 @@ impl<W: CrdtWriter> CrdtWrite<W> for OrderRange {
 }
 
 #[derive(Debug, Default, Clone, PartialEq)]
-pub struct DeleteSet(pub HashMap<Client, OrderRange>);
+pub struct DeleteSet(pub ClientMap<OrderRange>);
 
 impl Deref for DeleteSet {
-    type Target = HashMap<Client, OrderRange>;
+    type Target = ClientMap<OrderRange>;
 
     fn deref(&self) -> &Self::Target {
         &self.0
@@ -71,7 +71,7 @@ impl Deref for DeleteSet {
 
 impl<const N: usize> From<[(Client, Vec<Range<u64>>); N]> for DeleteSet {
     fn from(value: [(Client, Vec<Range<u64>>); N]) -> Self {
-        let mut map = HashMap::with_capacity(N);
+        let mut map = ClientMap::with_capacity(N);
         for (client, ranges) in value {
             map.insert(client, ranges.into());
         }
@@ -106,7 +106,6 @@ impl DeleteSet {
         }
     }
 
-    #[allow(dead_code)]
     pub fn batch_push(&mut self, client: Client, ranges: Vec<Range<u64>>) {
         match self.0.entry(client) {
             Entry::Occupied(e) => {
@@ -136,7 +135,7 @@ impl<R: CrdtReader> CrdtRead<R> for DeleteSet {
     fn read(decoder: &mut R) -> JwstCodecResult<Self> {
         let num_of_clients = decoder.read_var_u64()? as usize;
         // See: [HASHMAP_SAFE_CAPACITY]
-        let mut map = HashMap::with_capacity(num_of_clients.min(HASHMAP_SAFE_CAPACITY));
+        let mut map = ClientMap::with_capacity(num_of_clients.min(HASHMAP_SAFE_CAPACITY));
 
         for _ in 0..num_of_clients {
             let client = decoder.read_var_u64()?;
@@ -188,7 +187,7 @@ mod tests {
         {
             let mut delete_set = delete_set;
             delete_set.add(1, 5, 10);
-            assert_eq!(delete_set.get(&1), Some(&OrderRange::Fragment(vec![0..15, 20..30])));
+            assert_eq!(delete_set.get(&1), Some(&OrderRange::from(vec![0..15, 20..30])));
         }
     }
 
@@ -210,7 +209,7 @@ mod tests {
         {
             let mut delete_set = delete_set;
             delete_set.batch_push(1, vec![40..50, 10..20]);
-            assert_eq!(delete_set.get(&1), Some(&OrderRange::Fragment(vec![0..30, 40..50])));
+            assert_eq!(delete_set.get(&1), Some(&OrderRange::from(vec![0..30, 40..50])));
         }
     }
 

y-octo/src/doc/codec/update.rs

@@ -1,20 +1,17 @@
-use std::{
-    collections::{HashMap, VecDeque},
-    ops::Range,
-};
+use std::{collections::VecDeque, ops::Range};
 
 use super::*;
 use crate::doc::StateVector;
 
 #[derive(Debug, Default, Clone)]
 pub struct Update {
-    pub(crate) structs: HashMap<u64, VecDeque<Node>>,
+    pub(crate) structs: ClientMap<VecDeque<Node>>,
     pub(crate) delete_set: DeleteSet,
 
     /// all unapplicable items that we can't integrate into doc
     /// any item with inconsistent id clock or missing dependency will be put
     /// here
-    pub(crate) pending_structs: HashMap<Client, VecDeque<Node>>,
+    pub(crate) pending_structs: ClientMap<VecDeque<Node>>,
     /// missing state vector after applying updates
     pub(crate) missing_state: StateVector,
     /// all unapplicable delete set
@@ -26,7 +23,7 @@ impl<R: CrdtReader> CrdtRead<R> for Update {
         let num_of_clients = decoder.read_var_u64()? as usize;
 
         // See: [HASHMAP_SAFE_CAPACITY]
-        let mut map = HashMap::with_capacity(num_of_clients.min(HASHMAP_SAFE_CAPACITY));
+        let mut map = ClientMap::with_capacity(num_of_clients.min(HASHMAP_SAFE_CAPACITY));
         for _ in 0..num_of_clients {
             let num_of_structs = decoder.read_var_u64()? as usize;
             let client = decoder.read_var_u64()?;
@@ -531,7 +528,7 @@ mod tests {
     fn test_update_iterator() {
         loom_model!({
             let mut update = Update {
-                structs: HashMap::from([
+                structs: ClientMap::from_iter([
                     (
                         0,
                         VecDeque::from([
@@ -571,7 +568,7 @@ mod tests {
         loom_model!({
             let mut update = Update {
                 // an item with higher sequence id than local state
-                structs: HashMap::from([(0, VecDeque::from([struct_item((0, 4), 1)]))]),
+                structs: ClientMap::from_iter([(0, VecDeque::from([struct_item((0, 4), 1)]))]),
                 ..Update::default()
             };
 
@@ -623,7 +620,7 @@ mod tests {
     fn should_add_skip_when_clock_not_continuous() {
         loom_model!({
             let update = Update {
-                structs: HashMap::from([(
+                structs: ClientMap::from_iter([(
                     0,
                     VecDeque::from([
                         struct_item((0, 0), 1),
@@ -655,7 +652,7 @@ mod tests {
     fn merged_update_should_not_be_released_in_next_turn() {
         loom_model!({
             let update = Update {
-                structs: HashMap::from([(
+                structs: ClientMap::from_iter([(
                     0,
                     VecDeque::from([
                         struct_item((0, 0), 1),
@@ -670,7 +667,7 @@ mod tests {
             let merged = Update::merge([update]);
 
             let update2 = Update {
-                structs: HashMap::from([(
+                structs: ClientMap::from_iter([(
                     0,
                     VecDeque::from([struct_item((0, 30), 1), Node::new_gc((0, 32).into(), 1)]),
                 )]),

y-octo/src/doc/common/mod.rs

@@ -5,3 +5,5 @@ mod state;
 pub use range::*;
 pub use somr::*;
 pub use state::*;
+
+use super::*;