implement dynamic resizing async pool and server

Author: Turnalek

src/init/init.rs

@@ -98,21 +98,20 @@ async fn main() {
 		PIVOT_FILE.to_string(),
 	);
 
-	let start_port = 3;
-	let default_pool_size = 21; // 1 for qos-host, 20 for <app>-host, more are added as needed TODO: make the add logic dynamic
+	let start_port = 3; // used for qos-host only! others follow 4+ for the <app>-host
 	let core_pool = AsyncStreamPool::new(
-		(start_port..start_port + default_pool_size)
-			.into_iter()
-			.map(|p| SocketAddress::new_vsock(cid, p, VMADDR_NO_FLAGS)),
+		SocketAddress::new_vsock(cid, start_port, VMADDR_NO_FLAGS),
 		TimeVal::seconds(0),
-	);
+		1, // start at pool size 1, grow based on  manifest/args as necessary (see Reaper)
+	)
+	.expect("unable to create core pool");
 
 	let app_pool = AsyncStreamPool::new(
-		(0..default_pool_size)
-			.into_iter()
-			.map(|p| SocketAddress::new_unix(&format!("{SEC_APP_SOCK}_{p}"))),
+		SocketAddress::new_unix(SEC_APP_SOCK),
 		TimeVal::seconds(5),
-	);
+		1, // start at pool size 1, grow based on  manifest/args as necessary (see Reaper)
+	)
+	.expect("unable to create app pool");
 
 	Reaper::async_execute(&handles, Box::new(Nsm), core_pool, app_pool, None);
 

src/integration/src/bin/pivot_async_remote_tls.rs

@@ -110,26 +110,28 @@ async fn main() {
 	let proxy_path: &String = &args[2];
 
 	let pool = AsyncStreamPool::new(
-		std::iter::once(SocketAddress::new_unix(socket_path)),
+		SocketAddress::new_unix(socket_path),
 		TimeVal::seconds(0), // listener, no timeout
-	);
+		1,
+	)
+	.expect("unable to create async stream pool");
 
 	let proxy_pool = AsyncStreamPool::new(
-		std::iter::once(SocketAddress::new_unix(proxy_path)),
+		SocketAddress::new_unix(proxy_path),
 		TimeVal::seconds(ENCLAVE_APP_SOCKET_CLIENT_TIMEOUT_SECS),
+		1,
 	)
+	.expect("unable to create async stream pool")
 	.shared();
 
-	let tasks =
+	let server =
 		AsyncSocketServer::listen_all(pool, &Processor::new(proxy_pool))
 			.unwrap();
 
 	match tokio::signal::ctrl_c().await {
 		Ok(_) => {
 			eprintln!("pivot handling ctrl+c the tokio way");
-			for task in tasks {
-				task.abort();
-			}
+			server.terminate();
 		}
 		Err(err) => panic!("{err}"),
 	}

src/integration/src/lib.rs

@@ -57,6 +57,7 @@ pub const QOS_DIST_DIR: &str = "./mock/dist";
 pub const PCR3_PRE_IMAGE_PATH: &str = "./mock/namespaces/pcr3-preimage.txt";
 
 const MSG: &str = "msg";
+const POOL_SIZE: &str = "pool-size";
 
 /// Request/Response messages for "socket stress" pivot app.
 #[derive(BorshDeserialize, BorshSerialize, Debug, PartialEq, Eq)]
@@ -135,9 +136,17 @@ pub struct AdditionProofPayload {
 struct PivotParser;
 impl GetParserForOptions for PivotParser {
 	fn parser() -> Parser {
-		Parser::new().token(
-			Token::new(MSG, "A msg to write").takes_value(true).required(true),
-		)
+		Parser::new()
+			.token(
+				Token::new(MSG, "A msg to write")
+					.takes_value(true)
+					.required(true),
+			)
+			.token(
+				Token::new(POOL_SIZE, "App pool size")
+					.takes_value(true)
+					.required(false),
+			)
 	}
 }
 

src/integration/tests/async_boot.rs

@@ -30,20 +30,23 @@ async fn async_standard_boot_e2e() {
 	const PIVOT_HASH_PATH: &str = "/tmp/async_standard_boot_e2e-pivot-hash.txt";
 
 	let host_port = qos_test_primitives::find_free_port().unwrap();
-	let tmp: PathWrapper = "/tmp/boot-e2e".into();
+	let tmp: PathWrapper = "/tmp/async-boot-e2e".into();
 	let _: PathWrapper = PIVOT_OK5_SUCCESS_FILE.into();
 	let _: PathWrapper = PIVOT_HASH_PATH.into();
 	fs::create_dir_all(&*tmp).unwrap();
 
-	let usock: PathWrapper = "/tmp/boot-e2e/boot_e2e.sock".into();
-	let secret_path: PathWrapper = "/tmp/boot-e2e/boot_e2e.secret".into();
-	let pivot_path: PathWrapper = "/tmp/boot-e2e/boot_e2e.pivot".into();
-	let manifest_path: PathWrapper = "/tmp/boot-e2e/boot_e2e.manifest".into();
-	let eph_path: PathWrapper = "/tmp/boot-e2e/ephemeral_key.secret".into();
+	let usock: PathWrapper = "/tmp/async-boot-e2e/boot_e2e.sock".into();
+	let secret_path: PathWrapper = "/tmp/async-boot-e2e/boot_e2e.secret".into();
+	let pivot_path: PathWrapper = "/tmp/async-boot-e2e/boot_e2e.pivot".into();
+	let manifest_path: PathWrapper =
+		"/tmp/async-boot-e2e/boot_e2e.manifest".into();
+	let eph_path: PathWrapper =
+		"/tmp/async-boot-e2e/ephemeral_key.secret".into();
 
-	let boot_dir: PathWrapper = "/tmp/boot-e2e/boot-dir".into();
+	let boot_dir: PathWrapper = "/tmp/async-boot-e2e/boot-dir".into();
 	fs::create_dir_all(&*boot_dir).unwrap();
-	let attestation_dir: PathWrapper = "/tmp/boot-e2e/attestation-dir".into();
+	let attestation_dir: PathWrapper =
+		"/tmp/async-boot-e2e/attestation-dir".into();
 	fs::create_dir_all(&*attestation_dir).unwrap();
 	let attestation_doc_path = format!("{}/attestation_doc", &*attestation_dir);
 
@@ -65,7 +68,7 @@ async fn async_standard_boot_e2e() {
 
 	// -- CLIENT create manifest.
 	let msg = "testing420";
-	let pivot_args = format!("[--msg,{msg}]");
+	let pivot_args = format!("[--msg,{msg},--pool-size,20]");
 	let cli_manifest_path = format!("{}/manifest", &*boot_dir);
 
 	assert!(Command::new("../target/debug/qos_client")
@@ -131,7 +134,12 @@ async fn async_standard_boot_e2e() {
 	let pivot = PivotConfig {
 		hash: mock_pivot_hash,
 		restart: RestartPolicy::Never,
-		args: vec!["--msg".to_string(), msg.to_string()],
+		args: vec![
+			"--msg".to_string(),
+			msg.to_string(),
+			"--pool-size".to_string(),
+			"20".to_string(),
+		],
 	};
 	assert_eq!(manifest.pivot, pivot);
 	let manifest_set = ManifestSet { threshold: 2, members: members.clone() };
@@ -221,7 +229,7 @@ async fn async_standard_boot_e2e() {
 		);
 		assert_eq!(
 			&stdout.next().unwrap().unwrap(),
-			"[\"--msg\", \"testing420\"]?"
+			"[\"--msg\", \"testing420\", \"--pool-size\", \"20\"]?"
 		);
 		assert_eq!(&stdout.next().unwrap().unwrap(), "(y/n)");
 		stdin.write_all("y\n".as_bytes()).expect("Failed to write to stdin");