feat(common): LocalSystemParamManager for worker node

src/common/Cargo.toml

@@ -15,6 +15,7 @@ normal = ["workspace-hack"]
 
 [dependencies]
 anyhow = "1"
+arc-swap = "1"
 arrow-array = "33"
 arrow-schema = "33"
 async-trait = "0.1"

src/common/src/system_param/local_manager.rs

@@ -0,0 +1,86 @@
+// 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.
+
+use std::ops::Deref;
+use std::sync::Arc;
+
+use arc_swap::ArcSwap;
+use risingwave_pb::meta::SystemParams;
+use tokio::sync::watch::{channel, Receiver, Sender};
+
+use super::reader::SystemParamsReader;
+
+pub type SystemParamsReaderRef = Arc<ArcSwap<SystemParamsReader>>;
+
+/// 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.
+pub struct LocalSystemParamManager {
+    /// The latest parameters.
+    params: SystemParamsReaderRef,
+
+    /// Sender of the latest parameters.
+    tx: Sender<SystemParamsReaderRef>,
+}
+
+impl LocalSystemParamManager {
+    pub fn new(params: SystemParamsReader) -> Self {
+        let params = Arc::new(ArcSwap::from_pointee(params));
+        let (tx, _) = channel(params.clone());
+        Self { params, tx }
+    }
+
+    pub fn get_params(&self) -> SystemParamsReaderRef {
+        self.params.clone()
+    }
+
+    pub fn try_set_params(&self, new_params: SystemParams) {
+        let new_params_reader = SystemParamsReader::from(new_params);
+        if self.params.load().deref().deref() != &new_params_reader {
+            self.params.store(Arc::new(new_params_reader));
+            // Ignore no active receiver.
+            let _ = self.tx.send(self.params.clone());
+        }
+    }
+
+    pub fn watch_parmams(&self) -> Receiver<SystemParamsReaderRef> {
+        self.tx.subscribe()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_manager() {
+        let p = SystemParams::default().into();
+        let manager = LocalSystemParamManager::new(p);
+        let shared_params = manager.get_params();
+
+        let new_params = SystemParams {
+            sstable_size_mb: Some(1),
+            ..Default::default()
+        };
+
+        let mut params_rx = manager.watch_parmams();
+
+        manager.try_set_params(new_params.clone());
+        params_rx.changed().await.unwrap();
+        assert_eq!(**params_rx.borrow().load(), new_params.clone().into());
+        assert_eq!(**shared_params.load(), new_params.into());
+    }
+}

src/common/src/system_param/mod.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod local_manager;
 pub mod reader;
 
 use std::collections::HashSet;

src/common/src/system_param/reader.rs

@@ -21,7 +21,7 @@ use super::system_params_to_kv;
 /// - Avoid misuse of deprecated fields by hiding their getters.
 /// - Abstract fallback logic for fields that might not be provided by meta service due to backward
 ///   compatibility.
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq)]
 pub struct SystemParamsReader {
     prost: ProstSystemParams,
 }