mirrors/loro
1
0
Fork 0
mirror of https://github.com/loro-dev/loro.git synced 2024-11-25 04:38:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor: seal change exp impl

Browse source
This commit is contained in:
Zixuan Chen 2023-07-05 14:43:58 +08:00
parent 336bd1e497
commit f0f82fb581
22 changed files with 116 additions and 173 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
crates/bench-utils/src/lib.rs
View file

@ -69,7 +69,7 @@ pub fn gen_realtime_actions(action_num: usize, client_num: usize, seed: u64) ->
Action::Text { client, action } => {
*client %= client_num;
if !action.ins.is_empty() {
action.ins = (action.ins.as_bytes()[0] as u8).to_string();
action.ins = (action.ins.as_bytes()[0]).to_string();
}
}
Action::SyncAll => {

2
crates/loro-ffi/build.rs
View file

@ -2,7 +2,7 @@ fn main() {
let crate_dir = std::env::var("CARGO_MANIFEST_DIR").unwrap();
let config = cbindgen::Config::from_file("cbindgen.toml")
.expect("Unable to find cbindgen.toml configuration file");
cbindgen::generate_with_config(&crate_dir, config)
cbindgen::generate_with_config(crate_dir, config)
.unwrap()
.write_to_file("target/loro_ffi.h");
}

10
crates/loro-internal/benches/encode.rs
View file

@ -124,7 +124,7 @@ mod run {
mod import {
use criterion::Criterion;
use loro_internal::{change::ChangeMergeCfg, configure::Configure, LoroCore};
use loro_internal::{configure::Configure, LoroCore};
pub fn causal_iter(c: &mut Criterion) {
let mut b = c.benchmark_group("causal_iter");
@ -133,20 +133,12 @@ mod import {
b.iter(|| {
let mut c1 = LoroCore::new(
Configure {
change: ChangeMergeCfg {
max_change_length: 0,
max_change_interval: 0,
},
..Default::default()
},
Some(1),
);
let mut c2 = LoroCore::new(
Configure {
change: ChangeMergeCfg {
max_change_length: 0,
max_change_interval: 0,
},
..Default::default()
},
Some(2),

77
crates/loro-internal/src/change.rs
View file

@ -9,11 +9,11 @@ use crate::{
dag::DagNode,
id::{Counter, ID},
op::Op,
span::{HasId, HasIdSpan, HasLamport},
span::{HasId, HasLamport},
version::Frontiers,
};
use num::traits::AsPrimitive;
use rle::{HasIndex, HasLength, Mergable, Rle, RleVec, Sliceable};
use rle::{HasIndex, HasLength, Mergable, RleVec, Sliceable};
pub type Timestamp = i64;
pub type Lamport = u32;
@ -52,6 +52,14 @@ impl<O> Change<O> {
}
}
impl<O: Mergable + HasLength + HasIndex + Debug> HasIndex for Change<O> {
type Int = Counter;
fn get_start_index(&self) -> Self::Int {
self.id.counter
}
}
impl<O> HasId for Change<O> {
fn id_start(&self) -> ID {
self.id
@ -63,6 +71,23 @@ impl<O> HasLamport for Change<O> {
self.lamport
}
}
impl<O> Mergable for Change<O> {
fn is_mergable(&self, _other: &Self, _conf: &()) -> bool
where
Self: Sized,
{
false
}
fn merge(&mut self, _other: &Self, _conf: &())
where
Self: Sized,
{
unreachable!()
}
}
use std::fmt::Debug;
impl<O: Mergable + HasLength + HasIndex + Debug> HasLength for Change<O> {
fn content_len(&self) -> usize {
@ -70,54 +95,6 @@ impl<O: Mergable + HasLength + HasIndex + Debug> HasLength for Change<O> {
}
}
#[derive(Debug, Clone)]
pub struct ChangeMergeCfg {
pub max_change_length: usize,
pub max_change_interval: usize,
}
impl ChangeMergeCfg {
pub fn new() -> Self {
ChangeMergeCfg {
max_change_length: 1024,
max_change_interval: 60,
}
}
}
impl Default for ChangeMergeCfg {
fn default() -> Self {
Self {
max_change_length: 1024,
max_change_interval: 60,
}
}
}
impl<O: Rle + HasIndex> Mergable<ChangeMergeCfg> for Change<O> {
fn merge(&mut self, other: &Self, _: &ChangeMergeCfg) {
self.ops.merge(&other.ops, &());
}
fn is_mergable(&self, other: &Self, cfg: &ChangeMergeCfg) -> bool {
if other.deps.is_empty() || !(other.deps.len() == 1 && self.id_last() == other.deps[0]) {
return false;
}
if self.content_len() > cfg.max_change_length {
return false;
}
if other.timestamp - self.timestamp > cfg.max_change_interval as i64 {
return false;
}
self.id.peer == other.id.peer
&& self.id.counter + self.content_len() as Counter == other.id.counter
&& self.lamport + self.content_len() as Lamport == other.lamport
}
}
impl<O: Mergable + HasLength + Sliceable> Sliceable for Change<O> {
// TODO: feels slow, need to confirm whether this affects performance
fn slice(&self, from: usize, to: usize) -> Self {

5
crates/loro-internal/src/configure.rs
View file

@ -1,11 +1,10 @@
use std::{fmt::Debug, sync::Arc};
use crate::{change::ChangeMergeCfg, log_store::GcConfig, Timestamp};
use crate::{log_store::GcConfig, Timestamp};
use ring::rand::{SecureRandom, SystemRandom};
#[derive(Clone)]
pub struct Configure {
pub change: ChangeMergeCfg,
pub gc: GcConfig,
pub get_time: fn() -> Timestamp,
pub rand: Arc<dyn SecureRandomGenerator>,
@ -14,7 +13,6 @@ pub struct Configure {
impl Debug for Configure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Configure")
.field("change", &self.change)
.field("gc", &self.gc)
.field("get_time", &self.get_time)
.finish()
@ -57,7 +55,6 @@ impl SecureRandomGenerator for SystemRandom {
impl Default for Configure {
fn default() -> Self {
Self {
change: ChangeMergeCfg::default(),
gc: GcConfig::default(),
get_time: || 0,
rand: Arc::new(SystemRandom::new()),

1
crates/loro-internal/src/container/list/list_container.rs
View file

@ -27,7 +27,6 @@ use crate::{
prelim::Prelim,
transaction::Transaction,
value::LoroValue,
version::PatchedVersionVector,
LoroError, Transact, VersionVector,
};

5
crates/loro-internal/src/container/registry.rs
View file

@ -1,5 +1,5 @@
use std::{
ops::{Deref, DerefMut},
ops::Deref,
sync::{
atomic::{AtomicU32, Ordering},
Arc, Mutex, Weak,
@ -20,7 +20,6 @@ use crate::{
log_store::ImportContext,
op::{RemoteContent, RichOp},
transaction::Transaction,
version::PatchedVersionVector,
LoroError, LoroValue, Transact, VersionVector,
};
@ -346,6 +345,8 @@ impl ContainerRegistry {
#[cfg(feature = "test_utils")]
pub fn debug_inspect(&mut self) {
use std::ops::DerefMut;
for (_, ContainerAndId { container, id: _ }) in self.containers.iter_mut() {
if let ContainerInstance::Text(x) = container.try_lock().unwrap().deref_mut() {
x.debug_inspect()

6
crates/loro-internal/src/container/text/text_content.rs
View file

@ -5,11 +5,7 @@ use rle::{HasLength, Mergable, Sliceable};
use serde::{Deserialize, Serialize};
use smallvec::{smallvec, SmallVec};
use crate::{
delta::{DeltaItem, DeltaValue},
smstring::SmString,
LoroValue,
};
use crate::{delta::DeltaValue, smstring::SmString, LoroValue};
use super::string_pool::PoolString;

9
crates/loro-internal/src/dag/iter.rs
View file

@ -365,7 +365,6 @@ impl<'a, T: DagNode + 'a, D: Dag<Node = T>> Iterator for DagCausalIter<'a, D> {
#[cfg(test)]
mod test {
use crate::{
change::ChangeMergeCfg,
configure::Configure,
dag::DagUtils,
id::{Counter, ID},
@ -441,20 +440,12 @@ mod test {
fn parallel_case() {
let mut c1 = LoroCore::new(
Configure {
change: ChangeMergeCfg {
max_change_length: 0,
max_change_interval: 0,
},
..Default::default()
},
Some(1),
);
let mut c2 = LoroCore::new(
Configure {
change: ChangeMergeCfg {
max_change_length: 0,
max_change_interval: 0,
},
..Default::default()
},
Some(2),

39
crates/loro-internal/src/log_store.rs
View file

@ -16,10 +16,10 @@ use std::{
use fxhash::FxHashMap;
use rle::{HasLength, RleVec, RleVecWithIndex, Sliceable};
use rle::{HasLength, RleVec, Sliceable};
use crate::{
change::{Change, ChangeMergeCfg},
change::Change,
configure::Configure,
container::{
registry::{ContainerIdx, ContainerInstance, ContainerRegistry},
@ -59,7 +59,7 @@ impl GcConfig {
}
}
pub(crate) type ClientChanges = FxHashMap<PeerID, RleVecWithIndex<Change, ChangeMergeCfg>>;
pub(crate) type ClientChanges = FxHashMap<PeerID, RleVec<[Change; 0]>>;
pub(crate) type RemoteClientChanges = FxHashMap<PeerID, Vec<Change<RemoteOp>>>;
#[derive(Debug)]
@ -107,7 +107,7 @@ impl LogStore {
pub fn lookup_change(&self, id: ID) -> Option<&Change> {
self.changes.get(&id.peer).and_then(|changes| {
if id.counter <= changes.last().unwrap().id_last().counter {
Some(changes.get(id.counter as usize).unwrap().element)
Some(changes.get_by_atom_index(id.counter).unwrap().element)
} else {
None
}
@ -131,10 +131,7 @@ impl LogStore {
fn get_changes_slice(&self, id_span: IdSpan) -> Vec<Change> {
if let Some(changes) = self.changes.get(&id_span.client_id) {
let mut ans = Vec::with_capacity(id_span.atom_len() / 30);
for change in changes.slice_iter(
id_span.counter.min() as usize,
id_span.counter.norm_end() as usize,
) {
for change in changes.slice_iter(id_span.counter.min(), id_span.counter.norm_end()) {
let change = change.value.slice(change.start, change.end);
ans.push(change);
}
@ -273,10 +270,9 @@ impl LogStore {
self.latest_lamport = lamport + change.content_len() as u32 - 1;
self.latest_timestamp = timestamp;
self.vv.set_end(change.id_end());
let cfg = self.get_change_merge_cfg();
self.changes
.entry(self.this_client_id)
.or_insert_with(|| RleVecWithIndex::new_with_conf(cfg))
.or_default()
.push(change);
}
@ -295,7 +291,7 @@ impl LogStore {
self.changes
.get(&id.peer)
.map_or(0, |changes| changes.atom_len())
> id.counter as usize
> id.counter
}
#[inline]
@ -325,21 +321,6 @@ impl LogStore {
&self.vv
}
fn get_change_merge_cfg(&self) -> ChangeMergeCfg {
ChangeMergeCfg {
max_change_length: self.cfg.change.max_change_length,
max_change_interval: self.cfg.change.max_change_interval,
}
}
pub(crate) fn max_change_length(&mut self, max_change_length: usize) {
self.cfg.change.max_change_length = max_change_length
}
pub(crate) fn max_change_interval(&mut self, max_change_interval: usize) {
self.cfg.change.max_change_interval = max_change_interval
}
pub(crate) fn gc(&mut self, gc: bool) {
self.cfg.gc.gc = gc;
}
@ -355,12 +336,12 @@ impl LogStore {
self.changes.len(),
self.changes
.values()
.map(|v| format!("{}", v.vec().len()))
.map(|v| format!("{}", v.len()))
.collect::<Vec<_>>()
.join(", "),
self.changes
.values()
.map(|v| format!("{}", v.vec().iter().map(|x| x.ops.len()).sum::<usize>()))
.map(|v| format!("{}", v.iter().map(|x| x.ops.len()).sum::<usize>()))
.collect::<Vec<_>>()
.join(", "),
self.changes
@ -415,7 +396,7 @@ impl Dag for LogStore {
fn get(&self, id: ID) -> Option<&Self::Node> {
self.changes
.get(&id.peer)
.and_then(|x| x.get(id.counter as usize).map(|x| x.element))
.and_then(|x| x.get_by_atom_index(id.counter).map(|x| x.element))
}
fn frontier(&self) -> &[ID] {

16
crates/loro-internal/src/log_store/encoding/encode_snapshot.rs
View file

@ -2,13 +2,12 @@ use std::collections::VecDeque;
use fxhash::FxHashMap;
use itertools::Itertools;
use rle::{HasLength, RleVec, RleVecWithIndex};
use rle::{HasLength, RleVec, RleVecWithLen};
use serde::{Deserialize, Serialize};
use serde_columnar::{columnar, from_bytes, to_vec};
use smallvec::SmallVec;
use crate::{
change::{Change, ChangeMergeCfg},
change::Change,
container::text::text_content::SliceRange,
container::{
list::list_op::{DeleteSpan, InnerListOp},
@ -21,7 +20,7 @@ use crate::{
event::EventDiff,
hierarchy::Hierarchy,
id::{Counter, PeerID, ID},
log_store::{encoding::encode_changes::get_lamport_by_deps, ImportContext},
log_store::{encoding::encode_changes::get_lamport_by_deps, ClientChanges, ImportContext},
op::{InnerContent, Op},
span::HasLamportSpan,
version::{Frontiers, TotalOrderStamp},
@ -454,7 +453,7 @@ pub(super) fn decode_snapshot(
}
// calculate lamport
let mut lamport_map = FxHashMap::default();
let mut changes = FxHashMap::default();
let mut changes: ClientChanges = FxHashMap::default();
let mut client_ids: VecDeque<_> = changes_dq.keys().copied().collect();
let len = client_ids.len();
let mut loop_time = len;
@ -468,10 +467,7 @@ pub(super) fn decode_snapshot(
change.id.counter..change.id.counter + change.content_len() as Counter,
lamport,
));
changes
.entry(client_id)
.or_insert_with(|| RleVecWithIndex::new_with_conf(ChangeMergeCfg::new()))
.push(change);
changes.entry(client_id).or_default().push(change);
loop_time = len;
}
Err(_not_found_client) => {
@ -531,7 +527,7 @@ fn load_snapshot(
new_store: &mut LogStore,
new_hierarchy: &mut Hierarchy,
vv: VersionVector,
changes: FxHashMap<PeerID, RleVecWithIndex<Change, ChangeMergeCfg>>,
changes: ClientChanges,
containers: Vec<ContainerID>,
container_states: Vec<EncodedStateContent>,
keys: &[InternalString],

10
crates/loro-internal/src/log_store/import.rs
View file

@ -4,7 +4,6 @@ use crate::hierarchy::Hierarchy;
use crate::id::{Counter, PeerID, ID};
use crate::op::RemoteOp;
use crate::span::{CounterSpan, HasCounter, HasCounterSpan};
use crate::version::PatchedVersionVector;
use crate::LogStore;
use crate::{
container::registry::ContainerIdx,
@ -13,13 +12,12 @@ use crate::{
};
use itertools::Itertools;
use smallvec::{smallvec, SmallVec};
use std::sync::Arc;
use std::{collections::VecDeque, sync::MutexGuard};
use tracing::instrument;
use fxhash::{FxHashMap, FxHashSet};
use rle::{slice_vec_by, HasLength, RleVecWithIndex, Sliceable};
use rle::{slice_vec_by, HasLength, Sliceable};
use crate::{
container::{registry::ContainerInstance, ContainerID, ContainerTrait},
@ -173,7 +171,6 @@ impl LogStore {
next_vv.set_end(changes.last().unwrap().id_end());
}
// push changes to log stores
let cfg = self.get_change_merge_cfg();
for (client_id, changes) in changes.iter() {
let mut inner_changes = Vec::with_capacity(changes.len());
for change in changes.iter() {
@ -182,10 +179,7 @@ impl LogStore {
inner_changes.push(change);
}
let rle = self
.changes
.entry(*client_id)
.or_insert_with(|| RleVecWithIndex::new_cfg(cfg.clone()));
let rle = self.changes.entry(*client_id).or_default();
for change in inner_changes {
// if let Some(last) = rle.last() {
// assert_eq!(

17
crates/loro-internal/src/log_store/iter.rs
View file

@ -9,17 +9,15 @@ use crate::span::IdSpan;
use fxhash::FxHashMap;
use rle::HasLength;
use crate::change::ChangeMergeCfg;
use rle::RleVec;
use rle::RleVecWithLen;
use crate::change::Change;
use rle::RleVecWithIndex;
pub struct ClientOpIter<'a> {
pub(crate) change_index: usize,
pub(crate) op_index: usize,
pub(crate) changes: Option<&'a RleVecWithIndex<Change, ChangeMergeCfg>>,
pub(crate) changes: Option<&'a RleVec<[Change; 0]>>,
}
impl<'a> Iterator for ClientOpIter<'a> {
@ -50,14 +48,11 @@ pub struct OpSpanIter<'a> {
}
impl<'a> OpSpanIter<'a> {
pub fn new(
changes: &'a FxHashMap<PeerID, RleVecWithIndex<Change, ChangeMergeCfg>>,
target_span: IdSpan,
) -> Self {
pub fn new(changes: &'a FxHashMap<PeerID, RleVec<[Change; 0]>>, target_span: IdSpan) -> Self {
let rle_changes = changes.get(&target_span.client_id).unwrap();
let changes = rle_changes.vec();
let change_index = rle_changes
.get(target_span.id_start().counter as usize)
.get_by_atom_index(target_span.id_start().counter)
.map(|x| x.merged_index)
.unwrap_or(changes.len());
@ -67,7 +62,7 @@ impl<'a> OpSpanIter<'a> {
change_index,
op_index: rle_changes[change_index]
.ops
.get(target_span.counter.start)
.get_by_atom_index(target_span.counter.start)
.unwrap()
.merged_index,
}

13
crates/loro-internal/src/loro.rs
View file

@ -200,19 +200,6 @@ impl LoroCore {
}
// config
pub fn max_change_length(&mut self, max_change_length: usize) {
self.log_store
.write()
.unwrap()
.max_change_length(max_change_length);
}
pub fn max_change_interval(&mut self, max_change_interval: usize) {
self.log_store
.write()
.unwrap()
.max_change_interval(max_change_interval);
}
pub fn gc(&mut self, gc: bool) {
self.log_store.write().unwrap().gc(gc)

4
crates/loro-internal/src/refactor/arena.rs
View file

@ -1,6 +1,4 @@
use im::Vector;
use crate::container::{ContainerID, ContainerIdx};
use crate::container::ContainerID;
/// This is shared between [OpLog] and [AppState].
/// It uses a immutable data structure inside so that we have O(1) clone time.

2
crates/loro-internal/src/refactor/diff_calc.rs
View file

@ -1,6 +1,6 @@
use fxhash::FxHashMap;
use crate::container::{ContainerID, ContainerIdx};
use crate::container::ContainerID;
/// Calculate the diff between two versions. given [OpLog][super::oplog::OpLog]
/// and [AppState][super::state::AppState].

6
crates/loro-internal/src/refactor/oplog/dag.rs
View file

@ -72,7 +72,7 @@ impl Dag for AppDag {
} = id;
self.map
.get(&client_id)
.and_then(|rle| rle.get(counter).map(|x| x.element))
.and_then(|rle| rle.get_by_atom_index(counter).map(|x| x.element))
}
fn vv(&self) -> VersionVector {
@ -85,7 +85,7 @@ impl AppDag {
/// It's the version when the op is applied
pub fn get_vv(&self, id: ID) -> Option<ImVersionVector> {
self.map.get(&id.peer).and_then(|rle| {
rle.get(id.counter).map(|x| {
rle.get_by_atom_index(id.counter).map(|x| {
let mut vv = x.element.vv.clone();
vv.insert(id.peer, id.counter);
vv
@ -107,7 +107,7 @@ impl AppDag {
pub fn get_lamport(&self, id: &ID) -> Option<Lamport> {
self.map.get(&id.peer).and_then(|rle| {
rle.get(id.counter)
rle.get_by_atom_index(id.counter)
.map(|x| x.element.lamport + (id.counter - x.element.cnt) as Lamport)
})
}

7
crates/loro-internal/src/refactor/state.rs
View file

@ -1,12 +1,7 @@
use enum_dispatch::enum_dispatch;
use fxhash::FxHashMap;
use crate::{
container::{ContainerID, ContainerIdx},
event::Diff,
version::Frontiers,
VersionVector,
};
use crate::{container::ContainerID, event::Diff, version::Frontiers, VersionVector};
use super::arena::SharedArena;

1
crates/loro-wasm/src/lib.rs
View file

@ -112,7 +112,6 @@ impl Loro {
#[wasm_bindgen(constructor)]
pub fn new() -> Self {
let cfg: Configure = Configure {
change: Default::default(),
gc: GcConfig::default().with_gc(false),
get_time: || js_sys::Date::now() as i64,
rand: Arc::new(MathRandom),

1
crates/rle/src/range_map.rs
View file

@ -485,7 +485,6 @@ impl<Index: GlobalIndex, T: PartialEq + Eq> Mergable for WithStartEnd<Index, T>
#[cfg(test)]
mod test {
use std::ops::Range;
use super::*;
#[derive(Debug, PartialEq, Eq, Clone)]

54
crates/rle/src/rle_vec.rs
View file

@ -96,7 +96,7 @@ where
&self,
index: <A::Item as HasIndex>::Int,
) -> Option<SearchResult<'_, A::Item, <A::Item as HasIndex>::Int>> {
self.vec.get(index)
self.vec.get_by_atom_index(index)
}
pub fn iter_by_index(
@ -279,13 +279,13 @@ where
return SliceIterator::new_empty();
}
let start = self.get(from);
let start = self.get_by_atom_index(from);
if start.is_none() {
return SliceIterator::new_empty();
}
let start = start.unwrap();
let end = self.get(to);
let end = self.get_by_atom_index(to);
if let Some(end) = end {
SliceIterator {
vec: &self.vec,
@ -317,7 +317,7 @@ where
/// get the element at the given atom index.
/// return: (element, merged_index, offset)
pub fn get(
pub fn get_by_atom_index(
&self,
index: <A::Item as HasIndex>::Int,
) -> Option<SearchResult<'_, A::Item, <A::Item as HasIndex>::Int>> {
@ -355,10 +355,56 @@ where
})
}
pub fn slice_iter(
&self,
from: <A::Item as HasIndex>::Int,
to: <A::Item as HasIndex>::Int,
) -> SliceIterator<A::Item> {
if from == to || self.merged_len() == 0 {
return SliceIterator::new_empty();
}
let from_result = self.get_by_atom_index(from);
if from_result.is_none() {
return SliceIterator::new_empty();
}
let from_result = from_result.unwrap();
let to_result = if to == self.atom_len() {
None
} else {
self.get_by_atom_index(to)
};
if let Some(to_result) = to_result {
SliceIterator {
vec: &self.vec,
cur_index: from_result.merged_index,
cur_offset: from_result.offset.as_(),
end_index: Some(to_result.merged_index),
end_offset: Some(to_result.offset.as_()),
}
} else {
SliceIterator {
vec: &self.vec,
cur_index: from_result.merged_index,
cur_offset: from_result.offset.as_(),
end_index: None,
end_offset: None,
}
}
}
#[inline]
pub fn slice_merged(&self, range: Range<usize>) -> &[A::Item] {
&self.vec[range]
}
pub fn atom_len(&self) -> <A::Item as HasIndex>::Int {
self.vec
.last()
.map(|x| x.get_end_index())
.unwrap_or(<A::Item as HasIndex>::Int::from_usize(0).unwrap())
}
}
impl<A: Array> RleVec<A>
where

2
crates/rle/src/rle_vec_old.rs
View file

@ -345,7 +345,7 @@ impl<T, Cfg> Deref for RleVecWithIndex<T, Cfg> {
#[cfg(test)]
mod test {
mod prime_value {
use crate::{Mergable, RleVecWithIndex, Sliceable};
use crate::{Mergable, RleVecWithIndex};
impl Mergable for String {
fn is_mergable(&self, _: &Self, _: &()) -> bool {