add tui resource usage view

Author: Panaetius

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "coman"
-version = "0.7.0"
+version = "0.8.0"
 edition = "2024"
 description = "Compute Manager for managing HPC compute"
 authors = ["Ralf Grubenmann <ralf.grubenmann@sdsc.ethz.ch>"]

coman/Cargo.toml

@@ -5,6 +5,7 @@ pub enum Id {
     WorkloadList,
     WorkloadLogs,
     WorkloadDetails,
+    WorkloadResourceUsage,
     GlobalListener,
     Menu,
     InfoPopup,

coman/src/app/ids.rs

@@ -53,6 +53,7 @@ pub enum CscsMsg {
 #[derive(Debug, PartialEq)]
 pub enum JobMsg {
     Log(usize),
+    ResourceUsage(usize),
     Details(JobDetail),
     GetDetails(usize),
     Cancel(usize),

coman/src/app/messages.rs

@@ -21,12 +21,12 @@ use crate::{
     },
     components::{
         context_menu::ContextMenu, download_popup::DownloadTargetInput, error_popup::ErrorPopup, info_popup::InfoPopup,
-        login_popup::LoginPopup, system_select_popup::SystemSelectPopup, workload_details::WorkloadDetails,
-        workload_list::WorkloadList, workload_log::WorkloadLog,
+        login_popup::LoginPopup, resource_usage::ResourceUsage, system_select_popup::SystemSelectPopup,
+        workload_details::WorkloadDetails, workload_list::WorkloadList, workload_log::WorkloadLog,
     },
     cscs::{
         handlers::{cscs_login, cscs_system_set},
-        ports::{BackgroundTask, JobLogAction},
+        ports::{BackgroundTask, JobLogAction, JobResourceUsageAction},
     },
     trace_dbg,
     util::ui::{draw_area_in_absolute, draw_area_in_absolute_fixed_height},
@@ -58,6 +58,10 @@ where
     /// sending None stops watching
     pub job_log_tx: mpsc::Sender<JobLogAction>,
 
+    /// Triggers watching job logs
+    /// sending None stops watching
+    pub job_resource_usage_tx: mpsc::Sender<JobResourceUsageAction>,
+
     /// Allows creating user events based on messages
     pub user_event_tx: mpsc::Sender<UserEvent>,
 
@@ -69,12 +73,14 @@ impl<T> Model<T>
 where
     T: TerminalAdapter,
 {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         app: Application<Id, Msg, UserEvent>,
         bridge: TerminalBridge<T>,
         error_tx: mpsc::Sender<String>,
         select_system_tx: mpsc::Sender<()>,
         job_log_tx: mpsc::Sender<JobLogAction>,
+        job_resource_usage_tx: mpsc::Sender<JobResourceUsageAction>,
         user_event_tx: mpsc::Sender<UserEvent>,
         background_task_tx: mpsc::Sender<BackgroundTask>,
     ) -> Self {
@@ -87,6 +93,7 @@ where
             error_tx,
             select_system_tx,
             job_log_tx,
+            job_resource_usage_tx,
             user_event_tx,
             background_task_tx,
         }
@@ -151,6 +158,7 @@ where
         app.view(&Id::WorkloadList, frame, area);
         app.view(&Id::WorkloadLogs, frame, area);
         app.view(&Id::WorkloadDetails, frame, area);
+        app.view(&Id::WorkloadResourceUsage, frame, area);
     }
     fn view_files(app: &mut Application<Id, Msg, UserEvent>, frame: &mut Frame, area: Rect) {
         if app.mounted(&Id::FileView) {
@@ -314,6 +322,9 @@ where
                 if self.app.mounted(&Id::WorkloadDetails) {
                     assert!(self.app.umount(&Id::WorkloadDetails).is_ok());
                 }
+                if self.app.mounted(&Id::WorkloadResourceUsage) {
+                    assert!(self.app.umount(&Id::WorkloadResourceUsage).is_ok());
+                }
                 if !self.app.mounted(&Id::WorkloadLogs) {
                     assert!(
                         self.app
@@ -362,6 +373,12 @@ where
                             .is_ok()
                     );
                 }
+                if self.app.mounted(&Id::WorkloadResourceUsage) {
+                    assert!(self.app.umount(&Id::WorkloadResourceUsage).is_ok());
+                }
+                if self.app.mounted(&Id::WorkloadLogs) {
+                    assert!(self.app.umount(&Id::WorkloadLogs).is_ok());
+                }
                 if !self.app.mounted(&Id::WorkloadDetails) {
                     assert!(
                         self.app
@@ -372,6 +389,37 @@ where
                 assert!(self.app.active(&Id::WorkloadDetails).is_ok());
                 None
             }
+            JobMsg::ResourceUsage(jobid) => {
+                if self.app.mounted(&Id::WorkloadList) {
+                    assert!(
+                        self.app
+                            .attr(&Id::WorkloadList, Attribute::Display, AttrValue::Flag(false))
+                            .is_ok()
+                    );
+                }
+                if self.app.mounted(&Id::WorkloadDetails) {
+                    assert!(self.app.umount(&Id::WorkloadDetails).is_ok());
+                }
+                if self.app.mounted(&Id::WorkloadLogs) {
+                    assert!(self.app.umount(&Id::WorkloadLogs).is_ok());
+                }
+                if !self.app.mounted(&Id::WorkloadResourceUsage) {
+                    assert!(
+                        self.app
+                            .mount(Id::WorkloadResourceUsage, Box::new(ResourceUsage::default()), vec![])
+                            .is_ok()
+                    );
+                }
+                assert!(self.app.active(&Id::WorkloadResourceUsage).is_ok());
+                let job_resource_usage_tx = self.job_resource_usage_tx.clone();
+                tokio::spawn(async move {
+                    job_resource_usage_tx
+                        .send(JobResourceUsageAction::Job(jobid))
+                        .await
+                        .unwrap();
+                });
+                None
+            }
             JobMsg::Switch => {
                 let job_log_tx = self.job_log_tx.clone();
                 tokio::spawn(async move {
@@ -386,6 +434,9 @@ where
                 if self.app.mounted(&Id::WorkloadDetails) {
                     assert!(self.app.umount(&Id::WorkloadDetails).is_ok());
                 }
+                if self.app.mounted(&Id::WorkloadResourceUsage) {
+                    assert!(self.app.umount(&Id::WorkloadResourceUsage).is_ok());
+                }
                 if !self.app.mounted(&Id::WorkloadList) {
                     assert!(
                         self.app
@@ -404,6 +455,11 @@ where
                     // stopp polling for logs
                     job_log_tx.send(JobLogAction::Stop).await.unwrap();
                 });
+                let job_resource_usage_tx = self.job_resource_usage_tx.clone();
+                tokio::spawn(async move {
+                    // stopp polling for resource_usages
+                    job_resource_usage_tx.send(JobResourceUsageAction::Stop).await.unwrap();
+                });
                 None
             }
         }

coman/src/app/model.rs

@@ -1,5 +1,6 @@
 use crate::{
     app::messages::View,
+    cli::rpc::ResourceUsage,
     cscs::api_client::types::{Job, JobDetail, PathEntry, System},
 };
 
@@ -9,6 +10,7 @@ pub enum CscsEvent {
     GotWorkloadData(Vec<Job>),
     GotJobLog(String),
     GotJobDetails(JobDetail),
+    GotJobResourceUsage(ResourceUsage),
     SelectSystemList(Vec<System>),
     SystemSelected(String),
 }

coman/src/app/user_events.rs

@@ -11,7 +11,7 @@ use pid1::Pid1Settings;
 use rust_supervisor::{ChildType, Supervisor, SupervisorConfig};
 use tokio::{io::AsyncWriteExt, net::TcpStream};
 
-use crate::cli::rpc::RpcHandler;
+use crate::cli::rpc::{COMAN_RPC_ALPN, RpcHandler};
 
 const SECRET_KEY_ENV: &str = "COMAN_IROH_SECRET";
 const PORT_FORWARD_ENV: &str = "COMAN_FORWARDED_PORTS";
@@ -126,7 +126,7 @@ async fn port_forward() -> Result<()> {
 
     // add rpc server
     let rpc_handler = RpcHandler;
-    builder = builder.accept(b"/coman/rpc", rpc_handler);
+    builder = builder.accept(COMAN_RPC_ALPN, rpc_handler);
     let _router = builder.spawn();
     println!("port forwarding started");
 

coman/src/cli/exec.rs

@@ -11,14 +11,32 @@ use tokio_duplex::Duplex;
 
 use crate::cli::app::COMAN_VERSION;
 
-#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub const COMAN_RPC_ALPN: &[u8; 10] = b"/coman/rpc";
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq)]
 pub struct ResourceUsage {
     pub cpu: f32,
     pub rss: u64,
-    pub vss: u64,
+    pub vsz: u64,
     pub gpu: Option<Vec<(u64, u64)>>,
 }
 
+impl Eq for ResourceUsage {}
+impl Ord for ResourceUsage {
+    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
+        self.cpu.total_cmp(&other.cpu).then(
+            self.rss
+                .cmp(&other.rss)
+                .then(self.vsz.cmp(&other.vsz).then(self.gpu.cmp(&other.gpu))),
+        )
+    }
+}
+impl PartialOrd for ResourceUsage {
+    fn partial_cmp(&self, other: &ResourceUsage) -> Option<std::cmp::Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
 #[tarpc::service]
 pub trait ComanRPC {
     async fn version() -> String;
@@ -70,7 +88,7 @@ impl ComanRPC for RpcServer {
         ResourceUsage {
             cpu: process.cpu_usage() / sys.cpus().len() as f32,
             rss: process.memory(),
-            vss: process.virtual_memory(),
+            vsz: process.virtual_memory(),
             gpu: gpu_usage,
         }
     }

coman/src/cli/rpc.rs

@@ -5,6 +5,7 @@ pub(crate) mod file_tree;
 pub(crate) mod global_listener;
 pub(crate) mod info_popup;
 pub(crate) mod login_popup;
+pub(crate) mod resource_usage;
 pub(crate) mod status_bar;
 pub(crate) mod system_select_popup;
 pub(crate) mod toolbar;