grassrootseconomics/openethereum
5
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Reduce locking

Browse Source
This commit is contained in:
arkpar 2016-06-20 17:28:48 +02:00
parent 75a38500f1
commit 1ffe0c185c
4 changed files with 47 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
sync/src/chain.rs
View File

@ -1097,7 +1097,7 @@ impl ChainSync {
Ok(Some((RECEIPTS_PACKET, rlp_result))) Ok(Some((RECEIPTS_PACKET, rlp_result)))
} }
fn return_rlp<FRlp, FError>(&self, io: &mut SyncIo, rlp: &UntrustedRlp, peer: PeerId, rlp_func: FRlp, error_func: FError) -> Result<(), PacketDecodeError> fn return_rlp<FRlp, FError>(io: &mut SyncIo, rlp: &UntrustedRlp, peer: PeerId, rlp_func: FRlp, error_func: FError) -> Result<(), PacketDecodeError>
where FRlp : Fn(&SyncIo, &UntrustedRlp, PeerId) -> RlpResponseResult, where FRlp : Fn(&SyncIo, &UntrustedRlp, PeerId) -> RlpResponseResult,
FError : FnOnce(UtilError) -> String FError : FnOnce(UtilError) -> String
{ {
@ -1114,13 +1114,41 @@ impl ChainSync {
} }
/// Dispatch incoming requests and responses /// Dispatch incoming requests and responses
pub fn on_packet(&mut self, io: &mut SyncIo, peer: PeerId, packet_id: u8, data: &[u8]) { pub fn dispatch_packet(sync: &RwLock<ChainSync>, io: &mut SyncIo, peer: PeerId, packet_id: u8, data: &[u8]) {
let rlp = UntrustedRlp::new(data); let rlp = UntrustedRlp::new(data);
let result = match packet_id {
GET_BLOCK_BODIES_PACKET => ChainSync::return_rlp(io, &rlp, peer,
ChainSync::return_block_bodies,
|e| format!("Error sending block bodies: {:?}", e)),
GET_BLOCK_HEADERS_PACKET => ChainSync::return_rlp(io, &rlp, peer,
ChainSync::return_block_headers,
|e| format!("Error sending block headers: {:?}", e)),
GET_RECEIPTS_PACKET => ChainSync::return_rlp(io, &rlp, peer,
ChainSync::return_receipts,
|e| format!("Error sending receipts: {:?}", e)),
GET_NODE_DATA_PACKET => ChainSync::return_rlp(io, &rlp, peer,
ChainSync::return_node_data,
|e| format!("Error sending nodes: {:?}", e)),
_ => {
sync.write().unwrap().on_packet(io, peer, packet_id, data);
Ok(())
}
};
result.unwrap_or_else(|e| {
debug!(target:"sync", "{} -> Malformed packet {} : {}", peer, packet_id, e);
})
}
pub fn on_packet(&mut self, io: &mut SyncIo, peer: PeerId, packet_id: u8, data: &[u8]) {
if packet_id != STATUS_PACKET && !self.peers.contains_key(&peer) { if packet_id != STATUS_PACKET && !self.peers.contains_key(&peer) {
debug!(target:"sync", "Unexpected packet from unregistered peer: {}:{}", peer, io.peer_info(peer)); debug!(target:"sync", "Unexpected packet from unregistered peer: {}:{}", peer, io.peer_info(peer));
return; return;
} }
let rlp = UntrustedRlp::new(data);
let result = match packet_id { let result = match packet_id {
STATUS_PACKET => self.on_peer_status(io, peer, &rlp), STATUS_PACKET => self.on_peer_status(io, peer, &rlp),
TRANSACTIONS_PACKET => self.on_peer_transactions(io, peer, &rlp), TRANSACTIONS_PACKET => self.on_peer_transactions(io, peer, &rlp),
@ -1128,23 +1156,6 @@ impl ChainSync {
BLOCK_BODIES_PACKET => self.on_peer_block_bodies(io, peer, &rlp), BLOCK_BODIES_PACKET => self.on_peer_block_bodies(io, peer, &rlp),
NEW_BLOCK_PACKET => self.on_peer_new_block(io, peer, &rlp), NEW_BLOCK_PACKET => self.on_peer_new_block(io, peer, &rlp),
NEW_BLOCK_HASHES_PACKET => self.on_peer_new_hashes(io, peer, &rlp), NEW_BLOCK_HASHES_PACKET => self.on_peer_new_hashes(io, peer, &rlp),
GET_BLOCK_BODIES_PACKET => self.return_rlp(io, &rlp, peer,
ChainSync::return_block_bodies,
|e| format!("Error sending block bodies: {:?}", e)),
GET_BLOCK_HEADERS_PACKET => self.return_rlp(io, &rlp, peer,
ChainSync::return_block_headers,
|e| format!("Error sending block headers: {:?}", e)),
GET_RECEIPTS_PACKET => self.return_rlp(io, &rlp, peer,
ChainSync::return_receipts,
|e| format!("Error sending receipts: {:?}", e)),
GET_NODE_DATA_PACKET => self.return_rlp(io, &rlp, peer,
ChainSync::return_node_data,
|e| format!("Error sending nodes: {:?}", e)),
_ => { _ => {
debug!(target: "sync", "Unknown packet {}", packet_id); debug!(target: "sync", "Unknown packet {}", packet_id);
Ok(()) Ok(())
@ -1424,7 +1435,7 @@ mod tests {
fn return_receipts() { fn return_receipts() {
let mut client = TestBlockChainClient::new(); let mut client = TestBlockChainClient::new();
let mut queue = VecDeque::new(); let mut queue = VecDeque::new();
let mut sync = dummy_sync_with_peer(H256::new(), &client); let sync = dummy_sync_with_peer(H256::new(), &client);
let mut io = TestIo::new(&mut client, &mut queue, None); let mut io = TestIo::new(&mut client, &mut queue, None);
let mut receipt_list = RlpStream::new_list(4); let mut receipt_list = RlpStream::new_list(4);
@ -1445,7 +1456,7 @@ mod tests {
assert_eq!(603, rlp_result.unwrap().1.out().len()); assert_eq!(603, rlp_result.unwrap().1.out().len());
io.sender = Some(2usize); io.sender = Some(2usize);
sync.on_packet(&mut io, 0usize, super::GET_RECEIPTS_PACKET, &receipts_request); 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.queue.len());
} }
@ -1517,7 +1528,7 @@ mod tests {
fn return_nodes() { fn return_nodes() {
let mut client = TestBlockChainClient::new(); let mut client = TestBlockChainClient::new();
let mut queue = VecDeque::new(); let mut queue = VecDeque::new();
let mut sync = dummy_sync_with_peer(H256::new(), &client); let sync = dummy_sync_with_peer(H256::new(), &client);
let mut io = TestIo::new(&mut client, &mut queue, None); let mut io = TestIo::new(&mut client, &mut queue, None);
let mut node_list = RlpStream::new_list(3); let mut node_list = RlpStream::new_list(3);
@ -1537,7 +1548,8 @@ mod tests {
assert_eq!(34, rlp_result.unwrap().1.out().len()); assert_eq!(34, rlp_result.unwrap().1.out().len());
io.sender = Some(2usize); io.sender = Some(2usize);
sync.on_packet(&mut io, 0usize, super::GET_NODE_DATA_PACKET, &node_request);
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.queue.len());
} }

2
sync/src/lib.rs
View File

@ -169,7 +169,7 @@ impl NetworkProtocolHandler<SyncMessage> for EthSync {
} }
fn read(&self, io: &NetworkContext<SyncMessage>, peer: &PeerId, packet_id: u8, data: &[u8]) { fn read(&self, io: &NetworkContext<SyncMessage>, peer: &PeerId, packet_id: u8, data: &[u8]) {
self.sync.write().unwrap().on_packet(&mut NetSyncIo::new(io, self.chain.deref()) , *peer, packet_id, data); ChainSync::dispatch_packet(&self.sync, &mut NetSyncIo::new(io, self.chain.deref()) , *peer, packet_id, data);
} }
fn connected(&self, io: &NetworkContext<SyncMessage>, peer: &PeerId) { fn connected(&self, io: &NetworkContext<SyncMessage>, peer: &PeerId) {

6
sync/src/tests/chain.rs
View File

@ -47,7 +47,7 @@ fn status_after_sync() {
net.peer_mut(1).chain.add_blocks(1000, EachBlockWith::Uncle); net.peer_mut(1).chain.add_blocks(1000, EachBlockWith::Uncle);
net.peer_mut(2).chain.add_blocks(1000, EachBlockWith::Uncle); net.peer_mut(2).chain.add_blocks(1000, EachBlockWith::Uncle);
net.sync(); net.sync();
let status = net.peer(0).sync.status(); let status = net.peer(0).sync.read().unwrap().status();
assert_eq!(status.state, SyncState::Idle); assert_eq!(status.state, SyncState::Idle);
} }
@ -107,14 +107,14 @@ fn restart() {
assert!(net.peer(0).chain.chain_info().best_block_number > 100); assert!(net.peer(0).chain.chain_info().best_block_number > 100);
net.restart_peer(0); net.restart_peer(0);
let status = net.peer(0).sync.status(); let status = net.peer(0).sync.read().unwrap().status();
assert_eq!(status.state, SyncState::ChainHead); assert_eq!(status.state, SyncState::ChainHead);
} }
#[test] #[test]
fn status_empty() { fn status_empty() {
let net = TestNet::new(2); let net = TestNet::new(2);
assert_eq!(net.peer(0).sync.status().state, SyncState::Idle); assert_eq!(net.peer(0).sync.read().unwrap().status().state, SyncState::Idle);
} }
#[test] #[test]

16
sync/src/tests/helpers.rs
View File

@ -78,7 +78,7 @@ pub struct TestPacket {
pub struct TestPeer { pub struct TestPeer {
pub chain: TestBlockChainClient, pub chain: TestBlockChainClient,
pub sync: ChainSync, pub sync: RwLock<ChainSync>,
pub queue: VecDeque<TestPacket>, pub queue: VecDeque<TestPacket>,
} }
@ -97,7 +97,7 @@ impl TestNet {
let chain = TestBlockChainClient::new(); let chain = TestBlockChainClient::new();
let sync = ChainSync::new(SyncConfig::default(), &chain); let sync = ChainSync::new(SyncConfig::default(), &chain);
net.peers.push(TestPeer { net.peers.push(TestPeer {
sync: sync, sync: RwLock::new(sync),
chain: chain, chain: chain,
queue: VecDeque::new(), queue: VecDeque::new(),
}); });
@ -118,7 +118,7 @@ impl TestNet {
for client in 0..self.peers.len() { for client in 0..self.peers.len() {
if peer != client { if peer != client {
let mut p = self.peers.get_mut(peer).unwrap(); let mut p = self.peers.get_mut(peer).unwrap();
p.sync.on_peer_connected(&mut TestIo::new(&mut p.chain, &mut p.queue, Some(client as PeerId)), client as PeerId); p.sync.write().unwrap().on_peer_connected(&mut TestIo::new(&mut p.chain, &mut p.queue, Some(client as PeerId)), client as PeerId);
} }
} }
} }
@ -129,22 +129,22 @@ impl TestNet {
if let Some(packet) = self.peers[peer].queue.pop_front() { if let Some(packet) = self.peers[peer].queue.pop_front() {
let mut p = self.peers.get_mut(packet.recipient).unwrap(); let mut p = self.peers.get_mut(packet.recipient).unwrap();
trace!("--- {} -> {} ---", peer, packet.recipient); trace!("--- {} -> {} ---", peer, packet.recipient);
p.sync.on_packet(&mut TestIo::new(&mut p.chain, &mut p.queue, Some(peer as PeerId)), peer as PeerId, packet.packet_id, &packet.data); ChainSync::dispatch_packet(&p.sync, &mut TestIo::new(&mut p.chain, &mut p.queue, Some(peer as PeerId)), peer as PeerId, packet.packet_id, &packet.data);
trace!("----------------"); trace!("----------------");
} }
let mut p = self.peers.get_mut(peer).unwrap(); let mut p = self.peers.get_mut(peer).unwrap();
p.sync.maintain_sync(&mut TestIo::new(&mut p.chain, &mut p.queue, None)); p.sync.write().unwrap().maintain_sync(&mut TestIo::new(&mut p.chain, &mut p.queue, None));
} }
} }
pub fn sync_step_peer(&mut self, peer_num: usize) { pub fn sync_step_peer(&mut self, peer_num: usize) {
let mut peer = self.peer_mut(peer_num); let mut peer = self.peer_mut(peer_num);
peer.sync.maintain_sync(&mut TestIo::new(&mut peer.chain, &mut peer.queue, None)); peer.sync.write().unwrap().maintain_sync(&mut TestIo::new(&mut peer.chain, &mut peer.queue, None));
} }
pub fn restart_peer(&mut self, i: usize) { pub fn restart_peer(&mut self, i: usize) {
let peer = self.peer_mut(i); let peer = self.peer_mut(i);
peer.sync.restart(&mut TestIo::new(&mut peer.chain, &mut peer.queue, None)); peer.sync.write().unwrap().restart(&mut TestIo::new(&mut peer.chain, &mut peer.queue, None));
} }
pub fn sync(&mut self) -> u32 { pub fn sync(&mut self) -> u32 {
@ -173,6 +173,6 @@ impl TestNet {
pub fn trigger_chain_new_blocks(&mut self, peer_id: usize) { pub fn trigger_chain_new_blocks(&mut self, peer_id: usize) {
let mut peer = self.peer_mut(peer_id); let mut peer = self.peer_mut(peer_id);
peer.sync.chain_new_blocks(&mut TestIo::new(&mut peer.chain, &mut peer.queue, None), &[], &[], &[], &[]); peer.sync.write().unwrap().chain_new_blocks(&mut TestIo::new(&mut peer.chain, &mut peer.queue, None), &[], &[], &[], &[]);
} }
} }