feat: mutable barrier interval

proto/meta.proto

@@ -205,6 +205,7 @@ enum SubscribeType {
   FRONTEND = 1;
   HUMMOCK = 2;
   COMPACTOR = 3;
+  COMPUTE = 4;
 }
 
 // Below for notification service.
@@ -265,6 +266,7 @@ message SubscribeResponse {
     hummock.HummockVersionDeltas hummock_version_deltas = 15;
     MetaSnapshot snapshot = 16;
     backup_service.MetaBackupManifestId meta_backup_manifest_id = 17;
+    SystemParams system_params = 19;
   }
 }
 

src/common/common_service/src/observer_manager.rs

@@ -16,7 +16,6 @@ use std::time::Duration;
 
 use risingwave_common::bail;
 use risingwave_common::error::Result;
-use risingwave_pb::meta::meta_snapshot::SnapshotVersion;
 use risingwave_pb::meta::subscribe_response::Info;
 use risingwave_pb::meta::{SubscribeResponse, SubscribeType};
 use risingwave_rpc_client::error::RpcError;
@@ -49,6 +48,13 @@ impl SubscribeTypeEnum for SubscribeCompactor {
     }
 }
 
+pub struct SubscribeCompute {}
+impl SubscribeTypeEnum for SubscribeCompute {
+    fn subscribe_type() -> SubscribeType {
+        SubscribeType::Compute
+    }
+}
+
 /// `ObserverManager` is used to update data based on notification from meta.
 /// Call `start` to spawn a new asynchronous task
 /// We can write the notification logic by implementing `ObserverNodeImpl`.
@@ -109,12 +115,6 @@ where
             unreachable!();
         };
 
-        let SnapshotVersion {
-            catalog_version,
-            parallel_unit_mapping_version,
-            worker_node_version,
-        } = info.version.clone().unwrap();
-
         notification_vec.retain_mut(|notification| match notification.info.as_ref().unwrap() {
             Info::Database(_)
             | Info::Schema(_)
@@ -124,14 +124,21 @@ where
             | Info::Index(_)
             | Info::View(_)
             | Info::Function(_)
-            | Info::User(_) => notification.version > catalog_version,
-            Info::ParallelUnitMapping(_) => notification.version > parallel_unit_mapping_version,
-            Info::Node(_) => notification.version > worker_node_version,
+            | Info::User(_) => {
+                notification.version > info.version.as_ref().unwrap().catalog_version
+            }
+            Info::ParallelUnitMapping(_) => {
+                notification.version > info.version.as_ref().unwrap().parallel_unit_mapping_version
+            }
+            Info::Node(_) => {
+                notification.version > info.version.as_ref().unwrap().worker_node_version
+            }
             Info::HummockVersionDeltas(version_delta) => {
                 version_delta.version_deltas[0].id > info.hummock_version.as_ref().unwrap().id
             }
             Info::HummockSnapshot(_) => true,
             Info::MetaBackupManifestId(_) => true,
+            Info::SystemParams(_) => true,
             Info::Snapshot(_) => unreachable!(),
         });
 

src/common/src/system_param/local_manager.rs

@@ -22,12 +22,14 @@ use tokio::sync::watch::{channel, Receiver, Sender};
 use super::reader::SystemParamsReader;
 
 pub type SystemParamsReaderRef = Arc<ArcSwap<SystemParamsReader>>;
+pub type LocalSystemParamManagerRef = Arc<LocalSystemParamManager>;
 
 /// The system parameter manager on worker nodes. It provides two methods for other components to
 /// read the latest system parameters:
 /// - `get_params` returns a reference to the latest parameters that is atomically updated.
 /// - `watch_params` returns a channel on which calling `recv` will get the latest parameters.
 ///   Compared with `get_params`, the caller can be explicitly notified of parameter change.
+#[derive(Debug)]
 pub struct LocalSystemParamManager {
     /// The latest parameters.
     params: SystemParamsReaderRef,
@@ -56,7 +58,7 @@ impl LocalSystemParamManager {
         }
     }
 
-    pub fn watch_parmams(&self) -> Receiver<SystemParamsReaderRef> {
+    pub fn watch_params(&self) -> Receiver<SystemParamsReaderRef> {
         self.tx.subscribe()
     }
 }
@@ -76,7 +78,7 @@ mod tests {
             ..Default::default()
         };
 
-        let mut params_rx = manager.watch_parmams();
+        let mut params_rx = manager.watch_params();
 
         manager.try_set_params(new_params.clone());
         params_rx.changed().await.unwrap();

src/common/src/system_param/mod.rs

@@ -27,7 +27,7 @@ pub type SystemParamsError = String;
 type Result<T> = core::result::Result<T, SystemParamsError>;
 
 // Only includes undeprecated params.
-// Macro input is { field identifier, default value }
+// Macro input is { field identifier, type, default value }
 macro_rules! for_all_undeprecated_params {
     ($macro:ident) => {
         $macro! {
@@ -45,7 +45,7 @@ macro_rules! for_all_undeprecated_params {
 }
 
 // Only includes deprecated params. Used to define key constants.
-// Macro input is { field identifier, default value }
+// Macro input is { field identifier, type, default value }
 macro_rules! for_all_deprecated_params {
     ($macro:ident) => {
         $macro! {}
@@ -197,13 +197,25 @@ macro_rules! impl_set_system_param {
     };
 }
 
-for_all_undeprecated_params!(impl_system_params_from_kv);
+macro_rules! impl_default_system_params {
+    ($({ $field:ident, $type:ty, $default:expr },)*) => {
+        #[allow(clippy::needless_update)]
+        pub fn default_system_params() -> SystemParams {
+            SystemParams {
+                $(
+                    $field: Some($default),
+                )*
+                ..Default::default()
+            }
+        }
+    };
+}
 
+for_all_undeprecated_params!(impl_system_params_from_kv);
 for_all_undeprecated_params!(impl_system_params_to_kv);
-
 for_all_undeprecated_params!(impl_set_system_param);
-
 for_all_undeprecated_params!(impl_default_validation_on_set);
+for_all_undeprecated_params!(impl_default_system_params);
 
 struct OverrideValidateOnSet;
 impl ValidateOnSet for OverrideValidateOnSet {

src/compute/src/lib.rs

@@ -25,6 +25,7 @@
 extern crate tracing;
 
 pub mod memory_management;
+pub mod observer;
 pub mod rpc;
 pub mod server;
 

src/compute/src/memory_management/memory_manager.rs

@@ -16,10 +16,14 @@ use std::sync::atomic::AtomicU64;
 use std::sync::Arc;
 
 use risingwave_batch::task::BatchManager;
+use risingwave_common::system_param::local_manager::{
+    LocalSystemParamManagerRef, SystemParamsReaderRef,
+};
 #[cfg(target_os = "linux")]
 use risingwave_common::util::epoch::Epoch;
 use risingwave_stream::executor::monitor::StreamingMetrics;
 use risingwave_stream::task::LocalStreamManager;
+use tokio::select;
 
 /// The minimal memory requirement of computing tasks in megabytes.
 pub const MIN_COMPUTE_MEMORY_MB: usize = 512;
@@ -35,8 +39,8 @@ pub struct GlobalMemoryManager {
     watermark_epoch: Arc<AtomicU64>,
     /// Total memory can be allocated by the process.
     total_memory_available_bytes: usize,
-    /// Barrier interval.
-    barrier_interval_ms: u32,
+    /// Read the latest barrier interval from system parameter.
+    system_param_manager: LocalSystemParamManagerRef,
     metrics: Arc<StreamingMetrics>,
 }
 
@@ -50,17 +54,13 @@ impl GlobalMemoryManager {
 
     pub fn new(
         total_memory_available_bytes: usize,
-        barrier_interval_ms: u32,
+        system_param_manager: LocalSystemParamManagerRef,
         metrics: Arc<StreamingMetrics>,
     ) -> Arc<Self> {
-        // Arbitrarily set a minimal barrier interval in case it is too small,
-        // especially when it's 0.
-        let barrier_interval_ms = std::cmp::max(barrier_interval_ms, 10);
-
         Arc::new(Self {
             watermark_epoch: Arc::new(0.into()),
             total_memory_available_bytes,
-            barrier_interval_ms,
+            system_param_manager,
             metrics,
         })
     }
@@ -97,6 +97,7 @@ impl GlobalMemoryManager {
         use std::time::Duration;
 
         use tikv_jemalloc_ctl::{epoch as jemalloc_epoch, stats as jemalloc_stats};
+
         let mem_threshold_graceful =
             (self.total_memory_available_bytes as f64 * Self::EVICTION_THRESHOLD_GRACEFUL) as usize;
         let mem_threshold_aggressive = (self.total_memory_available_bytes as f64
@@ -109,12 +110,25 @@ impl GlobalMemoryManager {
         let jemalloc_epoch_mib = jemalloc_epoch::mib().unwrap();
         let jemalloc_allocated_mib = jemalloc_stats::allocated::mib().unwrap();
 
-        let mut tick_interval =
-            tokio::time::interval(Duration::from_millis(self.barrier_interval_ms as u64));
+        let mut barrier_interval_ms =
+            Self::get_eviction_check_interval(&self.system_param_manager.get_params());
+        let mut tick_interval = tokio::time::interval(Duration::from_millis(barrier_interval_ms));
+        let mut params_rx = self.system_param_manager.watch_params();
 
         loop {
             // Wait for a while to check if need eviction.
-            tick_interval.tick().await;
+            select! {
+                // Barrier interval has changed.
+                Ok(_) = params_rx.changed() => {
+                    let new_barrier_interval_ms = Self::get_eviction_check_interval(&params_rx.borrow());
+                    if new_barrier_interval_ms != barrier_interval_ms
+                     {
+                        tick_interval = tokio::time::interval(Duration::from_millis(new_barrier_interval_ms));
+                        barrier_interval_ms = new_barrier_interval_ms;
+                    }
+                }
+                _ = tick_interval.tick() => {},
+            }
 
             if let Err(e) = jemalloc_epoch_mib.advance() {
                 tracing::warn!("Jemalloc epoch advance failed! {:?}", e);
@@ -167,11 +181,11 @@ impl GlobalMemoryManager {
             // if watermark_time_ms + self.barrier_interval_ms as u64 * step > now, we do not
             // increase the step, and set the epoch to now time epoch.
             let physical_now = Epoch::physical_now();
-            if (physical_now - watermark_time_ms) / (self.barrier_interval_ms as u64) < step {
+            if (physical_now - watermark_time_ms) / (barrier_interval_ms) < step {
                 step = last_step;
                 watermark_time_ms = physical_now;
             } else {
-                watermark_time_ms += self.barrier_interval_ms as u64 * step;
+                watermark_time_ms += barrier_interval_ms * step;
             }
 
             self.metrics
@@ -190,4 +204,10 @@ impl GlobalMemoryManager {
             self.set_watermark_time_ms(watermark_time_ms);
         }
     }
+
+    fn get_eviction_check_interval(params: &SystemParamsReaderRef) -> u64 {
+        // Arbitrarily set a minimal barrier interval in case it is too small,
+        // especially when it's 0.
+        std::cmp::max(params.load().barrier_interval_ms() as u64, 10)
+    }
 }

src/compute/src/observer/mod.rs

@@ -0,0 +1,15 @@
+// Copyright 2023 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod observer_manager;