13196351565

Committed 07 Feb 2025 08:49AM UTC coverage: 91.378% (+0.4%) from 90.988%

Build # 13196351565

Build Type

Pull #165

github

Committed by

web-flow

Commit Message

Merge c0d5658d0 into 8c45a664c

Pull Request Pull Request #165: Implement DecodeWithMemTracking

Run Details

31 of 32 new or added lines in 7 files covered. (96.88%)

6 existing lines in 4 files now uncovered.

2798 of 3062 relevant lines covered (91.38%)

20975.19 hits per line

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

92.39

/src/testing.rs

// Copyright 2018-2019 Parity Technologies (UK) Ltd
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Helpers for testing

pub mod chain {
        use crate::{
                std::{collections::BTreeMap, vec::Vec},
                Chain, Error,
        };

        pub const GENESIS_HASH: &str = "genesis";
        const NULL_HASH: &str = "NULL";

        struct BlockRecord {
                number: u32,
                parent: &'static str,
        }

        pub struct DummyChain {
                inner: BTreeMap<&'static str, BlockRecord>,
                leaves: Vec<&'static str>,
                finalized: (&'static str, u32),
        }

        impl DummyChain {
                pub fn new() -> Self {
                        let mut inner = BTreeMap::new();
                        inner.insert(GENESIS_HASH, BlockRecord { number: 1, parent: NULL_HASH });

                        DummyChain { inner, leaves: vec![GENESIS_HASH], finalized: (GENESIS_HASH, 1) }
                }

                pub fn push_blocks(&mut self, mut parent: &'static str, blocks: &[&'static str]) {
                        if blocks.is_empty() {
                                return
                        }

                        let base_number = self.inner.get(parent).unwrap().number + 1;

                        if let Some(pos) = self.leaves.iter().position(|x| x == &parent) {
                                self.leaves.remove(pos);
                        }

                        for (i, descendent) in blocks.iter().enumerate() {
                                self.inner
                                        .insert(descendent, BlockRecord { number: base_number + i as u32, parent });

                                parent = descendent;
                        }

                        let new_leaf = blocks.last().unwrap();
                        let new_leaf_number = self.inner.get(new_leaf).unwrap().number;

                        let insertion_index = self
                                .leaves
                                .binary_search_by(|x| {
                                        self.inner.get(x).unwrap().number.cmp(&new_leaf_number).reverse()
                                })
                                .unwrap_or_else(|i| i);

                        self.leaves.insert(insertion_index, new_leaf);
                }

                pub fn number(&self, hash: &'static str) -> u32 {
                        self.inner.get(hash).unwrap().number
                }

                pub fn last_finalized(&self) -> (&'static str, u32) {
                        self.finalized
                }

                pub fn set_last_finalized(&mut self, last_finalized: (&'static str, u32)) {
                        self.finalized = last_finalized;
                }

                pub fn best_chain_containing(&self, base: &'static str) -> Option<(&'static str, u32)> {
                        let base_number = self.inner.get(base)?.number;

                        for leaf in &self.leaves {
                                // leaves are in descending order.
                                let leaf_number = self.inner.get(leaf).unwrap().number;
                                if leaf_number < base_number {
                                        break
                                }

                                if leaf == &base {
                                        return Some((leaf, leaf_number))
                                }

                                if self.ancestry(base, leaf).is_ok() {
                                        return Some((leaf, leaf_number))
                                }
                        }

                        None
                }
        }

        impl Chain<&'static str, u32> for DummyChain {
                fn ancestry(
                        &self,
                        base: &'static str,
                        mut block: &'static str,
                ) -> Result<Vec<&'static str>, Error> {
                        let mut ancestry = Vec::new();

                        loop {
                                match self.inner.get(block) {
                                        None => return Err(Error::NotDescendent),
                                        Some(record) => {
                                                block = record.parent;
                                        },
                                }

                                if block == NULL_HASH {
                                        return Err(Error::NotDescendent)
                                }
                                if block == base {
                                        break
                                }

                                ancestry.push(block);
                        }

                        Ok(ancestry)
                }
        }
}

#[cfg(feature = "std")]
pub mod environment {
        use super::chain::*;
        use crate::{
                round::State as RoundState,
                voter::{Callback, CommunicationIn, CommunicationOut, RoundData},
                Chain, Commit, Equivocation, Error, HistoricalVotes, Message, Precommit, Prevote,
                PrimaryPropose, SignedMessage,
        };
        use futures::{
                channel::mpsc::{self, UnboundedReceiver, UnboundedSender},
                prelude::*,
        };
        use futures_timer::Delay;
        use parking_lot::Mutex;
        use std::{
                collections::HashMap,
                pin::Pin,
                sync::Arc,
                task::{Context, Poll},
                time::Duration,
        };

        #[derive(Hash, Debug, Clone, Copy, PartialEq, Eq, Ord, PartialOrd)]
        pub struct Id(pub u32);

        #[derive(Debug, Clone, PartialEq, Eq)]
        pub struct Signature(pub u32);

        pub struct Environment {
                chain: Mutex<DummyChain>,
                local_id: Id,
                network: Network,
                listeners: Mutex<
                        Vec<UnboundedSender<(&'static str, u32, Commit<&'static str, u32, Signature, Id>)>>,
                >,
                last_completed_and_concluded: Mutex<(u64, u64)>,
        }

        impl Environment {
                pub fn new(network: Network, local_id: Id) -> Self {
                        Environment {
                                chain: Mutex::new(DummyChain::new()),
                                local_id,
                                network,
                                listeners: Mutex::new(Vec::new()),
                                last_completed_and_concluded: Mutex::new((0, 0)),
                        }
                }

                pub fn with_chain<F, U>(&self, f: F) -> U
                where
                        F: FnOnce(&mut DummyChain) -> U,
                {
                        let mut chain = self.chain.lock();
                        f(&mut chain)
                }

                /// Stream of finalized blocks.
                pub fn finalized_stream(
                        &self,
                ) -> UnboundedReceiver<(&'static str, u32, Commit<&'static str, u32, Signature, Id>)> {
                        let (tx, rx) = mpsc::unbounded();
                        self.listeners.lock().push(tx);
                        rx
                }

                /// Get the last completed and concluded rounds.
                pub fn last_completed_and_concluded(&self) -> (u64, u64) {
                        *self.last_completed_and_concluded.lock()
                }
        }

        impl Chain<&'static str, u32> for Environment {
                fn ancestry(
                        &self,
                        base: &'static str,
                        block: &'static str,
                ) -> Result<Vec<&'static str>, Error> {
                        self.chain.lock().ancestry(base, block)
                }
        }

        impl crate::voter::Environment<&'static str, u32> for Environment {
                type Timer = Box<dyn Future<Output = Result<(), Error>> + Unpin + Send>;
                type BestChain =
                        Box<dyn Future<Output = Result<Option<(&'static str, u32)>, Error>> + Unpin + Send>;
                type Id = Id;
                type Signature = Signature;
                type In = Box<
                        dyn Stream<Item = Result<SignedMessage<&'static str, u32, Signature, Id>, Error>>
                                + Unpin
                                + Send,
                >;
                type Out = Pin<Box<dyn Sink<Message<&'static str, u32>, Error = Error> + Send>>;
                type Error = Error;

                fn best_chain_containing(&self, base: &'static str) -> Self::BestChain {
                        Box::new(future::ok(self.chain.lock().best_chain_containing(base)))
                }

                fn round_data(&self, round: u64) -> RoundData<Self::Id, Self::Timer, Self::In, Self::Out> {
                        const GOSSIP_DURATION: Duration = Duration::from_millis(500);

                        let (incoming, outgoing) = self.network.make_round_comms(round, self.local_id);
                        RoundData {
                                voter_id: Some(self.local_id),
                                prevote_timer: Box::new(Delay::new(GOSSIP_DURATION).map(Ok)),
                                precommit_timer: Box::new(Delay::new(GOSSIP_DURATION + GOSSIP_DURATION).map(Ok)),
                                incoming: Box::new(incoming),
                                outgoing: Box::pin(outgoing),
                        }
                }

                fn round_commit_timer(&self) -> Self::Timer {
                        use rand::Rng;

                        const COMMIT_DELAY_MILLIS: u64 = 100;

                        let delay = Duration::from_millis(rand::thread_rng().gen_range(0..COMMIT_DELAY_MILLIS));

                        Box::new(Delay::new(delay).map(Ok))
                }

                fn completed(
                        &self,
                        round: u64,
                        _state: RoundState<&'static str, u32>,
                        _base: (&'static str, u32),
                        _votes: &HistoricalVotes<&'static str, u32, Self::Signature, Self::Id>,
                ) -> Result<(), Error> {
                        self.last_completed_and_concluded.lock().0 = round;
                        Ok(())
                }

                fn concluded(
                        &self,
                        round: u64,
                        _state: RoundState<&'static str, u32>,
                        _base: (&'static str, u32),
                        _votes: &HistoricalVotes<&'static str, u32, Self::Signature, Self::Id>,
                ) -> Result<(), Error> {
                        self.last_completed_and_concluded.lock().1 = round;
                        Ok(())
                }

                fn finalize_block(
                        &self,
                        hash: &'static str,
                        number: u32,
                        _round: u64,
                        commit: Commit<&'static str, u32, Signature, Id>,
                ) -> Result<(), Error> {
                        let mut chain = self.chain.lock();

                        let last_finalized = chain.last_finalized();
                        if number <= last_finalized.1 {
                                panic!("Attempted to finalize backwards")
                        }

                        assert!(
                                chain.ancestry(last_finalized.0, hash).is_ok(),
                                "Safety violation: reverting finalized block.",
                        );

                        chain.set_last_finalized((hash, number));
                        self.listeners
                                .lock()
                                .retain(|s| s.unbounded_send((hash, number as _, commit.clone())).is_ok());

                        Ok(())
                }

                fn proposed(
                        &self,
                        _round: u64,
                        _propose: PrimaryPropose<&'static str, u32>,
                ) -> Result<(), Self::Error> {
                        Ok(())
                }

                fn prevoted(
                        &self,
                        _round: u64,
                        _prevote: Prevote<&'static str, u32>,
                ) -> Result<(), Self::Error> {
                        Ok(())
                }

                fn precommitted(
                        &self,
                        _round: u64,
                        _precommit: Precommit<&'static str, u32>,
                ) -> Result<(), Self::Error> {
                        Ok(())
                }

                fn prevote_equivocation(
                        &self,
                        round: u64,
                        equivocation: Equivocation<Id, Prevote<&'static str, u32>, Signature>,
                ) {
                        panic!("Encountered equivocation in round {}: {:?}", round, equivocation);
                }

                fn precommit_equivocation(
                        &self,
                        round: u64,
                        equivocation: Equivocation<Id, Precommit<&'static str, u32>, Signature>,
                ) {
                        panic!("Encountered equivocation in round {}: {:?}", round, equivocation);
                }
        }

        // p2p network data for a round.
        struct BroadcastNetwork<M> {
                receiver: UnboundedReceiver<M>,
                raw_sender: UnboundedSender<M>,
                senders: Vec<UnboundedSender<M>>,
                history: Vec<M>,
        }

        impl<M: Clone> BroadcastNetwork<M> {
                fn new() -> Self {
                        let (tx, rx) = mpsc::unbounded();
                        BroadcastNetwork {
                                receiver: rx,
                                raw_sender: tx,
                                senders: Vec::new(),
                                history: Vec::new(),
                        }
                }

                pub fn send_message(&self, message: M) {
                        let _ = self.raw_sender.unbounded_send(message);
                }

                // add a node to the network for a round.
                fn add_node<N, F: Fn(N) -> M>(
                        &mut self,
                        f: F,
                ) -> (impl Stream<Item = Result<M, Error>>, impl Sink<N, Error = Error>) {
                        let (tx, rx) = mpsc::unbounded();
                        let messages_out = self
                                .raw_sender
                                .clone()
                                .sink_map_err(|e| panic!("Error sending messages: {:?}", e))
                                .with(move |message| future::ready(Ok(f(message))));

                        // get history to the node.
                        for prior_message in self.history.iter().cloned() {
                                let _ = tx.unbounded_send(prior_message);
                        }

                        self.senders.push(tx);

                        (rx.map(Ok), messages_out)
                }

                // do routing work
                fn route(&mut self, cx: &mut Context) -> Poll<()> {
                        loop {
                                match Stream::poll_next(Pin::new(&mut self.receiver), cx) {
                                        Poll::Pending => return Poll::Pending,
                                        Poll::Ready(None) => return Poll::Ready(()),
                                        Poll::Ready(Some(item)) => {
                                                self.history.push(item.clone());
                                                for sender in &self.senders {
                                                        let _ = sender.unbounded_send(item.clone());
                                                }
                                        },
                                }
                        }
                }
        }

        /// Make a test network.
        /// Give the network future to node environments and spawn the routing task
        /// to run.
        pub fn make_network() -> (Network, NetworkRouting) {
                let global_messages = Arc::new(Mutex::new(GlobalMessageNetwork::new()));
                let rounds = Arc::new(Mutex::new(HashMap::new()));
                (
                        Network { global_messages: global_messages.clone(), rounds: rounds.clone() },
                        NetworkRouting { global_messages, rounds },
                )
        }

        type RoundNetwork = BroadcastNetwork<SignedMessage<&'static str, u32, Signature, Id>>;
        type GlobalMessageNetwork = BroadcastNetwork<CommunicationIn<&'static str, u32, Signature, Id>>;

        /// A test network. Instantiate this with `make_network`,
        #[derive(Clone)]
        pub struct Network {
                rounds: Arc<Mutex<HashMap<u64, RoundNetwork>>>,
                global_messages: Arc<Mutex<GlobalMessageNetwork>>,
        }

        impl Network {
                pub fn make_round_comms(
                        &self,
                        round_number: u64,
                        node_id: Id,
                ) -> (
                        impl Stream<Item = Result<SignedMessage<&'static str, u32, Signature, Id>, Error>>,
                        impl Sink<Message<&'static str, u32>, Error = Error>,
                ) {
                        let mut rounds = self.rounds.lock();
                        rounds
                                .entry(round_number)
                                .or_insert_with(RoundNetwork::new)
                                .add_node(move |message| SignedMessage {
                                        message,
                                        signature: Signature(node_id.0),
                                        id: node_id,
                                })
                }

                pub fn make_global_comms(
                        &self,
                ) -> (
                        impl Stream<Item = Result<CommunicationIn<&'static str, u32, Signature, Id>, Error>>,
                        impl Sink<CommunicationOut<&'static str, u32, Signature, Id>, Error = Error>,
                ) {
                        let mut global_messages = self.global_messages.lock();
                        global_messages.add_node(|message| match message {
                                CommunicationOut::Commit(r, commit) =>
                                        CommunicationIn::Commit(r, commit.into(), Callback::Blank),
                        })
                }

                /// Send a message to all nodes.
                pub fn send_message(&self, message: CommunicationIn<&'static str, u32, Signature, Id>) {
                        self.global_messages.lock().send_message(message);
                }
        }

        /// the network routing task.
        pub struct NetworkRouting {
                rounds: Arc<Mutex<HashMap<u64, RoundNetwork>>>,
                global_messages: Arc<Mutex<GlobalMessageNetwork>>,
        }

        impl Future for NetworkRouting {
                type Output = ();

                fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<()> {
                        let mut rounds = self.rounds.lock();
                        rounds.retain(|_, round| match round.route(cx) {
                                Poll::Ready(()) => false,
                                Poll::Pending => true,
                        });

                        let mut global_messages = self.global_messages.lock();
                        let _ = global_messages.route(cx);

                        Poll::Pending
                }
        }
}

1	// Copyright 2018-2019 Parity Technologies (UK) Ltd
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	//! Helpers for testing
16
17	pub mod chain {
18	use crate::{
19	std::{collections::BTreeMap, vec::Vec},
20	Chain, Error,
21	};
22
23	pub const GENESIS_HASH: &str = "genesis";
24	const NULL_HASH: &str = "NULL";
25
26	struct BlockRecord {
27	number: u32,
28	parent: &'static str,
29	}
30
31	pub struct DummyChain {
32	inner: BTreeMap<&'static str, BlockRecord>,
33	leaves: Vec<&'static str>,
34	finalized: (&'static str, u32),
35	}
36
37	impl DummyChain {
38	pub fn new() -> Self {	76✔
39	let mut inner = BTreeMap::new();	76✔
40	inner.insert(GENESIS_HASH, BlockRecord { number: 1, parent: NULL_HASH });	76✔
41
42	DummyChain { inner, leaves: vec![GENESIS_HASH], finalized: (GENESIS_HASH, 1) }	76✔
43	}	76✔
44
45	pub fn push_blocks(&mut self, mut parent: &'static str, blocks: &[&'static str]) {	140✔
46	if blocks.is_empty() {	140✔
47	return
48	}
49
50	let base_number = self.inner.get(parent).unwrap().number + 1;	140✔
51
52	if let Some(pos) = self.leaves.iter().position(\|x\| x == &parent) {	306✔
53	self.leaves.remove(pos);	106✔
54	}
55
56	for (i, descendent) in blocks.iter().enumerate() {	698✔
57	self.inner	1,116✔
58	.insert(descendent, BlockRecord { number: base_number + i as u32, parent });	558✔
59
60	parent = descendent;	558✔
61	}
62
63	let new_leaf = blocks.last().unwrap();	140✔
64	let new_leaf_number = self.inner.get(new_leaf).unwrap().number;	140✔
65
66	let insertion_index = self	280✔
67	.leaves
68	.binary_search_by(\|x\| {	200✔
69	self.inner.get(x).unwrap().number.cmp(&new_leaf_number).reverse()	60✔
70	})	60✔
71	.unwrap_or_else(\|i\| i);	108✔
72
73	self.leaves.insert(insertion_index, new_leaf);	140✔
74	}	140✔
75
76	pub fn number(&self, hash: &'static str) -> u32 {	40✔
77	self.inner.get(hash).unwrap().number	40✔
78	}	40✔
79
80	pub fn last_finalized(&self) -> (&'static str, u32) {	84✔
81	self.finalized	84✔
82	}	84✔
83
84	pub fn set_last_finalized(&mut self, last_finalized: (&'static str, u32)) {	42✔
85	self.finalized = last_finalized;	42✔
86	}	42✔
87
88	pub fn best_chain_containing(&self, base: &'static str) -> Option<(&'static str, u32)> {	42✔
89	let base_number = self.inner.get(base)?.number;	42✔
90
91	for leaf in &self.leaves {	42✔
92	// leaves are in descending order.
93	let leaf_number = self.inner.get(leaf).unwrap().number;	42✔
94	if leaf_number < base_number {	42✔
95	break
96	}
97
98	if leaf == &base {	42✔
99	return Some((leaf, leaf_number))	4✔
100	}
101
102	if self.ancestry(base, leaf).is_ok() {	38✔
103	return Some((leaf, leaf_number))	38✔
104	}
105	}
106
107	None	×
108	}	42✔
109	}
110
111	impl Chain<&'static str, u32> for DummyChain {
112	fn ancestry(	1,386✔
113	&self,
114	base: &'static str,
115	mut block: &'static str,
116	) -> Result<Vec<&'static str>, Error> {
117	let mut ancestry = Vec::new();	1,386✔
118
119	loop {
120	match self.inner.get(block) {	6,156✔
121	None => return Err(Error::NotDescendent),	×
122	Some(record) => {	6,156✔
123	block = record.parent;	6,156✔
124	},
125	}
126
127	if block == NULL_HASH {	6,156✔
128	return Err(Error::NotDescendent)	×
129	}
130	if block == base {	6,156✔
131	break
132	}
133
134	ancestry.push(block);	4,770✔
135	}
136
137	Ok(ancestry)	1,386✔
138	}	1,386✔
139	}
140	}
141
142	#[cfg(feature = "std")]
143	pub mod environment {
144	use super::chain::*;
145	use crate::{
146	round::State as RoundState,
147	voter::{Callback, CommunicationIn, CommunicationOut, RoundData},
148	Chain, Commit, Equivocation, Error, HistoricalVotes, Message, Precommit, Prevote,
149	PrimaryPropose, SignedMessage,
150	};
151	use futures::{
152	channel::mpsc::{self, UnboundedReceiver, UnboundedSender},
153	prelude::*,
154	};
155	use futures_timer::Delay;
156	use parking_lot::Mutex;
157	use std::{
158	collections::HashMap,
159	pin::Pin,
160	sync::Arc,
161	task::{Context, Poll},
162	time::Duration,
163	};
164
165	#[derive(Hash, Debug, Clone, Copy, PartialEq, Eq, Ord, PartialOrd)]	73,386✔
166	pub struct Id(pub u32);	24,294✔
167
168	#[derive(Debug, Clone, PartialEq, Eq)]	45,760✔
169	pub struct Signature(pub u32);	22,880✔
170
171	pub struct Environment {
172	chain: Mutex<DummyChain>,
173	local_id: Id,
174	network: Network,
175	listeners: Mutex<
176	Vec<UnboundedSender<(&'static str, u32, Commit<&'static str, u32, Signature, Id>)>>,
177	>,
178	last_completed_and_concluded: Mutex<(u64, u64)>,
179	}
180
181	impl Environment {
182	pub fn new(network: Network, local_id: Id) -> Self {	42✔
183	Environment {	42✔
184	chain: Mutex::new(DummyChain::new()),	42✔
185	local_id,
186	network,
187	listeners: Mutex::new(Vec::new()),	42✔
188	last_completed_and_concluded: Mutex::new((0, 0)),	42✔
189	}
190	}	42✔
191
192	pub fn with_chain<F, U>(&self, f: F) -> U	42✔
193	where
194	F: FnOnce(&mut DummyChain) -> U,
195	{
196	let mut chain = self.chain.lock();	42✔
197	f(&mut chain)	42✔
198	}	42✔
199
200	/// Stream of finalized blocks.
201	pub fn finalized_stream(	36✔
202	&self,
203	) -> UnboundedReceiver<(&'static str, u32, Commit<&'static str, u32, Signature, Id>)> {
204	let (tx, rx) = mpsc::unbounded();	36✔
205	self.listeners.lock().push(tx);	36✔
206	rx
207	}	36✔
208
209	/// Get the last completed and concluded rounds.
210	pub fn last_completed_and_concluded(&self) -> (u64, u64) {	2✔
211	*self.last_completed_and_concluded.lock()	2✔
212	}	2✔
213	}
214
215	impl Chain<&'static str, u32> for Environment {
216	fn ancestry(	1,088✔
217	&self,
218	base: &'static str,
219	block: &'static str,
220	) -> Result<Vec<&'static str>, Error> {
221	self.chain.lock().ancestry(base, block)	1,088✔
222	}	1,088✔
223	}
224
225	impl crate::voter::Environment<&'static str, u32> for Environment {
226	type Timer = Box<dyn Future<Output = Result<(), Error>> + Unpin + Send>;
227	type BestChain =
228	Box<dyn Future<Output = Result<Option<(&'static str, u32)>, Error>> + Unpin + Send>;
229	type Id = Id;
230	type Signature = Signature;
231	type In = Box<
232	dyn Stream<Item = Result<SignedMessage<&'static str, u32, Signature, Id>, Error>>
233	+ Unpin
234	+ Send,
235	>;
236	type Out = Pin<Box<dyn Sink<Message<&'static str, u32>, Error = Error> + Send>>;
237	type Error = Error;
238
239	fn best_chain_containing(&self, base: &'static str) -> Self::BestChain {	42✔
240	Box::new(future::ok(self.chain.lock().best_chain_containing(base)))	42✔
241	}	42✔
242
243	fn round_data(&self, round: u64) -> RoundData<Self::Id, Self::Timer, Self::In, Self::Out> {	90✔
244	const GOSSIP_DURATION: Duration = Duration::from_millis(500);
245
246	let (incoming, outgoing) = self.network.make_round_comms(round, self.local_id);	90✔
247	RoundData {	90✔
248	voter_id: Some(self.local_id),	90✔
249	prevote_timer: Box::new(Delay::new(GOSSIP_DURATION).map(Ok)),	90✔
250	precommit_timer: Box::new(Delay::new(GOSSIP_DURATION + GOSSIP_DURATION).map(Ok)),	90✔
251	incoming: Box::new(incoming),	90✔
252	outgoing: Box::pin(outgoing),	90✔
253	}
254	}	90✔
255
256	fn round_commit_timer(&self) -> Self::Timer {	44✔
257	use rand::Rng;
258
259	const COMMIT_DELAY_MILLIS: u64 = 100;
260
261	let delay = Duration::from_millis(rand::thread_rng().gen_range(0..COMMIT_DELAY_MILLIS));	44✔
262
263	Box::new(Delay::new(delay).map(Ok))	44✔
264	}	44✔
265
266	fn completed(	40✔
267	&self,
268	round: u64,
269	_state: RoundState<&'static str, u32>,
270	_base: (&'static str, u32),
271	_votes: &HistoricalVotes<&'static str, u32, Self::Signature, Self::Id>,
272	) -> Result<(), Error> {
273	self.last_completed_and_concluded.lock().0 = round;	40✔
274	Ok(())	40✔
275	}	40✔
276
277	fn concluded(	6✔
278	&self,
279	round: u64,
280	_state: RoundState<&'static str, u32>,
281	_base: (&'static str, u32),
282	_votes: &HistoricalVotes<&'static str, u32, Self::Signature, Self::Id>,
283	) -> Result<(), Error> {
284	self.last_completed_and_concluded.lock().1 = round;	6✔
285	Ok(())	6✔
286	}	6✔
287
288	fn finalize_block(	42✔
289	&self,
290	hash: &'static str,
291	number: u32,
292	_round: u64,
293	commit: Commit<&'static str, u32, Signature, Id>,
294	) -> Result<(), Error> {
295	let mut chain = self.chain.lock();	42✔
296
297	let last_finalized = chain.last_finalized();	42✔
298	if number <= last_finalized.1 {	42✔
299	panic!("Attempted to finalize backwards")	×
300	}
301
302	assert!(	42✔
303	chain.ancestry(last_finalized.0, hash).is_ok(),	42✔
304	"Safety violation: reverting finalized block.",
305	);
306
307	chain.set_last_finalized((hash, number));	42✔
308	self.listeners	84✔
309	.lock()
310	.retain(\|s\| s.unbounded_send((hash, number as _, commit.clone())).is_ok());	120✔
311
312	Ok(())	42✔
313	}	42✔
314
315	fn proposed(	×
316	&self,
317	_round: u64,
318	_propose: PrimaryPropose<&'static str, u32>,
319	) -> Result<(), Self::Error> {
UNCOV 320	Ok(())	×
321	}	×
322
323	fn prevoted(	42✔
324	&self,
325	_round: u64,
326	_prevote: Prevote<&'static str, u32>,
327	) -> Result<(), Self::Error> {
328	Ok(())	42✔
329	}	42✔
330
331	fn precommitted(	38✔
332	&self,
333	_round: u64,
334	_precommit: Precommit<&'static str, u32>,
335	) -> Result<(), Self::Error> {
336	Ok(())	38✔
337	}	38✔
338
339	fn prevote_equivocation(	×
340	&self,
341	round: u64,
342	equivocation: Equivocation<Id, Prevote<&'static str, u32>, Signature>,
343	) {
344	panic!("Encountered equivocation in round {}: {:?}", round, equivocation);	×
345	}
346
347	fn precommit_equivocation(	×
348	&self,
349	round: u64,
350	equivocation: Equivocation<Id, Precommit<&'static str, u32>, Signature>,
351	) {
352	panic!("Encountered equivocation in round {}: {:?}", round, equivocation);	×
353	}
354	}
355
356	// p2p network data for a round.
357	struct BroadcastNetwork<M> {
358	receiver: UnboundedReceiver<M>,
359	raw_sender: UnboundedSender<M>,
360	senders: Vec<UnboundedSender<M>>,
361	history: Vec<M>,
362	}
363
364	impl<M: Clone> BroadcastNetwork<M> {
365	fn new() -> Self {	60✔
366	let (tx, rx) = mpsc::unbounded();	60✔
367	BroadcastNetwork {	60✔
368	receiver: rx,
369	raw_sender: tx,
370	senders: Vec::new(),	60✔
371	history: Vec::new(),	60✔
372	}
373	}	60✔
374
375	pub fn send_message(&self, message: M) {	2✔
376	let _ = self.raw_sender.unbounded_send(message);	2✔
377	}	2✔
378
379	// add a node to the network for a round.
380	fn add_node<N, F: Fn(N) -> M>(	278✔
381	&mut self,
382	f: F,
383	) -> (impl Stream<Item = Result<M, Error>>, impl Sink<N, Error = Error>) {
384	let (tx, rx) = mpsc::unbounded();	278✔
385	let messages_out = self	278✔
386	.raw_sender
387	.clone()
388	.sink_map_err(\|e\| panic!("Error sending messages: {:?}", e))	×
389	.with(move \|message\| future::ready(Ok(f(message))));	364✔
390
391	// get history to the node.
392	for prior_message in self.history.iter().cloned() {	278✔
393	let _ = tx.unbounded_send(prior_message);	×
394	}
395
396	self.senders.push(tx);	278✔
397
398	(rx.map(Ok), messages_out)	278✔
399	}	278✔
400
401	// do routing work
402	fn route(&mut self, cx: &mut Context) -> Poll<()> {	170✔
403	loop {	536✔
404	match Stream::poll_next(Pin::new(&mut self.receiver), cx) {	536✔
405	Poll::Pending => return Poll::Pending,	170✔
406	Poll::Ready(None) => return Poll::Ready(()),	×
407	Poll::Ready(Some(item)) => {	366✔
408	self.history.push(item.clone());	366✔
409	for sender in &self.senders {	19,310✔
410	let _ = sender.unbounded_send(item.clone());	18,944✔
411	}
412	},	12✔
413	}
414	}	536✔
415	}	170✔
416	}
417
418	/// Make a test network.
419	/// Give the network future to node environments and spawn the routing task
420	/// to run.
421	pub fn make_network() -> (Network, NetworkRouting) {	18✔
422	let global_messages = Arc::new(Mutex::new(GlobalMessageNetwork::new()));	18✔
423	let rounds = Arc::new(Mutex::new(HashMap::new()));	18✔
424	(	18✔
425	Network { global_messages: global_messages.clone(), rounds: rounds.clone() },	18✔
426	NetworkRouting { global_messages, rounds },	18✔
427	)
428	}	18✔
429
430	type RoundNetwork = BroadcastNetwork<SignedMessage<&'static str, u32, Signature, Id>>;
431	type GlobalMessageNetwork = BroadcastNetwork<CommunicationIn<&'static str, u32, Signature, Id>>;
432
433	/// A test network. Instantiate this with `make_network`,
434	#[derive(Clone)]	64✔
435	pub struct Network {
436	rounds: Arc<Mutex<HashMap<u64, RoundNetwork>>>,	32✔
437	global_messages: Arc<Mutex<GlobalMessageNetwork>>,	32✔
438	}
439
440	impl Network {
441	pub fn make_round_comms(	230✔
442	&self,
443	round_number: u64,
444	node_id: Id,
445	) -> (
446	impl Stream<Item = Result<SignedMessage<&'static str, u32, Signature, Id>, Error>>,
447	impl Sink<Message<&'static str, u32>, Error = Error>,
448	) {
449	let mut rounds = self.rounds.lock();	230✔
450	rounds	230✔
451	.entry(round_number)
452	.or_insert_with(RoundNetwork::new)
453	.add_node(move \|message\| SignedMessage {	708✔
454	message,
455	signature: Signature(node_id.0),	354✔
456	id: node_id,	354✔
457	})	354✔
458	}	230✔
459
460	pub fn make_global_comms(	48✔
461	&self,
462	) -> (
463	impl Stream<Item = Result<CommunicationIn<&'static str, u32, Signature, Id>, Error>>,
464	impl Sink<CommunicationOut<&'static str, u32, Signature, Id>, Error = Error>,
465	) {
466	let mut global_messages = self.global_messages.lock();	48✔
467	global_messages.add_node(\|message\| match message {	58✔
468	CommunicationOut::Commit(r, commit) =>	10✔
469	CommunicationIn::Commit(r, commit.into(), Callback::Blank),	10✔
470	})	10✔
471	}	48✔
472
473	/// Send a message to all nodes.
474	pub fn send_message(&self, message: CommunicationIn<&'static str, u32, Signature, Id>) {	2✔
475	self.global_messages.lock().send_message(message);	2✔
476	}	2✔
477	}
478
479	/// the network routing task.
480	pub struct NetworkRouting {
481	rounds: Arc<Mutex<HashMap<u64, RoundNetwork>>>,
482	global_messages: Arc<Mutex<GlobalMessageNetwork>>,
483	}
484
485	impl Future for NetworkRouting {
486	type Output = ();
487
488	fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<()> {	66✔
489	let mut rounds = self.rounds.lock();	66✔
490	rounds.retain(\|_, round\| match round.route(cx) {	170✔
491	Poll::Ready(()) => false,	×
492	Poll::Pending => true,	104✔
493	});	104✔
494
495	let mut global_messages = self.global_messages.lock();	66✔
496	let _ = global_messages.route(cx);	66✔
497
498	Poll::Pending	66✔
499	}	66✔
500	}
501	}

paritytech / finality-grandpa / 13196351565

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