From c777362d02e0eb93137c17cb2a0a6ec3a973294b Mon Sep 17 00:00:00 2001 From: arkpar Date: Sun, 11 Dec 2016 12:32:01 +0100 Subject: [PATCH] Sync channel for consensus test --- ethcore/src/client/client.rs | 9 +- ethcore/src/client/test_client.rs | 2 +- ethcore/src/engines/mod.rs | 3 +- ethcore/src/engines/tendermint/mod.rs | 6 +- ethcore/src/service.rs | 3 +- parity/informant.rs | 2 +- sync/src/chain.rs | 154 +++++++++++++------------- sync/src/tests/chain.rs | 2 +- sync/src/tests/consensus.rs | 113 ++++++++++++------- sync/src/tests/helpers.rs | 62 ++++++----- util/io/src/service.rs | 44 ++++++-- 11 files changed, 232 insertions(+), 168 deletions(-) diff --git a/ethcore/src/client/client.rs b/ethcore/src/client/client.rs index 8586eec30..f63274ccb 100644 --- a/ethcore/src/client/client.rs +++ b/ethcore/src/client/client.rs @@ -580,6 +580,12 @@ impl Client { self.miner.clone() } + + /// Replace io channel. Useful for testing. + pub fn set_io_channel(&self, io_channel: IoChannel) { + *self.io_channel.lock() = io_channel; + } + /// Attempt to get a copy of a specific block's final state. /// /// This will not fail if given BlockId::Latest. @@ -1289,7 +1295,8 @@ impl BlockChainClient for Client { } fn queue_consensus_message(&self, message: Bytes) { - if let Err(e) = self.io_channel.lock().send(ClientIoMessage::NewMessage(message)) { + let channel = self.io_channel.lock().clone(); + if let Err(e) = channel.send(ClientIoMessage::NewMessage(message)) { debug!("Ignoring the message, error queueing: {}", e); } } diff --git a/ethcore/src/client/test_client.rs b/ethcore/src/client/test_client.rs index 6dd59f43d..462241ed1 100644 --- a/ethcore/src/client/test_client.rs +++ b/ethcore/src/client/test_client.rs @@ -678,7 +678,7 @@ impl BlockChainClient for TestBlockChainClient { } fn queue_consensus_message(&self, message: Bytes) { - self.spec.engine.handle_message(UntrustedRlp::new(&message)).unwrap(); + self.spec.engine.handle_message(&message).unwrap(); } fn pending_transactions(&self) -> Vec { diff --git a/ethcore/src/engines/mod.rs b/ethcore/src/engines/mod.rs index 7793d636d..03d5c1a43 100644 --- a/ethcore/src/engines/mod.rs +++ b/ethcore/src/engines/mod.rs @@ -28,7 +28,6 @@ pub use self::basic_authority::BasicAuthority; pub use self::authority_round::AuthorityRound; pub use self::tendermint::Tendermint; -use rlp::UntrustedRlp; use util::*; use account_provider::AccountProvider; use block::ExecutedBlock; @@ -178,7 +177,7 @@ pub trait Engine : Sync + Send { /// Handle any potential consensus messages; /// updating consensus state and potentially issuing a new one. - fn handle_message(&self, _message: UntrustedRlp) -> Result<(), Error> { Err(EngineError::UnexpectedMessage.into()) } + fn handle_message(&self, _message: &[u8]) -> Result<(), Error> { Err(EngineError::UnexpectedMessage.into()) } // TODO: builtin contract routing - to do this properly, it will require removing the built-in configuration-reading logic // from Spec into here and removing the Spec::builtins field. diff --git a/ethcore/src/engines/tendermint/mod.rs b/ethcore/src/engines/tendermint/mod.rs index 2b779950a..ae63ad8d1 100644 --- a/ethcore/src/engines/tendermint/mod.rs +++ b/ethcore/src/engines/tendermint/mod.rs @@ -146,7 +146,8 @@ impl Tendermint { } fn broadcast_message(&self, message: Bytes) { - if let Some(ref channel) = *self.message_channel.lock() { + let channel = self.message_channel.lock().clone(); + if let Some(ref channel) = channel { match channel.send(ClientIoMessage::BroadcastMessage(message)) { Ok(_) => trace!(target: "poa", "broadcast_message: BroadcastMessage message sent."), Err(err) => warn!(target: "poa", "broadcast_message: Could not send a sealing message {}.", err), @@ -449,7 +450,8 @@ impl Engine for Tendermint { } } - fn handle_message(&self, rlp: UntrustedRlp) -> Result<(), Error> { + fn handle_message(&self, rlp: &[u8]) -> Result<(), Error> { + let rlp = UntrustedRlp::new(rlp); let message: ConsensusMessage = try!(rlp.as_val()); if !self.votes.is_old_or_known(&message) { let sender = public_to_address(&try!(recover(&message.signature.into(), &try!(rlp.at(1)).as_raw().sha3()))); diff --git a/ethcore/src/service.rs b/ethcore/src/service.rs index c2cb1889e..9b178fe04 100644 --- a/ethcore/src/service.rs +++ b/ethcore/src/service.rs @@ -17,7 +17,6 @@ //! Creates and registers client and network services. use util::*; -use rlp::{UntrustedRlp, View}; use io::*; use spec::Spec; use error::*; @@ -228,7 +227,7 @@ impl IoHandler for ClientIoHandler { ClientIoMessage::UpdateSealing => self.client.update_sealing(), ClientIoMessage::SubmitSeal(ref hash, ref seal) => self.client.submit_seal(*hash, seal.clone()), ClientIoMessage::BroadcastMessage(ref message) => self.client.broadcast_message(message.clone()), - ClientIoMessage::NewMessage(ref message) => if let Err(e) = self.client.engine().handle_message(UntrustedRlp::new(message)) { + ClientIoMessage::NewMessage(ref message) => if let Err(e) = self.client.engine().handle_message(message) { trace!(target: "poa", "Invalid message received: {}", e); }, _ => {} // ignore other messages diff --git a/parity/informant.rs b/parity/informant.rs index 3d922ada2..ef6bea0b9 100644 --- a/parity/informant.rs +++ b/parity/informant.rs @@ -183,7 +183,7 @@ impl ChainNotify for Informant { let ripe = Instant::now() > *last_import + Duration::from_secs(1) && !importing; let txs_imported = imported.iter() .take(imported.len().saturating_sub(if ripe { 1 } else { 0 })) - .filter_map(|h| self.client.block(BlockID::Hash(*h))) + .filter_map(|h| self.client.block(BlockId::Hash(*h))) .map(|b| BlockView::new(&b).transactions_count()) .sum(); diff --git a/sync/src/chain.rs b/sync/src/chain.rs index a692c4004..b1fa0d2a9 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -2041,7 +2041,8 @@ impl ChainSync { } /// Called when peer sends us new consensus packet - fn on_consensus_packet(io: &mut SyncIo, _peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> { + fn on_consensus_packet(io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> { + trace!(target: "sync", "Received consensus packet from {:?}", peer_id); io.chain().queue_consensus_message(r.as_raw().to_vec()); Ok(()) } @@ -2116,9 +2117,9 @@ mod tests { #[test] fn return_receipts_empty() { let mut client = TestBlockChainClient::new(); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let ss = TestSnapshotService::new(); - let io = TestIo::new(&mut client, &ss, &mut queue, None); + let io = TestIo::new(&mut client, &ss, &queue, None); let result = ChainSync::return_receipts(&io, &UntrustedRlp::new(&[0xc0]), 0); @@ -2128,10 +2129,10 @@ mod tests { #[test] fn return_receipts() { let mut client = TestBlockChainClient::new(); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let sync = dummy_sync_with_peer(H256::new(), &client); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let mut receipt_list = RlpStream::new_list(4); receipt_list.append(&H256::from("0000000000000000000000000000000000000000000000005555555555555555")); @@ -2152,7 +2153,7 @@ mod tests { io.sender = Some(2usize); ChainSync::dispatch_packet(&RwLock::new(sync), &mut io, 0usize, super::GET_RECEIPTS_PACKET, &receipts_request); - assert_eq!(1, io.queue.len()); + assert_eq!(1, io.packets.len()); } #[test] @@ -2185,9 +2186,9 @@ mod tests { let headers: Vec<_> = blocks.iter().map(|b| Rlp::new(b).at(0).as_raw().to_vec()).collect(); let hashes: Vec<_> = headers.iter().map(|h| HeaderView::new(h).sha3()).collect(); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let ss = TestSnapshotService::new(); - let io = TestIo::new(&mut client, &ss, &mut queue, None); + let io = TestIo::new(&mut client, &ss, &queue, None); let unknown: H256 = H256::new(); let result = ChainSync::return_block_headers(&io, &UntrustedRlp::new(&make_hash_req(&unknown, 1, 0, false)), 0); @@ -2223,10 +2224,10 @@ mod tests { #[test] fn return_nodes() { let mut client = TestBlockChainClient::new(); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let sync = dummy_sync_with_peer(H256::new(), &client); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let mut node_list = RlpStream::new_list(3); node_list.append(&H256::from("0000000000000000000000000000000000000000000000005555555555555555")); @@ -2249,7 +2250,7 @@ mod tests { io.sender = Some(2usize); ChainSync::dispatch_packet(&RwLock::new(sync), &mut io, 0usize, super::GET_NODE_DATA_PACKET, &node_request); - assert_eq!(1, io.queue.len()); + assert_eq!(1, io.packets.len()); } fn dummy_sync_with_peer(peer_latest_hash: H256, client: &BlockChainClient) -> ChainSync { @@ -2280,11 +2281,11 @@ mod tests { fn finds_lagging_peers() { let mut client = TestBlockChainClient::new(); client.add_blocks(100, EachBlockWith::Uncle); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(10), &client); let chain_info = client.chain_info(); let ss = TestSnapshotService::new(); - let io = TestIo::new(&mut client, &ss, &mut queue, None); + let io = TestIo::new(&mut client, &ss, &queue, None); let lagging_peers = sync.get_peers(&chain_info, &io, PeerStatus::Lagging); let current_peers = sync.get_peers(&chain_info, &io, PeerStatus::Current); @@ -2297,11 +2298,11 @@ mod tests { fn finds_current_peers() { let mut client = TestBlockChainClient::new(); client.add_blocks(100, EachBlockWith::Uncle); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let mut sync = dummy_sync_with_peer(client.block_hash(BlockId::Latest).unwrap(), &client); let chain_info = client.chain_info(); let ss = TestSnapshotService::new(); - let io = TestIo::new(&mut client, &ss, &mut queue, None); + let io = TestIo::new(&mut client, &ss, &queue, None); let current_peers = sync.get_peers(&chain_info, &io, PeerStatus::Current); let lagging_peers = sync.get_peers(&chain_info, &io, PeerStatus::Lagging); @@ -2331,79 +2332,79 @@ mod tests { fn sends_new_hashes_to_lagging_peer() { let mut client = TestBlockChainClient::new(); client.add_blocks(100, EachBlockWith::Uncle); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client); let chain_info = client.chain_info(); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let peers = sync.get_peers(&chain_info, &io, PeerStatus::Lagging); let peer_count = sync.propagate_new_hashes(&chain_info, &mut io, &peers); // 1 message should be send - assert_eq!(1, io.queue.len()); + assert_eq!(1, io.packets.len()); // 1 peer should be updated assert_eq!(1, peer_count); // NEW_BLOCK_HASHES_PACKET - assert_eq!(0x01, io.queue[0].packet_id); + assert_eq!(0x01, io.packets[0].packet_id); } #[test] fn sends_latest_block_to_lagging_peer() { let mut client = TestBlockChainClient::new(); client.add_blocks(100, EachBlockWith::Uncle); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client); let chain_info = client.chain_info(); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let peers = sync.get_peers(&chain_info, &io, PeerStatus::Lagging); let peer_count = sync.propagate_blocks(&chain_info, &mut io, &[], &peers); // 1 message should be send - assert_eq!(1, io.queue.len()); + assert_eq!(1, io.packets.len()); // 1 peer should be updated assert_eq!(1, peer_count); // NEW_BLOCK_PACKET - assert_eq!(0x07, io.queue[0].packet_id); + assert_eq!(0x07, io.packets[0].packet_id); } #[test] fn sends_sealed_block() { let mut client = TestBlockChainClient::new(); client.add_blocks(100, EachBlockWith::Uncle); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let hash = client.block_hash(BlockId::Number(99)).unwrap(); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client); let chain_info = client.chain_info(); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let peers = sync.get_peers(&chain_info, &io, PeerStatus::Lagging); let peer_count = sync.propagate_blocks(&chain_info, &mut io, &[hash.clone()], &peers); // 1 message should be send - assert_eq!(1, io.queue.len()); + assert_eq!(1, io.packets.len()); // 1 peer should be updated assert_eq!(1, peer_count); // NEW_BLOCK_PACKET - assert_eq!(0x07, io.queue[0].packet_id); + assert_eq!(0x07, io.packets[0].packet_id); } #[test] fn sends_proposed_block() { let mut client = TestBlockChainClient::new(); client.add_blocks(100, EachBlockWith::Uncle); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let block = client.block(BlockId::Latest).unwrap(); let mut sync = dummy_sync_with_peer(client.block_hash(BlockId::Latest).unwrap(), &client); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); sync.propagate_proposed_blocks(&mut io, &[block]); // 1 message should be send - assert_eq!(1, io.queue.len()); + assert_eq!(1, io.packets.len()); // NEW_BLOCK_PACKET - assert_eq!(0x07, io.queue[0].packet_id); + assert_eq!(0x07, io.packets[0].packet_id); } #[test] @@ -2412,9 +2413,9 @@ mod tests { client.add_blocks(100, EachBlockWith::Uncle); client.insert_transaction_to_queue(); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(1), &client); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let peer_count = sync.propagate_new_transactions(&mut io); // Try to propagate same transactions for the second time let peer_count2 = sync.propagate_new_transactions(&mut io); @@ -2424,13 +2425,13 @@ mod tests { let peer_count3 = sync.propagate_new_transactions(&mut io); // 1 message should be send - assert_eq!(1, io.queue.len()); + assert_eq!(1, io.packets.len()); // 1 peer should be updated but only once assert_eq!(1, peer_count); assert_eq!(0, peer_count2); assert_eq!(0, peer_count3); // TRANSACTIONS_PACKET - assert_eq!(0x02, io.queue[0].packet_id); + assert_eq!(0x02, io.packets[0].packet_id); } #[test] @@ -2439,21 +2440,21 @@ mod tests { client.add_blocks(100, EachBlockWith::Uncle); client.insert_transaction_to_queue(); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(1), &client); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let peer_count = sync.propagate_new_transactions(&mut io); io.chain.insert_transaction_to_queue(); // New block import should trigger propagation. sync.chain_new_blocks(&mut io, &[], &[], &[], &[], &[], &[]); // 2 message should be send - assert_eq!(2, io.queue.len()); + assert_eq!(2, io.packets.len()); // 1 peer should receive the message assert_eq!(1, peer_count); // TRANSACTIONS_PACKET - assert_eq!(0x02, io.queue[0].packet_id); - assert_eq!(0x02, io.queue[1].packet_id); + assert_eq!(0x02, io.packets[0].packet_id); + assert_eq!(0x02, io.packets[1].packet_id); } #[test] @@ -2462,31 +2463,34 @@ mod tests { client.add_blocks(100, EachBlockWith::Uncle); client.insert_transaction_to_queue(); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(1), &client); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let ss = TestSnapshotService::new(); // should sent some { - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let peer_count = sync.propagate_new_transactions(&mut io); - assert_eq!(1, io.queue.len()); + assert_eq!(1, io.packets.len()); assert_eq!(1, peer_count); } // Insert some more client.insert_transaction_to_queue(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); - // Propagate new transactions - let peer_count2 = sync.propagate_new_transactions(&mut io); - // And now the peer should have all transactions - let peer_count3 = sync.propagate_new_transactions(&mut io); + let (peer_count2, peer_count3) = { + let mut io = TestIo::new(&mut client, &ss, &queue, None); + // Propagate new transactions + let peer_count2 = sync.propagate_new_transactions(&mut io); + // And now the peer should have all transactions + let peer_count3 = sync.propagate_new_transactions(&mut io); + (peer_count2, peer_count3) + }; // 2 message should be send (in total) - assert_eq!(2, io.queue.len()); + assert_eq!(2, queue.read().len()); // 1 peer should be updated but only once after inserting new transaction assert_eq!(1, peer_count2); assert_eq!(0, peer_count3); // TRANSACTIONS_PACKET - assert_eq!(0x02, io.queue[0].packet_id); - assert_eq!(0x02, io.queue[1].packet_id); + assert_eq!(0x02, queue.read()[0].packet_id); + assert_eq!(0x02, queue.read()[1].packet_id); } #[test] @@ -2495,9 +2499,9 @@ mod tests { client.add_blocks(100, EachBlockWith::Uncle); client.insert_transaction_to_queue(); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(1), &client); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); sync.propagate_new_transactions(&mut io); let stats = sync.transactions_stats(); @@ -2511,11 +2515,11 @@ mod tests { let block_data = get_dummy_block(11, client.chain_info().best_block_hash); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client); //sync.have_common_block = true; let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let block = UntrustedRlp::new(&block_data); @@ -2531,10 +2535,10 @@ mod tests { let block_data = get_dummy_blocks(11, client.chain_info().best_block_hash); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let block = UntrustedRlp::new(&block_data); @@ -2547,10 +2551,10 @@ mod tests { fn handles_peer_new_block_empty() { let mut client = TestBlockChainClient::new(); client.add_blocks(10, EachBlockWith::Uncle); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let empty_data = vec![]; let block = UntrustedRlp::new(&empty_data); @@ -2564,10 +2568,10 @@ mod tests { fn handles_peer_new_hashes() { let mut client = TestBlockChainClient::new(); client.add_blocks(10, EachBlockWith::Uncle); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let hashes_data = get_dummy_hashes(); let hashes_rlp = UntrustedRlp::new(&hashes_data); @@ -2581,10 +2585,10 @@ mod tests { fn handles_peer_new_hashes_empty() { let mut client = TestBlockChainClient::new(); client.add_blocks(10, EachBlockWith::Uncle); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let empty_hashes_data = vec![]; let hashes_rlp = UntrustedRlp::new(&empty_hashes_data); @@ -2600,16 +2604,16 @@ mod tests { fn hashes_rlp_mutually_acceptable() { let mut client = TestBlockChainClient::new(); client.add_blocks(100, EachBlockWith::Uncle); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client); let chain_info = client.chain_info(); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let peers = sync.get_peers(&chain_info, &io, PeerStatus::Lagging); sync.propagate_new_hashes(&chain_info, &mut io, &peers); - let data = &io.queue[0].data.clone(); + let data = &io.packets[0].data.clone(); let result = sync.on_peer_new_hashes(&mut io, 0, &UntrustedRlp::new(data)); assert!(result.is_ok()); } @@ -2620,16 +2624,16 @@ mod tests { fn block_rlp_mutually_acceptable() { let mut client = TestBlockChainClient::new(); client.add_blocks(100, EachBlockWith::Uncle); - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client); let chain_info = client.chain_info(); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); let peers = sync.get_peers(&chain_info, &io, PeerStatus::Lagging); sync.propagate_blocks(&chain_info, &mut io, &[], &peers); - let data = &io.queue[0].data.clone(); + let data = &io.packets[0].data.clone(); let result = sync.on_peer_new_block(&mut io, 0, &UntrustedRlp::new(data)); assert!(result.is_ok()); } @@ -2657,9 +2661,9 @@ mod tests { // when { - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); io.chain.miner.chain_new_blocks(io.chain, &[], &[], &[], &good_blocks); sync.chain_new_blocks(&mut io, &[], &[], &[], &good_blocks, &[], &[]); assert_eq!(io.chain.miner.status().transactions_in_future_queue, 0); @@ -2672,9 +2676,9 @@ mod tests { client.set_nonce(view.transactions()[0].sender().unwrap(), U256::from(1)); } { - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&client, &ss, &queue, None); io.chain.miner.chain_new_blocks(io.chain, &[], &[], &good_blocks, &retracted_blocks); sync.chain_new_blocks(&mut io, &[], &[], &good_blocks, &retracted_blocks, &[], &[]); } @@ -2697,9 +2701,9 @@ mod tests { let good_blocks = vec![client.block_hash_delta_minus(2)]; let retracted_blocks = vec![client.block_hash_delta_minus(1)]; - let mut queue = VecDeque::new(); + let queue = RwLock::new(VecDeque::new()); let ss = TestSnapshotService::new(); - let mut io = TestIo::new(&mut client, &ss, &mut queue, None); + let mut io = TestIo::new(&mut client, &ss, &queue, None); // when sync.chain_new_blocks(&mut io, &[], &[], &[], &good_blocks, &[], &[]); diff --git a/sync/src/tests/chain.rs b/sync/src/tests/chain.rs index 02b9063fa..c099bd0aa 100644 --- a/sync/src/tests/chain.rs +++ b/sync/src/tests/chain.rs @@ -101,7 +101,7 @@ fn forked_with_misbehaving_peer() { ::env_logger::init().ok(); let mut net = TestNet::new(3); // peer 0 is on a totally different chain with higher total difficulty - net.peer_mut(0).chain = TestBlockChainClient::new_with_extra_data(b"fork".to_vec()); + net.peer_mut(0).chain = Arc::new(TestBlockChainClient::new_with_extra_data(b"fork".to_vec())); net.peer(0).chain.add_blocks(50, EachBlockWith::Nothing); net.peer(1).chain.add_blocks(10, EachBlockWith::Nothing); net.peer(2).chain.add_blocks(10, EachBlockWith::Nothing); diff --git a/sync/src/tests/consensus.rs b/sync/src/tests/consensus.rs index 89297ba1f..e9cefc045 100644 --- a/sync/src/tests/consensus.rs +++ b/sync/src/tests/consensus.rs @@ -15,7 +15,9 @@ // along with Parity. If not, see . use util::*; -use ethcore::client::BlockChainClient; +use io::{IoHandler, IoContext, IoChannel}; +use ethcore::client::{BlockChainClient, Client, MiningBlockChainClient}; +use ethcore::service::{ClientIoMessage}; use ethcore::spec::Spec; use ethcore::miner::MinerService; use ethcore::transaction::*; @@ -24,23 +26,62 @@ use ethkey::KeyPair; use super::helpers::*; use SyncConfig; +struct TestIoHandler { + client: Arc, +} + +impl IoHandler for TestIoHandler { + fn message(&self, _io: &IoContext, net_message: &ClientIoMessage) { + match *net_message { + ClientIoMessage::UpdateSealing => self.client.update_sealing(), + ClientIoMessage::SubmitSeal(ref hash, ref seal) => self.client.submit_seal(*hash, seal.clone()), + ClientIoMessage::BroadcastMessage(ref message) => self.client.broadcast_message(message.clone()), + ClientIoMessage::NewMessage(ref message) => if let Err(e) = self.client.engine().handle_message(message) { + panic!("Invalid message received: {}", e); + }, + _ => {} // ignore other messages + } + } +} + #[test] -fn test_authority_round() { - let s1 = KeyPair::from_secret("1".sha3()).unwrap(); - let s2 = KeyPair::from_secret("0".sha3()).unwrap(); +fn authority_round() { + let s0 = KeyPair::from_secret("1".sha3()).unwrap(); + let s1 = KeyPair::from_secret("0".sha3()).unwrap(); let spec_factory = || { let spec = Spec::new_test_round(); let account_provider = AccountProvider::transient_provider(); + account_provider.insert_account(s0.secret().clone(), "").unwrap(); account_provider.insert_account(s1.secret().clone(), "").unwrap(); - account_provider.insert_account(s2.secret().clone(), "").unwrap(); spec.engine.register_account_provider(Arc::new(account_provider)); spec }; - let mut net = TestNet::new_with_spec(2, SyncConfig::default(), spec_factory); + let mut net = TestNet::with_spec(2, SyncConfig::default(), spec_factory); let mut net = &mut *net; - // Push transaction to both clients. Only one of them gets lucky to mine a block. - net.peer(0).chain.miner().set_engine_signer(s1.address(), "".to_owned()).unwrap(); - net.peer(1).chain.miner().set_engine_signer(s2.address(), "".to_owned()).unwrap(); + let io_handler0: Arc> = Arc::new(TestIoHandler { client: net.peer(0).chain.clone() }); + let io_handler1: Arc> = Arc::new(TestIoHandler { client: net.peer(1).chain.clone() }); + // Push transaction to both clients. Only one of them gets lucky to produce a block. + net.peer(0).chain.miner().set_engine_signer(s0.address(), "".to_owned()).unwrap(); + net.peer(1).chain.miner().set_engine_signer(s1.address(), "".to_owned()).unwrap(); + net.peer(0).chain.engine().register_message_channel(IoChannel::to_handler(Arc::downgrade(&io_handler0))); + net.peer(1).chain.engine().register_message_channel(IoChannel::to_handler(Arc::downgrade(&io_handler1))); + net.peer(0).chain.set_io_channel(IoChannel::to_handler(Arc::downgrade(&io_handler1))); + net.peer(1).chain.set_io_channel(IoChannel::to_handler(Arc::downgrade(&io_handler1))); + let tx0 = Transaction { + nonce: 0.into(), + gas_price: 0.into(), + gas: 21000.into(), + action: Action::Call(Address::default()), + value: 0.into(), + data: Vec::new(), + }.sign(s0.secret(), None); + // exchange statuses + net.sync(); + net.peer(0).chain.miner().import_own_transaction(&*net.peer(0).chain, tx0).unwrap(); + net.sync(); + assert_eq!(net.peer(0).chain.chain_info().best_block_number, 1); + assert_eq!(net.peer(1).chain.chain_info().best_block_number, 1); + let tx1 = Transaction { nonce: 0.into(), gas_price: 0.into(), @@ -49,59 +90,49 @@ fn test_authority_round() { value: 0.into(), data: Vec::new(), }.sign(s1.secret(), None); - // exhange statuses - net.sync_steps(5); - net.peer(0).chain.miner().import_own_transaction(&net.peer(0).chain, tx1).unwrap(); - net.sync(); - assert_eq!(net.peer(0).chain.chain_info().best_block_number, 1); - assert_eq!(net.peer(1).chain.chain_info().best_block_number, 1); - - let tx2 = Transaction { - nonce: 0.into(), - gas_price: 0.into(), - gas: 21000.into(), - action: Action::Call(Address::default()), - value: 0.into(), - data: Vec::new(), - }.sign(s2.secret(), None); - net.peer(1).chain.miner().import_own_transaction(&net.peer(1).chain, tx2).unwrap(); + net.peer(1).chain.miner().import_own_transaction(&*net.peer(1).chain, tx1).unwrap(); net.peer(1).chain.engine().step(); - net.peer(1).chain.miner().update_sealing(&net.peer(1).chain); net.sync(); assert_eq!(net.peer(0).chain.chain_info().best_block_number, 2); assert_eq!(net.peer(1).chain.chain_info().best_block_number, 2); } #[test] -fn test_tendermint() { +fn tendermint() { ::env_logger::init().ok(); - let s1 = KeyPair::from_secret("1".sha3()).unwrap(); - let s2 = KeyPair::from_secret("0".sha3()).unwrap(); + let s0 = KeyPair::from_secret("1".sha3()).unwrap(); + let s1 = KeyPair::from_secret("0".sha3()).unwrap(); let spec_factory = || { let spec = Spec::new_test_tendermint(); let account_provider = AccountProvider::transient_provider(); + account_provider.insert_account(s0.secret().clone(), "").unwrap(); account_provider.insert_account(s1.secret().clone(), "").unwrap(); - account_provider.insert_account(s2.secret().clone(), "").unwrap(); spec.engine.register_account_provider(Arc::new(account_provider)); spec }; - let mut net = TestNet::new_with_spec(2, SyncConfig::default(), spec_factory); + let mut net = TestNet::with_spec(2, SyncConfig::default(), spec_factory); let mut net = &mut *net; + let io_handler0: Arc> = Arc::new(TestIoHandler { client: net.peer(0).chain.clone() }); + let io_handler1: Arc> = Arc::new(TestIoHandler { client: net.peer(1).chain.clone() }); // Push transaction to both clients. Only one of them issues a proposal. - net.peer(0).chain.miner().set_engine_signer(s1.address(), "".to_owned()).unwrap(); - net.peer(1).chain.miner().set_engine_signer(s2.address(), "".to_owned()).unwrap(); - let tx1 = Transaction { + net.peer(0).chain.miner().set_engine_signer(s0.address(), "".to_owned()).unwrap(); + net.peer(1).chain.miner().set_engine_signer(s1.address(), "".to_owned()).unwrap(); + net.peer(0).chain.engine().register_message_channel(IoChannel::to_handler(Arc::downgrade(&io_handler0))); + net.peer(1).chain.engine().register_message_channel(IoChannel::to_handler(Arc::downgrade(&io_handler1))); + net.peer(0).chain.set_io_channel(IoChannel::to_handler(Arc::downgrade(&io_handler1))); + net.peer(1).chain.set_io_channel(IoChannel::to_handler(Arc::downgrade(&io_handler1))); + let tx0 = Transaction { nonce: 0.into(), gas_price: 0.into(), gas: 21000.into(), action: Action::Call(Address::default()), value: 0.into(), data: Vec::new(), - }.sign(s1.secret(), None); + }.sign(s0.secret(), None); // exhange statuses net.sync_steps(5); - net.peer(0).chain.miner().import_own_transaction(&net.peer(0).chain, tx1).unwrap(); + net.peer(0).chain.miner().import_own_transaction(&*net.peer(0).chain, tx0).unwrap(); // Propose net.sync(); // Propose timeout @@ -110,23 +141,19 @@ fn test_tendermint() { // Precommit net.sync(); - net.sync_steps(5); - net.sync(); - ::std::thread::sleep(::std::time::Duration::from_millis(100)); assert_eq!(net.peer(0).chain.chain_info().best_block_number, 1); assert_eq!(net.peer(1).chain.chain_info().best_block_number, 1); - let tx2 = Transaction { + let tx1 = Transaction { nonce: 0.into(), gas_price: 0.into(), gas: 21000.into(), action: Action::Call(Address::default()), value: 0.into(), data: Vec::new(), - }.sign(s2.secret(), None); - net.peer(1).chain.miner().import_own_transaction(&net.peer(1).chain, tx2).unwrap(); + }.sign(s1.secret(), None); + net.peer(1).chain.miner().import_own_transaction(&*net.peer(1).chain, tx1).unwrap(); net.peer(1).chain.engine().step(); - net.peer(1).chain.miner().update_sealing(&net.peer(1).chain); net.sync(); assert_eq!(net.peer(0).chain.chain_info().best_block_number, 2); assert_eq!(net.peer(1).chain.chain_info().best_block_number, 2); diff --git a/sync/src/tests/helpers.rs b/sync/src/tests/helpers.rs index 9be250ba0..c6b2ec949 100644 --- a/sync/src/tests/helpers.rs +++ b/sync/src/tests/helpers.rs @@ -45,14 +45,15 @@ impl FlushingBlockChainClient for TestBlockChainClient {} pub struct TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p { pub chain: &'p C, pub snapshot_service: &'p TestSnapshotService, - pub queue: &'p mut VecDeque, + pub queue: &'p RwLock>, pub sender: Option, pub to_disconnect: HashSet, + pub packets: Vec, overlay: RwLock>, } impl<'p, C> TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p { - pub fn new(chain: &'p C, ss: &'p TestSnapshotService, queue: &'p mut VecDeque, sender: Option) -> TestIo<'p, C> { + pub fn new(chain: &'p C, ss: &'p TestSnapshotService, queue: &'p RwLock>, sender: Option) -> TestIo<'p, C> { TestIo { chain: chain, snapshot_service: ss, @@ -60,10 +61,17 @@ impl<'p, C> TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p { sender: sender, to_disconnect: HashSet::new(), overlay: RwLock::new(HashMap::new()), + packets: Vec::new(), } } } +impl<'p, C> Drop for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p { + fn drop(&mut self) { + self.queue.write().extend(self.packets.drain(..)); + } +} + impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p { fn disable_peer(&mut self, peer_id: PeerId) { self.disconnect_peer(peer_id); @@ -78,7 +86,7 @@ impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p { } fn respond(&mut self, packet_id: PacketId, data: Vec) -> Result<(), NetworkError> { - self.queue.push_back(TestPacket { + self.packets.push(TestPacket { data: data, packet_id: packet_id, recipient: self.sender.unwrap() @@ -87,7 +95,7 @@ impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p { } fn send(&mut self, peer_id: PeerId, packet_id: PacketId, data: Vec) -> Result<(), NetworkError> { - self.queue.push_back(TestPacket { + self.packets.push(TestPacket { data: data, packet_id: packet_id, recipient: peer_id, @@ -100,7 +108,7 @@ impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p { } fn chain(&self) -> &BlockChainClient { - self.chain + &*self.chain } fn snapshot_service(&self) -> &SnapshotService { @@ -131,7 +139,7 @@ pub struct TestPacket { } pub struct TestPeer where C: FlushingBlockChainClient { - pub chain: C, + pub chain: Arc, pub snapshot_service: Arc, pub sync: RwLock, pub queue: RwLock>, @@ -167,7 +175,7 @@ impl TestNet { net.peers.push(Arc::new(TestPeer { sync: RwLock::new(sync), snapshot_service: ss, - chain: chain, + chain: Arc::new(chain), queue: RwLock::new(VecDeque::new()), })); } @@ -176,7 +184,7 @@ impl TestNet { } impl TestNet { - pub fn new_with_spec(n: usize, config: SyncConfig, spec_factory: F) -> GuardedTempResult> + pub fn with_spec(n: usize, config: SyncConfig, spec_factory: F) -> GuardedTempResult> where F: Fn() -> Spec { let mut net = TestNet { @@ -192,17 +200,17 @@ impl TestNet { let db_config = DatabaseConfig::with_columns(NUM_COLUMNS); let spec = spec_factory(); - let client = Arc::try_unwrap(EthcoreClient::new( + let client = EthcoreClient::new( ClientConfig::default(), &spec, client_dir.as_path(), Arc::new(Miner::with_spec(&spec)), IoChannel::disconnected(), &db_config - ).unwrap()).ok().unwrap(); + ).unwrap(); let ss = Arc::new(TestSnapshotService::new()); - let sync = ChainSync::new(config.clone(), &client); + let sync = ChainSync::new(config.clone(), &*client); let peer = Arc::new(TestPeer { sync: RwLock::new(sync), snapshot_service: ss, @@ -233,8 +241,8 @@ impl TestNet where C: FlushingBlockChainClient { for client in 0..self.peers.len() { if peer != client { let p = &self.peers[peer]; - p.sync.write().update_targets(&p.chain); - p.sync.write().on_peer_connected(&mut TestIo::new(&p.chain, &p.snapshot_service, &mut p.queue.write(), Some(client as PeerId)), client as PeerId); + p.sync.write().update_targets(&*p.chain); + p.sync.write().on_peer_connected(&mut TestIo::new(&*p.chain, &p.snapshot_service, &p.queue, Some(client as PeerId)), client as PeerId); } } } @@ -246,16 +254,15 @@ impl TestNet where C: FlushingBlockChainClient { if let Some(packet) = packet { let disconnecting = { let p = &self.peers[packet.recipient]; - let mut queue = p.queue.write(); trace!("--- {} -> {} ---", peer, packet.recipient); let to_disconnect = { - let mut io = TestIo::new(&p.chain, &p.snapshot_service, &mut queue, Some(peer as PeerId)); + let mut io = TestIo::new(&*p.chain, &p.snapshot_service, &p.queue, Some(peer as PeerId)); ChainSync::dispatch_packet(&p.sync, &mut io, peer as PeerId, packet.packet_id, &packet.data); - io.to_disconnect + io.to_disconnect.clone() }; for d in &to_disconnect { // notify this that disconnecting peers are disconnecting - let mut io = TestIo::new(&p.chain, &p.snapshot_service, &mut queue, Some(*d)); + let mut io = TestIo::new(&*p.chain, &p.snapshot_service, &p.queue, Some(*d)); p.sync.write().on_peer_aborting(&mut io, *d); self.disconnect_events.push((peer, *d)); } @@ -264,8 +271,7 @@ impl TestNet where C: FlushingBlockChainClient { for d in &disconnecting { // notify other peers that this peer is disconnecting let p = &self.peers[*d]; - let mut queue = p.queue.write(); - let mut io = TestIo::new(&p.chain, &p.snapshot_service, &mut queue, Some(peer as PeerId)); + let mut io = TestIo::new(&*p.chain, &p.snapshot_service, &p.queue, Some(peer as PeerId)); p.sync.write().on_peer_aborting(&mut io, peer as PeerId); } } @@ -277,15 +283,14 @@ impl TestNet where C: FlushingBlockChainClient { pub fn sync_step_peer(&mut self, peer_num: usize) { let peer = self.peer(peer_num); peer.chain.flush(); - let mut queue = peer.queue.write(); - peer.sync.write().maintain_peers(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None)); - peer.sync.write().maintain_sync(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None)); - peer.sync.write().propagate_new_transactions(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None)); + peer.sync.write().maintain_peers(&mut TestIo::new(&*peer.chain, &peer.snapshot_service, &peer.queue, None)); + peer.sync.write().maintain_sync(&mut TestIo::new(&*peer.chain, &peer.snapshot_service, &peer.queue, None)); + peer.sync.write().propagate_new_transactions(&mut TestIo::new(&*peer.chain, &peer.snapshot_service, &peer.queue, None)); } pub fn restart_peer(&mut self, i: usize) { let peer = self.peer(i); - peer.sync.write().restart(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut peer.queue.write(), None)); + peer.sync.write().restart(&mut TestIo::new(&*peer.chain, &peer.snapshot_service, &peer.queue, None)); } pub fn sync(&mut self) -> u32 { @@ -314,8 +319,7 @@ impl TestNet where C: FlushingBlockChainClient { pub fn trigger_chain_new_blocks(&mut self, peer_id: usize) { let peer = self.peer(peer_id); - let mut queue = peer.queue.write(); - peer.sync.write().chain_new_blocks(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None), &[], &[], &[], &[], &[], &[]); + peer.sync.write().chain_new_blocks(&mut TestIo::new(&*peer.chain, &peer.snapshot_service, &peer.queue, None), &[], &[], &[], &[], &[], &[]); } } @@ -329,8 +333,7 @@ impl ChainNotify for TestPeer { proposed: Vec, _duration: u64) { - let mut queue = self.queue.write(); - let mut io = TestIo::new(&self.chain, &self.snapshot_service, &mut queue, None); + let mut io = TestIo::new(&*self.chain, &self.snapshot_service, &self.queue, None); self.sync.write().chain_new_blocks( &mut io, &imported, @@ -346,8 +349,7 @@ impl ChainNotify for TestPeer { fn stop(&self) {} fn broadcast(&self, message: Vec) { - let mut queue = self.queue.write(); - let mut io = TestIo::new(&self.chain, &self.snapshot_service, &mut queue, None); + let mut io = TestIo::new(&*self.chain, &self.snapshot_service, &self.queue, None); self.sync.write().propagate_consensus_packet(&mut io, message.clone()); } } diff --git a/util/io/src/service.rs b/util/io/src/service.rs index 6086acadd..d99a2cb1e 100644 --- a/util/io/src/service.rs +++ b/util/io/src/service.rs @@ -329,11 +329,18 @@ impl Handler for IoManager where Message: Send + Clone + Sync } } +#[derive(Clone)] +enum Handlers where Message: Send + Clone { + SharedCollection(Weak>, HandlerId>>>), + Single(Weak>), +} + /// Allows sending messages into the event loop. All the IO handlers will get the message /// in the `message` callback. pub struct IoChannel where Message: Send + Clone{ channel: Option>>, - handlers: Weak>, HandlerId>>>, + handlers: Handlers, + } impl Clone for IoChannel where Message: Send + Clone + Sync + 'static { @@ -348,19 +355,29 @@ impl Clone for IoChannel where Message: Send + Clone + Sync + impl IoChannel where Message: Send + Clone + Sync + 'static { /// Send a message through the channel pub fn send(&self, message: Message) -> Result<(), IoError> { - if let Some(ref channel) = self.channel { - try!(channel.send(IoMessage::UserMessage(message))); + match self.channel { + Some(ref channel) => try!(channel.send(IoMessage::UserMessage(message))), + None => try!(self.send_sync(message)) } Ok(()) } /// Send a message through the channel and handle it synchronously pub fn send_sync(&self, message: Message) -> Result<(), IoError> { - if let Some(handlers) = self.handlers.upgrade() { - for id in 0 .. MAX_HANDLERS { - if let Some(h) = handlers.read().get(id) { - let handler = h.clone(); - handler.message(&IoContext::new(self.clone(), id), &message); + match self.handlers { + Handlers::SharedCollection(ref handlers) => { + if let Some(handlers) = handlers.upgrade() { + for id in 0 .. MAX_HANDLERS { + if let Some(h) = handlers.read().get(id) { + let handler = h.clone(); + handler.message(&IoContext::new(self.clone(), id), &message); + } + } + } + }, + Handlers::Single(ref handler) => { + if let Some(handler) = handler.upgrade() { + handler.message(&IoContext::new(self.clone(), 0), &message); } } } @@ -378,14 +395,21 @@ impl IoChannel where Message: Send + Clone + Sync + 'static { pub fn disconnected() -> IoChannel { IoChannel { channel: None, - handlers: Weak::default(), + handlers: Handlers::SharedCollection(Weak::default()), } } + /// Create a new synchronous channel to a given handler. + pub fn to_handler(handler: Weak>) -> IoChannel { + IoChannel { + channel: None, + handlers: Handlers::Single(handler), + } + } fn new(channel: Sender>, handlers: Weak>, HandlerId>>>) -> IoChannel { IoChannel { channel: Some(channel), - handlers: handlers, + handlers: Handlers::SharedCollection(handlers), } } }