rocketmq_runtime/
config.rs

1// Copyright 2023 The RocketMQ Rust Authors
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
15use std::time::Duration;
16
17use crate::blocking::BlockingLanePolicies;
18use crate::error::RuntimeContractViolation;
19use crate::RuntimeContractPolicy;
20
21/// Minimum global blocking capacity that can reserve one slot per lane.
22pub const MIN_ENTRYPOINT_BLOCKING_THREADS: usize = 3;
23/// Maximum supported global blocking capacity.
24pub const MAX_ENTRYPOINT_BLOCKING_THREADS: usize = 512;
25/// Shutdown budget used when no lifecycle deadline applies.
26///
27/// A process that runs a [`ServiceLifecycle`](crate::ServiceLifecycle) takes
28/// its shutdown deadline from the lifecycle configuration instead.
29pub(crate) const DEFAULT_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);
30
31/// Thread, driver and shutdown policy for one owned Tokio runtime.
32///
33/// [`Default`] uses the available CPU parallelism, falling back to four
34/// workers. [`Self::for_parallelism`] derives a saturating, capped blocking
35/// limit and aligns lane concurrency with it. Lanes share the global limit;
36/// their individual maxima are not independent thread pools.
37///
38/// Validation performs no I/O or thread creation. Disabling a driver is a
39/// caller capability choice: networking needs I/O, while timers, scheduling
40/// and deadline-based waits need the time driver.
41#[derive(Debug, Clone)]
42pub struct RuntimeConfig {
43    /// Number of asynchronous worker threads; must be positive.
44    pub worker_threads: usize,
45    /// Managed blocking capacity shared by the blocking lanes.
46    ///
47    /// The Tokio blocking pool holds [`Self::tokio_blocking_threads`] threads:
48    /// this capacity plus headroom for blocking work outside the lanes.
49    pub max_blocking_threads: usize,
50    /// Nonblank prefix used for runtime worker threads.
51    pub thread_name: String,
52    /// Optional stack size in bytes; `None` retains Tokio's default.
53    pub thread_stack_size: Option<usize>,
54    /// Idle blocking-thread lifetime.
55    pub thread_keep_alive: Duration,
56    /// Fallback budget for owner shutdown entrypoints called without a deadline.
57    ///
58    /// A process that runs a `ServiceLifecycle` passes the lifecycle's
59    /// deadline instead, so this budget applies only to owners shut down
60    /// without one. Calls that accept an explicit absolute deadline use that
61    /// deadline.
62    /// This is not a per-task execution timeout and cannot stop a blocking
63    /// closure already running on an operating-system thread.
64    pub shutdown_timeout: Duration,
65    /// Per-lane admission, queueing and deadline policies.
66    pub blocking_lane_policies: BlockingLanePolicies,
67    /// Enables Tokio's I/O driver; enabled in the default profile.
68    pub enable_io: bool,
69    /// Enables Tokio's timer driver; enabled in the default profile.
70    pub enable_time: bool,
71}
72
73impl RuntimeConfig {
74    /// Builds a validated entrypoint profile from the effective CPU quota.
75    ///
76    /// `available_parallelism` should be the cgroup-aware value reported by
77    /// [`std::thread::available_parallelism`]. Keeping it explicit makes the
78    /// derivation deterministic in bootstrap tests and container profiles.
79    pub fn for_parallelism(thread_name: impl Into<String>, available_parallelism: usize) -> Self {
80        let parallelism = available_parallelism.max(1);
81        let max_blocking_threads = parallelism
82            .saturating_mul(4)
83            .clamp(MIN_ENTRYPOINT_BLOCKING_THREADS, MAX_ENTRYPOINT_BLOCKING_THREADS);
84        let mut blocking_lane_policies = BlockingLanePolicies::for_parallelism(parallelism);
85        blocking_lane_policies.cap_concurrency(max_blocking_threads);
86        Self {
87            worker_threads: parallelism,
88            max_blocking_threads,
89            thread_name: thread_name.into(),
90            thread_stack_size: None,
91            thread_keep_alive: Duration::from_secs(30),
92            shutdown_timeout: DEFAULT_SHUTDOWN_TIMEOUT,
93            blocking_lane_policies,
94            enable_io: true,
95            enable_time: true,
96        }
97    }
98
99    /// Applies an explicit global blocking limit and aligns every lane ceiling.
100    ///
101    /// # Errors
102    ///
103    /// Returns a contract violation when `limit` cannot reserve one slot for
104    /// each blocking lane or exceeds the supported process bound.
105    pub fn with_max_blocking_threads(mut self, limit: usize) -> Result<Self, RuntimeContractViolation> {
106        if !(MIN_ENTRYPOINT_BLOCKING_THREADS..=MAX_ENTRYPOINT_BLOCKING_THREADS).contains(&limit) {
107            return Err(RuntimeContractViolation::InvalidConfiguration {
108                policy: RuntimeContractPolicy::MaxBlockingThreadsWithinSupportedRange,
109            });
110        }
111        self.max_blocking_threads = limit;
112        self.blocking_lane_policies.cap_concurrency(limit);
113        self.validate()?;
114        Ok(self)
115    }
116
117    /// Tokio blocking threads beyond the managed capacity: one eighth of it,
118    /// and at least two.
119    ///
120    /// Direct `spawn_blocking` calls, DNS resolution and `tokio::fs` share the
121    /// Tokio pool with admitted lane work. Without headroom they could take the
122    /// threads that admitted work is counting on, and that work would queue
123    /// inside Tokio where admission cannot see it.
124    pub fn blocking_thread_headroom(&self) -> usize {
125        (self.max_blocking_threads / 8).max(2)
126    }
127
128    /// Size of the Tokio blocking pool: the managed capacity plus
129    /// [`Self::blocking_thread_headroom`].
130    pub fn tokio_blocking_threads(&self) -> usize {
131        self.max_blocking_threads
132            .saturating_add(self.blocking_thread_headroom())
133    }
134
135    /// Creates the server default value.
136    pub fn server_default(thread_name: impl Into<String>) -> Self {
137        Self {
138            thread_name: thread_name.into(),
139            ..Self::default()
140        }
141    }
142
143    /// Creates the broker default value.
144    pub fn broker_default() -> Self {
145        Self::server_default("rocketmq-broker")
146    }
147
148    /// Creates the namesrv default value.
149    pub fn namesrv_default() -> Self {
150        Self::server_default("rocketmq-namesrv")
151    }
152
153    /// Creates the proxy default value.
154    pub fn proxy_default() -> Self {
155        Self::server_default("rocketmq-proxy")
156    }
157
158    /// Creates the controller default value.
159    pub fn controller_default() -> Self {
160        Self::server_default("rocketmq-controller")
161    }
162
163    /// Validates this value.
164    ///
165    /// # Errors
166    ///
167    /// Returns a deterministic contract violation when a runtime setting is
168    /// structurally invalid. It performs no I/O and never creates a runtime.
169    pub fn validate(&self) -> Result<(), RuntimeContractViolation> {
170        if self.worker_threads == 0 {
171            return Err(RuntimeContractViolation::InvalidConfiguration {
172                policy: RuntimeContractPolicy::WorkerThreadsPositive,
173            });
174        }
175        if self.max_blocking_threads == 0 {
176            return Err(RuntimeContractViolation::InvalidConfiguration {
177                policy: RuntimeContractPolicy::MaxBlockingThreadsPositive,
178            });
179        }
180        if self.max_blocking_threads > MAX_ENTRYPOINT_BLOCKING_THREADS {
181            return Err(RuntimeContractViolation::InvalidConfiguration {
182                policy: RuntimeContractPolicy::MaxBlockingThreadsWithinSupportedRange,
183            });
184        }
185        if self.thread_name.trim().is_empty() {
186            return Err(RuntimeContractViolation::InvalidConfiguration {
187                policy: RuntimeContractPolicy::ThreadNameNotBlank,
188            });
189        }
190        if matches!(self.thread_stack_size, Some(0)) {
191            return Err(RuntimeContractViolation::InvalidConfiguration {
192                policy: RuntimeContractPolicy::ThreadStackSizePositive,
193            });
194        }
195        self.blocking_lane_policies
196            .validate_for_global_capacity(self.max_blocking_threads)?;
197        Ok(())
198    }
199}
200
201impl Default for RuntimeConfig {
202    fn default() -> Self {
203        let parallelism = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4);
204        Self::for_parallelism("rocketmq-runtime", parallelism)
205    }
206}
207
208#[cfg(test)]
209mod tests {
210    use super::*;
211
212    #[test]
213    fn entrypoint_profiles_derive_bounded_cpu_and_lane_budgets() {
214        for (parallelism, expected_workers, expected_blocking) in [(0, 1, 4), (1, 1, 4), (2, 2, 8), (32, 32, 128)] {
215            let config = RuntimeConfig::for_parallelism("profile-test", parallelism);
216            assert_eq!(config.worker_threads, expected_workers);
217            assert_eq!(config.max_blocking_threads, expected_blocking);
218            assert!(config.blocking_lane_policies.storage_io.max_concurrency <= expected_blocking);
219            assert!(config.blocking_lane_policies.metadata_io.max_concurrency <= expected_blocking);
220            assert!(config.blocking_lane_policies.cpu_crypto.max_concurrency <= expected_blocking);
221            config.validate().expect("derived entrypoint profile is valid");
222        }
223
224        let capped = RuntimeConfig::for_parallelism("large-profile-test", usize::MAX);
225        assert_eq!(capped.max_blocking_threads, MAX_ENTRYPOINT_BLOCKING_THREADS);
226        assert_eq!(
227            RuntimeConfig::for_parallelism("headroom-test", 32).tokio_blocking_threads(),
228            128 + 16
229        );
230    }
231
232    #[test]
233    fn explicit_blocking_limit_has_safe_bounds_and_caps_lanes() {
234        let config = RuntimeConfig::for_parallelism("override-test", 16)
235            .with_max_blocking_threads(6)
236            .expect("explicit safe limit");
237        assert_eq!(config.max_blocking_threads, 6);
238        assert_eq!(config.tokio_blocking_threads(), 8);
239        assert_eq!(config.blocking_lane_policies.storage_io.max_concurrency, 6);
240        assert_eq!(config.blocking_lane_policies.metadata_io.max_concurrency, 6);
241        assert!(config.blocking_lane_policies.cpu_crypto.max_concurrency <= 6);
242
243        for invalid in [0, 1, 2, MAX_ENTRYPOINT_BLOCKING_THREADS + 1] {
244            assert!(RuntimeConfig::for_parallelism("invalid-override", 4)
245                .with_max_blocking_threads(invalid)
246                .is_err());
247        }
248    }
249}
rocketmq_runtime/config.rs

rocketmq_runtime/
config.rs
