implement fully async reaper

src/init/init.rs

@@ -83,7 +83,7 @@ async fn main() {
 	)
 	.expect("unable to create app pool");
 
-	Reaper::execute(&handles, Box::new(Nsm), core_pool, app_pool, None);
+	Reaper::execute(&handles, Box::new(Nsm), core_pool, app_pool, None).await;
 
 	reboot();
 }

src/integration/examples/boot_enclave.rs

@@ -437,7 +437,7 @@ async fn main() {
 	}
 
 	// Give the enclave time to start the pivot
-	std::thread::sleep(std::time::Duration::from_secs(2));
+	tokio::time::sleep(std::time::Duration::from_secs(2)).await;
 
 	let enclave_info_url =
 		format!("http://{LOCAL_HOST}:{}/qos/enclave-info", host_port);

src/integration/src/bin/pivot_socket_stress.rs

@@ -21,7 +21,7 @@ impl Processor {
 impl RequestProcessor for Processor {
 	async fn process(&self, request: &[u8]) -> Vec<u8> {
 		// Simulate just some baseline lag for all requests
-		std::thread::sleep(std::time::Duration::from_secs(1));
+		tokio::time::sleep(std::time::Duration::from_secs(1)).await;
 
 		let msg = PivotSocketStressMsg::try_from_slice(request)
 			.expect("Received invalid message - test is broken");

src/integration/tests/boot.rs

@@ -445,18 +445,18 @@ async fn standard_boot_e2e() {
 			.success());
 	}
 
-	// Give the enclave time to start the pivot
-	std::thread::sleep(std::time::Duration::from_secs(2));
-
-	// Check that the pivot executed
-	let contents = std::fs::read(PIVOT_OK2_SUCCESS_FILE).unwrap();
-	assert_eq!(std::str::from_utf8(&contents).unwrap(), msg);
-
 	let enclave_info_url =
 		format!("http://{LOCAL_HOST}:{}/qos/enclave-info", host_port);
 	let enclave_info: EnclaveInfo =
 		ureq::get(&enclave_info_url).call().unwrap().into_json().unwrap();
 	assert_eq!(enclave_info.phase, ProtocolPhase::QuorumKeyProvisioned);
 
+	// Give the enclave time to start the pivot
+	tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+
+	// Check that the pivot executed
+	let contents = std::fs::read(PIVOT_OK2_SUCCESS_FILE).unwrap();
+	assert_eq!(std::str::from_utf8(&contents).unwrap(), msg);
+
 	fs::remove_file(PIVOT_OK2_SUCCESS_FILE).unwrap();
 }

src/integration/tests/dev_boot.rs

@@ -77,7 +77,7 @@ async fn dev_boot_e2e() {
 		.unwrap();
 
 	// Give the coordinator time to pivot
-	std::thread::sleep(std::time::Duration::from_secs(2));
+	tokio::time::sleep(std::time::Duration::from_secs(2)).await;
 
 	// Make sure pivot ran
 	assert!(Path::new(PIVOT_OK3_SUCCESS_FILE).exists());

src/integration/tests/enclave_app_client_socket_stress.rs

@@ -1,3 +1,5 @@
+use std::time::Duration;
+
 use borsh::BorshDeserialize;
 use integration::{
 	wait_for_usock, PivotSocketStressMsg, PIVOT_SOCKET_STRESS_PATH,
@@ -14,7 +16,7 @@ use qos_core::{
 		},
 		ProtocolError, ProtocolPhase, ENCLAVE_APP_SOCKET_CLIENT_TIMEOUT_SECS,
 	},
-	reaper::{Reaper, REAPER_RESTART_DELAY_IN_SECONDS},
+	reaper::{Reaper, REAPER_RESTART_DELAY},
 };
 use qos_nsm::mock::MockNsm;
 use qos_p256::P256Pair;
@@ -76,12 +78,12 @@ async fn enclave_app_client_socket_stress() {
 	handles.put_quorum_key(&p256_pair).unwrap();
 
 	let enclave_pool =
-		StreamPool::new(SocketAddress::new_unix(ENCLAVE_SOCK), 1).unwrap();
+		StreamPool::single(SocketAddress::new_unix(ENCLAVE_SOCK)).unwrap();
 
 	let app_pool =
-		StreamPool::new(SocketAddress::new_unix(APP_SOCK), 1).unwrap();
+		StreamPool::single(SocketAddress::new_unix(APP_SOCK)).unwrap();
 
-	std::thread::spawn(move || {
+	tokio::spawn(async move {
 		Reaper::execute(
 			&handles,
 			Box::new(MockNsm),
@@ -91,13 +93,14 @@ async fn enclave_app_client_socket_stress() {
 			// works
 			Some(ProtocolPhase::QuorumKeyProvisioned),
 		)
+		.await;
 	});
 
 	// Make sure the pivot has some time to start up
 	wait_for_usock(APP_SOCK).await;
 
 	let enclave_client_pool =
-		StreamPool::new(SocketAddress::new_unix(ENCLAVE_SOCK), 1).unwrap();
+		StreamPool::single(SocketAddress::new_unix(ENCLAVE_SOCK)).unwrap();
 	let enclave_client = SocketClient::new(
 		enclave_client_pool.shared(),
 		TimeVal::seconds(ENCLAVE_APP_SOCKET_CLIENT_TIMEOUT_SECS + 3), // needs to be bigger than the slow request below + some time for recovery
@@ -118,10 +121,7 @@ async fn enclave_app_client_socket_stress() {
 		)
 	);
 
-	tokio::time::sleep(std::time::Duration::from_secs(
-		REAPER_RESTART_DELAY_IN_SECONDS + 1,
-	))
-	.await;
+	tokio::time::sleep(REAPER_RESTART_DELAY + Duration::from_secs(1)).await;
 	// The pivot panicked and should have been restarted.
 	let app_request =
 		borsh::to_vec(&PivotSocketStressMsg::OkRequest(1)).unwrap();

src/integration/tests/reaper.rs

@@ -5,13 +5,13 @@ use qos_core::{
 	handles::Handles,
 	io::{SocketAddress, StreamPool},
 	protocol::services::boot::ManifestEnvelope,
-	reaper::{Reaper, REAPER_EXIT_DELAY_IN_SECONDS},
+	reaper::{Reaper, REAPER_EXIT_DELAY},
 };
 use qos_nsm::mock::MockNsm;
 use qos_test_primitives::PathWrapper;
 
-#[test]
-fn reaper_works() {
+#[tokio::test]
+async fn reaper_works() {
 	let secret_path: PathWrapper = "/tmp/reaper_works.secret".into();
 	// let eph_path = "reaper_works.eph.key";
 	let usock: PathWrapper = "/tmp/reaper_works.sock".into();
@@ -43,18 +43,19 @@ fn reaper_works() {
 	let app_pool =
 		StreamPool::new(SocketAddress::new_unix("./never.sock"), 1).unwrap();
 
-	let reaper_handle = std::thread::spawn(move || {
+	let reaper_handle = tokio::spawn(async move {
 		Reaper::execute(
 			&handles,
 			Box::new(MockNsm),
 			enclave_pool,
 			app_pool,
 			None,
 		)
+		.await;
 	});
 
 	// Give the enclave server time to bind to the socket
-	std::thread::sleep(std::time::Duration::from_secs(1));
+	tokio::time::sleep(std::time::Duration::from_secs(1)).await;
 
 	// Check that the reaper is still running, presumably waiting for
 	// the secret.
@@ -65,14 +66,14 @@ fn reaper_works() {
 	fs::write(&*secret_path, b"super dank tank secret tech").unwrap();
 
 	// Make the sure the reaper executed successfully.
-	reaper_handle.join().unwrap();
+	reaper_handle.await.unwrap();
 	let contents = fs::read(integration::PIVOT_OK_SUCCESS_FILE).unwrap();
 	assert_eq!(std::str::from_utf8(&contents).unwrap(), msg);
 	assert!(fs::remove_file(integration::PIVOT_OK_SUCCESS_FILE).is_ok());
 }
 
-#[test]
-fn reaper_handles_non_zero_exits() {
+#[tokio::test]
+async fn reaper_handles_non_zero_exits() {
 	let secret_path: PathWrapper =
 		"/tmp/reaper_handles_non_zero_exits.secret".into();
 	let usock: PathWrapper = "/tmp/reaper_handles_non_zero_exits.sock".into();
@@ -100,18 +101,19 @@ fn reaper_handles_non_zero_exits() {
 	let app_pool =
 		StreamPool::new(SocketAddress::new_unix("./never.sock"), 1).unwrap();
 
-	let reaper_handle = std::thread::spawn(move || {
+	let reaper_handle = tokio::spawn(async move {
 		Reaper::execute(
 			&handles,
 			Box::new(MockNsm),
 			enclave_pool,
 			app_pool,
 			None,
 		)
+		.await;
 	});
 
 	// Give the enclave server time to bind to the socket
-	std::thread::sleep(std::time::Duration::from_secs(1));
+	tokio::time::sleep(std::time::Duration::from_secs(1)).await;
 
 	// Check that the reaper is still running, presumably waiting for
 	// the secret.
@@ -123,15 +125,13 @@ fn reaper_handles_non_zero_exits() {
 
 	// Ensure the reaper has enough time to detect the secret, launch the
 	// pivot, and let the pivot exit.
-	std::thread::sleep(std::time::Duration::from_secs(
-		REAPER_EXIT_DELAY_IN_SECONDS * 2,
-	));
+	tokio::time::sleep(REAPER_EXIT_DELAY * 2).await;
 
 	assert!(reaper_handle.is_finished());
 }
 
-#[test]
-fn reaper_handles_panic() {
+#[tokio::test]
+async fn reaper_handles_panic() {
 	let secret_path: PathWrapper = "/tmp/reaper_handles_panics.secret".into();
 	let usock: PathWrapper = "/tmp/reaper_handles_panics.sock".into();
 	let manifest_path: PathWrapper =
@@ -158,18 +158,19 @@ fn reaper_handles_panic() {
 	let app_pool =
 		StreamPool::new(SocketAddress::new_unix("./never.sock"), 1).unwrap();
 
-	let reaper_handle = std::thread::spawn(move || {
+	let reaper_handle = tokio::spawn(async move {
 		Reaper::execute(
 			&handles,
 			Box::new(MockNsm),
 			enclave_pool,
 			app_pool,
 			None,
 		)
+		.await;
 	});
 
 	// Give the enclave server time to bind to the socket
-	std::thread::sleep(std::time::Duration::from_secs(1));
+	tokio::time::sleep(std::time::Duration::from_secs(1)).await;
 
 	// Check that the reaper is still running, presumably waiting for
 	// the secret.
@@ -181,9 +182,7 @@ fn reaper_handles_panic() {
 
 	// Ensure the reaper has enough time to detect the secret, launch the
 	// pivot, and let the pivot exit.
-	std::thread::sleep(std::time::Duration::from_secs(
-		REAPER_EXIT_DELAY_IN_SECONDS * 2,
-	));
+	tokio::time::sleep(REAPER_EXIT_DELAY * 2).await;
 
 	assert!(reaper_handle.is_finished());
 }

src/qos_core/Cargo.toml

@@ -21,7 +21,7 @@ serde_bytes = { workspace = true }
 serde = { workspace = true }
 
 futures = { workspace = true }
-tokio = { workspace = true, features = ["io-util", "macros", "net", "rt-multi-thread", "time", "signal"], default-features = false }
+tokio = { workspace = true, features = ["io-util", "macros", "net", "rt-multi-thread", "time", "signal", "process"], default-features = false }
 tokio-vsock = { workspace = true }
 
 [dev-dependencies]

src/qos_core/src/cli.rs

@@ -151,8 +151,8 @@ impl CLI {
 		} else if opts.parsed.help() {
 			println!("{}", opts.parsed.info());
 		} else {
-			// start reaper in a thread so we can terminate on ctrl+c properly
-			std::thread::spawn(move || {
+			// start reaper in a task so we can terminate on ctrl+c properly
+			tokio::spawn(async move {
 				Reaper::execute(
 					&Handles::new(
 						opts.ephemeral_file(),
@@ -165,7 +165,8 @@ impl CLI {
 						.expect("Unable to create enclave socket pool"),
 					opts.app_pool().expect("Unable to create enclave app pool"),
 					None,
-				);
+				)
+				.await;
 			});
 
 			eprintln!("qos_core: Reaper running, press ctrl+c to quit");