jason/socktop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add processes window cleanup

Browse Source 
- refactor code
- add unit test
- fix warnings.
This commit is contained in:
jasonwitty
2025-10-05 00:07:27 -07:00
parent e66008f341
commit e857cfc665
11 changed files with 3570 additions and 1586 deletions
Download Patch File Download Diff File Expand all files Collapse all files
socktop_agent/src/metrics.rs
+11 -16
View File
@@ -608,18 +608,13 @@ fn enumerate_child_processes_lightweight(
// This is much faster than refresh_processes_specifics(All)
if let Ok(entries) = fs::read_dir("/proc") {
for entry in entries.flatten() {
if let Ok(file_name) = entry.file_name().into_string() {
if let Ok(pid) = file_name.parse::<u32>() {
// Check if this process is a child of our target
if let Some(child_parent_pid) = read_parent_pid_from_proc(pid) {
if child_parent_pid == parent_pid {
// Found a child! Collect its details from /proc
if let Some(child_info) = collect_process_info_from_proc(pid, system) {
children.push(child_info);
}
}
}
}
if let Ok(file_name) = entry.file_name().into_string()
&& let Ok(pid) = file_name.parse::<u32>()
&& let Some(child_parent_pid) = read_parent_pid_from_proc(pid)
&& child_parent_pid == parent_pid
&& let Some(child_info) = collect_process_info_from_proc(pid, system)
{
children.push(child_info);
}
}
}
@@ -673,10 +668,10 @@ fn collect_process_info_from_proc(
if let Some(kb) = value.split_whitespace().next() {
mem_bytes = kb.parse::<u64>().unwrap_or(0) * 1024;
}
} else if let Some(value) = line.strip_prefix("VmSize:") {
if let Some(kb) = value.split_whitespace().next() {
virtual_mem_bytes = kb.parse::<u64>().unwrap_or(0) * 1024;
}
} else if let Some(value) = line.strip_prefix("VmSize:")
&& let Some(kb) = value.split_whitespace().next()
{
virtual_mem_bytes = kb.parse::<u64>().unwrap_or(0) * 1024;
}
}
socktop_agent/src/ws.rs
+64 -66
View File
@@ -114,87 +114,85 @@ async fn handle_socket(mut socket: WebSocket, state: AppState) {
drop(cache); // Explicit drop to release mutex early
}
Message::Text(ref text) if text.starts_with("get_process_metrics:") => {
if let Some(pid_str) = text.strip_prefix("get_process_metrics:") {
if let Ok(pid) = pid_str.parse::<u32>() {
let ttl = std::time::Duration::from_millis(250); // 250ms TTL
if let Some(pid_str) = text.strip_prefix("get_process_metrics:")
&& let Ok(pid) = pid_str.parse::<u32>()
{
let ttl = std::time::Duration::from_millis(250); // 250ms TTL
// Check cache first
// Check cache first
{
let cache = state.cache_process_metrics.lock().await;
if let Some(entry) = cache.get(&pid)
&& entry.is_fresh(ttl)
&& let Some(cached_response) = entry.get()
{
let cache = state.cache_process_metrics.lock().await;
if let Some(entry) = cache.get(&pid) {
if entry.is_fresh(ttl)
&& let Some(cached_response) = entry.get()
{
let _ = send_json(&mut socket, cached_response).await;
continue;
}
}
let _ = send_json(&mut socket, cached_response).await;
continue;
}
}
// Collect fresh data
match crate::metrics::collect_process_metrics(pid, &state).await {
Ok(response) => {
// Cache the response
{
let mut cache = state.cache_process_metrics.lock().await;
cache
.entry(pid)
.or_insert_with(crate::state::CacheEntry::new)
.set(response.clone());
}
let _ = send_json(&mut socket, &response).await;
}
Err(err) => {
let error_response = serde_json::json!({
"error": err,
"request": "get_process_metrics",
"pid": pid
});
let _ = send_json(&mut socket, &error_response).await;
// Collect fresh data
match crate::metrics::collect_process_metrics(pid, &state).await {
Ok(response) => {
// Cache the response
{
let mut cache = state.cache_process_metrics.lock().await;
cache
.entry(pid)
.or_insert_with(crate::state::CacheEntry::new)
.set(response.clone());
}
let _ = send_json(&mut socket, &response).await;
}
Err(err) => {
let error_response = serde_json::json!({
"error": err,
"request": "get_process_metrics",
"pid": pid
});
let _ = send_json(&mut socket, &error_response).await;
}
}
}
}
Message::Text(ref text) if text.starts_with("get_journal_entries:") => {
if let Some(pid_str) = text.strip_prefix("get_journal_entries:") {
if let Ok(pid) = pid_str.parse::<u32>() {
let ttl = std::time::Duration::from_secs(1); // 1s TTL
if let Some(pid_str) = text.strip_prefix("get_journal_entries:")
&& let Ok(pid) = pid_str.parse::<u32>()
{
let ttl = std::time::Duration::from_secs(1); // 1s TTL
// Check cache first
// Check cache first
{
let cache = state.cache_journal_entries.lock().await;
if let Some(entry) = cache.get(&pid)
&& entry.is_fresh(ttl)
&& let Some(cached_response) = entry.get()
{
let cache = state.cache_journal_entries.lock().await;
if let Some(entry) = cache.get(&pid) {
if entry.is_fresh(ttl)
&& let Some(cached_response) = entry.get()
{
let _ = send_json(&mut socket, cached_response).await;
continue;
}
}
let _ = send_json(&mut socket, cached_response).await;
continue;
}
}
// Collect fresh data
match crate::metrics::collect_journal_entries(pid) {
Ok(response) => {
// Cache the response
{
let mut cache = state.cache_journal_entries.lock().await;
cache
.entry(pid)
.or_insert_with(crate::state::CacheEntry::new)
.set(response.clone());
}
let _ = send_json(&mut socket, &response).await;
}
Err(err) => {
let error_response = serde_json::json!({
"error": err,
"request": "get_journal_entries",
"pid": pid
});
let _ = send_json(&mut socket, &error_response).await;
// Collect fresh data
match crate::metrics::collect_journal_entries(pid) {
Ok(response) => {
// Cache the response
{
let mut cache = state.cache_journal_entries.lock().await;
cache
.entry(pid)
.or_insert_with(crate::state::CacheEntry::new)
.set(response.clone());
}
let _ = send_json(&mut socket, &response).await;
}
Err(err) => {
let error_response = serde_json::json!({
"error": err,
"request": "get_journal_entries",
"pid": pid
});
let _ = send_json(&mut socket, &error_response).await;
}
}
}