Skip to content
This repository was archived by the owner on Nov 6, 2020. It is now read-only.

refactor(fetch) : light use only one DNS thread #9647

Merged
niklasad1 merged 2 commits into from
Sep 28, 2018
Merged

refactor(fetch) : light use only one DNS thread #9647

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ad958e1
refactor(fetch) : light use only one `DNS` thread
niklasad1 Sep 25, 2018
ac3e528
grumbles(fetch) : pass number of threads directly
niklasad1 Sep 26, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions parity/run.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,13 @@ const SNAPSHOT_HISTORY: u64 = 100;
const GAS_CORPUS_EXPIRATION_MINUTES: u64 = 60 * 6;

// Pops along with error messages when a password is missing or invalid.
const VERIFY_PASSWORD_HINT: &'static str = "Make sure valid password is present in files passed using `--password` or in the configuration file.";
const VERIFY_PASSWORD_HINT: &str = "Make sure valid password is present in files passed using `--password` or in the configuration file.";

// Full client number of DNS threads
const FETCH_FULL_NUM_DNS_THREADS: usize = 4;
Copy link
Copy Markdown
Collaborator

@tomusdrw tomusdrw Sep 27, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I doubt we actually need all 4 for the full node either. But fine with keeping it the same for backward compatibility.


// Light client number of DNS threads
const FETCH_LIGHT_NUM_DNS_THREADS: usize = 1;

#[derive(Debug, PartialEq)]
pub struct RunCmd {
Expand Down Expand Up @@ -283,7 +289,7 @@ fn execute_light_impl(cmd: RunCmd, logger: Arc<RotatingLogger>) -> Result<Runnin
let cpu_pool = CpuPool::new(4);

// fetch service
let fetch = fetch::Client::new().map_err(|e| format!("Error starting fetch client: {:?}", e))?;
let fetch = fetch::Client::new(FETCH_LIGHT_NUM_DNS_THREADS).map_err(|e| format!("Error starting fetch client: {:?}", e))?;
let passwords = passwords_from_files(&cmd.acc_conf.password_files)?;

// prepare account provider
Expand Down Expand Up @@ -477,7 +483,7 @@ fn execute_impl<Cr, Rr>(cmd: RunCmd, logger: Arc<RotatingLogger>, on_client_rq:
let event_loop = EventLoop::spawn();

// fetch service
let fetch = fetch::Client::new().map_err(|e| format!("Error starting fetch client: {:?}", e))?;
let fetch = fetch::Client::new(FETCH_FULL_NUM_DNS_THREADS).map_err(|e| format!("Error starting fetch client: {:?}", e))?;

let txpool_size = cmd.miner_options.pool_limits.max_count;
// create miner
Expand Down
35 changes: 23 additions & 12 deletions util/fetch/src/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,11 +170,11 @@ impl Drop for Client {

impl Client {
/// Create a new fetch client.
pub fn new() -> Result<Self, Error> {
pub fn new(num_dns_threads: usize) -> Result<Self, Error> {
let (tx_start, rx_start) = std::sync::mpsc::sync_channel(1);
let (tx_proto, rx_proto) = mpsc::channel(64);

Client::background_thread(tx_start, rx_proto)?;
Client::background_thread(tx_start, rx_proto, num_dns_threads)?;

match rx_start.recv_timeout(Duration::from_secs(10)) {
Err(RecvTimeoutError::Timeout) => {
Expand All @@ -199,7 +199,7 @@ impl Client {
})
}

fn background_thread(tx_start: TxStartup, rx_proto: mpsc::Receiver<ChanItem>) -> io::Result<thread::JoinHandle<()>> {
fn background_thread(tx_start: TxStartup, rx_proto: mpsc::Receiver<ChanItem>, num_dns_threads: usize) -> io::Result<thread::JoinHandle<()>> {
thread::Builder::new().name("fetch".into()).spawn(move || {
let mut core = match reactor::Core::new() {
Ok(c) => c,
Expand All @@ -208,7 +208,7 @@ impl Client {

let handle = core.handle();
let hyper = hyper::Client::configure()
.connector(hyper_rustls::HttpsConnector::new(4, &core.handle()))
.connector(hyper_rustls::HttpsConnector::new(num_dns_threads, &core.handle()))
.build(&core.handle());

let future = rx_proto.take_while(|item| Ok(item.is_some()))
Expand Down Expand Up @@ -640,7 +640,18 @@ mod test {
#[test]
fn it_should_fetch() {
let server = TestServer::run();
let client = Client::new().unwrap();
let client = Client::new(4).unwrap();
let future = client.get(&format!("http://{}?123", server.addr()), Default::default());
let resp = future.wait().unwrap();
assert!(resp.is_success());
let body = resp.concat2().wait().unwrap();
assert_eq!(&body[..], b"123")
}

#[test]
fn it_should_fetch_in_light_mode() {
let server = TestServer::run();
let client = Client::new(1).unwrap();
let future = client.get(&format!("http://{}?123", server.addr()), Default::default());
let resp = future.wait().unwrap();
assert!(resp.is_success());
Expand All @@ -651,7 +662,7 @@ mod test {
#[test]
fn it_should_timeout() {
let server = TestServer::run();
let client = Client::new().unwrap();
let client = Client::new(4).unwrap();
let abort = Abort::default().with_max_duration(Duration::from_secs(1));
match client.get(&format!("http://{}/delay?3", server.addr()), abort).wait() {
Err(Error::Timeout) => {}
Expand All @@ -662,7 +673,7 @@ mod test {
#[test]
fn it_should_follow_redirects() {
let server = TestServer::run();
let client = Client::new().unwrap();
let client = Client::new(4).unwrap();
let abort = Abort::default();
let future = client.get(&format!("http://{}/redirect?http://{}/", server.addr(), server.addr()), abort);
assert!(future.wait().unwrap().is_success())
Expand All @@ -671,7 +682,7 @@ mod test {
#[test]
fn it_should_follow_relative_redirects() {
let server = TestServer::run();
let client = Client::new().unwrap();
let client = Client::new(4).unwrap();
let abort = Abort::default().with_max_redirects(4);
let future = client.get(&format!("http://{}/redirect?/", server.addr()), abort);
assert!(future.wait().unwrap().is_success())
Expand All @@ -680,7 +691,7 @@ mod test {
#[test]
fn it_should_not_follow_too_many_redirects() {
let server = TestServer::run();
let client = Client::new().unwrap();
let client = Client::new(4).unwrap();
let abort = Abort::default().with_max_redirects(3);
match client.get(&format!("http://{}/loop", server.addr()), abort).wait() {
Err(Error::TooManyRedirects) => {}
Expand All @@ -691,7 +702,7 @@ mod test {
#[test]
fn it_should_read_data() {
let server = TestServer::run();
let client = Client::new().unwrap();
let client = Client::new(4).unwrap();
let abort = Abort::default();
let future = client.get(&format!("http://{}?abcdefghijklmnopqrstuvwxyz", server.addr()), abort);
let resp = future.wait().unwrap();
Expand All @@ -702,7 +713,7 @@ mod test {
#[test]
fn it_should_not_read_too_much_data() {
let server = TestServer::run();
let client = Client::new().unwrap();
let client = Client::new(4).unwrap();
let abort = Abort::default().with_max_size(3);
let resp = client.get(&format!("http://{}/?1234", server.addr()), abort).wait().unwrap();
assert!(resp.is_success());
Expand All @@ -715,7 +726,7 @@ mod test {
#[test]
fn it_should_not_read_too_much_data_sync() {
let server = TestServer::run();
let client = Client::new().unwrap();
let client = Client::new(4).unwrap();
let abort = Abort::default().with_max_size(3);
let resp = client.get(&format!("http://{}/?1234", server.addr()), abort).wait().unwrap();
assert!(resp.is_success());
Expand Down