Merge pull request #31 from alecmocatta/thread-scoped-unwind-catching

Replace global panic hook with abort_on_unwind on constellation threads

Author: mergify[bot]

constellation-internal/Cargo.toml

@@ -26,6 +26,7 @@ cargo_metadata = { version = "0.8", default-features = false }
 either = { version = "1.5", features = ["serde"] }
 palaver = "0.2"
 rand = { version = "0.7", features = ["small_rng"] }
+replace_with = "0.1.4"
 serde = { version = "1.0", features = ["derive"] }
 serde_bytes = "0.11"
 serde_json = "1.0"

constellation-internal/src/lib.rs

@@ -26,13 +26,13 @@ mod format;
 pub mod msg;
 
 #[cfg(unix)]
-use nix::{fcntl, sys::signal, unistd};
+use nix::{fcntl, libc, sys::signal, unistd};
 use palaver::file::{copy, memfd_create};
 use serde::{Deserialize, Serialize};
 use std::{
-	convert::TryInto, env, ffi::{CString, OsString}, fmt::{self, Debug, Display}, fs::File, io::{self, Read, Seek}, net, ops, os::unix::{
-		ffi::OsStringExt, io::{AsRawFd, FromRawFd}
-	}, sync::{Arc, Mutex}
+	convert::TryInto, env, ffi::{CString, OsString}, fmt::{self, Debug, Display}, fs::File, io::{self, Read, Seek, Write}, net, ops, os::unix::{
+		ffi::OsStringExt, io::{AsRawFd, FromRawFd, IntoRawFd}
+	}, process::abort, sync::{Arc, Mutex}
 };
 
 #[cfg(target_family = "unix")]
@@ -552,12 +552,12 @@ pub enum ProcessInputEvent {
 
 #[allow(missing_debug_implementations)]
 #[derive(Clone)]
-pub struct Trace<W: io::Write> {
+pub struct Trace<W: Write> {
 	stdout: Arc<Mutex<W>>,
 	format: Format,
 	verbose: bool,
 }
-impl<W: io::Write> Trace<W> {
+impl<W: Write> Trace<W> {
 	pub fn new(stdout: W, format: Format, verbose: bool) -> Self {
 		Self {
 			stdout: Arc::new(Mutex::new(stdout)),
@@ -638,6 +638,41 @@ where
 
 //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
+struct StdErr;
+impl Write for StdErr {
+	fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+		let mut file = unsafe { File::from_raw_fd(libc::STDERR_FILENO) };
+		let ret = file.write(buf);
+		let _ = file.into_raw_fd();
+		ret
+	}
+	fn flush(&mut self) -> io::Result<()> {
+		Ok(())
+	}
+}
+
+#[inline]
+fn abort_on_unwind_<F: FnOnce() -> T, T>(f: F) -> T {
+	replace_with::on_unwind(f, || {
+		let _ = StdErr.write_all(b"Constellation: detected unexpected panic; aborting\n");
+		abort();
+	})
+}
+
+#[must_use]
+#[inline]
+pub fn abort_on_unwind<F: FnOnce() -> T, T>(f: F) -> impl FnOnce() -> T {
+	|| abort_on_unwind_(f)
+}
+
+#[must_use]
+#[inline]
+pub fn abort_on_unwind_1<F: FnOnce(&A) -> T, T, A>(f: F) -> impl FnOnce(&A) -> T {
+	|a| abort_on_unwind_(|| f(a))
+}
+
+//////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
 pub mod cargo_metadata {
 	use cargo_metadata::Target;
 	use serde::Deserialize;

src/bin/constellation/bridge.rs

@@ -36,7 +36,7 @@ use std::{
 
 use constellation::FutureExt1;
 use constellation_internal::{
-	forbid_alloc, map_bincode_err, msg::{bincode_deserialize_from, bincode_serialize_into, BridgeRequest, FabricRequest}, BufferedStream, DeployInputEvent, DeployOutputEvent, ExitStatus, Fd, Pid, ProcessInputEvent, ProcessOutputEvent, Resources
+	abort_on_unwind, abort_on_unwind_1, forbid_alloc, map_bincode_err, msg::{bincode_deserialize_from, bincode_serialize_into, BridgeRequest, FabricRequest}, BufferedStream, DeployInputEvent, DeployOutputEvent, ExitStatus, Fd, Pid, ProcessInputEvent, ProcessOutputEvent, Resources
 };
 
 const SCHEDULER_FD: Fd = 4;
@@ -87,7 +87,9 @@ fn monitor_process(
 					let sender_ = sender_.clone();
 					let _ = thread::Builder::new()
 						.name(String::from("d"))
-						.spawn(move || monitor_process(new_pid, sender_, receiver1))
+						.spawn(abort_on_unwind(move || {
+							monitor_process(new_pid, sender_, receiver1)
+						}))
 						.unwrap();
 				}
 				ProcessOutputEvent::Output(fd, output) => {
@@ -207,10 +209,10 @@ fn recce(
 	nix::unistd::close(writer).unwrap();
 	// let _ = thread::Builder::new()
 	// 	.name(String::from(""))
-	// 	.spawn(move || {
+	// 	.spawn(abort_on_unwind(move || {
 	// 		thread::sleep(RECCE_TIMEOUT);
 	// 		let _ = nix::sys::signal::kill(child, nix::sys::signal::Signal::SIGKILL);
-	// 	})
+	// 	}))
 	// 	.unwrap();
 	// TODO: do this without waitpid/kill race
 	loop {
@@ -307,12 +309,14 @@ fn manage_connection(
 		let (sender1, receiver1) = futures::channel::mpsc::channel(0);
 		let _ = thread::Builder::new()
 			.name(String::from("c"))
-			.spawn(move || monitor_process(pid, sender, receiver1))
+			.spawn(abort_on_unwind(move || {
+				monitor_process(pid, sender, receiver1)
+			}))
 			.unwrap();
 		let hashmap = &Mutex::new(HashMap::new());
 		let _ = hashmap.lock().unwrap().insert(pid, sender1);
 		crossbeam::scope(|scope| {
-			let _ = scope.spawn(move |_scope| {
+			let _ = scope.spawn(abort_on_unwind_1(move |_scope| {
 				loop {
 					let event: Result<DeployInputEvent, _> =
 						bincode::deserialize_from(&mut stream_read).map_err(map_bincode_err);
@@ -340,7 +344,7 @@ fn manage_connection(
 				for (_, process) in x.iter_mut() {
 					let _unchecked_error = process.send(InputEventInt::Kill).block();
 				}
-			});
+			}));
 			for event in receiver.iter() {
 				let event = match event {
 					OutputEventInt::Spawn(pid, new_pid, sender) => {
@@ -387,16 +391,18 @@ pub fn main() {
 	let (sender, receiver) = mpsc::sync_channel(0);
 	let _ = thread::Builder::new()
 		.name(String::from("a"))
-		.spawn(move || {
+		.spawn(abort_on_unwind(move || {
 			for stream in listener.incoming() {
 				trace!("BRIDGE: accepted");
 				let sender = sender.clone();
 				let _ = thread::Builder::new()
 					.name(String::from("b"))
-					.spawn(|| manage_connection(stream.unwrap(), sender))
+					.spawn(abort_on_unwind(|| {
+						manage_connection(stream.unwrap(), sender)
+					}))
 					.unwrap();
 			}
-		})
+		}))
 		.unwrap();
 
 	for (request, sender) in receiver {

src/bin/constellation/main.rs

@@ -90,7 +90,7 @@ use std::{
 };
 
 use constellation_internal::{
-	forbid_alloc, map_bincode_err, msg::{bincode_deserialize_from, FabricRequest}, BufferedStream, FabricOutputEvent, Fd, Format, Pid, PidInternal, Trace
+	abort_on_unwind, abort_on_unwind_1, forbid_alloc, map_bincode_err, msg::{bincode_deserialize_from, FabricRequest}, BufferedStream, FabricOutputEvent, Fd, Format, Pid, PidInternal, Trace
 };
 
 #[derive(PartialEq, Debug)]
@@ -151,7 +151,7 @@ fn main() {
 				.set_port(fabric.local_addr().unwrap().port());
 			let _ = thread::Builder::new()
 				.name(String::from("master"))
-				.spawn(move || {
+				.spawn(abort_on_unwind(move || {
 					master::run(
 						SocketAddr::new(listen.ip(), master_addr.port()),
 						Pid::new(master_addr.ip(), master_addr.port()),
@@ -167,7 +167,7 @@ fn main() {
 							)
 							.collect::<HashMap<_, _>>(),
 					); // TODO: error on clash
-				})
+				}))
 				.unwrap();
 			(listen.ip(), fabric)
 		}
@@ -352,7 +352,7 @@ fn main() {
 				{
 					break;
 				}
-				let _ = scope.spawn(move |_scope| {
+				let _ = scope.spawn(abort_on_unwind_1(move |_scope| {
 					loop {
 						match wait::waitpid(child, None) {
 							Err(nix::Error::Sys(nix::errno::Errno::EINTR)) => (),
@@ -380,7 +380,7 @@ fn main() {
 						&Either::Right::<Pid, Pid>(process_id),
 					)
 					.map_err(map_bincode_err);
-				});
+				}));
 			}
 			for (&_job, &pid) in pending.read().unwrap().iter() {
 				// TODO: this is racey

src/bin/constellation/master.rs

@@ -5,7 +5,7 @@ use std::{
 };
 
 use constellation_internal::{
-	map_bincode_err, msg::{bincode_deserialize_from, FabricRequest}, BufferedStream, Pid, Resources
+	abort_on_unwind, abort_on_unwind_1, map_bincode_err, msg::{bincode_deserialize_from, FabricRequest}, BufferedStream, Pid, Resources
 };
 
 #[derive(Debug)]
@@ -60,19 +60,19 @@ pub fn run(
 			let stream = TcpStream::connect(&fabric).unwrap();
 			let sender1 = sender.clone();
 			let _ = thread::Builder::new()
-				.spawn(move || {
+				.spawn(abort_on_unwind(move || {
 					let (receiver, sender) = (receiver_a, sender1);
 					let (mut stream_read, mut stream_write) =
 						(BufferedStream::new(&stream), BufferedStream::new(&stream));
 					bincode::serialize_into::<_, IpAddr>(&mut stream_write, &fabric.ip()).unwrap();
 					let _ip = bincode::deserialize_from::<_, IpAddr>(&mut stream_read).unwrap();
 					crossbeam::scope(|scope| {
-						let _ = scope.spawn(|_spawn| {
+						let _ = scope.spawn(abort_on_unwind_1(|_spawn| {
 							for request in receiver {
 								bincode::serialize_into(&mut stream_write.write(), &request)
 									.unwrap();
 							}
-						});
+						}));
 						while let Ok(done) =
 							bincode::deserialize_from::<_, Either<Pid, Pid>>(&mut stream_read)
 								.map_err(map_bincode_err)
@@ -81,11 +81,11 @@ pub fn run(
 						}
 					})
 					.unwrap();
-				})
+				}))
 				.unwrap();
 			let sender = sender.clone();
 			let _ = thread::Builder::new()
-				.spawn(move || {
+				.spawn(abort_on_unwind(move || {
 					#[cfg(feature = "distribute_binaries")]
 					let binary = {
 						let mut binary = Vec::new();
@@ -115,21 +115,21 @@ pub fn run(
 						.unwrap();
 					let _pid: Pid = receiver.recv().unwrap().unwrap();
 					// println!("bridge at {:?}", pid);
-				})
+				}))
 				.unwrap();
 			(sender_a, node, fabric.ip(), VecDeque::new())
 		})
 		.collect::<Vec<_>>();
 
 	let listener = TcpListener::bind(bind_addr).unwrap();
 	let _ = thread::Builder::new()
-		.spawn(move || {
+		.spawn(abort_on_unwind(move || {
 			for stream in listener.incoming() {
 				// println!("accepted");
 				let stream = stream.unwrap();
 				let sender = sender.clone();
 				let _ = thread::Builder::new()
-					.spawn(move || {
+					.spawn(abort_on_unwind(move || {
 						let (mut stream_read, mut stream_write) =
 							(BufferedStream::new(&stream), &stream);
 						while let Ok(request) =
@@ -144,10 +144,10 @@ pub fn run(
 								break;
 							}
 						}
-					})
+					}))
 					.unwrap();
 			}
-		})
+		}))
 		.unwrap();
 
 	let mut processes: HashMap<(usize, Pid), Resources> = HashMap::new();

src/bin/deploy.rs

@@ -35,7 +35,7 @@ use std::{
 };
 
 use constellation_internal::{
-	map_bincode_err, msg::{bincode_serialize_into, BridgeRequest}, BufferedStream, DeployInputEvent, DeployOutputEvent, Envs, ExitStatus, Format, Formatter, Pid, StyleSupport
+	abort_on_unwind_1, map_bincode_err, msg::{bincode_serialize_into, BridgeRequest}, BufferedStream, DeployInputEvent, DeployOutputEvent, Envs, ExitStatus, Format, Formatter, Pid, StyleSupport
 };
 
 const USAGE: &str = "Run a binary on a constellation cluster.
@@ -112,7 +112,7 @@ fn main() {
 		panic!("Deploy failed due to not being able to allocate process to any of the nodes or constellation::init() not being called immediately inside main()")
 	}); // TODO get resources from bridge
 	crossbeam::scope(|scope| {
-		let _ = scope.spawn(|_scope| {
+		let _ = scope.spawn(abort_on_unwind_1(|_scope| {
 			let mut stdin = io::stdin();
 			loop {
 				let mut buf = MaybeUninit::<[u8; 1024]>::uninit();
@@ -130,7 +130,7 @@ fn main() {
 					break;
 				}
 			}
-		});
+		}));
 		let mut exit_code = ExitStatus::Success;
 		let mut ref_count = 1;
 		let mut pids = HashSet::new();

src/channel/mod.rs

@@ -11,6 +11,8 @@ use std::{
 };
 use tcp_typed::{Connection, Listener};
 
+use constellation_internal::abort_on_unwind;
+
 use super::Fd;
 
 pub use self::{inner::*, inner_states::*};
@@ -114,7 +116,7 @@ impl Reactor {
 		let mut triggeree = Some(triggeree);
 		let tcp_thread = thread::Builder::new()
 			.name(String::from("tcp-thread"))
-			.spawn(move || {
+			.spawn(abort_on_unwind(move || {
 				let context = context();
 				let context = context.borrow();
 				let mut listener = context.listener.try_write().unwrap();
@@ -418,7 +420,7 @@ impl Reactor {
 					});
 				}
 				// trace!("/close"); // called after rust runtime exited, not sure what trace does
-			})
+			}))
 			.unwrap();
 		Handle {
 			triggerer: Some(triggerer),