Unverified Commit fd037238 authored by Roman Zeyde's avatar Roman Zeyde
Browse files

Use std::thread instead of crossbeam

parent b69219a5
......@@ -9,7 +9,6 @@ arrayref = "0.3"
base64 = "0.9"
bincode = "1.0"
bitcoin = "0.13"
crossbeam = "0.3"
error-chain = "0.11"
hex = "0.3"
futures = "0.2"
......
= Electrum
Update height to -1 for txns with any unconfirmed input (https://electrumx.readthedocs.io/en/latest/protocol-basics.html#status)
Snapshot DB after successful indexing - and run queries on this snapshot
Use futures for electrum RPC serving and cancellation
Handle multiple RPC client in parallel
Figure out graceful shutting down and error logging for RPC server
......
use argparse::{ArgumentParser, StoreTrue};
use bitcoin::util::hash::Sha256dHash;
use crossbeam;
use std::fs::OpenOptions;
use std::net::SocketAddr;
use std::sync::Arc;
......@@ -98,15 +97,13 @@ fn run_server(config: &Config) {
});
let query = Arc::new(query::Query::new(app.clone()));
crossbeam::scope(|scope| {
let poll_delay = Duration::from_secs(5);
scope.spawn(|| rpc::serve(&config.rpc_addr, query.clone()));
loop {
thread::sleep(poll_delay);
query.update_mempool().unwrap();
tip = app.update_index(tip);
}
});
let poll_delay = Duration::from_secs(5);
rpc::start(&config.rpc_addr, query.clone());
loop {
thread::sleep(poll_delay);
query.update_mempool().unwrap();
tip = app.update_index(tip);
}
}
fn setup_logging(config: &Config) {
......
......@@ -4,7 +4,6 @@ extern crate argparse;
extern crate base64;
extern crate bincode;
extern crate bitcoin;
extern crate crossbeam;
extern crate crypto;
extern crate futures;
extern crate hex;
......
......@@ -2,14 +2,15 @@ use bitcoin::blockdata::block::BlockHeader;
use bitcoin::blockdata::transaction::Transaction;
use bitcoin::network::serialize::{deserialize, serialize};
use bitcoin::util::hash::Sha256dHash;
use crossbeam;
use error_chain::ChainedError;
use hex;
use serde_json::{from_str, Number, Value};
use std::collections::HashMap;
use std::io::{BufRead, BufReader, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream};
use std::sync::Arc;
use std::sync::mpsc::{sync_channel, Receiver, RecvTimeoutError, SyncSender};
use std::thread;
use std::time::Duration;
use query::Query;
......@@ -294,12 +295,20 @@ impl Connection {
pub fn run(mut self) {
let reader = BufReader::new(self.stream.try_clone().expect("failed to clone TcpStream"));
crossbeam::scope(|scope| {
let chan = Channel::new();
let tx = chan.sender();
scope.spawn(|| Connection::handle_requests(reader, tx));
self.handle_replies(&chan).unwrap();
});
let chan = Channel::new();
let tx = chan.sender();
let child = thread::spawn(|| Connection::handle_requests(reader, tx));
if let Err(e) = self.handle_replies(&chan) {
error!(
"[{}] connection handling failed: {}",
self.addr,
e.display_chain().to_string()
);
}
let _ = self.stream.shutdown(Shutdown::Both);
if child.join().is_err() {
error!("[{}] receiver panicked", self.addr);
}
}
}
......@@ -329,13 +338,13 @@ impl Channel {
}
}
pub fn serve(addr: &SocketAddr, query: Arc<Query>) {
let listener = TcpListener::bind(addr).unwrap();
pub fn start(addr: &SocketAddr, query: Arc<Query>) -> thread::JoinHandle<()> {
let listener = TcpListener::bind(addr).expect(&format!("bind({}) failed", addr));
info!("RPC server running on {}", addr);
loop {
let (stream, addr) = listener.accept().unwrap();
thread::spawn(move || loop {
let (stream, addr) = listener.accept().expect("accept failed");
info!("[{}] connected peer", addr);
Connection::new(query.clone(), stream, addr).run();
info!("[{}] disconnected peer", addr);
}
})
}
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment