GRIN/grin-node
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[2.x.x] Use blocking IO in P2P to reduce CPU load (#2855)

Browse Source 
* Use blocking IO in P2P to reduce CPU load
This commit is contained in:
hashmap
2019-07-12 23:17:38 +02:00
committed by GitHub
parent 1395074a25
commit d3dbafa80b
11 changed files with 138 additions and 240 deletions
Download Patch File Download Diff File Expand all files Collapse all files
p2p/src/conn.rs
+95 -69
View File
@@ -20,24 +20,26 @@
//! forces us to go through some additional gymnastic to loop over the async
//! stream and make sure we get the right number of bytes out.
use std::fs::File;
use std::io::{self, Read, Write};
use std::net::{Shutdown, TcpStream};
use std::sync::{mpsc, Arc};
use std::{
cmp,
thread::{self, JoinHandle},
time,
};
use crate::core::ser::{self, FixedLength, ProtocolVersion};
use crate::core::ser;
use crate::core::ser::{FixedLength, ProtocolVersion};
use crate::msg::{
read_body, read_discard, read_header, read_item, write_to_buf, MsgHeader, MsgHeaderWrapper,
Type,
};
use crate::types::Error;
use crate::util::read_write::{read_exact, write_all};
use crate::util::{RateCounter, RwLock};
use std::fs::File;
use std::io::{self, Read, Write};
use std::net::{Shutdown, TcpStream};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{mpsc, Arc};
use std::time::Duration;
use std::{
cmp,
thread::{self, JoinHandle},
};
const IO_TIMEOUT: Duration = Duration::from_millis(1000);
/// A trait to be implemented in order to receive messages from the
/// connection. Allows providing an optional response.
@@ -56,7 +58,11 @@ macro_rules! try_break {
($inner:expr) => {
match $inner {
Ok(v) => Some(v),
Err(Error::Connection(ref e)) if e.kind() == io::ErrorKind::WouldBlock => None,
Err(Error::Connection(ref e))
if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut =>
{
None
}
Err(Error::Store(_))
| Err(Error::Chain(_))
| Err(Error::Internal)
@@ -106,12 +112,7 @@ impl<'a> Message<'a> {
while written < len {
let read_len = cmp::min(8000, len - written);
let mut buf = vec![0u8; read_len];
read_exact(
&mut self.stream,
&mut buf[..],
time::Duration::from_secs(10),
true,
)?;
self.stream.read_exact(&mut buf[..])?;
writer.write_all(&mut buf)?;
written += read_len;
}
@@ -151,7 +152,7 @@ impl<'a> Response<'a> {
self.version,
)?;
msg.append(&mut self.body);
write_all(&mut self.stream, &msg[..], time::Duration::from_secs(10))?;
self.stream.write_all(&msg[..])?;
tracker.inc_sent(msg.len() as u64);
if let Some(mut file) = self.attachment {
@@ -160,7 +161,7 @@ impl<'a> Response<'a> {
match file.read(&mut buf[..]) {
Ok(0) => break,
Ok(n) => {
write_all(&mut self.stream, &buf[..n], time::Duration::from_secs(10))?;
self.stream.write_all(&buf[..n])?;
// Increase sent bytes "quietly" without incrementing the counter.
// (In a loop here for the single attachment).
tracker.inc_quiet_sent(n as u64);
@@ -181,34 +182,39 @@ pub const SEND_CHANNEL_CAP: usize = 10;
pub struct StopHandle {
/// Channel to close the connection
pub close_channel: mpsc::Sender<()>,
stopped: Arc<AtomicBool>,
// we need Option to take ownhership of the handle in stop()
peer_thread: Option<JoinHandle<()>>,
reader_thread: Option<JoinHandle<()>>,
writer_thread: Option<JoinHandle<()>>,
}
impl StopHandle {
/// Schedule this connection to safely close via the async close_channel.
pub fn stop(&self) {
if self.close_channel.send(()).is_err() {
debug!("peer's close_channel is disconnected, must be stopped already");
return;
}
self.stopped.store(true, Ordering::Relaxed);
}
pub fn wait(&mut self) {
if let Some(peer_thread) = self.peer_thread.take() {
// wait only if other thread is calling us, eg shutdown
if thread::current().id() != peer_thread.thread().id() {
debug!("waiting for thread {:?} exit", peer_thread.thread().id());
if let Err(e) = peer_thread.join() {
error!("failed to wait for peer thread to stop: {:?}", e);
}
} else {
debug!(
"attempt to wait for thread {:?} from itself",
peer_thread.thread().id()
);
if let Some(reader_thread) = self.reader_thread.take() {
self.join_thread(reader_thread);
}
if let Some(writer_thread) = self.writer_thread.take() {
self.join_thread(writer_thread);
}
}
fn join_thread(&self, peer_thread: JoinHandle<()>) {
// wait only if other thread is calling us, eg shutdown
if thread::current().id() != peer_thread.thread().id() {
debug!("waiting for thread {:?} exit", peer_thread.thread().id());
if let Err(e) = peer_thread.join() {
error!("failed to stop peer thread: {:?}", e);
}
} else {
debug!(
"attempt to stop thread {:?} from itself",
peer_thread.thread().id()
);
}
}
}
@@ -277,20 +283,27 @@ where
H: MessageHandler,
{
let (send_tx, send_rx) = mpsc::sync_channel(SEND_CHANNEL_CAP);
let (close_tx, close_rx) = mpsc::channel();
stream
.set_nonblocking(true)
.expect("Non-blocking IO not available.");
let peer_thread = poll(stream, version, handler, send_rx, close_rx, tracker)?;
.set_read_timeout(Some(IO_TIMEOUT))
.expect("can't set read timeout");
stream
.set_write_timeout(Some(IO_TIMEOUT))
.expect("can't set read timeout");
let stopped = Arc::new(AtomicBool::new(false));
let (reader_thread, writer_thread) =
poll(stream, version, handler, send_rx, stopped.clone(), tracker)?;
Ok((
ConnHandle {
send_channel: send_tx,
},
StopHandle {
close_channel: close_tx,
peer_thread: Some(peer_thread),
stopped,
reader_thread: Some(reader_thread),
writer_thread: Some(writer_thread),
},
))
}
@@ -300,24 +313,24 @@ fn poll<H>(
version: ProtocolVersion,
handler: H,
send_rx: mpsc::Receiver<Vec<u8>>,
close_rx: mpsc::Receiver<()>,
stopped: Arc<AtomicBool>,
tracker: Arc<Tracker>,
) -> io::Result<JoinHandle<()>>
) -> io::Result<(JoinHandle<()>, JoinHandle<()>)>
where
H: MessageHandler,
{
// Split out tcp stream out into separate reader/writer halves.
let mut reader = conn.try_clone().expect("clone conn for reader failed");
let mut writer = conn.try_clone().expect("clone conn for writer failed");
let mut responder = conn.try_clone().expect("clone conn for writer failed");
let reader_stopped = stopped.clone();
thread::Builder::new()
.name("peer".to_string())
let reader_thread = thread::Builder::new()
.name("peer_read".to_string())
.spawn(move || {
let sleep_time = time::Duration::from_millis(5);
let mut retry_send = Err(());
loop {
// check the read end
match try_break!(read_header(&mut reader, version, None)) {
match try_break!(read_header(&mut reader, version)) {
Some(MsgHeaderWrapper::Known(header)) => {
let msg = Message::from_header(header, &mut reader, version);
@@ -331,7 +344,7 @@ where
tracker.inc_received(MsgHeader::LEN as u64 + msg.header.msg_len);
if let Some(Some(resp)) =
try_break!(handler.consume(msg, &mut writer, tracker.clone()))
try_break!(handler.consume(msg, &mut responder, tracker.clone()))
{
try_break!(resp.write(tracker.clone()));
}
@@ -345,35 +358,48 @@ where
None => {}
}
// check the write end, use or_else so try_recv is lazily eval'd
let maybe_data = retry_send.or_else(|_| send_rx.try_recv());
// check the close channel
if reader_stopped.load(Ordering::Relaxed) {
break;
}
}
debug!(
"Shutting down reader connection with {}",
reader
.peer_addr()
.map(|a| a.to_string())
.unwrap_or("?".to_owned())
);
let _ = reader.shutdown(Shutdown::Both);
})?;
let writer_thread = thread::Builder::new()
.name("peer_read".to_string())
.spawn(move || {
let mut retry_send = Err(());
loop {
let maybe_data = retry_send.or_else(|_| send_rx.recv_timeout(IO_TIMEOUT));
retry_send = Err(());
if let Ok(data) = maybe_data {
let written = try_break!(write_all(
&mut writer,
&data[..],
std::time::Duration::from_secs(10)
)
.map_err(&From::from));
let written = try_break!(writer.write_all(&data[..]).map_err(&From::from));
if written.is_none() {
retry_send = Ok(data);
}
}
// check the close channel
if let Ok(_) = close_rx.try_recv() {
if stopped.load(Ordering::Relaxed) {
break;
}
thread::sleep(sleep_time);
}
debug!(
"Shutting down connection with {}",
conn.peer_addr()
"Shutting down reader connection with {}",
writer
.peer_addr()
.map(|a| a.to_string())
.unwrap_or("?".to_owned())
);
let _ = conn.shutdown(Shutdown::Both);
})
})?;
Ok((reader_thread, writer_thread))
}