chore(storage): remove compactor workload (risingwavelabs#12005)

Little-Wallace · Sep 6, 2023 · 61ef35b · 61ef35b
1 parent a76e519
commit 61ef35b
Show file tree

Hide file tree

Showing 4 changed files with 1 addition and 37 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/proto/hummock.proto b/proto/hummock.proto
@@ -374,11 +374,6 @@ message CompactTaskProgress {
   uint64 num_pending_write_io = 6;
 }
 
-// The measurement of the workload on a compactor to determine whether it is idle.
-message CompactorWorkload {
-  uint32 cpu = 1;
-}
-
 message SubscribeCompactionEventRequest {
   // Register provides the context_id of the corresponding Compactor.
   message Register {

diff --git a/src/storage/Cargo.toml b/src/storage/Cargo.toml
@@ -53,7 +53,6 @@ scopeguard = "1"
 sled = "0.34.7"
 spin = "0.9"
 sync-point = { path = "../utils/sync-point" }
-sysinfo = { version = "0.29", default-features = false }
 tempfile = "3"
 thiserror = "1"
 # tikv-client = { git = "https://github.com/tikv/client-rust", rev = "5714b2", optional = true }

diff --git a/src/storage/src/hummock/compactor/mod.rs b/src/storage/src/hummock/compactor/mod.rs
@@ -23,7 +23,6 @@ pub(super) mod task_progress;
 
 use std::collections::HashMap;
 use std::marker::PhantomData;
-use std::ops::Div;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, SystemTime};
@@ -47,12 +46,10 @@ use risingwave_pb::hummock::subscribe_compaction_event_request::{
 };
 use risingwave_pb::hummock::subscribe_compaction_event_response::Event as ResponseEvent;
 use risingwave_pb::hummock::{
-    CompactTaskProgress, CompactorWorkload, SubscribeCompactionEventRequest,
-    SubscribeCompactionEventResponse,
+    CompactTaskProgress, SubscribeCompactionEventRequest, SubscribeCompactionEventResponse,
 };
 use risingwave_rpc_client::HummockMetaClient;
 pub use shared_buffer_compact::{compact, merge_imms_in_memory};
-use sysinfo::{CpuRefreshKind, ProcessExt, ProcessRefreshKind, RefreshKind, System, SystemExt};
 use tokio::sync::oneshot::Sender;
 use tokio::task::JoinHandle;
 use tokio::time::Instant;
@@ -321,9 +318,6 @@ pub fn start_compactor(
     let task_progress = compactor_context.task_progress_manager.clone();
     let periodic_event_update_interval = Duration::from_millis(1000);
     let cpu_core_num = compactor_context.compaction_executor.worker_num() as u32;
-    let mut system =
-        System::new_with_specifics(RefreshKind::new().with_cpu(CpuRefreshKind::everything()));
-    let pid = sysinfo::get_current_pid().unwrap();
     let running_task_count = compactor_context.running_task_count.clone();
     let pull_task_ack = Arc::new(AtomicBool::new(true));
 
@@ -339,7 +333,6 @@ pub fn start_compactor(
         let shutdown_map = CompactionShutdownMap::default();
         let mut min_interval = tokio::time::interval(stream_retry_interval);
         let mut periodic_event_interval = tokio::time::interval(periodic_event_update_interval);
-        let mut workload_collect_interval = tokio::time::interval(Duration::from_secs(60));
 
         // This outer loop is to recreate stream.
         'start_stream: loop {
@@ -375,7 +368,6 @@ pub fn start_compactor(
 
             let executor = compactor_context.compaction_executor.clone();
             let sstable_object_id_manager = sstable_object_id_manager.clone();
-            let mut last_workload = CompactorWorkload::default();
 
             // This inner loop is to consume stream or report task progress.
             let mut event_loop_iteration_now = Instant::now();
@@ -454,7 +446,6 @@ pub fn start_compactor(
                         }
 
                         tracing::info!(
-                            cpu = %last_workload.cpu,
                             running_task_count = %running_task_count.load(Ordering::Relaxed),
                             pull_task_ack = %pull_task_ack.load(Ordering::Relaxed),
                             pending_pull_task_count = %pending_pull_task_count
@@ -463,26 +454,6 @@ pub fn start_compactor(
                         continue;
                     }
 
-                    _ = workload_collect_interval.tick() => {
-                        let refresh_result = system.refresh_process_specifics(pid, ProcessRefreshKind::new().with_cpu());
-                        debug_assert!(refresh_result);
-                        let cpu = if let Some(process) = system.process(pid) {
-                            process.cpu_usage().div(cpu_core_num as f32) as u32
-                        } else {
-                            tracing::warn!("fail to get process pid {:?}", pid);
-                            0
-                        };
-
-                        tracing::debug!("compactor cpu usage {cpu}");
-                        let workload = CompactorWorkload {
-                            cpu,
-                        };
-
-                        last_workload = workload.clone();
-
-                        continue;
-                    }
-
                     event = response_event_stream.next() => {
                         event
                     }