fix(storage): fix compactor task parallelism race

src/compute/src/server.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 use std::net::SocketAddr;
-use std::sync::atomic::AtomicU32;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -217,12 +216,6 @@ pub async fn compute_node_serve(
             ));
 
             let compaction_executor = Arc::new(CompactionExecutor::new(Some(1)));
-            let max_task_parallelism = Arc::new(AtomicU32::new(
-                (compaction_executor.worker_num() as f32
-                    * storage_opts.compactor_max_task_multiplier)
-                    .ceil() as u32,
-            ));
-
             let compactor_context = CompactorContext {
                 storage_opts,
                 sstable_store: storage.sstable_store(),
@@ -235,8 +228,6 @@ pub async fn compute_node_serve(
                 await_tree_reg: await_tree_config
                     .clone()
                     .map(new_compaction_await_tree_reg_ref),
-                running_task_parallelism: Arc::new(AtomicU32::new(0)),
-                max_task_parallelism,
             };
 
             let (handle, shutdown_sender) = start_compactor(

src/storage/compactor/src/server.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 use std::net::SocketAddr;
-use std::sync::atomic::AtomicU32;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -247,10 +246,6 @@ pub async fn compactor_serve(
     let compaction_executor = Arc::new(CompactionExecutor::new(
         opts.compaction_worker_threads_number,
     ));
-    let max_task_parallelism = Arc::new(AtomicU32::new(
-        (compaction_executor.worker_num() as f32 * storage_opts.compactor_max_task_multiplier)
-            .ceil() as u32,
-    ));
 
     let compactor_context = CompactorContext {
         storage_opts,
@@ -262,8 +257,6 @@ pub async fn compactor_serve(
 
         task_progress_manager: Default::default(),
         await_tree_reg: await_tree_reg.clone(),
-        running_task_parallelism: Arc::new(AtomicU32::new(0)),
-        max_task_parallelism,
     };
     let mut sub_tasks = vec![
         MetaClient::start_heartbeat_loop(
@@ -383,10 +376,6 @@ pub async fn shared_compactor_serve(
     let compaction_executor = Arc::new(CompactionExecutor::new(
         opts.compaction_worker_threads_number,
     ));
-    let max_task_parallelism = Arc::new(AtomicU32::new(
-        (compaction_executor.worker_num() as f32 * storage_opts.compactor_max_task_multiplier)
-            .ceil() as u32,
-    ));
     let compactor_context = CompactorContext {
         storage_opts,
         sstable_store,
@@ -396,8 +385,6 @@ pub async fn shared_compactor_serve(
         memory_limiter,
         task_progress_manager: Default::default(),
         await_tree_reg,
-        running_task_parallelism: Arc::new(AtomicU32::new(0)),
-        max_task_parallelism,
     };
     let join_handle = tokio::spawn(async move {
         tonic::transport::Server::builder()

src/storage/hummock_test/src/compactor_tests.rs

@@ -17,7 +17,6 @@ pub(crate) mod tests {
 
     use std::collections::{BTreeMap, BTreeSet, VecDeque};
     use std::ops::Bound;
-    use std::sync::atomic::AtomicU32;
     use std::sync::Arc;
 
     use bytes::{BufMut, Bytes, BytesMut};
@@ -200,12 +199,6 @@ pub(crate) mod tests {
         storage_opts: Arc<StorageOpts>,
         sstable_store: SstableStoreRef,
     ) -> CompactorContext {
-        let compaction_executor = Arc::new(CompactionExecutor::new(Some(1)));
-        let max_task_parallelism = Arc::new(AtomicU32::new(
-            (compaction_executor.worker_num() as f32 * storage_opts.compactor_max_task_multiplier)
-                .ceil() as u32,
-        ));
-
         CompactorContext {
             storage_opts,
             sstable_store,
@@ -215,8 +208,6 @@ pub(crate) mod tests {
             memory_limiter: MemoryLimiter::unlimit(),
             task_progress_manager: Default::default(),
             await_tree_reg: None,
-            running_task_parallelism: Arc::new(AtomicU32::new(0)),
-            max_task_parallelism,
         }
     }