Merge branch 'master' of github.com:ethcore/parity into jsonrpc2
This commit is contained in:
commit
484a4d8bdd
@ -11,7 +11,7 @@ env_logger = "0.3"
|
||||
rustc-serialize = "0.3"
|
||||
docopt = "0.6"
|
||||
docopt_macros = "0.6"
|
||||
ctrlc = "1.0"
|
||||
ctrlc = { git = "https://github.com/tomusdrw/rust-ctrlc.git" }
|
||||
clippy = "0.0.37"
|
||||
ethcore-util = { path = "util" }
|
||||
ethcore = { path = "ethcore" }
|
||||
|
@ -26,6 +26,7 @@ use views::*;
|
||||
use header::*;
|
||||
use service::*;
|
||||
use client::BlockStatus;
|
||||
use util::panics::*;
|
||||
|
||||
/// Block queue status
|
||||
#[derive(Debug)]
|
||||
@ -59,6 +60,7 @@ impl BlockQueueInfo {
|
||||
/// A queue of blocks. Sits between network or other I/O and the BlockChain.
|
||||
/// Sorts them ready for blockchain insertion.
|
||||
pub struct BlockQueue {
|
||||
panic_handler: Arc<PanicHandler>,
|
||||
engine: Arc<Box<Engine>>,
|
||||
more_to_verify: Arc<Condvar>,
|
||||
verification: Arc<Mutex<Verification>>,
|
||||
@ -113,6 +115,7 @@ impl BlockQueue {
|
||||
let ready_signal = Arc::new(QueueSignal { signalled: AtomicBool::new(false), message_channel: message_channel });
|
||||
let deleting = Arc::new(AtomicBool::new(false));
|
||||
let empty = Arc::new(Condvar::new());
|
||||
let panic_handler = PanicHandler::new_in_arc();
|
||||
|
||||
let mut verifiers: Vec<JoinHandle<()>> = Vec::new();
|
||||
let thread_count = max(::num_cpus::get(), 3) - 2;
|
||||
@ -123,11 +126,21 @@ impl BlockQueue {
|
||||
let ready_signal = ready_signal.clone();
|
||||
let empty = empty.clone();
|
||||
let deleting = deleting.clone();
|
||||
verifiers.push(thread::Builder::new().name(format!("Verifier #{}", i)).spawn(move || BlockQueue::verify(verification, engine, more_to_verify, ready_signal, deleting, empty))
|
||||
.expect("Error starting block verification thread"));
|
||||
let panic_handler = panic_handler.clone();
|
||||
verifiers.push(
|
||||
thread::Builder::new()
|
||||
.name(format!("Verifier #{}", i))
|
||||
.spawn(move || {
|
||||
panic_handler.catch_panic(move || {
|
||||
BlockQueue::verify(verification, engine, more_to_verify, ready_signal, deleting, empty)
|
||||
}).unwrap()
|
||||
})
|
||||
.expect("Error starting block verification thread")
|
||||
);
|
||||
}
|
||||
BlockQueue {
|
||||
engine: engine,
|
||||
panic_handler: panic_handler,
|
||||
ready_signal: ready_signal.clone(),
|
||||
more_to_verify: more_to_verify.clone(),
|
||||
verification: verification.clone(),
|
||||
@ -324,6 +337,12 @@ impl BlockQueue {
|
||||
}
|
||||
}
|
||||
|
||||
impl MayPanic for BlockQueue {
|
||||
fn on_panic<F>(&self, closure: F) where F: OnPanicListener {
|
||||
self.panic_handler.on_panic(closure);
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for BlockQueue {
|
||||
fn drop(&mut self) {
|
||||
self.clear();
|
||||
|
@ -859,7 +859,7 @@ mod tests {
|
||||
let transactions = bc.transactions(&b1_hash).unwrap();
|
||||
assert_eq!(transactions.len(), 7);
|
||||
for t in transactions {
|
||||
assert_eq!(bc.transaction(TransactionId::Hash(t.hash())).unwrap(), t);
|
||||
assert_eq!(bc.transaction(&t.hash()).unwrap(), t);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -17,6 +17,7 @@
|
||||
//! Blockchain database client.
|
||||
|
||||
use util::*;
|
||||
use util::panics::*;
|
||||
use rocksdb::{Options, DB, DBCompactionStyle};
|
||||
use blockchain::{BlockChain, BlockProvider, CacheSize};
|
||||
use views::BlockView;
|
||||
@ -167,7 +168,8 @@ pub struct Client {
|
||||
state_db: Mutex<JournalDB>,
|
||||
block_queue: RwLock<BlockQueue>,
|
||||
report: RwLock<ClientReport>,
|
||||
import_lock: Mutex<()>
|
||||
import_lock: Mutex<()>,
|
||||
panic_handler: Arc<PanicHandler>,
|
||||
}
|
||||
|
||||
const HISTORY: u64 = 1000;
|
||||
@ -214,13 +216,19 @@ impl Client {
|
||||
if state_db.is_empty() && engine.spec().ensure_db_good(&mut state_db) {
|
||||
state_db.commit(0, &engine.spec().genesis_header().hash(), None).expect("Error commiting genesis state to state DB");
|
||||
}
|
||||
|
||||
let block_queue = BlockQueue::new(engine.clone(), message_channel);
|
||||
let panic_handler = PanicHandler::new_in_arc();
|
||||
panic_handler.forward_from(&block_queue);
|
||||
|
||||
Ok(Arc::new(Client {
|
||||
chain: chain,
|
||||
engine: engine.clone(),
|
||||
engine: engine,
|
||||
state_db: Mutex::new(state_db),
|
||||
block_queue: RwLock::new(BlockQueue::new(engine, message_channel)),
|
||||
block_queue: RwLock::new(block_queue),
|
||||
report: RwLock::new(Default::default()),
|
||||
import_lock: Mutex::new(()),
|
||||
panic_handler: panic_handler
|
||||
}))
|
||||
}
|
||||
|
||||
@ -435,3 +443,9 @@ impl BlockChainClient for Client {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl MayPanic for Client {
|
||||
fn on_panic<F>(&self, closure: F) where F: OnPanicListener {
|
||||
self.panic_handler.on_panic(closure);
|
||||
}
|
||||
}
|
||||
|
@ -17,6 +17,7 @@
|
||||
//! Creates and registers client and network services.
|
||||
|
||||
use util::*;
|
||||
use util::panics::*;
|
||||
use spec::Spec;
|
||||
use error::*;
|
||||
use std::env;
|
||||
@ -38,17 +39,22 @@ pub type NetSyncMessage = NetworkIoMessage<SyncMessage>;
|
||||
pub struct ClientService {
|
||||
net_service: NetworkService<SyncMessage>,
|
||||
client: Arc<Client>,
|
||||
panic_handler: Arc<PanicHandler>
|
||||
}
|
||||
|
||||
impl ClientService {
|
||||
/// Start the service in a separate thread.
|
||||
pub fn start(spec: Spec, net_config: NetworkConfiguration) -> Result<ClientService, Error> {
|
||||
let panic_handler = PanicHandler::new_in_arc();
|
||||
let mut net_service = try!(NetworkService::start(net_config));
|
||||
panic_handler.forward_from(&net_service);
|
||||
|
||||
info!("Starting {}", net_service.host_info());
|
||||
info!("Configured for {} using {} engine", spec.name, spec.engine_name);
|
||||
let mut dir = env::home_dir().unwrap();
|
||||
dir.push(".parity");
|
||||
let client = try!(Client::new(spec, &dir, net_service.io().channel()));
|
||||
panic_handler.forward_from(client.deref());
|
||||
let client_io = Arc::new(ClientIoHandler {
|
||||
client: client.clone()
|
||||
});
|
||||
@ -57,6 +63,7 @@ impl ClientService {
|
||||
Ok(ClientService {
|
||||
net_service: net_service,
|
||||
client: client,
|
||||
panic_handler: panic_handler,
|
||||
})
|
||||
}
|
||||
|
||||
@ -81,6 +88,12 @@ impl ClientService {
|
||||
}
|
||||
}
|
||||
|
||||
impl MayPanic for ClientService {
|
||||
fn on_panic<F>(&self, closure: F) where F: OnPanicListener {
|
||||
self.panic_handler.on_panic(closure);
|
||||
}
|
||||
}
|
||||
|
||||
/// IO interface for the Client handler
|
||||
struct ClientIoHandler {
|
||||
client: Arc<Client>
|
||||
|
117
parity/main.rs
117
parity/main.rs
@ -40,6 +40,7 @@ use rlog::{LogLevelFilter};
|
||||
use env_logger::LogBuilder;
|
||||
use ctrlc::CtrlC;
|
||||
use util::*;
|
||||
use util::panics::MayPanic;
|
||||
use ethcore::spec::*;
|
||||
use ethcore::client::*;
|
||||
use ethcore::service::{ClientService, NetSyncMessage};
|
||||
@ -105,10 +106,17 @@ fn setup_rpc_server(client: Arc<Client>, sync: Arc<EthSync>, url: &str) {
|
||||
fn setup_rpc_server(_client: Arc<Client>, _sync: Arc<EthSync>, _url: &str) {
|
||||
}
|
||||
|
||||
fn main() {
|
||||
let args: Args = Args::docopt().decode().unwrap_or_else(|e| e.exit());
|
||||
struct Configuration {
|
||||
args: Args
|
||||
}
|
||||
impl Configuration {
|
||||
fn parse() -> Self {
|
||||
Configuration {
|
||||
args: Args::docopt().decode().unwrap_or_else(|e| e.exit())
|
||||
}
|
||||
}
|
||||
|
||||
if args.flag_version {
|
||||
fn print_version(&self) {
|
||||
println!("\
|
||||
Parity version {} ({}-{}-{})
|
||||
Copyright 2015, 2016 Ethcore (UK) Limited
|
||||
@ -118,51 +126,102 @@ There is NO WARRANTY, to the extent permitted by law.
|
||||
|
||||
By Wood/Paronyan/Kotewicz/Drwięga/Volf.\
|
||||
", env!("CARGO_PKG_VERSION"), Target::arch(), Target::env(), Target::os());
|
||||
return;
|
||||
}
|
||||
|
||||
setup_log(&args.flag_logging);
|
||||
unsafe { ::fdlimit::raise_fd_limit(); }
|
||||
|
||||
let spec = match args.flag_chain.as_ref() {
|
||||
fn get_spec(&self) -> Spec {
|
||||
match self.args.flag_chain.as_ref() {
|
||||
"frontier" | "mainnet" => ethereum::new_frontier(),
|
||||
"morden" | "testnet" => ethereum::new_morden(),
|
||||
"olympic" => ethereum::new_olympic(),
|
||||
f => Spec::from_json_utf8(contents(f).expect("Couldn't read chain specification file. Sure it exists?").as_ref()),
|
||||
};
|
||||
let init_nodes = match args.arg_enode.len() {
|
||||
}
|
||||
}
|
||||
|
||||
fn get_init_nodes(&self, spec: &Spec) -> Vec<String> {
|
||||
match self.args.arg_enode.len() {
|
||||
0 => spec.nodes().clone(),
|
||||
_ => args.arg_enode.clone(),
|
||||
};
|
||||
let mut net_settings = NetworkConfiguration::new();
|
||||
net_settings.boot_nodes = init_nodes;
|
||||
match args.flag_address {
|
||||
_ => self.args.arg_enode.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
fn get_net_addresses(&self) -> (SocketAddr, SocketAddr) {
|
||||
let listen_address;
|
||||
let public_address;
|
||||
|
||||
match self.args.flag_address {
|
||||
None => {
|
||||
net_settings.listen_address = SocketAddr::from_str(args.flag_listen_address.as_ref()).expect("Invalid listen address given with --listen-address");
|
||||
net_settings.public_address = SocketAddr::from_str(args.flag_public_address.as_ref()).expect("Invalid public address given with --public-address");
|
||||
listen_address = SocketAddr::from_str(self.args.flag_listen_address.as_ref()).expect("Invalid listen address given with --listen-address");
|
||||
public_address = SocketAddr::from_str(self.args.flag_public_address.as_ref()).expect("Invalid public address given with --public-address");
|
||||
}
|
||||
Some(ref a) => {
|
||||
net_settings.public_address = SocketAddr::from_str(a.as_ref()).expect("Invalid listen/public address given with --address");
|
||||
net_settings.listen_address = net_settings.public_address.clone();
|
||||
public_address = SocketAddr::from_str(a.as_ref()).expect("Invalid listen/public address given with --address");
|
||||
listen_address = public_address.clone();
|
||||
}
|
||||
}
|
||||
let mut service = ClientService::start(spec, net_settings).unwrap();
|
||||
let client = service.client().clone();
|
||||
client.configure_cache(args.flag_cache_pref_size, args.flag_cache_max_size);
|
||||
let sync = EthSync::register(service.network(), client);
|
||||
if args.flag_jsonrpc {
|
||||
setup_rpc_server(service.client(), sync.clone(), &args.flag_jsonrpc_url);
|
||||
}
|
||||
let io_handler = Arc::new(ClientIoHandler { client: service.client(), info: Default::default(), sync: sync });
|
||||
service.io().register_handler(io_handler).expect("Error registering IO handler");
|
||||
};
|
||||
|
||||
(listen_address, public_address)
|
||||
}
|
||||
}
|
||||
|
||||
fn wait_for_exit(client_service: &ClientService) {
|
||||
let exit = Arc::new(Condvar::new());
|
||||
// Handle possible exits
|
||||
let e = exit.clone();
|
||||
CtrlC::set_handler(move || { e.notify_all(); });
|
||||
let e = exit.clone();
|
||||
client_service.on_panic(move |_reason| { e.notify_all(); });
|
||||
// Wait for signal
|
||||
let mutex = Mutex::new(());
|
||||
let _ = exit.wait(mutex.lock().unwrap()).unwrap();
|
||||
}
|
||||
|
||||
fn main() {
|
||||
let conf = Configuration::parse();
|
||||
if conf.args.flag_version {
|
||||
conf.print_version();
|
||||
return;
|
||||
}
|
||||
|
||||
let spec = conf.get_spec();
|
||||
|
||||
// Setup logging
|
||||
setup_log(&conf.args.flag_logging);
|
||||
// Raise fdlimit
|
||||
unsafe { ::fdlimit::raise_fd_limit(); }
|
||||
|
||||
// Configure network
|
||||
let init_nodes = conf.get_init_nodes(&spec);
|
||||
let (listen, public) = conf.get_net_addresses();
|
||||
let mut net_settings = NetworkConfiguration::new();
|
||||
net_settings.boot_nodes = init_nodes;
|
||||
net_settings.listen_address = listen;
|
||||
net_settings.public_address = public;
|
||||
|
||||
// Build client
|
||||
let mut service = ClientService::start(spec, net_settings).unwrap();
|
||||
let client = service.client().clone();
|
||||
client.configure_cache(conf.args.flag_cache_pref_size, conf.args.flag_cache_max_size);
|
||||
|
||||
// Sync
|
||||
let sync = EthSync::register(service.network(), client);
|
||||
|
||||
// Setup rpc
|
||||
if conf.args.flag_jsonrpc {
|
||||
setup_rpc_server(service.client(), sync.clone(), &conf.args.flag_jsonrpc_url);
|
||||
}
|
||||
|
||||
// Register IO handler
|
||||
let io_handler = Arc::new(ClientIoHandler {
|
||||
client: service.client(),
|
||||
info: Default::default(),
|
||||
sync: sync
|
||||
});
|
||||
service.io().register_handler(io_handler).expect("Error registering IO handler");
|
||||
|
||||
// Handle exit
|
||||
wait_for_exit(&service);
|
||||
}
|
||||
|
||||
struct Informant {
|
||||
chain_info: RwLock<Option<BlockChainInfo>>,
|
||||
cache_info: RwLock<Option<CacheSize>>,
|
||||
|
@ -25,6 +25,7 @@ use io::{IoError, IoHandler};
|
||||
use arrayvec::*;
|
||||
use crossbeam::sync::chase_lev;
|
||||
use io::worker::{Worker, Work, WorkType};
|
||||
use panics::*;
|
||||
|
||||
/// Timer ID
|
||||
pub type TimerToken = usize;
|
||||
@ -159,13 +160,21 @@ pub struct IoManager<Message> where Message: Send + Sync {
|
||||
|
||||
impl<Message> IoManager<Message> where Message: Send + Sync + Clone + 'static {
|
||||
/// Creates a new instance and registers it with the event loop.
|
||||
pub fn start(event_loop: &mut EventLoop<IoManager<Message>>) -> Result<(), UtilError> {
|
||||
pub fn start(panic_handler: Arc<PanicHandler>, event_loop: &mut EventLoop<IoManager<Message>>) -> Result<(), UtilError> {
|
||||
let (worker, stealer) = chase_lev::deque();
|
||||
let num_workers = 4;
|
||||
let work_ready_mutex = Arc::new(Mutex::new(()));
|
||||
let work_ready = Arc::new(Condvar::new());
|
||||
let workers = (0..num_workers).map(|i|
|
||||
Worker::new(i, stealer.clone(), IoChannel::new(event_loop.channel()), work_ready.clone(), work_ready_mutex.clone())).collect();
|
||||
Worker::new(
|
||||
i,
|
||||
stealer.clone(),
|
||||
IoChannel::new(event_loop.channel()),
|
||||
work_ready.clone(),
|
||||
work_ready_mutex.clone(),
|
||||
panic_handler.clone()
|
||||
)
|
||||
).collect();
|
||||
|
||||
let mut io = IoManager {
|
||||
timers: Arc::new(RwLock::new(HashMap::new())),
|
||||
@ -306,19 +315,32 @@ impl<Message> IoChannel<Message> where Message: Send + Clone {
|
||||
/// General IO Service. Starts an event loop and dispatches IO requests.
|
||||
/// 'Message' is a notification message type
|
||||
pub struct IoService<Message> where Message: Send + Sync + Clone + 'static {
|
||||
panic_handler: Arc<PanicHandler>,
|
||||
thread: Option<JoinHandle<()>>,
|
||||
host_channel: Sender<IoMessage<Message>>,
|
||||
}
|
||||
|
||||
impl<Message> MayPanic for IoService<Message> where Message: Send + Sync + Clone + 'static {
|
||||
fn on_panic<F>(&self, closure: F) where F: OnPanicListener {
|
||||
self.panic_handler.on_panic(closure);
|
||||
}
|
||||
}
|
||||
|
||||
impl<Message> IoService<Message> where Message: Send + Sync + Clone + 'static {
|
||||
/// Starts IO event loop
|
||||
pub fn start() -> Result<IoService<Message>, UtilError> {
|
||||
let panic_handler = PanicHandler::new_in_arc();
|
||||
let mut event_loop = EventLoop::new().unwrap();
|
||||
let channel = event_loop.channel();
|
||||
let panic = panic_handler.clone();
|
||||
let thread = thread::spawn(move || {
|
||||
IoManager::<Message>::start(&mut event_loop).unwrap(); //TODO:
|
||||
let p = panic.clone();
|
||||
panic.catch_panic(move || {
|
||||
IoManager::<Message>::start(p, &mut event_loop).unwrap();
|
||||
}).unwrap()
|
||||
});
|
||||
Ok(IoService {
|
||||
panic_handler: panic_handler,
|
||||
thread: Some(thread),
|
||||
host_channel: channel
|
||||
})
|
||||
|
@ -21,6 +21,7 @@ use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
|
||||
use crossbeam::sync::chase_lev;
|
||||
use io::service::{HandlerId, IoChannel, IoContext};
|
||||
use io::{IoHandler};
|
||||
use panics::*;
|
||||
|
||||
pub enum WorkType<Message> {
|
||||
Readable,
|
||||
@ -51,7 +52,9 @@ impl Worker {
|
||||
stealer: chase_lev::Stealer<Work<Message>>,
|
||||
channel: IoChannel<Message>,
|
||||
wait: Arc<Condvar>,
|
||||
wait_mutex: Arc<Mutex<()>>) -> Worker
|
||||
wait_mutex: Arc<Mutex<()>>,
|
||||
panic_handler: Arc<PanicHandler>
|
||||
) -> Worker
|
||||
where Message: Send + Sync + Clone + 'static {
|
||||
let deleting = Arc::new(AtomicBool::new(false));
|
||||
let mut worker = Worker {
|
||||
@ -60,7 +63,11 @@ impl Worker {
|
||||
deleting: deleting.clone(),
|
||||
};
|
||||
worker.thread = Some(thread::Builder::new().name(format!("IO Worker #{}", index)).spawn(
|
||||
move || Worker::work_loop(stealer, channel.clone(), wait, wait_mutex.clone(), deleting))
|
||||
move || {
|
||||
panic_handler.catch_panic(move || {
|
||||
Worker::work_loop(stealer, channel.clone(), wait, wait_mutex.clone(), deleting)
|
||||
}).unwrap()
|
||||
})
|
||||
.expect("Error creating worker thread"));
|
||||
worker
|
||||
}
|
||||
|
@ -21,6 +21,8 @@
|
||||
#![feature(plugin)]
|
||||
#![plugin(clippy)]
|
||||
#![allow(needless_range_loop, match_bool)]
|
||||
#![feature(catch_panic)]
|
||||
|
||||
//! Ethcore-util library
|
||||
//!
|
||||
//! ### Rust version:
|
||||
@ -129,6 +131,7 @@ pub mod semantic_version;
|
||||
pub mod io;
|
||||
pub mod network;
|
||||
pub mod log;
|
||||
pub mod panics;
|
||||
|
||||
pub use common::*;
|
||||
pub use misc::*;
|
||||
|
@ -16,6 +16,7 @@
|
||||
|
||||
use std::sync::*;
|
||||
use error::*;
|
||||
use panics::*;
|
||||
use network::{NetworkProtocolHandler, NetworkConfiguration};
|
||||
use network::error::{NetworkError};
|
||||
use network::host::{Host, NetworkIoMessage, ProtocolId};
|
||||
@ -27,13 +28,17 @@ use io::*;
|
||||
pub struct NetworkService<Message> where Message: Send + Sync + Clone + 'static {
|
||||
io_service: IoService<NetworkIoMessage<Message>>,
|
||||
host_info: String,
|
||||
stats: Arc<NetworkStats>
|
||||
stats: Arc<NetworkStats>,
|
||||
panic_handler: Arc<PanicHandler>
|
||||
}
|
||||
|
||||
impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'static {
|
||||
/// Starts IO event loop
|
||||
pub fn start(config: NetworkConfiguration) -> Result<NetworkService<Message>, UtilError> {
|
||||
let panic_handler = PanicHandler::new_in_arc();
|
||||
let mut io_service = try!(IoService::<NetworkIoMessage<Message>>::start());
|
||||
panic_handler.forward_from(&io_service);
|
||||
|
||||
let host = Arc::new(Host::new(config));
|
||||
let stats = host.stats().clone();
|
||||
let host_info = host.client_version();
|
||||
@ -43,6 +48,7 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
|
||||
io_service: io_service,
|
||||
host_info: host_info,
|
||||
stats: stats,
|
||||
panic_handler: panic_handler
|
||||
})
|
||||
}
|
||||
|
||||
@ -72,3 +78,9 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
impl<Message> MayPanic for NetworkService<Message> where Message: Send + Sync + Clone + 'static {
|
||||
fn on_panic<F>(&self, closure: F) where F: OnPanicListener {
|
||||
self.panic_handler.on_panic(closure);
|
||||
}
|
||||
}
|
||||
|
183
util/src/panics.rs
Normal file
183
util/src/panics.rs
Normal file
@ -0,0 +1,183 @@
|
||||
// Copyright 2015, 2016 Ethcore (UK) Ltd.
|
||||
// This file is part of Parity.
|
||||
|
||||
// Parity is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
|
||||
// Parity is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
//! Panic utilities
|
||||
|
||||
use std::thread;
|
||||
use std::ops::DerefMut;
|
||||
use std::any::Any;
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
/// Thread-safe closure for handling possible panics
|
||||
pub trait OnPanicListener: Send + Sync + 'static {
|
||||
/// Invoke listener
|
||||
fn call(&mut self, arg: &str);
|
||||
}
|
||||
|
||||
/// Forwards panics from child
|
||||
pub trait ForwardPanic {
|
||||
/// Attach `on_panic` listener to `child` and rethrow all panics
|
||||
fn forward_from<S>(&self, child: &S) where S : MayPanic;
|
||||
}
|
||||
|
||||
/// Trait indicating that the structure catches some of the panics (most probably from spawned threads)
|
||||
/// and it's possbile to be notified when one of the threads panics.
|
||||
pub trait MayPanic {
|
||||
/// `closure` will be invoked whenever panic in thread is caught
|
||||
fn on_panic<F>(&self, closure: F) where F: OnPanicListener;
|
||||
}
|
||||
|
||||
/// Structure that allows to catch panics and notify listeners
|
||||
pub struct PanicHandler {
|
||||
listeners: Mutex<Vec<Box<OnPanicListener>>>
|
||||
}
|
||||
|
||||
impl PanicHandler {
|
||||
/// Creates new `PanicHandler` wrapped in `Arc`
|
||||
pub fn new_in_arc() -> Arc<PanicHandler> {
|
||||
Arc::new(Self::new())
|
||||
}
|
||||
|
||||
/// Creates new `PanicHandler`
|
||||
pub fn new() -> PanicHandler {
|
||||
PanicHandler {
|
||||
listeners: Mutex::new(vec![])
|
||||
}
|
||||
}
|
||||
|
||||
/// Invoke closure and catch any possible panics.
|
||||
/// In case of panic notifies all listeners about it.
|
||||
#[allow(deprecated)]
|
||||
// TODO [todr] catch_panic is deprecated but panic::recover has different bounds (not allowing mutex)
|
||||
pub fn catch_panic<G, R>(&self, g: G) -> thread::Result<R> where G: FnOnce() -> R + Send + 'static {
|
||||
let result = thread::catch_panic(g);
|
||||
|
||||
if let Err(ref e) = result {
|
||||
let res = convert_to_string(e);
|
||||
if let Some(r) = res {
|
||||
self.notify_all(r);
|
||||
}
|
||||
}
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
fn notify_all(&self, r: String) {
|
||||
let mut listeners = self.listeners.lock().unwrap();
|
||||
for listener in listeners.deref_mut() {
|
||||
listener.call(&r);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl MayPanic for PanicHandler {
|
||||
fn on_panic<F>(&self, closure: F) where F: OnPanicListener {
|
||||
self.listeners.lock().unwrap().push(Box::new(closure));
|
||||
}
|
||||
}
|
||||
|
||||
impl ForwardPanic for Arc<PanicHandler> {
|
||||
fn forward_from<S>(&self, child: &S) where S : MayPanic {
|
||||
let p = self.clone();
|
||||
child.on_panic(move |t| p.notify_all(t));
|
||||
}
|
||||
}
|
||||
|
||||
impl<F> OnPanicListener for F
|
||||
where F: FnMut(String) + Send + Sync + 'static {
|
||||
fn call(&mut self, arg: &str) {
|
||||
self(arg.to_owned())
|
||||
}
|
||||
}
|
||||
|
||||
fn convert_to_string(t: &Box<Any + Send>) -> Option<String> {
|
||||
let as_str = t.downcast_ref::<&'static str>().map(|t| t.clone().to_owned());
|
||||
let as_string = t.downcast_ref::<String>().cloned();
|
||||
|
||||
as_str.or(as_string)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn should_notify_listeners_about_panic () {
|
||||
use std::sync::RwLock;
|
||||
// given
|
||||
let invocations = Arc::new(RwLock::new(vec![]));
|
||||
let i = invocations.clone();
|
||||
let p = PanicHandler::new();
|
||||
p.on_panic(move |t| i.write().unwrap().push(t));
|
||||
|
||||
// when
|
||||
p.catch_panic(|| panic!("Panic!")).unwrap_err();
|
||||
|
||||
// then
|
||||
assert!(invocations.read().unwrap()[0] == "Panic!");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn should_notify_listeners_about_panic_when_string_is_dynamic () {
|
||||
use std::sync::RwLock;
|
||||
// given
|
||||
let invocations = Arc::new(RwLock::new(vec![]));
|
||||
let i = invocations.clone();
|
||||
let p = PanicHandler::new();
|
||||
p.on_panic(move |t| i.write().unwrap().push(t));
|
||||
|
||||
// when
|
||||
p.catch_panic(|| panic!("Panic: {}", 1)).unwrap_err();
|
||||
|
||||
// then
|
||||
assert!(invocations.read().unwrap()[0] == "Panic: 1");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn should_notify_listeners_about_panic_in_other_thread () {
|
||||
use std::thread;
|
||||
use std::sync::RwLock;
|
||||
|
||||
// given
|
||||
let invocations = Arc::new(RwLock::new(vec![]));
|
||||
let i = invocations.clone();
|
||||
let p = PanicHandler::new();
|
||||
p.on_panic(move |t| i.write().unwrap().push(t));
|
||||
|
||||
// when
|
||||
let t = thread::spawn(move ||
|
||||
p.catch_panic(|| panic!("Panic!")).unwrap()
|
||||
);
|
||||
t.join().unwrap_err();
|
||||
|
||||
// then
|
||||
assert!(invocations.read().unwrap()[0] == "Panic!");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn should_forward_panics () {
|
||||
use std::sync::RwLock;
|
||||
// given
|
||||
let invocations = Arc::new(RwLock::new(vec![]));
|
||||
let i = invocations.clone();
|
||||
let p = PanicHandler::new_in_arc();
|
||||
p.on_panic(move |t| i.write().unwrap().push(t));
|
||||
|
||||
let p2 = PanicHandler::new();
|
||||
p.forward_from(&p2);
|
||||
|
||||
// when
|
||||
p2.catch_panic(|| panic!("Panic!")).unwrap_err();
|
||||
|
||||
// then
|
||||
assert!(invocations.read().unwrap()[0] == "Panic!");
|
||||
}
|
Loading…
Reference in New Issue
Block a user