risingwave_compute/
lib.rs

1// Copyright 2022 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15#![feature(coroutines)]
16#![feature(type_alias_impl_trait)]
17#![feature(impl_trait_in_assoc_type)]
18#![cfg_attr(coverage, feature(coverage_attribute))]
19#![warn(clippy::large_futures, clippy::large_stack_frames)]
20
21#[macro_use]
22extern crate tracing;
23
24pub mod memory;
25pub mod observer;
26pub mod rpc;
27pub mod server;
28pub mod telemetry;
29
30use std::future::Future;
31use std::pin::Pin;
32use std::sync::Arc;
33
34use clap::Parser;
35use risingwave_common::config::{AsyncStackTraceOption, MetricLevel, OverrideConfig, Role};
36use risingwave_common::util::meta_addr::MetaAddressStrategy;
37use risingwave_common::util::resource_util::cpu::total_cpu_available;
38use risingwave_common::util::resource_util::memory::system_memory_available_bytes;
39use risingwave_common::util::tokio_util::sync::CancellationToken;
40use risingwave_common::util::worker_util::DEFAULT_RESOURCE_GROUP;
41
42/// If `total_memory_bytes` is not specified, the default memory limit will be set to
43/// the system memory limit multiplied by this proportion
44const DEFAULT_MEMORY_PROPORTION: f64 = 0.7;
45
46/// Command-line arguments for compute-node.
47#[derive(Parser, Clone, Debug, OverrideConfig)]
48#[command(
49    version,
50    about = "The worker node that executes query plans and handles data ingestion and output"
51)]
52pub struct ComputeNodeOpts {
53    // TODO: rename to listen_addr and separate out the port.
54    /// The address that this service listens to.
55    /// Usually the localhost + desired port.
56    #[clap(long, env = "RW_LISTEN_ADDR", default_value = "127.0.0.1:5688")]
57    pub listen_addr: String,
58
59    /// The address for contacting this instance of the service.
60    /// This would be synonymous with the service's "public address"
61    /// or "identifying address".
62    /// Optional, we will use `listen_addr` if not specified.
63    #[clap(long, env = "RW_ADVERTISE_ADDR")]
64    pub advertise_addr: Option<String>,
65
66    /// We will start a http server at this address via `MetricsManager`.
67    /// Then the prometheus instance will poll the metrics from this address.
68    #[clap(
69        long,
70        env = "RW_PROMETHEUS_LISTENER_ADDR",
71        default_value = "127.0.0.1:1222"
72    )]
73    pub prometheus_listener_addr: String,
74
75    #[clap(long, env = "RW_META_ADDR", default_value = "http://127.0.0.1:5690")]
76    pub meta_address: MetaAddressStrategy,
77
78    /// The path of `risingwave.toml` configuration file.
79    ///
80    /// If empty, default configuration values will be used.
81    #[clap(long, env = "RW_CONFIG_PATH", default_value = "")]
82    pub config_path: String,
83
84    /// Total available memory for the compute node in bytes. Used by both computing and storage.
85    #[clap(long, env = "RW_TOTAL_MEMORY_BYTES", default_value_t = default_total_memory_bytes())]
86    pub total_memory_bytes: usize,
87
88    /// Reserved memory for the compute node in bytes.
89    /// If not set, a portion (default to 30% for the first 16GB and 20% for the rest)
90    /// for the `total_memory_bytes` will be used as the reserved memory.
91    ///
92    /// The total memory compute and storage can use is `total_memory_bytes` - `reserved_memory_bytes`.
93    #[clap(long, env = "RW_RESERVED_MEMORY_BYTES")]
94    pub reserved_memory_bytes: Option<usize>,
95
96    /// Target memory usage for Memory Manager.
97    /// If not set, the default value is `total_memory_bytes` - `reserved_memory_bytes`
98    ///
99    /// It's strongly recommended to set it for standalone deployment.
100    ///
101    /// ## Why need this?
102    ///
103    /// Our [`crate::memory::manager::MemoryManager`] works by reading the memory statistics from
104    /// Jemalloc. This is fine when running the compute node alone; while for standalone mode,
105    /// the memory usage of **all nodes** are counted. Thus, we need to pass a reasonable total
106    /// usage so that the memory is kept around this value.
107    #[clap(long, env = "RW_MEMORY_MANAGER_TARGET_BYTES")]
108    pub memory_manager_target_bytes: Option<usize>,
109
110    /// The parallelism that the compute node will register to the scheduler of the meta service.
111    #[clap(long, env = "RW_PARALLELISM", default_value_t = default_parallelism())]
112    #[override_opts(if_absent, path = streaming.actor_runtime_worker_threads_num)]
113    pub parallelism: usize,
114
115    /// Resource group for scheduling, default value is "default"
116    #[clap(long, env = "RW_RESOURCE_GROUP", default_value_t = default_resource_group())]
117    pub resource_group: String,
118
119    /// Decides whether the compute node can be used for streaming and serving.
120    #[clap(long, env = "RW_COMPUTE_NODE_ROLE", value_enum, default_value_t = default_role())]
121    pub role: Role,
122
123    /// Used for control the metrics level, similar to log level.
124    ///
125    /// level = 0: disable metrics
126    /// level > 0: enable metrics
127    #[clap(long, hide = true, env = "RW_METRICS_LEVEL")]
128    #[override_opts(path = server.metrics_level)]
129    pub metrics_level: Option<MetricLevel>,
130
131    /// Path to data file cache data directory.
132    /// Left empty to disable file cache.
133    #[clap(long, hide = true, env = "RW_DATA_FILE_CACHE_DIR")]
134    #[override_opts(path = storage.data_file_cache.dir)]
135    pub data_file_cache_dir: Option<String>,
136
137    /// Path to meta file cache data directory.
138    /// Left empty to disable file cache.
139    #[clap(long, hide = true, env = "RW_META_FILE_CACHE_DIR")]
140    #[override_opts(path = storage.meta_file_cache.dir)]
141    pub meta_file_cache_dir: Option<String>,
142
143    /// Enable async stack tracing through `await-tree` for risectl.
144    #[clap(long, hide = true, env = "RW_ASYNC_STACK_TRACE", value_enum)]
145    #[override_opts(path = streaming.async_stack_trace)]
146    pub async_stack_trace: Option<AsyncStackTraceOption>,
147
148    /// Enable heap profile dump when memory usage is high.
149    #[clap(long, hide = true, env = "RW_HEAP_PROFILING_DIR")]
150    #[override_opts(path = server.heap_profiling.dir)]
151    pub heap_profiling_dir: Option<String>,
152
153    /// Endpoint of the connector node.
154    #[deprecated = "connector node has been deprecated."]
155    #[clap(long, hide = true, env = "RW_CONNECTOR_RPC_ENDPOINT")]
156    pub connector_rpc_endpoint: Option<String>,
157
158    /// The path of the temp secret file directory.
159    #[clap(
160        long,
161        hide = true,
162        env = "RW_TEMP_SECRET_FILE_DIR",
163        default_value = "./secrets"
164    )]
165    pub temp_secret_file_dir: String,
166}
167
168impl risingwave_common::opts::Opts for ComputeNodeOpts {
169    fn name() -> &'static str {
170        "compute"
171    }
172
173    fn meta_addr(&self) -> MetaAddressStrategy {
174        self.meta_address.clone()
175    }
176}
177
178fn validate_opts(opts: &ComputeNodeOpts) {
179    let system_memory_available_bytes = system_memory_available_bytes();
180    if opts.total_memory_bytes > system_memory_available_bytes {
181        let error_msg = format!(
182            "total_memory_bytes {} is larger than the total memory available bytes {} that can be acquired.",
183            opts.total_memory_bytes, system_memory_available_bytes
184        );
185        tracing::error!(error_msg);
186        panic!("{}", error_msg);
187    }
188    if opts.parallelism == 0 {
189        let error_msg = "parallelism should not be zero";
190        tracing::error!(error_msg);
191        panic!("{}", error_msg);
192    }
193    let total_cpu_available = total_cpu_available().ceil() as usize;
194    if opts.parallelism > total_cpu_available {
195        let error_msg = format!(
196            "parallelism {} is larger than the total cpu available {} that can be acquired.",
197            opts.parallelism, total_cpu_available
198        );
199        tracing::warn!(error_msg);
200    }
201}
202
203use crate::server::compute_node_serve;
204
205/// Start compute node
206pub fn start(
207    opts: ComputeNodeOpts,
208    shutdown: CancellationToken,
209) -> Pin<Box<dyn Future<Output = ()> + Send>> {
210    // WARNING: don't change the function signature. Making it `async fn` will cause
211    // slow compile in release mode.
212    Box::pin(async move {
213        tracing::info!("options: {:?}", opts);
214        validate_opts(&opts);
215
216        let listen_addr = opts.listen_addr.parse().unwrap();
217
218        let advertise_addr = opts
219            .advertise_addr
220            .as_ref()
221            .unwrap_or_else(|| {
222                tracing::warn!("advertise addr is not specified, defaulting to listen_addr");
223                &opts.listen_addr
224            })
225            .parse()
226            .unwrap();
227        tracing::info!("advertise addr is {}", advertise_addr);
228
229        compute_node_serve(listen_addr, advertise_addr, Arc::new(opts), shutdown).await;
230    })
231}
232
233pub fn default_total_memory_bytes() -> usize {
234    (system_memory_available_bytes() as f64 * DEFAULT_MEMORY_PROPORTION) as usize
235}
236
237pub fn default_parallelism() -> usize {
238    total_cpu_available().ceil() as usize
239}
240
241pub fn default_resource_group() -> String {
242    DEFAULT_RESOURCE_GROUP.to_owned()
243}
244
245pub fn default_role() -> Role {
246    Role::Both
247}
248
249#[cfg(test)]
250mod tests {
251    use super::*;
252
253    #[test]
254    fn test_compute_role_rejects_none() {
255        assert!(ComputeNodeOpts::try_parse_from(["compute", "--role", "none"]).is_err());
256    }
257}
risingwave_compute/lib.rs

risingwave_compute/
lib.rs
