WIP

Author: zwang28

src/storage/src/hummock/compactor/table_change_log_compactor_runner.rs

@@ -83,11 +83,49 @@ pub(crate) async fn compact_table_change_log(
         compact_task_to_string(&compact_task),
     );
     let group_label = compact_task.compaction_group_id.to_string();
+    let mut task_status = TaskStatus::Success;
     let timer = context
         .compactor_metrics
         .compact_task_duration
         .with_label_values(&[&group_label, &"0".to_string()])
         .start_timer();
+    let memory_detector = context
+        .memory_limiter
+        .try_require_memory(task_memory_capacity_with_parallelism);
+    if memory_detector.is_none() {
+        tracing::warn!(
+            "Not enough memory to serve the task {} task_memory_capacity_with_parallelism {}  memory_usage {} memory_quota {}",
+            compact_task.task_id,
+            task_memory_capacity_with_parallelism,
+            context.memory_limiter.get_memory_usage(),
+            context.memory_limiter.quota()
+        );
+        task_status = TaskStatus::NoAvailMemoryResourceCanceled;
+        return (
+            seal_table_change_log_compaction_task(
+                compact_task,
+                context.clone(),
+                vec![],
+                vec![],
+                task_status,
+            ),
+            memory_detector,
+        );
+    }
+
+    context.compactor_metrics.compact_task_pending_num.inc();
+    context
+        .compactor_metrics
+        .compact_task_pending_parallelism
+        .add(parallelism as _);
+    let _release_metrics_guard =
+        scopeguard::guard((parallelism, context.clone()), |(parallelism, context)| {
+            context.compactor_metrics.compact_task_pending_num.dec();
+            context
+                .compactor_metrics
+                .compact_task_pending_parallelism
+                .sub(parallelism as _);
+        });
     let mut compaction_futures = vec![];
     let mut abort_handles = vec![];
     let task_progress_guard =
@@ -125,7 +163,6 @@ pub(crate) async fn compact_table_change_log(
         compaction_futures.push(handle);
     }
     let mut buffered = stream::iter(compaction_futures).buffer_unordered(parallelism);
-    let mut task_status = TaskStatus::Success;
     loop {
         tokio::select! {
             _ = &mut shutdown_rx => {
@@ -190,8 +227,10 @@ pub(crate) async fn compact_table_change_log(
         cost_time,
         compact_task_output_to_string(&compact_task)
     );
-    // TODO(ZW): impl memory_detector
-    ((compact_task, table_stats, object_timestamps), None)
+    (
+        (compact_task, table_stats, object_timestamps),
+        memory_detector,
+    )
 }
 
 fn estimate_table_change_log_task_output_capacity(