1 // Copyright (c) 2023 Huawei Device Co., Ltd.
2 // Licensed under the Apache License, Version 2.0 (the "License");
3 // you may not use this file except in compliance with the License.
4 // You may obtain a copy of the License at
5 //
6 // http://www.apache.org/licenses/LICENSE-2.0
7 //
8 // Unless required by applicable law or agreed to in writing, software
9 // distributed under the License is distributed on an "AS IS" BASIS,
10 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11 // See the License for the specific language governing permissions and
12 // limitations under the License.
13
14 //! A builder to configure the runtime, and thread pool of the runtime.
15 //!
16 //! Ylong-runtime provides two kinds of runtime.
17 //! `CurrentThread`: Runtime which runs on the current thread.
18 //! `MultiThread`: Runtime which runs on multiple threads.
19 //!
20 //! After configuring the builder, a call to `build` will return the actual
21 //! runtime instance. [`MultiThreadBuilder`] could also be used for configuring
22 //! the global singleton runtime.
23 //!
24 //! For thread pool, the builder allows the user to set the thread number, stack
25 //! size and name prefix of each thread.
26
27 pub(crate) mod common_builder;
28 #[cfg(feature = "current_thread_runtime")]
29 pub(crate) mod current_thread_builder;
30 pub(crate) mod multi_thread_builder;
31
32 use std::fmt::Debug;
33 use std::sync::Arc;
34
35 #[cfg(feature = "current_thread_runtime")]
36 pub use current_thread_builder::CurrentThreadBuilder;
37 pub use multi_thread_builder::MultiThreadBuilder;
38
39 pub(crate) use crate::builder::common_builder::CommonBuilder;
40 use crate::error::ScheduleError;
41 use crate::executor::blocking_pool::BlockPoolSpawner;
42
43 cfg_not_ffrt!(
44 use crate::executor::async_pool::AsyncPoolSpawner;
45 use std::io;
46 );
47
48 /// A callback function to be executed in different stages of a thread's
49 /// life-cycle
50 pub type CallbackHook = Arc<dyn Fn() + Send + Sync + 'static>;
51
52 /// Schedule Policy.
53 #[derive(Debug, Clone, Copy, PartialOrd, PartialEq, Eq)]
54 pub enum ScheduleAlgo {
55 /// Bounded local queues which adopts FIFO order.
56 FifoBound,
57 }
58
59 /// Builder to build the runtime. Provides methods to customize the runtime,
60 /// such as setting thread pool size, worker thread stack size, work thread
61 /// prefix and etc.
62 ///
63 /// If `multi_instance_runtime` or `current_thread_runtime` feature is turned
64 /// on: After setting the RuntimeBuilder, a call to build will initialize the
65 /// actual runtime and returns its instance. If there is an invalid parameter
66 /// during the build, an error would be returned.
67 ///
68 /// Otherwise:
69 /// RuntimeBuilder will not have the `build()` method, instead, this builder
70 /// should be passed to set the global executor.
71 ///
72 /// # Examples
73 ///
74 /// ```no run
75 /// #![cfg(feature = "multi_instance_runtime")]
76 ///
77 /// use ylong_runtime::builder::RuntimeBuilder;
78 /// use ylong_runtime::executor::Runtime;
79 ///
80 /// let runtime = RuntimeBuilder::new_multi_thread()
81 /// .worker_num(4)
82 /// .worker_stack_size(1024 * 300)
83 /// .build()
84 /// .unwrap();
85 /// ```
86 pub struct RuntimeBuilder;
87
88 impl RuntimeBuilder {
89 /// Initializes a new RuntimeBuilder with current_thread settings.
90 ///
91 /// All tasks will run on the current thread, which means it does not create
92 /// any other worker threads.
93 ///
94 /// # Examples
95 ///
96 /// ```
97 /// use ylong_runtime::builder::RuntimeBuilder;
98 ///
99 /// let runtime = RuntimeBuilder::new_current_thread()
100 /// .worker_stack_size(1024 * 3)
101 /// .max_blocking_pool_size(4);
102 /// ```
103 #[cfg(feature = "current_thread_runtime")]
new_current_thread() -> CurrentThreadBuilder104 pub fn new_current_thread() -> CurrentThreadBuilder {
105 CurrentThreadBuilder::new()
106 }
107
108 /// Initializes a new RuntimeBuilder with multi_thread settings.
109 ///
110 /// When running, worker threads will be created according to the builder
111 /// configuration, and tasks will be allocated and run in the newly
112 /// created thread pool.
113 ///
114 /// # Examples
115 ///
116 /// ```
117 /// use ylong_runtime::builder::RuntimeBuilder;
118 ///
119 /// let runtime = RuntimeBuilder::new_multi_thread().max_blocking_pool_size(4);
120 /// ```
new_multi_thread() -> MultiThreadBuilder121 pub fn new_multi_thread() -> MultiThreadBuilder {
122 MultiThreadBuilder::new()
123 }
124 }
125
126 #[cfg(not(feature = "ffrt"))]
initialize_async_spawner( builder: &MultiThreadBuilder, ) -> io::Result<AsyncPoolSpawner>127 pub(crate) fn initialize_async_spawner(
128 builder: &MultiThreadBuilder,
129 ) -> io::Result<AsyncPoolSpawner> {
130 AsyncPoolSpawner::new(builder)
131 }
132
initialize_blocking_spawner( builder: &CommonBuilder, ) -> Result<BlockPoolSpawner, ScheduleError>133 pub(crate) fn initialize_blocking_spawner(
134 builder: &CommonBuilder,
135 ) -> Result<BlockPoolSpawner, ScheduleError> {
136 let blocking_spawner = BlockPoolSpawner::new(builder);
137 blocking_spawner.create_permanent_threads()?;
138 Ok(blocking_spawner)
139 }
140
141 #[cfg(test)]
142 mod test {
143 use crate::builder::RuntimeBuilder;
144 #[cfg(not(feature = "ffrt"))]
145 use crate::builder::ScheduleAlgo;
146
147 /// UT test cases for RuntimeBuilder::new_multi_thread()
148 ///
149 /// # Brief
150 /// 1. Checks if the object name property is None
151 /// 2. Checks if the object core_pool_size property is None
152 /// 3. Checks if the object is_steal property is true
153 /// 4. Checks if the object is_affinity property is true
154 /// 5. Checks if the object permanent_blocking_thread_num property is 4
155 /// 6. Checks if the object max_pool_size property is Some(50)
156 /// 7. Checks if the object keep_alive_time property is None
157 /// 8. Checks if the object schedule_algo property is
158 /// ScheduleAlgo::FifoBound
159 /// 9. Checks if the object stack_size property is None
160 /// 10. Checks if the object after_start property is None
161 /// 11. Checks if the object before_stop property is None
162 #[test]
ut_thread_pool_builder_new()163 fn ut_thread_pool_builder_new() {
164 let thread_pool_builder = RuntimeBuilder::new_multi_thread();
165 assert_eq!(thread_pool_builder.common.worker_name, None);
166 assert_eq!(thread_pool_builder.common.blocking_permanent_thread_num, 0);
167 assert_eq!(thread_pool_builder.common.max_blocking_pool_size, Some(50));
168 assert_eq!(thread_pool_builder.common.keep_alive_time, None);
169 #[cfg(not(feature = "ffrt"))]
170 {
171 assert_eq!(thread_pool_builder.core_thread_size, None);
172 assert_eq!(
173 thread_pool_builder.common.schedule_algo,
174 ScheduleAlgo::FifoBound
175 );
176 }
177 assert_eq!(thread_pool_builder.common.stack_size, None);
178 }
179
180 /// UT test cases for RuntimeBuilder::name()
181 ///
182 /// # Brief
183 /// 1. Checks if the object name property is modified value
184 #[test]
ut_thread_pool_builder_name()185 fn ut_thread_pool_builder_name() {
186 let name = String::from("worker_name");
187 let thread_pool_builder = RuntimeBuilder::new_multi_thread().worker_name(name.clone());
188 assert_eq!(thread_pool_builder.common.worker_name, Some(name));
189 }
190
191 /// UT test cases for RuntimeBuilder::core_pool_size()
192 ///
193 /// # Brief
194 /// 1. core_pool_size set to 1, Check if the return value is Some(1)
195 /// 2. core_pool_size set to 64, Check if the return value is Some(64)
196 /// 3. core_pool_size set to 0, Check if the return value is Some(1)
197 /// 4. core_pool_size set to 65, Check if the return value is Some(64)
198 #[test]
199 #[cfg(not(feature = "ffrt"))]
ut_thread_pool_builder_core_pool_size()200 fn ut_thread_pool_builder_core_pool_size() {
201 let thread_pool_builder = RuntimeBuilder::new_multi_thread().worker_num(1);
202 assert_eq!(thread_pool_builder.core_thread_size, Some(1));
203
204 let thread_pool_builder = RuntimeBuilder::new_multi_thread().worker_num(64);
205 assert_eq!(thread_pool_builder.core_thread_size, Some(64));
206
207 let thread_pool_builder = RuntimeBuilder::new_multi_thread().worker_num(0);
208 assert_eq!(thread_pool_builder.core_thread_size, Some(1));
209
210 let thread_pool_builder = RuntimeBuilder::new_multi_thread().worker_num(65);
211 assert_eq!(thread_pool_builder.core_thread_size, Some(64));
212 }
213
214 /// UT test cases for RuntimeBuilder::stack_size()
215 ///
216 /// # Brief
217 /// 1. stack_size set to 0, Check if the return value is Some(1)
218 /// 2. stack_size set to 1, Check if the return value is Some(1)
219 #[test]
ut_thread_pool_builder_stack_size()220 fn ut_thread_pool_builder_stack_size() {
221 let thread_pool_builder = RuntimeBuilder::new_multi_thread().worker_stack_size(0);
222 assert_eq!(thread_pool_builder.common.stack_size.unwrap(), 1);
223
224 let thread_pool_builder = RuntimeBuilder::new_multi_thread().worker_stack_size(1);
225 assert_eq!(thread_pool_builder.common.stack_size.unwrap(), 1);
226 }
227 }
228
229 #[cfg(test)]
230 #[cfg(feature = "current_thread_runtime")]
231 mod current_thread_test {
232 use crate::builder::RuntimeBuilder;
233
234 /// UT test cases for new_current_thread.
235 ///
236 /// # Brief
237 /// 1. Verify the result when multiple tasks are inserted to the current
238 /// thread at a time.
239 /// 2. Insert the task for multiple times, wait until the task is complete,
240 /// verify the result, and then perform the operation again.
241 /// 3. Spawn nest thread.
242 #[test]
ut_thread_pool_builder_current_thread()243 fn ut_thread_pool_builder_current_thread() {
244 let runtime = RuntimeBuilder::new_current_thread().build().unwrap();
245 let mut handles = vec![];
246 for index in 0..1000 {
247 let handle = runtime.spawn(async move { index });
248 handles.push(handle);
249 }
250 for (index, handle) in handles.into_iter().enumerate() {
251 let result = runtime.block_on(handle).unwrap();
252 assert_eq!(result, index);
253 }
254
255 let runtime = RuntimeBuilder::new_current_thread().build().unwrap();
256 for index in 0..1000 {
257 let handle = runtime.spawn(async move { index });
258 let result = runtime.block_on(handle).unwrap();
259 assert_eq!(result, index);
260 }
261
262 let runtime = RuntimeBuilder::new_current_thread().build().unwrap();
263 let handle = runtime.spawn(async move {
264 let runtime = RuntimeBuilder::new_current_thread().build().unwrap();
265 let handle = runtime.spawn(async move { 1_usize });
266 let result = runtime.block_on(handle).unwrap();
267 assert_eq!(result, 1);
268 result
269 });
270 let result = runtime.block_on(handle).unwrap();
271 assert_eq!(result, 1);
272 }
273 }
274
275 #[cfg(not(feature = "ffrt"))]
276 #[cfg(test)]
277 mod ylong_executor_test {
278 use crate::builder::{RuntimeBuilder, ScheduleAlgo};
279
280 /// UT test cases for ThreadPoolBuilder::is_affinity()
281 ///
282 /// # Brief
283 /// 1. is_affinity set to true, check if it is a modified value
284 /// 2. is_affinity set to false, check if it is a modified value
285 #[test]
ut_thread_pool_builder_is_affinity()286 fn ut_thread_pool_builder_is_affinity() {
287 let thread_pool_builder = RuntimeBuilder::new_multi_thread().is_affinity(true);
288 assert!(thread_pool_builder.common.is_affinity);
289
290 let thread_pool_builder = RuntimeBuilder::new_multi_thread().is_affinity(false);
291 assert!(!thread_pool_builder.common.is_affinity);
292 }
293
294 /// UT test cases for RuntimeBuilder::blocking_permanent_thread_num()
295 ///
296 /// # Brief
297 /// 1. permanent_blocking_thread_num set to 1, check if the return value is
298 /// 1.
299 /// 2. permanent_blocking_thread_num set to max_thread_num, check if the
300 /// return value is max_blocking_pool_size.
301 /// 3. permanent_blocking_thread_num set to 0, check if the return value is
302 /// 1.
303 /// 4. permanent_blocking_thread_num set to max_thread_num + 1, Check if the
304 /// return value O is max_blocking_pool_size.
305 #[test]
ut_thread_pool_builder_permanent_blocking_thread_num()306 fn ut_thread_pool_builder_permanent_blocking_thread_num() {
307 let thread_pool_builder =
308 RuntimeBuilder::new_multi_thread().blocking_permanent_thread_num(1);
309 assert_eq!(thread_pool_builder.common.blocking_permanent_thread_num, 1);
310
311 let blocking_permanent_thread_num =
312 thread_pool_builder.common.max_blocking_pool_size.unwrap();
313 let thread_pool_builder = RuntimeBuilder::new_multi_thread()
314 .blocking_permanent_thread_num(blocking_permanent_thread_num);
315 assert_eq!(
316 thread_pool_builder.common.blocking_permanent_thread_num,
317 blocking_permanent_thread_num
318 );
319
320 let thread_pool_builder =
321 RuntimeBuilder::new_multi_thread().blocking_permanent_thread_num(0);
322 assert_eq!(thread_pool_builder.common.blocking_permanent_thread_num, 0);
323
324 let permanent_blocking_thread_num =
325 thread_pool_builder.common.max_blocking_pool_size.unwrap() + 1;
326 let thread_pool_builder = RuntimeBuilder::new_multi_thread()
327 .blocking_permanent_thread_num(permanent_blocking_thread_num);
328 assert_eq!(
329 thread_pool_builder.common.blocking_permanent_thread_num,
330 thread_pool_builder.common.max_blocking_pool_size.unwrap()
331 );
332 }
333
334 /// UT test cases for RuntimeBuilder::max_pool_size()
335 ///
336 /// # Brief
337 /// 1. max_pool_size set to 1, check if the return value is Some(1)
338 /// 2. max_pool_size set to 64, check if the return value is Some(64)
339 /// 3. max_pool_size set to 0, check if the return value is Some(1)
340 /// 4. max_pool_size set to 65, check if the return value is Some(64)
341 #[test]
ut_thread_pool_builder_max_pool_size()342 fn ut_thread_pool_builder_max_pool_size() {
343 let thread_pool_builder = RuntimeBuilder::new_multi_thread().max_blocking_pool_size(1);
344 assert_eq!(
345 thread_pool_builder.common.max_blocking_pool_size.unwrap(),
346 1
347 );
348
349 let thread_pool_builder = RuntimeBuilder::new_multi_thread().max_blocking_pool_size(64);
350 assert_eq!(
351 thread_pool_builder.common.max_blocking_pool_size.unwrap(),
352 64
353 );
354
355 let thread_pool_builder = RuntimeBuilder::new_multi_thread().max_blocking_pool_size(0);
356 assert_eq!(
357 thread_pool_builder.common.max_blocking_pool_size.unwrap(),
358 1
359 );
360
361 let thread_pool_builder = RuntimeBuilder::new_multi_thread().max_blocking_pool_size(65);
362 assert_eq!(
363 thread_pool_builder.common.max_blocking_pool_size.unwrap(),
364 64
365 );
366 }
367
368 /// UT test cases for RuntimeBuilder::keep_alive_time()
369 ///
370 /// # Brief
371 /// 1. keep_alive_time set to 0, check if the return value is
372 /// Some(Duration::from_secs(0))
373 /// 2. keep_alive_time set to 1, check if the return value is
374 /// Some(Duration::from_secs(1))
375 #[test]
ut_thread_pool_builder_keep_alive_time()376 fn ut_thread_pool_builder_keep_alive_time() {
377 use std::time::Duration;
378
379 let keep_alive_time = Duration::from_secs(0);
380 let thread_pool_builder =
381 RuntimeBuilder::new_multi_thread().keep_alive_time(keep_alive_time);
382 assert_eq!(
383 thread_pool_builder.common.keep_alive_time.unwrap(),
384 keep_alive_time
385 );
386
387 let keep_alive_time = Duration::from_secs(1);
388 let thread_pool_builder =
389 RuntimeBuilder::new_multi_thread().keep_alive_time(keep_alive_time);
390 assert_eq!(
391 thread_pool_builder.common.keep_alive_time.unwrap(),
392 keep_alive_time
393 );
394 }
395
396 /// UT test cases for RuntimeBuilder::schedule_algo()
397 ///
398 /// # Brief
399 /// 1. schedule_algo set to FifoBound, check if it is the modified value
400 #[cfg(not(feature = "ffrt"))]
401 #[test]
ut_thread_pool_builder_schedule_algo_test()402 fn ut_thread_pool_builder_schedule_algo_test() {
403 let schedule_algo = ScheduleAlgo::FifoBound;
404 let thread_pool_builder = RuntimeBuilder::new_multi_thread().schedule_algo(schedule_algo);
405 assert_eq!(thread_pool_builder.common.schedule_algo, schedule_algo);
406 }
407 }
408