tmt/test: Sort tests by time taken, use mpsc channels

Sort tests in descending order of time taken for completion so longer
tests get scheduled together. Also, update to use mpsc channels for
communication between threads

Signed-off-by: Pragyan Poudyal <pragyanpoudyal41999@gmail.com>

Author: Johan-Liebert1

crates/xtask/src/tmt.rs

@@ -1,5 +1,6 @@
-use std::thread::JoinHandle;
+use std::sync::mpsc;
 use std::time::Duration;
+use std::usize;
 
 use anyhow::{Context, Result};
 use camino::{Utf8Path, Utf8PathBuf};
@@ -39,6 +40,36 @@ const ENV_BOOTC_UPGRADE_IMAGE: &str = "BOOTC_upgrade_image";
 // Distro identifiers
 const DISTRO_CENTOS_9: &str = "centos-9";
 
+// Tests sorted by time taken (descending)
+const TESTS_SORTED_BY_TIME: [&str; 23] = [
+    // 10+ mins
+    "multi-device-esp",
+    "composefs-gc-uki",
+    "composefs-gc",
+    // 5+ mins
+    "loader-entries-source",
+    "download-only-upgrade",
+    "bib-build",
+    "rollback",
+    "logically-bound-switch",
+    "soft-reboot",
+    "switch-to-unified",
+    "image-pushpull-upgrade",
+    "install-no-boot-dir",
+    "upgrade-tag",
+    "custom-selinux-policy",
+    "factory-reset",
+    // 3+ mins
+    "upgrade-check-status",
+    "soft-reboot-selinux-policy",
+    "install-bootloader-none",
+    "install-outside-container",
+    "install-unified-flag",
+    "usroverlay",
+    "image-upgrade-reboot",
+    "install-karg-delete",
+];
+
 // Import the argument types from xtask.rs
 use crate::bcvk::BcvkInstallOpts;
 use crate::{RunTmtArgs, SealState, TmtProvisionArgs};
@@ -653,6 +684,13 @@ pub(crate) fn run_tmt(sh: &Shell, args: &RunTmtArgs) -> Result<()> {
         return Ok(());
     }
 
+    plans.sort_by_key(|full_plan_name| {
+        TESTS_SORTED_BY_TIME
+            .iter()
+            .position(|test_time| full_plan_name.contains(test_time))
+            .unwrap_or(usize::MAX)
+    });
+
     println!("Found {} test plan(s): {:?}", plans.len(), plans);
 
     let mut install_opts = Vec::new();
@@ -696,7 +734,7 @@ pub(crate) fn run_tmt(sh: &Shell, args: &RunTmtArgs) -> Result<()> {
     // Environment variables to pass to tmt (in addition to args.env)
     let mut tmt_env_vars = Vec::new();
 
-    let mut handles: Vec<JoinHandle<RunPlanResult>> = vec![];
+    let mut active_threads = 0;
 
     let num_cpu = std::thread::available_parallelism()
         .map(|c| c.get())
@@ -715,6 +753,8 @@ pub(crate) fn run_tmt(sh: &Shell, args: &RunTmtArgs) -> Result<()> {
 
     println!("parallel_vms: {parallel_vms}");
 
+    let (tx, rx) = mpsc::channel::<RunPlanResult>();
+
     // Run each plan in its own VM
     for plan in plans {
         let plan_name = sanitize_plan_name(plan);
@@ -773,8 +813,9 @@ pub(crate) fn run_tmt(sh: &Shell, args: &RunTmtArgs) -> Result<()> {
         let vm_mem = vm_mem.to_string();
         let vm_cpu = vm_cpu.to_string();
 
-        let handle = std::thread::spawn(move || {
-            run_plan(
+        let tx_clone = tx.clone();
+        std::thread::spawn(move || {
+            let result = run_plan(
                 cloned_plan,
                 cloned_vm_name,
                 image,
@@ -786,36 +827,44 @@ pub(crate) fn run_tmt(sh: &Shell, args: &RunTmtArgs) -> Result<()> {
                 preserve_vm,
                 vm_cpu,
                 vm_mem,
-            )
-        });
+            );
 
-        handles.push(handle);
+            if let Err(e) = tx_clone.send(result) {
+                eprintln!("Failed to send result through channel: {}", e);
+            }
+        });
 
-        if handles.len() >= parallel_vms {
-            let e = handles.remove(0).join();
+        active_threads += 1;
 
-            match e {
+        // wait for a thread to complete if we've reached the parallel limit
+        if active_threads >= parallel_vms {
+            match rx.recv() {
                 Ok(plan_result) => {
                     test_results.push(plan_result);
+                    active_threads -= 1;
                 }
-
                 Err(e) => {
-                    eprintln!("Join failed: {e:?}");
+                    eprintln!("Failed to receive result from channel: {}", e);
+                    // still decrement to avoid infinite loop
+                    // in theory this shouldn't happen as we loop over plans, but
+                    // for sanity
+                    active_threads -= 1;
                 }
             }
         }
     }
 
-    for h in handles {
-        let e = h.join();
+    // drop the sender to signal no more messages
+    drop(tx);
 
-        match e {
+    // remaining results from channel
+    for _ in 0..active_threads {
+        match rx.recv() {
             Ok(plan_result) => {
                 test_results.push(plan_result);
             }
-
             Err(e) => {
-                eprintln!("Join failed: {e:?}");
+                eprintln!("Failed to receive remaining result from channel: {}", e);
             }
         }
     }