516025458

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/maidsafe/crdt_tree/commit/<a class=hub.com/maidsafe/crdt_tree/commit/c266218129694f1541d37757b8599068f9bb95b4">c26621812<a href="https://github.com/maidsafe/crdt_tree/commit/c266218129694f1541d37757b8599068f9bb95b4">">Merge </a><a class="double-link" href="https://github.com/maidsafe/crdt_tree/commit/<a class="double-link" href="https://github.com/maidsafe/crdt_tree/commit/36a927f0cb6c0ba0ea2ce4de660d64a80ff88976">36a927f0c</a>">36a927f0c</a><a href="https://github.com/maidsafe/crdt_tree/commit/c266218129694f1541d37757b8599068f9bb95b4"> into 9d63596f7">9d63596f7</a>

Pull Request Pull Request #26: Adds a higher-level API with built-in lamport clock and causal threshold detection.

Run Details

0 of 53 new or added lines in 1 file covered. (0.0%)

101 of 324 relevant lines covered (31.17%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/treereplica.rs

// Copyright 2020 MaidSafe.net limited.
//
// This SAFE Network Software is licensed to you under the MIT license <LICENSE-MIT
// http://opensource.org/licenses/MIT> or the Modified BSD license <LICENSE-BSD
// https://opensource.org/licenses/BSD-3-Clause>, at your option. This file may not be copied,
// modified, or distributed except according to those terms. Please review the Licences for the
// specific language governing permissions and limitations relating to use of the SAFE Network
// Software.

extern crate crdts;

use super::{Clock, LogOpMove, OpMove, State, Tree, TreeId, TreeMeta};
use crdts::Actor;
use log::debug;
use std::collections::HashMap;

/// `TreeReplica` holds tree `State` plus lamport timestamp (actor + counter)
///
/// It can optionally keep track of the latest timestamp for each
/// replica which is needed for calculating the causally stable threshold which
/// is in turn needed for log truncation.
///
/// `TreeReplica` is a higher-level interface to the Tree CRDT and is tied to a
/// particular actor/peer.
///
/// `State` is a lower-level interface to the Tree CRDT and is not tied to any
/// actor/peer.
#[derive(Debug)]
pub struct TreeReplica<ID: TreeId, TM: TreeMeta, A: Actor> {
    state: State<ID, TM, A>, // Tree state
    time: Clock<A>,          // Lamport Clock for this replica/tree.

    latest_time_by_replica: HashMap<A, Clock<A>>,
}

impl<ID: TreeId, TM: TreeMeta, A: Actor + std::fmt::Debug> TreeReplica<ID, TM, A> {
    /// returns new TreeReplica
    pub fn new(id: A) -> Self {
        Self {
            state: State::new(),
            time: Clock::<A>::new(id, None),
            latest_time_by_replica: HashMap::<A, Clock<A>>::new(),
        }
    }

    /// Generates an OpMove
    ///
    /// Note that OpMove::timestamp is incremented from TreeReplica::time.
    /// TreeReplica::time is not updated until ::apply_op() is called.
    ///
    /// Therefore, multiple ops generated with this method may share the same
    /// timestamp, and only one can be sucessfully applied.
    ///
    /// To generate multiple ops before calling ::apply_op(), use ::opmoves() instead.
    pub fn opmove(&self, parent_id: ID, metadata: TM, child_id: ID) -> OpMove<ID, TM, A> {
        OpMove::new(self.time.inc(), parent_id, metadata, child_id)
    }

    /// Generates a list of OpMove from a list of tuples (child_id, metadata, parent_id)
    ///
    /// Each OpMove::timestamp will be greater than the previous op in the returned list.
    /// Therefore, these operations can be successfully applied via ::apply_op() without
    /// timestamp collision.
    pub fn opmoves(&self, ops: Vec<(ID, TM, ID)>) -> Vec<OpMove<ID, TM, A>> {
        let mut time = self.time.clone();

        let mut opmoves = vec![];

        for op in ops {
            opmoves.push(OpMove::new(time.tick(), op.0, op.1, op.2));
        }
        opmoves
    }

    /// Returns actor ID for this replica
    #[inline]
    pub fn id(&self) -> &A {
        self.time.actor_id()
    }

    /// Returns the latest lamport time seen by this replica
    #[inline]
    pub fn time(&self) -> &Clock<A> {
        &self.time
    }

    /// Returns Tree State reference
    #[inline]
    pub fn state(&self) -> &State<ID, TM, A> {
        &self.state
    }

    /// Returns Tree reference
    #[inline]
    pub fn tree(&self) -> &Tree<ID, TM> {
        self.state.tree()
    }

    /// Returns mutable Tree reference
    ///
    /// Warning: this is dangerous.  Normally the `Tree` should
    /// not be mutated directly.
    ///
    /// See the demo_move_to_trash in examples/demo.rs for a
    /// use-case, only after log truncation has been performed.
    #[inline]
    pub fn tree_mut(&mut self) -> &mut Tree<ID, TM> {
        self.state.tree_mut()
    }

    /// Applies single operation to `State` and updates our time clock
    ///
    /// Also records latest timestamp for each replica if
    /// track_causally_stable_threshold flag is set.
    pub fn apply_op(&mut self, op: OpMove<ID, TM, A>) {
        self.time = self.time.merge(op.timestamp());

        // store latest timestamp for this actor.
        // This is only needed for calculation of causally_stable_threshold.
        let id = op.timestamp().actor_id();
        match self.latest_time_by_replica.get(id) {
            Some(latest) if (op.timestamp() <= latest) => {
                debug!(
                    "Clock not increased, current timestamp {:?}, provided is {:?}, dropping op!",
                    latest,
                    op.timestamp()
                );
            }
            _ => {
                self.latest_time_by_replica
                    .insert(op.timestamp().actor_id().clone(), op.timestamp().clone());
            }
        };

        self.state.apply_op(op);
    }

    /// Applies list of operations
    pub fn apply_ops(&mut self, ops: Vec<OpMove<ID, TM, A>>) {
        for op in ops {
            self.apply_op(op);
        }
    }

    /// Applies list of operations without taking ownership
    pub fn apply_ops_byref(&mut self, ops: &[OpMove<ID, TM, A>]) {
        self.apply_ops(ops.to_vec())
    }

    /// applies op from a log.  useful for log replay.
    pub fn apply_log_op(&mut self, log_op: LogOpMove<ID, TM, A>) {
        self.apply_op(log_op.into());
    }

    /// applies ops from a log.  useful for log replay.
    pub fn apply_log_ops(&mut self, log_ops: Vec<LogOpMove<ID, TM, A>>) {
        for log_op in log_ops {
            self.apply_log_op(log_op);
        }
    }

    /// returns the causally stable threshold
    pub fn causally_stable_threshold(&self) -> Option<&Clock<A>> {
        // The minimum of latest timestamp from each replica
        // is the causally stable threshold.

        let mut v: Vec<&Clock<A>> = self.latest_time_by_replica.values().collect();
        v.sort();
        v.reverse(); // reverse, so last is lowest.
        v.pop()
    }

    /// truncates log
    pub fn truncate_log(&mut self) -> bool {
        let result = self.causally_stable_threshold();
        match result.cloned() {
            Some(t) => self.state.truncate_log_before(&t),
            None => false,
        }
    }
}

1	// Copyright 2020 MaidSafe.net limited.
2	//
3	// This SAFE Network Software is licensed to you under the MIT license <LICENSE-MIT
4	// http://opensource.org/licenses/MIT> or the Modified BSD license <LICENSE-BSD
5	// https://opensource.org/licenses/BSD-3-Clause>, at your option. This file may not be copied,
6	// modified, or distributed except according to those terms. Please review the Licences for the
7	// specific language governing permissions and limitations relating to use of the SAFE Network
8	// Software.
9
10	extern crate crdts;
11
12	use super::{Clock, LogOpMove, OpMove, State, Tree, TreeId, TreeMeta};
13	use crdts::Actor;
14	use log::debug;
15	use std::collections::HashMap;
16
17	/// `TreeReplica` holds tree `State` plus lamport timestamp (actor + counter)
18	///
19	/// It can optionally keep track of the latest timestamp for each
20	/// replica which is needed for calculating the causally stable threshold which
21	/// is in turn needed for log truncation.
22	///
23	/// `TreeReplica` is a higher-level interface to the Tree CRDT and is tied to a
24	/// particular actor/peer.
25	///
26	/// `State` is a lower-level interface to the Tree CRDT and is not tied to any
27	/// actor/peer.
28	#[derive(Debug)]
29	pub struct TreeReplica<ID: TreeId, TM: TreeMeta, A: Actor> {
30	state: State<ID, TM, A>, // Tree state
31	time: Clock<A>, // Lamport Clock for this replica/tree.
32
33	latest_time_by_replica: HashMap<A, Clock<A>>,
34	}
35
36	impl<ID: TreeId, TM: TreeMeta, A: Actor + std::fmt::Debug> TreeReplica<ID, TM, A> {
37	/// returns new TreeReplica
NEW 38	pub fn new(id: A) -> Self {	×
39	Self {
NEW 40	state: State::new(),	×
NEW 41	time: Clock::<A>::new(id, None),	×
NEW 42	latest_time_by_replica: HashMap::<A, Clock<A>>::new(),	×
43	}
44	}
45
46	/// Generates an OpMove
47	///
48	/// Note that OpMove::timestamp is incremented from TreeReplica::time.
49	/// TreeReplica::time is not updated until ::apply_op() is called.
50	///
51	/// Therefore, multiple ops generated with this method may share the same
52	/// timestamp, and only one can be sucessfully applied.
53	///
54	/// To generate multiple ops before calling ::apply_op(), use ::opmoves() instead.
NEW 55	pub fn opmove(&self, parent_id: ID, metadata: TM, child_id: ID) -> OpMove<ID, TM, A> {	×
NEW 56	OpMove::new(self.time.inc(), parent_id, metadata, child_id)	×
57	}
58
59	/// Generates a list of OpMove from a list of tuples (child_id, metadata, parent_id)
60	///
61	/// Each OpMove::timestamp will be greater than the previous op in the returned list.
62	/// Therefore, these operations can be successfully applied via ::apply_op() without
63	/// timestamp collision.
NEW 64	pub fn opmoves(&self, ops: Vec<(ID, TM, ID)>) -> Vec<OpMove<ID, TM, A>> {	×
NEW 65	let mut time = self.time.clone();	×
66
NEW 67	let mut opmoves = vec![];	×
68
NEW 69	for op in ops {	×
NEW 70	opmoves.push(OpMove::new(time.tick(), op.0, op.1, op.2));	×
71	}
NEW 72	opmoves	×
73	}
74
75	/// Returns actor ID for this replica
76	#[inline]
NEW 77	pub fn id(&self) -> &A {	×
NEW 78	self.time.actor_id()	×
79	}
80
81	/// Returns the latest lamport time seen by this replica
82	#[inline]
NEW 83	pub fn time(&self) -> &Clock<A> {	×
NEW 84	&self.time	×
85	}
86
87	/// Returns Tree State reference
88	#[inline]
NEW 89	pub fn state(&self) -> &State<ID, TM, A> {	×
NEW 90	&self.state	×
91	}
92
93	/// Returns Tree reference
94	#[inline]
NEW 95	pub fn tree(&self) -> &Tree<ID, TM> {	×
NEW 96	self.state.tree()	×
97	}
98
99	/// Returns mutable Tree reference
100	///
101	/// Warning: this is dangerous. Normally the `Tree` should
102	/// not be mutated directly.
103	///
104	/// See the demo_move_to_trash in examples/demo.rs for a
105	/// use-case, only after log truncation has been performed.
106	#[inline]
NEW 107	pub fn tree_mut(&mut self) -> &mut Tree<ID, TM> {	×
NEW 108	self.state.tree_mut()	×
109	}
110
111	/// Applies single operation to `State` and updates our time clock
112	///
113	/// Also records latest timestamp for each replica if
114	/// track_causally_stable_threshold flag is set.
NEW 115	pub fn apply_op(&mut self, op: OpMove<ID, TM, A>) {	×
NEW 116	self.time = self.time.merge(op.timestamp());	×
117
118	// store latest timestamp for this actor.
119	// This is only needed for calculation of causally_stable_threshold.
NEW 120	let id = op.timestamp().actor_id();	×
NEW 121	match self.latest_time_by_replica.get(id) {	×
NEW 122	Some(latest) if (op.timestamp() <= latest) => {	×
NEW 123	debug!(	×
124	"Clock not increased, current timestamp {:?}, provided is {:?}, dropping op!",
NEW 125	latest,	×
NEW 126	op.timestamp()	×
127	);
128	}
NEW 129	_ => {	×
NEW 130	self.latest_time_by_replica	×
NEW 131	.insert(op.timestamp().actor_id().clone(), op.timestamp().clone());	×
132	}
133	};
134
NEW 135	self.state.apply_op(op);	×
136	}
137
138	/// Applies list of operations
NEW 139	pub fn apply_ops(&mut self, ops: Vec<OpMove<ID, TM, A>>) {	×
NEW 140	for op in ops {	×
NEW 141	self.apply_op(op);	×
142	}
143	}
144
145	/// Applies list of operations without taking ownership
NEW 146	pub fn apply_ops_byref(&mut self, ops: &[OpMove<ID, TM, A>]) {	×
NEW 147	self.apply_ops(ops.to_vec())	×
148	}
149
150	/// applies op from a log. useful for log replay.
NEW 151	pub fn apply_log_op(&mut self, log_op: LogOpMove<ID, TM, A>) {	×
NEW 152	self.apply_op(log_op.into());	×
153	}
154
155	/// applies ops from a log. useful for log replay.
NEW 156	pub fn apply_log_ops(&mut self, log_ops: Vec<LogOpMove<ID, TM, A>>) {	×
NEW 157	for log_op in log_ops {	×
NEW 158	self.apply_log_op(log_op);	×
159	}
160	}
161
162	/// returns the causally stable threshold
NEW 163	pub fn causally_stable_threshold(&self) -> Option<&Clock<A>> {	×
164	// The minimum of latest timestamp from each replica
165	// is the causally stable threshold.
166
NEW 167	let mut v: Vec<&Clock<A>> = self.latest_time_by_replica.values().collect();	×
NEW 168	v.sort();	×
169	v.reverse(); // reverse, so last is lowest.
NEW 170	v.pop()	×
171	}
172
173	/// truncates log
NEW 174	pub fn truncate_log(&mut self) -> bool {	×
NEW 175	let result = self.causally_stable_threshold();	×
NEW 176	match result.cloned() {	×
NEW 177	Some(t) => self.state.truncate_log_before(&t),	×
NEW 178	None => false,	×
179	}
180	}
181	}

maidsafe / crdt_tree / 516025458

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous