Implementing PanicHandlers for all places when new thread is spawned. Handling Client panics

ethcore/src/block_queue.rs

@@ -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: SafeStringPanicHandler,
 	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 = StringPanicHandler::new_thread_safe();
 
 		let mut verifiers: Vec<JoinHandle<()>> = Vec::new();
 		let thread_count = max(::num_cpus::get(), 3) - 2;
@@ -123,11 +126,22 @@ 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))
+			let panic_handler = panic_handler.clone();
-				.expect("Error starting block verification thread"));
+			verifiers.push(
+				thread::Builder::new()
+				.name(format!("Verifier #{}", i))
+				.spawn(move || {
+					let mut panic = panic_handler.lock().unwrap();
+					panic.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 +338,12 @@ impl BlockQueue {
 	}
 }
 
+impl MayPanic<String> for BlockQueue {
+	fn on_panic<F>(&self, closure: F) where F: OnPanicListener<String> {
+		self.panic_handler.on_panic(closure);
+	}
+}
+
 impl Drop for BlockQueue {
 	fn drop(&mut self) {
 		self.clear();

ethcore/src/client.rs

@@ -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;
@@ -157,7 +158,8 @@ pub struct Client {
 	state_db: Mutex<JournalDB>,
 	block_queue: RwLock<BlockQueue>,
 	report: RwLock<ClientReport>,
-	import_lock: Mutex<()>
+	import_lock: Mutex<()>,
+	panic_handler: SafeStringPanicHandler,
 }
 
 const HISTORY: u64 = 1000;
@@ -204,13 +206,20 @@ 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 = StringPanicHandler::new_thread_safe();
+		let panic = panic_handler.clone();
+		block_queue.on_panic(move |t: &String| panic.lock().unwrap().notify_all(t));
+
 		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
 		}))
 	}
 
@@ -423,3 +432,9 @@ impl BlockChainClient for Client {
 		}
 	}
 }
+
+impl MayPanic<String> for Client {
+	fn on_panic<F>(&self, closure: F) where F: OnPanicListener<String> {
+		self.panic_handler.on_panic(closure);
+	}
+}

parity/main.rs

@@ -39,9 +39,11 @@ use rlog::{LogLevelFilter};
 use env_logger::LogBuilder;
 use ctrlc::CtrlC;
 use util::*;
+use util::panics::MayPanic;
 use ethcore::client::*;
 use ethcore::service::{ClientService, NetSyncMessage};
 use ethcore::ethereum;
+use ethcore::spec;
 use ethcore::blockchain::CacheSize;
 use ethsync::EthSync;
 
@@ -97,46 +99,96 @@ 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() {
+struct Configuration {
-	let args: Args = Args::docopt().decode().unwrap_or_else(|e| e.exit());
+	args: Args
-
+}
-	setup_log(&args.flag_logging);
+impl Configuration {
-	unsafe { ::fdlimit::raise_fd_limit(); }
+	fn parse() -> Self {
-
+		Configuration {
-	let spec = ethereum::new_frontier();
+			args: Args::docopt().decode().unwrap_or_else(|e| e.exit())
-	let init_nodes = match 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 {
-		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");
-		}
-		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();
 		}
 	}
-	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");
 
+	fn get_init_nodes(&self, spec: &spec::Spec) -> Vec<String> {
+		match self.args.arg_enode.len() {
+			0 => spec.nodes().clone(),
+			_ => self.args.arg_enode.clone(),
+		}
+	}
+
+	fn get_net_addresses(&self) -> (SocketAddr, SocketAddr) {
+		let listen_address;
+		let public_address;
+
+		match self.args.flag_address {
+			None => {
+				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) => {
+				public_address = SocketAddr::from_str(a.as_ref()).expect("Invalid listen/public address given with --address");
+				listen_address = public_address.clone();
+			}
+		};
+
+		(listen_address, public_address)
+	}
+}
+
+fn wait_for_exit(client: Arc<Client>) {
 	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.on_panic(move |_t: &String| { e.notify_all(); });
+	// Wait for signal
 	let mutex = Mutex::new(());
 	let _ = exit.wait(mutex.lock().unwrap()).unwrap();
 }
 
+fn main() {
+	let conf = Configuration::parse();
+	let spec = ethereum::new_frontier();
+
+	// 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.client());
+}
+
 struct Informant {
 	chain_info: RwLock<Option<BlockChainInfo>>,
 	cache_info: RwLock<Option<CacheSize>>,

util/src/io/mod.rs

@@ -31,16 +31,16 @@
 //!
 //! impl IoHandler<MyMessage> for MyHandler {
 //! 	fn initialize(&self, io: &IoContext<MyMessage>) {
-//!		io.register_timer(0, 1000).unwrap();
+//!			io.register_timer(0, 1000).unwrap();
-//!	}
+//!		}
 //!
-//!	fn timeout(&self, _io: &IoContext<MyMessage>, timer: TimerToken) {
+//!		fn timeout(&self, _io: &IoContext<MyMessage>, timer: TimerToken) {
-//!		println!("Timeout {}", timer);
+//!			println!("Timeout {}", timer);
-//!	}
+//!		}
 //!
-//!	fn message(&self, _io: &IoContext<MyMessage>, message: &MyMessage) {
+//!		fn message(&self, _io: &IoContext<MyMessage>, message: &MyMessage) {
-//!		println!("Message {}", message.data);
+//!			println!("Message {}", message.data);
-//!	}
+//!		}
 //! }
 //!
 //! fn main () {

util/src/io/service.rs

@@ -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;
@@ -306,19 +307,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: SafeStringPanicHandler,
 	thread: Option<JoinHandle<()>>,
 	host_channel: Sender<IoMessage<Message>>,
 }
 
+impl<Message> MayPanic<String> for IoService<Message> where Message: Send + Sync + Clone + 'static {
+	fn on_panic<F>(&self, closure: F) where F: OnPanicListener<String> {
+		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 = StringPanicHandler::new_thread_safe();
 		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 mut panic = panic.lock().unwrap();
+			panic.catch_panic(move || {
+				IoManager::<Message>::start(&mut event_loop).unwrap();
+			}).unwrap()
 		});
 		Ok(IoService {
+			panic_handler: panic_handler,
 			thread: Some(thread),
 			host_channel: channel
 		})

util/src/io/worker.rs

@@ -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,
@@ -43,6 +44,7 @@ pub struct Worker {
 	thread: Option<JoinHandle<()>>,
 	wait: Arc<Condvar>,
 	deleting: Arc<AtomicBool>,
+	panic_handler: SafeStringPanicHandler,
 }
 
 impl Worker {
@@ -53,14 +55,22 @@ impl Worker {
 						wait: Arc<Condvar>,
 						wait_mutex: Arc<Mutex<()>>) -> Worker
 						where Message: Send + Sync + Clone + 'static {
+		let panic_handler = StringPanicHandler::new_thread_safe();
 		let deleting = Arc::new(AtomicBool::new(false));
 		let mut worker = Worker {
+			panic_handler: panic_handler.clone(),
 			thread: None,
 			wait: wait.clone(),
 			deleting: deleting.clone(),
 		};
+		let panic_handler = panic_handler.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 || {
+				let mut panic = panic_handler.lock().unwrap();
+				panic.catch_panic(move || {
+					Worker::work_loop(stealer, channel.clone(), wait, wait_mutex.clone(), deleting)
+				}).unwrap()
+			})
 			.expect("Error creating worker thread"));
 		worker
 	}
@@ -105,6 +115,12 @@ impl Worker {
 	}
 }
 
+impl MayPanic<String> for Worker {
+	fn on_panic<F>(&self, closure: F) where F: OnPanicListener<String> {
+		self.panic_handler.on_panic(closure);
+	}
+}
+
 impl Drop for Worker {
 	fn drop(&mut self) {
 		self.deleting.store(true, AtomicOrdering::Relaxed);

util/src/lib.rs

@@ -21,7 +21,8 @@
 #![feature(plugin)]
 #![plugin(clippy)]
 #![allow(needless_range_loop, match_bool)]
-#![feature(std_panic, recover)]
+#![feature(catch_panic)]
+
 //! Ethcore-util library
 //!
 //! ### Rust version:

util/src/panics.rs

@@ -17,10 +17,9 @@
 //! Panic utilities
 
 use std::thread;
-use std::panic;
-use std::sync::Mutex;
-use std::any::Any;
 use std::ops::DerefMut;
+use std::any::Any;
+use std::sync::{Arc, Mutex};
 
 pub trait OnPanicListener<T>: Send + Sync + 'static {
 	fn call(&mut self, arg: &T);
@@ -33,26 +32,37 @@ impl<F, T> OnPanicListener<T> for F
 	}
 }
 
-pub trait ArgsConverter<T> {
+pub trait ArgsConverter<T> : Send + Sync {
 	fn convert(&self, t: &Box<Any + Send>) -> Option<T>;
 }
 
 pub trait MayPanic<T> {
-	fn on_panic<F>(&mut self, closure: F)
+	fn on_panic<F>(&self, closure: F)
 		where F: OnPanicListener<T>;
 }
 
 pub trait PanicHandler<T, C: ArgsConverter<T>> : MayPanic<T>{
-	fn new(converter: C) -> Self;
+	fn with_converter(converter: C) -> Self;
 	fn catch_panic<G, R>(&mut self, g: G) -> thread::Result<R>
-		where G: FnOnce() -> R + panic::RecoverSafe;
+		where G: FnOnce() -> R + Send + 'static;
+	fn notify_all(&mut self, &T);
 }
 
+pub type SafeStringPanicHandler = Arc<Mutex<StringPanicHandler>>;
+
+impl MayPanic<String> for SafeStringPanicHandler {
+	fn on_panic<F>(&self, closure: F) where F: OnPanicListener<String> {
+		self.lock().unwrap().on_panic(closure);
+	}
+}
 
 pub struct StringConverter;
 impl ArgsConverter<String> for StringConverter {
 	fn convert(&self, t: &Box<Any + Send>) -> Option<String> {
-		t.downcast_ref::<&'static str>().map(|t| t.clone().to_owned())
+		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)
 	}
 }
 
@@ -62,86 +72,152 @@ pub struct BasePanicHandler<T, C>
 	listeners: Mutex<Vec<Box<OnPanicListener<T>>>>
 }
 
-impl<T, C> BasePanicHandler<T, C>
-	where C: ArgsConverter<T>, T: 'static {
-	fn notify_all(&mut self, res: Option<T>) {
-		if let None = res {
-			return;
-		}
-		let r = res.unwrap();
-		let mut listeners = self.listeners.lock().unwrap();
-		for listener in listeners.deref_mut() {
-			listener.call(&r);
-		}
-	}
-}
-
 impl<T, C> PanicHandler<T, C> for BasePanicHandler<T, C>
 	where C: ArgsConverter<T>, T: 'static {
 
-	fn new(converter: C) -> Self {
+	fn with_converter(converter: C) -> Self {
 		BasePanicHandler {
 			converter: converter,
 			listeners: Mutex::new(vec![])
 		}
 	}
 
-	fn catch_panic<G, R>(&mut self, g: G) -> thread::Result<R>
+	#[allow(deprecated)]
-		where G: FnOnce() -> R + panic::RecoverSafe {
+	// TODO [todr] catch_panic is deprecated but panic::recover has different bounds (not allowing mutex)
-			let result = panic::recover(g);
+	fn catch_panic<G, R>(&mut self, g: G) -> thread::Result<R> where G: FnOnce() -> R + Send + 'static {
+		let result = thread::catch_panic(g);
 
-			println!("After calling function");
+		if let Err(ref e) = result {
-			if let Err(ref e) = result {
+			let res = self.converter.convert(e);
-				let res = self.converter.convert(e);
+			if let Some(r) = res {
-				println!("Got error. Notifying");
+				self.notify_all(&r);
-				self.notify_all(res);
 			}
-
-			result
 		}
+
+		result
+	}
+
+	fn notify_all(&mut self, r: &T) {
+		let mut listeners = self.listeners.lock().unwrap();
+		for listener in listeners.deref_mut() {
+			listener.call(r);
+		}
+	}
 }
 
 impl<T, C> MayPanic<T> for BasePanicHandler<T, C>
 	where C: ArgsConverter<T>, T: 'static {
-	fn on_panic<F>(&mut self, closure: F)
+	fn on_panic<F>(&self, closure: F)
 		where F: OnPanicListener<T> {
 		self.listeners.lock().unwrap().push(Box::new(closure));
 	}
 }
 
+pub struct StringPanicHandler {
+	handler: BasePanicHandler<String, StringConverter>
+}
+
+impl StringPanicHandler {
+	pub fn new_thread_safe() -> SafeStringPanicHandler {
+		Arc::new(Mutex::new(Self::new()))
+	}
+
+	pub fn new () -> Self {
+		Self::with_converter(StringConverter)
+	}
+}
+
+impl PanicHandler<String, StringConverter> for StringPanicHandler {
+
+	fn with_converter(converter: StringConverter) -> Self {
+		StringPanicHandler {
+			handler: BasePanicHandler::with_converter(converter)
+		}
+	}
+
+	fn catch_panic<G, R>(&mut self, g: G) -> thread::Result<R> where G: FnOnce() -> R + Send + 'static {
+		self.handler.catch_panic(g)
+	}
+
+	fn notify_all(&mut self, r: &String) {
+		self.handler.notify_all(r);
+	}
+}
+
+impl MayPanic<String> for StringPanicHandler {
+	fn on_panic<F>(&self, closure: F)
+		where F: OnPanicListener<String> {
+			self.handler.on_panic(closure)
+		}
+}
+
 #[test]
 fn should_notify_listeners_about_panic () {
-	use std::sync::{Arc, RwLock};
+	use std::sync::RwLock;
-
 	// given
 	let invocations = Arc::new(RwLock::new(vec![]));
 	let i = invocations.clone();
-	let mut p = BasePanicHandler::new(StringConverter);
+	let mut p = StringPanicHandler::new();
 	p.on_panic(move |t: &String| i.write().unwrap().push(t.clone()));
 
 	// when
-	p.catch_panic(|| panic!("Panic!"));
+	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 mut p = StringPanicHandler::new();
+	p.on_panic(move |t: &String| i.write().unwrap().push(t.clone()));
+
+	// 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::{Arc, RwLock};
+	use std::sync::RwLock;
 
 	// given
 	let invocations = Arc::new(RwLock::new(vec![]));
 	let i = invocations.clone();
-	let mut p = BasePanicHandler::new(StringConverter);
+	let mut p = StringPanicHandler::new();
 	p.on_panic(move |t: &String| i.write().unwrap().push(t.clone()));
 
 	// when
 	let t = thread::spawn(move ||
-		p.catch_panic(|| panic!("Panic!"))
+		p.catch_panic(|| panic!("Panic!")).unwrap()
 	);
-	t.join();
+	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 mut p = StringPanicHandler::new();
+	p.on_panic(move |t: &String| i.write().unwrap().push(t.clone()));
+
+	let mut p2 = StringPanicHandler::new();
+	p2.on_panic(move |t: &String| p.notify_all(t));
+
+	// when
+	p2.catch_panic(|| panic!("Panic!")).unwrap_err();
 
 	// then
 	assert!(invocations.read().unwrap()[0] == "Panic!");