perf: improve apply update performance (#34)

this PR partially solves #33 and greatly improves the performance of
apply update
test cases show a performance improvement of about 40-50%

.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::*;