use thread with Reaper::async_execute

fixes a loop problem with async
also adds integration test to test out qos_host and qos_core (aka Reaper)

Author: Turnalek

src/init/init.rs

@@ -116,8 +116,7 @@ async fn main() {
 		TimeVal::seconds(5),
 	);
 
-	Reaper::async_execute(&handles, Box::new(Nsm), core_pool, app_pool, None)
-		.await;
+	Reaper::async_execute(&handles, Box::new(Nsm), core_pool, app_pool, None);
 
 	reboot();
 }

src/integration/tests/async_client.rs

@@ -0,0 +1,84 @@
+use qos_core::{
+	async_client::AsyncClient,
+	async_server::{AsyncRequestProcessor, AsyncSocketServer},
+	io::{AsyncStreamPool, SocketAddress, TimeVal, TimeValLike},
+	server::SocketServerError,
+};
+use tokio::task::JoinHandle;
+
+#[derive(Clone)]
+struct EchoProcessor;
+
+impl AsyncRequestProcessor for EchoProcessor {
+	async fn process(&self, request: Vec<u8>) -> Vec<u8> {
+		request
+	}
+}
+
+async fn run_echo_server(
+	socket_path: &str,
+) -> Result<Vec<JoinHandle<Result<(), SocketServerError>>>, SocketServerError> {
+	let timeout = TimeVal::milliseconds(50);
+	let pool = AsyncStreamPool::new(
+		std::iter::once(SocketAddress::new_unix(socket_path)),
+		timeout,
+	);
+	let tasks = AsyncSocketServer::listen_all(pool, &EchoProcessor)?;
+
+	Ok(tasks)
+}
+
+#[tokio::test]
+async fn direct_connect_works() {
+	let socket_path = "/tmp/async_client_test_direct_connect_works.sock";
+	let sockets = std::iter::once(SocketAddress::new_unix(socket_path));
+	let timeout = TimeVal::milliseconds(50);
+	let pool = AsyncStreamPool::new(sockets, timeout).shared();
+
+	let client = AsyncClient::new(pool);
+
+	let server_tasks = run_echo_server(socket_path).await.unwrap();
+
+	let r = client.call(&[0]).await;
+	assert!(r.is_ok());
+
+	for task in server_tasks {
+		task.abort();
+	}
+}
+
+#[tokio::test]
+async fn times_out_properly() {
+	let socket_path = "/tmp/async_client_test_times_out_properly.sock";
+	let sockets = std::iter::once(SocketAddress::new_unix(socket_path));
+	let timeout = TimeVal::milliseconds(50);
+	let pool = AsyncStreamPool::new(sockets, timeout).shared();
+	let client = AsyncClient::new(pool);
+
+	let r = client.call(&[0]).await;
+	assert!(r.is_err());
+}
+
+#[tokio::test]
+async fn repeat_connect_works() {
+	let socket_path = "/tmp/async_client_test_repeat_connect_works.sock";
+	let sockets = std::iter::once(SocketAddress::new_unix(socket_path));
+	let timeout = TimeVal::milliseconds(50);
+	let pool = AsyncStreamPool::new(sockets, timeout).shared();
+	let client = AsyncClient::new(pool);
+
+	// server not running yet, expect a connection error
+	let r = client.call(&[0]).await;
+	assert!(r.is_err());
+
+	// start server
+	let server_tasks = run_echo_server(socket_path).await.unwrap();
+
+	// server running, expect success
+	let r = client.call(&[0]).await;
+	assert!(r.is_ok());
+
+	for task in server_tasks {
+		task.abort();
+	}
+}

src/integration/tests/async_qos_host.rs

@@ -0,0 +1,66 @@
+use std::{process::Command, time::Duration};
+
+use integration::PIVOT_OK_PATH;
+use qos_test_primitives::{ChildWrapper, PathWrapper};
+
+const TEST_ENCLAVE_SOCKET: &str = "/tmp/async_qos_host_test.enclave.sock";
+const POOL_SIZE: &str = "1";
+
+#[tokio::test]
+async fn connects_and_gets_info() {
+	let _qos_host: ChildWrapper =
+		Command::new("../target/debug/async_qos_host")
+			.arg("--usock")
+			.arg(TEST_ENCLAVE_SOCKET)
+			.arg("--pool-size")
+			.arg(POOL_SIZE)
+			.arg("--host-ip")
+			.arg("127.0.0.1")
+			.arg("--host-port")
+			.arg("3323")
+			.arg("--socket-timeout")
+			.arg("50") // ms
+			.spawn()
+			.unwrap()
+			.into();
+
+	tokio::time::sleep(Duration::from_millis(100)).await; // let the qos_host start
+
+	let r = ureq::get("http://127.0.0.1:3323/qos/enclave-info").call();
+	assert!(r.is_err()); // expect 500 here
+
+	let enclave_socket = format!("{TEST_ENCLAVE_SOCKET}_0"); // manually pick the 1st one
+	let secret_path: PathWrapper = "./async_qos_host_test.secret".into();
+	// let eph_path = "reaper_works.eph.key";
+	let manifest_path: PathWrapper = "async_qos_host_test..manifest".into();
+
+	// For our sanity, ensure the secret does not yet exist
+	drop(std::fs::remove_file(&*secret_path));
+	// Remove the socket file if it exists as well, in case of bad crashes
+	drop(std::fs::remove_file(&enclave_socket));
+
+	let mut _enclave_child_process: ChildWrapper =
+		Command::new("../target/debug/async_qos_core")
+			.args([
+				"--usock",
+				TEST_ENCLAVE_SOCKET,
+				"--quorum-file",
+				&*secret_path,
+				"--pivot-file",
+				PIVOT_OK_PATH,
+				"--ephemeral-file",
+				"eph_path",
+				"--mock",
+				"--manifest-file",
+				&*manifest_path,
+			])
+			.spawn()
+			.unwrap()
+			.into();
+
+	// Give the enclave server time to bind to the socket
+	tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+
+	let r = ureq::get("http://127.0.0.1:3323/qos/enclave-info").call();
+	assert!(r.is_ok()); // expect 200 here
+}

src/integration/tests/qos_host.rs

@@ -0,0 +1,59 @@
+use std::{process::Command, time::Duration};
+
+use integration::PIVOT_OK_PATH;
+use qos_test_primitives::{ChildWrapper, PathWrapper};
+
+const TEST_ENCLAVE_SOCKET: &str = "/tmp/qos_host_test.enclave.sock";
+
+#[test]
+fn connects_and_gets_info() {
+	let _qos_host: ChildWrapper = Command::new("../target/debug/qos_host")
+		.arg("--usock")
+		.arg(TEST_ENCLAVE_SOCKET)
+		.arg("--host-ip")
+		.arg("127.0.0.1")
+		.arg("--host-port")
+		.arg("3323")
+		.arg("--socket-timeout")
+		.arg("50") // ms
+		.spawn()
+		.unwrap()
+		.into();
+
+	std::thread::sleep(Duration::from_millis(100)); // let the qos_host start
+
+	let r = ureq::get("http://127.0.0.1:3323/qos/enclave-info").call();
+	assert!(r.is_err()); // expect 500 here
+
+	let secret_path: PathWrapper = "./reaper_works.secret".into();
+	// let eph_path = "reaper_works.eph.key";
+	let manifest_path: PathWrapper = "reaper_works.manifest".into();
+
+	// For our sanity, ensure the secret does not yet exist
+	drop(std::fs::remove_file(&*secret_path));
+
+	let mut _enclave_child_process: ChildWrapper =
+		Command::new("../target/debug/qos_core")
+			.args([
+				"--usock",
+				TEST_ENCLAVE_SOCKET,
+				"--quorum-file",
+				&*secret_path,
+				"--pivot-file",
+				PIVOT_OK_PATH,
+				"--ephemeral-file",
+				"eph_path",
+				"--mock",
+				"--manifest-file",
+				&*manifest_path,
+			])
+			.spawn()
+			.unwrap()
+			.into();
+
+	// Give the enclave server time to bind to the socket
+	std::thread::sleep(std::time::Duration::from_millis(200));
+
+	let r = ureq::get("http://127.0.0.1:3323/qos/enclave-info").call();
+	assert!(r.is_ok()); // expect 200 here
+}

src/qos_core/src/bin/async_qos_core.rs

@@ -1,6 +1,5 @@
 use qos_core::cli::CLI;
 
-#[tokio::main]
-async fn main() {
-	CLI::async_execute().await;
+fn main() {
+	CLI::async_execute();
 }

src/qos_core/src/cli.rs

@@ -203,7 +203,7 @@ impl CLI {
 
 	/// Execute the enclave server CLI with the environment args using tokio/async
 	#[cfg(feature = "async")]
-	pub async fn async_execute() {
+	pub fn async_execute() {
 		let mut args: Vec<String> = env::args().collect();
 		let opts = EnclaveOpts::new(&mut args);
 
@@ -223,8 +223,7 @@ impl CLI {
 				opts.async_pool(false),
 				opts.async_pool(true),
 				None,
-			)
-			.await;
+			);
 		}
 	}
 }

src/qos_core/src/io/async_stream.rs

@@ -136,7 +136,6 @@ impl AsyncStream {
 		if self.inner.is_none() {
 			self.connect().await?;
 		}
-
 		self.send(req_buf).await?;
 		self.recv().await
 	}
@@ -256,7 +255,7 @@ pub struct AsyncListener {
 impl AsyncListener {
 	/// Bind and listen on the given address.
 	pub(crate) fn listen(addr: &SocketAddress) -> Result<Self, IOError> {
-		let listener = match addr {
+		let listener = match *addr {
 			SocketAddress::Unix(uaddr) => {
 				let path =
 					uaddr.path().ok_or(IOError::ConnectAddressInvalid)?;
@@ -265,8 +264,6 @@ impl AsyncListener {
 			}
 			#[cfg(feature = "vm")]
 			SocketAddress::Vsock(vaddr) => {
-				let vaddr =
-					tokio_vsock::VsockAddr::new(vaddr.cid(), vaddr.port());
 				let inner = InnerListener::Vsock(VsockListener::bind(vaddr)?);
 				Self { inner }
 			}
@@ -326,17 +323,26 @@ async fn retry_unix_connect(
 		let socket = UnixSocket::new_stream()?;
 
 		eprintln!("Attempting USOCK connect to: {:?}", addr.path());
-		match tokio::time::timeout(timeout, socket.connect(path)).await? {
-			Ok(stream) => {
-				eprintln!("Connected to USOCK at: {:?}", addr.path());
-				return Ok(stream);
-			}
-			Err(err) => {
-				eprintln!("Error connecting to USOCK: {err}");
-				if SystemTime::now() > eot {
-					return Err(err);
+		let tr = tokio::time::timeout(timeout, socket.connect(path)).await;
+		match tr {
+			Ok(r) => match r {
+				Ok(stream) => {
+					eprintln!("Connected to USOCK at: {:?}", addr.path());
+					return Ok(stream);
 				}
-				tokio::time::sleep(sleep_time).await;
+				Err(err) => {
+					eprintln!("Error connecting to USOCK: {err}");
+					if SystemTime::now() > eot {
+						return Err(err);
+					}
+					tokio::time::sleep(sleep_time).await;
+				}
+			},
+			Err(err) => {
+				eprintln!(
+					"Connecting to USOCK failed with timeout error: {err}"
+				);
+				return Err(err.into());
 			}
 		}
 	}
@@ -354,18 +360,27 @@ async fn retry_vsock_connect(
 
 	loop {
 		eprintln!("Attempting VSOCK connect to: {:?}", addr);
-		match tokio::time::timeout(timeout, VsockStream::connect(*addr)).await?
-		{
-			Ok(stream) => {
-				eprintln!("Connected to VSOCK at: {:?}", addr);
-				return Ok(stream);
-			}
-			Err(err) => {
-				eprintln!("Error connecting to VSOCK: {}", err);
-				if SystemTime::now() > eot {
-					return Err(err);
+		let tr =
+			tokio::time::timeout(timeout, VsockStream::connect(*addr)).await;
+		match tr {
+			Ok(r) => match r {
+				Ok(stream) => {
+					eprintln!("Connected to VSOCK at: {:?}", addr);
+					return Ok(stream);
+				}
+				Err(err) => {
+					eprintln!("Error connecting to VSOCK: {}", err);
+					if SystemTime::now() > eot {
+						return Err(err);
+					}
+					tokio::time::sleep(sleep_time).await;
 				}
-				tokio::time::sleep(sleep_time).await;
+			},
+			Err(err) => {
+				eprintln!(
+					"Connecting to VSOCK failed with timeout error: {err}"
+				);
+				return Err(err.into());
 			}
 		}
 	}

src/qos_core/src/lib.rs

@@ -65,7 +65,8 @@ pub const SEC_APP_SOCK: &str = "./local-enclave/sec_app.sock";
 /// Default socket for enclave <-> secure app communication.
 #[cfg(feature = "vm")]
 pub const SEC_APP_SOCK: &str = "/sec_app.sock";
-
+/// Default socket connect timeout in milliseconds
+pub const DEFAULT_SOCKET_TIMEOUT: &str = "5000";
 /// Default socket pool size is 20
 #[cfg(feature = "async")]
 pub const DEFAULT_POOL_SIZE: &str = "1"; // DEBUG: ales - set to something real after debugging