rocketmq_runtime/blocking/
policy.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 serde::Serialize;
18
19use crate::error::RuntimeContractViolation;
20use crate::RuntimeContractPolicy;
21
22/// Represents blocking pool policy.
23#[derive(Debug, Clone)]
24pub struct BlockingPoolPolicy {
25    /// The name value.
26    pub name: String,
27    /// The max concurrency value.
28    pub max_concurrency: usize,
29    /// The max queue depth value.
30    pub max_queue_depth: usize,
31    /// The queue timeout value.
32    pub queue_timeout: Duration,
33    /// The task timeout value.
34    pub task_timeout: Duration,
35    /// The warn after value.
36    pub warn_after: Duration,
37}
38
39impl BlockingPoolPolicy {
40    /// Validates this value.
41    ///
42    /// # Errors
43    ///
44    /// Returns a contract violation when concurrency or queue depth is zero,
45    /// or when the combined queue and task timeout cannot be represented.
46    pub fn validate(&self) -> Result<(), RuntimeContractViolation> {
47        if self.max_concurrency == 0 {
48            return Err(RuntimeContractViolation::InvalidConfiguration {
49                policy: RuntimeContractPolicy::BlockingMaxConcurrencyPositive,
50            });
51        }
52        if self.max_queue_depth == 0 {
53            return Err(RuntimeContractViolation::InvalidConfiguration {
54                policy: RuntimeContractPolicy::BlockingMaxQueueDepthPositive,
55            });
56        }
57        let Some(total_timeout) = self.queue_timeout.checked_add(self.task_timeout) else {
58            return Err(RuntimeContractViolation::InvalidConfiguration {
59                policy: RuntimeContractPolicy::BlockingTimeoutRepresentable,
60            });
61        };
62        if std::time::Instant::now().checked_add(total_timeout).is_none() {
63            return Err(RuntimeContractViolation::InvalidConfiguration {
64                policy: RuntimeContractPolicy::BlockingTimeoutRepresentable,
65            });
66        }
67        Ok(())
68    }
69}
70
71impl Default for BlockingPoolPolicy {
72    fn default() -> Self {
73        Self {
74            name: "rocketmq-blocking".to_string(),
75            max_concurrency: 64,
76            max_queue_depth: 256,
77            queue_timeout: Duration::from_secs(5),
78            task_timeout: Duration::from_secs(30),
79            warn_after: Duration::from_secs(1),
80        }
81    }
82}
83
84/// Capacity-isolated blocking work owned by a service root.
85#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)]
86#[serde(rename_all = "kebab-case")]
87pub enum BlockingLane {
88    /// Represents the storage io case.
89    StorageIo,
90    /// Represents the metadata io case.
91    MetadataIo,
92    /// Represents the cpu crypto case.
93    CpuCrypto,
94}
95
96impl BlockingLane {
97    pub(crate) const ALL: [Self; 3] = [Self::StorageIo, Self::MetadataIo, Self::CpuCrypto];
98
99    pub(crate) const fn index(self) -> usize {
100        match self {
101            Self::StorageIo => 0,
102            Self::MetadataIo => 1,
103            Self::CpuCrypto => 2,
104        }
105    }
106}
107
108/// Represents blocking lane policies.
109#[derive(Debug, Clone)]
110pub struct BlockingLanePolicies {
111    /// The storage io value.
112    pub storage_io: BlockingPoolPolicy,
113    /// The metadata io value.
114    pub metadata_io: BlockingPoolPolicy,
115    /// The cpu crypto value.
116    pub cpu_crypto: BlockingPoolPolicy,
117}
118
119impl BlockingLanePolicies {
120    pub(crate) fn for_parallelism(parallelism: usize) -> Self {
121        let parallelism = parallelism.max(1);
122        Self {
123            storage_io: BlockingPoolPolicy {
124                name: "rocketmq-blocking.storage-io".to_string(),
125                max_concurrency: parallelism.saturating_mul(4).max(4),
126                max_queue_depth: parallelism.saturating_mul(16).max(16),
127                ..BlockingPoolPolicy::default()
128            },
129            metadata_io: BlockingPoolPolicy {
130                name: "rocketmq-blocking.metadata-io".to_string(),
131                max_concurrency: parallelism.saturating_mul(2).max(2),
132                max_queue_depth: parallelism.saturating_mul(8).max(8),
133                ..BlockingPoolPolicy::default()
134            },
135            cpu_crypto: BlockingPoolPolicy {
136                name: "rocketmq-blocking.cpu-crypto".to_string(),
137                max_concurrency: parallelism.max(2),
138                max_queue_depth: parallelism.saturating_mul(4).max(8),
139                ..BlockingPoolPolicy::default()
140            },
141        }
142    }
143
144    pub(crate) fn cap_concurrency(&mut self, global_capacity: usize) {
145        self.storage_io.max_concurrency = self.storage_io.max_concurrency.min(global_capacity).max(1);
146        self.metadata_io.max_concurrency = self.metadata_io.max_concurrency.min(global_capacity).max(1);
147        self.cpu_crypto.max_concurrency = self.cpu_crypto.max_concurrency.min(global_capacity).max(1);
148    }
149
150    /// Validates this value.
151    ///
152    /// # Errors
153    ///
154    /// Returns the first contract violation from one of the configured lanes.
155    pub fn validate(&self) -> Result<(), RuntimeContractViolation> {
156        self.storage_io.validate()?;
157        self.metadata_io.validate()?;
158        self.cpu_crypto.validate()
159    }
160
161    pub(crate) fn validate_for_global_capacity(&self, global_capacity: usize) -> Result<(), RuntimeContractViolation> {
162        self.validate()?;
163        if global_capacity < BlockingLane::ALL.len() {
164            return Err(RuntimeContractViolation::InvalidConfiguration {
165                policy: RuntimeContractPolicy::BlockingGlobalCapacityCoversLanes,
166            });
167        }
168        Ok(())
169    }
170
171    /// Returns the policy in force for one lane.
172    ///
173    /// Lane capacity, phase timeouts, and the warn threshold are owned here for
174    /// every consumer of the lane, so a component that accepts its own timeout
175    /// fields can report which value actually applies.
176    #[must_use]
177    pub fn policy(&self, lane: BlockingLane) -> &BlockingPoolPolicy {
178        match lane {
179            BlockingLane::StorageIo => &self.storage_io,
180            BlockingLane::MetadataIo => &self.metadata_io,
181            BlockingLane::CpuCrypto => &self.cpu_crypto,
182        }
183    }
184
185    pub(crate) fn max_concurrency(&self, lane: BlockingLane) -> usize {
186        match lane {
187            BlockingLane::StorageIo => self.storage_io.max_concurrency,
188            BlockingLane::MetadataIo => self.metadata_io.max_concurrency,
189            BlockingLane::CpuCrypto => self.cpu_crypto.max_concurrency,
190        }
191    }
192
193    pub(crate) fn total_max_concurrency(&self) -> usize {
194        self.storage_io
195            .max_concurrency
196            .saturating_add(self.metadata_io.max_concurrency)
197            .saturating_add(self.cpu_crypto.max_concurrency)
198    }
199
200    /// Creates the uniform value.
201    pub fn uniform(policy: BlockingPoolPolicy) -> Self {
202        let mut storage_io = policy.clone();
203        storage_io.name = format!("{}.storage-io", policy.name);
204        let mut metadata_io = policy.clone();
205        metadata_io.name = format!("{}.metadata-io", policy.name);
206        let mut cpu_crypto = policy;
207        cpu_crypto.name = format!("{}.cpu-crypto", cpu_crypto.name);
208        Self {
209            storage_io,
210            metadata_io,
211            cpu_crypto,
212        }
213    }
214}
215
216impl Default for BlockingLanePolicies {
217    fn default() -> Self {
218        let parallelism = std::thread::available_parallelism()
219            .map(|value| value.get())
220            .unwrap_or(4);
221        Self::for_parallelism(parallelism)
222    }
223}
224
225/// Identifies the blocking kind state.
226#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
227pub enum BlockingKind {
228    /// Represents the short io case.
229    ShortIo,
230    /// Represents the cpu bound case.
231    CpuBound,
232    /// Represents the long running case.
233    LongRunning,
234}
235
236/// Represents blocking task id.
237#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)]
238pub struct BlockingTaskId(pub(crate) u64);
239
240impl BlockingTaskId {
241    /// Borrows this value as u64.
242    pub fn as_u64(self) -> u64 {
243        self.0
244    }
245}
246
247/// Identifies the blocking task state state.
248#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
249pub enum BlockingTaskState {
250    /// Represents the queued case.
251    Queued,
252    /// Represents the running case.
253    Running,
254    /// Represents the completed case.
255    Completed,
256    /// Represents the join failed case.
257    JoinFailed,
258    /// Represents the timed out still running case.
259    TimedOutStillRunning,
260}
rocketmq_runtime/blocking/policy.rs

rocketmq_runtime/blocking/
policy.rs
