pi_async_rt/rt/
serial.rs

1use std::thread;
2use std::any::Any;
3use std::pin::Pin;
4use std::ptr::null_mut;
5use std::vec::IntoIter;
6use std::time::Duration;
7use std::future::Future;
8use std::marker::PhantomData;
9use std::ops::{Deref, DerefMut};
10use std::cell::{RefCell, UnsafeCell};
11use std::task::{Poll, Waker, Context};
12use std::io::{Error, Result, ErrorKind};
13use std::fmt::{Debug, Formatter, Result as FmtResult};
14use std::sync::{Arc, atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering}};
15
16use futures::{future::{FutureExt, LocalBoxFuture},
17              stream::{Stream, StreamExt, LocalBoxStream},
18              task::ArcWake};
19use parking_lot::{Mutex, Condvar};
20use crossbeam_queue::ArrayQueue;
21use crossbeam_channel::{Sender, Receiver, unbounded};
22use flume::{Sender as AsyncSender, Receiver as AsyncReceiver};
23#[cfg(not(target_arch = "wasm32"))]
24use polling::Poller;
25use num_cpus;
26
27use pi_cancel_timer::Timer;
28use slotmap::{Key, KeyData};
29use quanta::{Clock, Instant as QInstant};
30
31use crate::{lock::spin,
32            rt::{PI_ASYNC_LOCAL_THREAD_ASYNC_RUNTIME, TaskId, AsyncPipelineResult,
33                 TimeoutWaiter, wake_thread_waker, wake_waiting_worker,
34                 serial_local_thread::{LocalTaskRunner, LocalTaskRuntime},
35                 serial_single_thread::SingleTaskRuntime,
36                 serial_worker_thread::{WorkerTaskRunner, WorkerRuntime}}};
37
38///
39/// 顺序执行的异步任务
40///
41pub struct AsyncTask<
42    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O>,
43    O: Default + 'static = (),
44> {
45    uid:        TaskId,                                     //任务唯一id
46    future:     Mutex<Option<LocalBoxFuture<'static, O>>>,  //异步任务
47    pool:       Arc<P>,                                     //异步任务池
48    priority:   usize,                                      //异步任务优先级
49    context:    Option<UnsafeCell<Box<dyn Any>>>,           //异步任务上下文
50}
51
52unsafe impl<
53    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O>,
54    O: Default + 'static,
55> Send for AsyncTask<P, O> {}
56unsafe impl<
57    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O>,
58    O: Default + 'static,
59> Sync for AsyncTask<P, O> {}
60
61impl<
62    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O, Pool = P>,
63    O: Default + 'static,
64> ArcWake for AsyncTask<P, O> {
65    fn wake_by_ref(arc_self: &Arc<Self>) {
66        let pool = arc_self.get_pool();
67        let _ = pool.push_keep(arc_self.clone());
68
69        if let Some(waits) = pool.get_waits() {
70            //当前任务属于多线程异步运行时
71            let _ = wake_waiting_worker(waits);
72        } else {
73            //当前线程属于单线程异步运行时
74            if let Some(thread_waker) = pool.get_thread_waker() {
75                let _ = wake_thread_waker(thread_waker);
76            }
77        }
78    }
79}
80
81impl<
82    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O, Pool = P>,
83    O: Default + 'static,
84> AsyncTask<P, O> {
85    /// 构建单线程任务
86    pub fn new(uid: TaskId,
87               pool: Arc<P>,
88               priority: usize,
89               future: Option<LocalBoxFuture<'static, O>>) -> AsyncTask<P, O> {
90        AsyncTask {
91            uid,
92            future: Mutex::new(future),
93            pool,
94            priority,
95            context: None,
96        }
97    }
98
99    /// 使用指定上下文构建单线程任务
100    pub fn with_context<C: 'static>(uid: TaskId,
101                                    pool: Arc<P>,
102                                    priority: usize,
103                                    future: Option<LocalBoxFuture<'static, O>>,
104                                    context: C) -> AsyncTask<P, O> {
105        let any = Box::new(context);
106
107        AsyncTask {
108            uid,
109            future: Mutex::new(future),
110            pool,
111            priority,
112            context: Some(UnsafeCell::new(any)),
113        }
114    }
115
116    /// 使用指定异步运行时和上下文构建单线程任务
117    pub fn with_runtime_and_context<RT, C>(runtime: &RT,
118                                           priority: usize,
119                                           future: Option<LocalBoxFuture<'static, O>>,
120                                           context: C) -> AsyncTask<P, O>
121        where RT: AsyncRuntime<O, Pool = P>,
122              C: 'static {
123        let any = Box::new(context);
124
125        AsyncTask {
126            uid: runtime.alloc::<O>(),
127            future: Mutex::new(future),
128            pool: runtime.shared_pool(),
129            priority,
130            context: Some(UnsafeCell::new(any)),
131        }
132    }
133
134    /// 检查是否允许唤醒
135    pub fn is_enable_wakeup(&self) -> bool {
136        self.uid.exist_waker::<O>()
137    }
138
139    /// 获取内部任务
140    pub fn get_inner(&self) -> Option<LocalBoxFuture<'static, O>> {
141        self.future.lock().take()
142    }
143
144    /// 设置内部任务
145    pub fn set_inner(&self, inner: Option<LocalBoxFuture<'static, O>>) {
146        *self.future.lock() = inner;
147    }
148
149    /// 获取任务的所有者
150    #[inline]
151    pub fn owner(&self) -> usize {
152        unsafe {
153            *self.uid.0.get() as usize
154        }
155    }
156
157    /// 获取异步任务优先级
158    pub fn priority(&self) -> usize {
159        self.priority
160    }
161
162    //判断异步任务是否有上下文
163    pub fn exist_context(&self) -> bool {
164        self.context.is_some()
165    }
166
167    //获取异步任务上下文的只读引用
168    pub fn get_context<C: 'static>(&self) -> Option<&C> {
169        if let Some(context) = &self.context {
170            //存在上下文
171            let any = unsafe { &*context.get() };
172            return <dyn Any>::downcast_ref::<C>(&**any);
173        }
174
175        None
176    }
177
178    //获取异步任务上下文的可写引用
179    pub fn get_context_mut<C: 'static>(&self) -> Option<&mut C> {
180        if let Some(context) = &self.context {
181            //存在上下文
182            let any = unsafe { &mut *context.get() };
183            return <dyn Any>::downcast_mut::<C>(&mut **any);
184        }
185
186        None
187    }
188
189    //设置异步任务上下文，返回上一个异步任务上下文
190    pub fn set_context<C: 'static>(&self, new: C) {
191        if let Some(context) = &self.context {
192            //存在上一个上下文，则释放上一个上下文
193            let _ = unsafe { &*context.get() };
194
195            //设置新的上下文
196            let any: Box<dyn Any + 'static> = Box::new(new);
197            unsafe { *context.get() = any; }
198        }
199    }
200
201    //获取异步任务的任务池
202    pub fn get_pool(&self) -> &P {
203        self.pool.as_ref()
204    }
205}
206
207///
208/// 异步任务池
209///
210pub trait AsyncTaskPool<O: Default + 'static = ()>: Default + 'static {
211    type Pool: AsyncTaskPoolExt<O> + AsyncTaskPool<O>;
212
213    /// 获取绑定的线程唯一id
214    fn get_thread_id(&self) -> usize;
215
216    /// 获取当前异步任务池内任务数量
217    fn len(&self) -> usize;
218
219    /// 将异步任务加入异步任务池
220    fn push(&self, task: Arc<AsyncTask<Self::Pool, O>>) -> Result<()>;
221
222    /// 将异步任务加入本地异步任务池
223    fn push_local(&self, task: Arc<AsyncTask<Self::Pool, O>>) -> Result<()>;
224
225    /// 将指定了优先级的异步任务加入任务池
226    fn push_priority(&self, priority: usize, task: Arc<AsyncTask<Self::Pool, O>>) -> Result<()>;
227
228    /// 异步任务被唤醒时，将异步任务继续加入异步任务池
229    fn push_keep(&self, task: Arc<AsyncTask<Self::Pool, O>>) -> Result<()>;
230
231    /// 尝试从异步任务池中弹出一个异步任务
232    fn try_pop(&self) -> Option<Arc<AsyncTask<Self::Pool, O>>>;
233
234    /// 尝试从异步任务池中弹出所有异步任务
235    fn try_pop_all(&self) -> IntoIter<Arc<AsyncTask<Self::Pool, O>>>;
236
237    /// 获取本地线程的唤醒器
238    fn get_thread_waker(&self) -> Option<&Arc<(AtomicBool, Mutex<()>, Condvar)>> {
239        None
240    }
241}
242
243///
244/// 异步任务池扩展
245///
246pub trait AsyncTaskPoolExt<O: Default + 'static = ()>: 'static {
247    /// 设置待唤醒的工作者唤醒器队列
248    fn set_waits(&mut self,
249                 _waits: Arc<ArrayQueue<Arc<(AtomicBool, Mutex<()>, Condvar)>>>) {}
250
251    /// 获取待唤醒的工作者唤醒器队列
252    fn get_waits(&self) -> Option<&Arc<ArrayQueue<Arc<(AtomicBool, Mutex<()>, Condvar)>>>> {
253        //默认没有待唤醒的工作者唤醒器队列
254        None
255    }
256
257    /// 获取空闲的工作者的数量，这个数量大于0，表示可以新开线程来运行可分派的工作者
258    fn idler_len(&self) -> usize {
259        //默认不分派
260        0
261    }
262
263    /// 分派一个空闲的工作者
264    fn spawn_worker(&self) -> Option<usize> {
265        //默认不分派
266        None
267    }
268
269    /// 获取工作者的数量
270    fn worker_len(&self) -> usize {
271        //默认工作者数量和本机逻辑核数相同
272        #[cfg(not(target_arch = "wasm32"))]
273        return num_cpus::get();
274        #[cfg(target_arch = "wasm32")]
275        return 1;
276    }
277
278    /// 获取缓冲区的任务数量，缓冲区任务是未分配给工作者的任务
279    fn buffer_len(&self) -> usize {
280        //默认没有缓冲区
281        0
282    }
283
284    /// 设置当前绑定本地线程的唤醒器
285    fn set_thread_waker(&mut self, _thread_waker: Arc<(AtomicBool, Mutex<()>, Condvar)>) {
286        //默认不设置
287    }
288
289    /// 复制当前绑定本地线程的唤醒器
290    fn clone_thread_waker(&self) -> Option<Arc<(AtomicBool, Mutex<()>, Condvar)>> {
291        //默认不复制
292        None
293    }
294
295    /// 关闭当前工作者
296    fn close_worker(&self) {
297        //默认不允许关闭工作者
298    }
299}
300
301///
302/// 顺序执行任务的异步运行时
303///
304pub trait AsyncRuntime<O: Default + 'static = ()>: Clone + Send + Sync + 'static {
305    type Pool: AsyncTaskPoolExt<O> + AsyncTaskPool<O, Pool = Self::Pool>;
306
307    /// 共享运行时内部任务池
308    fn shared_pool(&self) -> Arc<Self::Pool>;
309
310    /// 获取当前异步运行时的唯一id
311    fn get_id(&self) -> usize;
312
313    /// 获取当前异步运行时待处理任务数量
314    fn wait_len(&self) -> usize;
315
316    /// 获取当前异步运行时任务数量
317    fn len(&self) -> usize;
318
319    /// 分配异步任务的唯一id
320    fn alloc<R: 'static>(&self) -> TaskId;
321
322    /// 派发一个指定的异步任务到异步运行时
323    fn spawn<F>(&self, future: F) -> Result<TaskId>
324        where F: Future<Output = O> + 'static;
325
326    /// 派发一个异步任务到本地异步运行时，如果本地没有本异步运行时，则会派发到当前运行时中
327    fn spawn_local<F>(&self, future: F) -> Result<TaskId>
328        where F: Future<Output = O> + 'static;
329
330    /// 派发一个指定优先级的异步任务到异步运行时
331    fn spawn_priority<F>(&self, priority: usize, future: F) -> Result<TaskId>
332        where F: Future<Output = O> + 'static;
333
334    /// 派发一个异步任务到异步运行时，并立即让出任务的当前运行
335    fn spawn_yield<F>(&self, future: F) -> Result<TaskId>
336        where F: Future<Output = O> + 'static;
337
338    /// 派发一个在指定时间后执行的异步任务到异步运行时，时间单位ms
339    fn spawn_timing<F>(&self, future: F, time: usize) -> Result<TaskId>
340        where F: Future<Output = O> + 'static;
341
342    /// 派发一个指定任务唯一id的异步任务到异步运行时
343    fn spawn_by_id<F>(&self, task_id: TaskId, future: F) -> Result<()>
344        where F: Future<Output = O> + 'static;
345
346    /// 派发一个指定任务唯一id的异步任务到本地异步运行时，如果本地没有本异步运行时，则会派发到当前运行时中
347    fn spawn_local_by_id<F>(&self, task_id: TaskId, future: F) -> Result<()>
348        where F: Future<Output = O> + 'static;
349
350    /// 派发一个指定任务唯一id和任务优先级的异步任务到异步运行时
351    fn spawn_priority_by_id<F>(&self,
352                               task_id: TaskId,
353                               priority: usize,
354                               future: F) -> Result<()>
355        where F: Future<Output = O> + 'static;
356
357    /// 派发一个指定任务唯一id的异步任务到异步运行时，并立即让出任务的当前运行
358    fn spawn_yield_by_id<F>(&self, task_id: TaskId, future: F) -> Result<()>
359        where F: Future<Output = O> + 'static;
360
361    /// 派发一个指定任务唯一id和在指定时间后执行的异步任务到异步运行时，时间单位ms
362    fn spawn_timing_by_id<F>(&self,
363                             task_id: TaskId,
364                             future: F,
365                             time: usize) -> Result<()>
366        where F: Future<Output = O> + 'static;
367
368    /// 挂起指定唯一id的异步任务
369    fn pending<Output: 'static>(&self, task_id: &TaskId, waker: Waker) -> Poll<Output>;
370
371    /// 唤醒指定唯一id的异步任务
372    fn wakeup<Output: 'static>(&self, task_id: &TaskId);
373
374    /// 挂起当前异步运行时的当前任务，并在指定的其它运行时上派发一个指定的异步任务，等待其它运行时上的异步任务完成后，唤醒当前运行时的当前任务，并返回其它运行时上的异步任务的值
375    fn wait<V: 'static>(&self) -> AsyncWait<V>;
376
377    /// 挂起当前异步运行时的当前任务，并在多个其它运行时上执行多个其它任务，其中任意一个任务完成，则唤醒当前运行时的当前任务，并返回这个已完成任务的值，而其它未完成的任务的值将被忽略
378    fn wait_any<V: 'static>(&self, capacity: usize) -> AsyncWaitAny<V>;
379
380    /// 挂起当前异步运行时的当前任务，并在多个其它运行时上执行多个其它任务，任务返回后需要通过用户指定的检查回调进行检查，其中任意一个任务检查通过，则唤醒当前运行时的当前任务，并返回这个已完成任务的值，而其它未完成或未检查通过的任务的值将被忽略，如果所有任务都未检查通过，则强制唤醒当前运行时的当前任务
381    fn wait_any_callback<V: 'static>(&self, capacity: usize) -> AsyncWaitAnyCallback<V>;
382
383    /// 构建用于派发多个异步任务到指定运行时的映射归并，需要指定映射归并的容量
384    fn map_reduce<V: 'static>(&self, capacity: usize) -> AsyncMapReduce<V>;
385
386    /// 挂起当前异步运行时的当前任务，等待指定的时间后唤醒当前任务
387    fn timeout(&self, timeout: usize) -> LocalBoxFuture<'static, ()>;
388
389    /// 立即让出当前任务的执行
390    fn yield_now(&self) -> LocalBoxFuture<'static, ()>;
391
392    /// 生成一个异步管道，输入指定流，输入流的每个值通过过滤器生成输出流的值
393    fn pipeline<S, SO, F, FO>(&self, input: S, filter: F) -> LocalBoxStream<'static, FO>
394        where S: Stream<Item = SO> + 'static,
395              SO: 'static,
396              F: FnMut(SO) -> AsyncPipelineResult<FO> + 'static,
397              FO: 'static;
398
399    /// 关闭异步运行时，返回请求关闭是否成功
400    fn close(&self) -> bool;
401}
402
403///
404/// 顺序执行的异步运行时扩展
405///
406pub trait AsyncRuntimeExt<O: Default + 'static = ()> {
407    /// 派发一个指定的异步任务到异步运行时，并指定异步任务的初始化上下文
408    fn spawn_with_context<F, C>(&self,
409                                task_id: TaskId,
410                                future: F,
411                                context: C) -> Result<()>
412        where F: Future<Output = O> + 'static,
413              C: 'static;
414
415    /// 派发一个在指定时间后执行的异步任务到异步运行时，并指定异步任务的初始化上下文，时间单位ms
416    fn spawn_timing_with_context<F, C>(&self,
417                                       task_id: TaskId,
418                                       future: F,
419                                       context: C,
420                                       time: usize) -> Result<()>
421        where F: Future<Output = O> + 'static,
422              C: 'static;
423
424    /// 立即创建一个指定任务池的异步运行时，并执行指定的异步任务，阻塞当前线程，等待异步任务完成后返回
425    fn block_on<F>(&self, future: F) -> Result<F::Output>
426        where F: Future + 'static,
427              <F as Future>::Output: Default + 'static;
428}
429
430///
431/// 异步运行时构建器
432///
433pub struct AsyncRuntimeBuilder<O: Default + 'static = ()>(PhantomData<O>);
434
435impl<O: Default + 'static> AsyncRuntimeBuilder<O> {
436    /// 构建默认的本地异步任务运行时
437    pub fn default_local_thread(name: Option<&str>,
438                                stack_size: Option<usize>) -> LocalTaskRuntime<O> {
439        let runner = LocalTaskRunner::new();
440
441        let thread_name = if let Some(name) = name {
442            name
443        } else {
444            //默认的线程名称
445            "Default-Local-RT"
446        };
447        let thread_stack_size = if let Some(size) = stack_size {
448            size
449        } else {
450            //默认的线程堆栈大小
451            2 * 1024 * 1024
452        };
453
454        runner.startup(thread_name, thread_stack_size)
455    }
456
457    /// 构建默认的工作者异步运行时
458    pub fn default_worker_thread(worker_name: Option<&str>,
459                                 worker_stack_size: Option<usize>,
460                                 worker_sleep_timeout: Option<u64>,
461                                 worker_loop_interval: Option<Option<u64>>) -> WorkerRuntime<O> {
462        let runner = WorkerTaskRunner::default();
463
464        let thread_name = if let Some(name) = worker_name {
465            name
466        } else {
467            //默认的线程名称
468            "Default-Single-Worker"
469        };
470        let thread_stack_size = if let Some(size) = worker_stack_size {
471            size
472        } else {
473            //默认的线程堆栈大小
474            2 * 1024 * 1024
475        };
476        let sleep_timeout = if let Some(timeout) = worker_sleep_timeout {
477            timeout
478        } else {
479            //默认的线程休眠时长
480            1
481        };
482        let loop_interval = if let Some(interval) = worker_loop_interval {
483            interval
484        } else {
485            //默认的线程循环间隔时长
486            None
487        };
488
489        //创建线程并在线程中执行异步运行时
490        let clock = Clock::new();
491        let runner_copy = runner.clone();
492        let rt_copy = runner.get_runtime();
493        let rt = runner.startup(
494            thread_name,
495            thread_stack_size,
496            sleep_timeout,
497            loop_interval,
498            move || {
499                let now = clock.recent();
500                match runner_copy.run_once() {
501                    Err(e) => {
502                        panic!("Run runner failed, reason: {:?}", e);
503                    },
504                    Ok(len) => {
505                        (len == 0,
506                         clock
507                             .recent()
508                             .duration_since(now))
509                    },
510                }
511            },
512            move || {
513                rt_copy.wait_len() + rt_copy.len()
514            },
515        );
516
517        rt
518    }
519
520    /// 构建自定义的本地异步任务运行时
521    #[cfg(not(target_arch = "wasm32"))]
522    pub fn custom_local_thread(name: Option<&str>,
523                               stack_size: Option<usize>,
524                               poller: Option<Arc<Poller>>,
525                               try_count: Option<usize>,
526                               timeout: Option<Duration>,) -> LocalTaskRuntime<O> {
527        let poller = if let Some(poller) = poller {
528            poller
529        } else {
530            Arc::new(Poller::new().expect("Failed to create poller"))
531        };
532        let runner = LocalTaskRunner::with_poll(poller);
533
534        let thread_name = if let Some(name) = name {
535            name
536        } else {
537            //默认的线程名称
538            "Custom-Local-RT"
539        };
540        let thread_stack_size = if let Some(size) = stack_size {
541            size
542        } else {
543            //默认的线程堆栈大小
544            2 * 1024 * 1024
545        };
546        let try_count = try_count.unwrap_or(3);
547
548        runner.startup_with_poll(
549            thread_name,
550            thread_stack_size,
551            try_count,
552            timeout
553        )
554    }
555
556    /// 构建自定义的工作者异步运行时
557    pub fn custom_worker_thread<P, F0, F1>(pool: P,
558                                           worker_handle: Arc<AtomicBool>,
559                                           worker_condvar: Arc<(AtomicBool, Mutex<()>, Condvar)>,
560                                           thread_name: &str,
561                                           thread_stack_size: usize,
562                                           sleep_timeout: u64,
563                                           loop_interval: Option<u64>,
564                                           loop_func: F0,
565                                           get_queue_len: F1) -> WorkerRuntime<O, P>
566        where P: AsyncTaskPoolExt<O> + AsyncTaskPool<O, Pool = P>,
567              F0: Fn() -> (bool, Duration) + Send + 'static,
568              F1: Fn() -> usize + Send + 'static {
569        let runner = WorkerTaskRunner::new(pool,
570                                           worker_handle,
571                                           worker_condvar);
572
573        //创建线程并在线程中执行异步运行时
574        let rt_copy = runner.get_runtime();
575        let rt = runner.startup(
576            thread_name,
577            thread_stack_size,
578            sleep_timeout,
579            loop_interval,
580            loop_func,
581            move || {
582                rt_copy.wait_len() + get_queue_len()
583            },
584        );
585
586        rt
587    }
588}
589
590/// 绑定指定异步运行时到本地线程
591pub fn bind_local_thread<O: Default + 'static>(runtime: LocalAsyncRuntime<O>) {
592    match PI_ASYNC_LOCAL_THREAD_ASYNC_RUNTIME.try_with(move |rt| {
593        let raw = Arc::into_raw(Arc::new(runtime)) as *mut LocalAsyncRuntime<O> as *mut ();
594        rt.store(raw, Ordering::Relaxed);
595    }) {
596        Err(e) => {
597            panic!("Bind single runtime to local thread failed, reason: {:?}", e);
598        },
599        Ok(_) => (),
600    }
601}
602
603/// 从本地线程解绑单线程异步任务执行器
604pub fn unbind_local_thread() {
605    let _ = PI_ASYNC_LOCAL_THREAD_ASYNC_RUNTIME.try_with(move |rt| {
606        rt.store(null_mut(), Ordering::Relaxed);
607    });
608}
609
610///
611/// 本地线程绑定的异步运行时
612///
613pub struct LocalAsyncRuntime<O: Default + 'static> {
614    inner:              *const (),                                                          //内部运行时指针
615    get_id_func:        fn(*const ()) -> usize,                                             //获取本地运行时的id的函数
616    spawn_func:         fn(*const (), LocalBoxFuture<'static, O>) -> Result<()>,            //本地派发函数
617    spawn_local_func:   fn(*const (), LocalBoxFuture<'static, O>) -> Result<()>,            //本地派发函数
618    spawn_timing_func:  fn(*const (), LocalBoxFuture<'static, O>, usize) -> Result<()>,     //定时派发函数
619    timeout_func:       fn(*const (), usize) -> LocalBoxFuture<'static, ()>,                //超时函数
620}
621
622unsafe impl<O: Default + 'static> Send for LocalAsyncRuntime<O> {}
623unsafe impl<O: Default + 'static> Sync for LocalAsyncRuntime<O> {}
624
625impl<O: Default + 'static> LocalAsyncRuntime<O> {
626    /// 创建本地线程绑定的异步运行时
627    pub fn new(inner: *const (),
628               get_id_func: fn(*const ()) -> usize,
629               spawn_func: fn(*const (), LocalBoxFuture<'static, O>) -> Result<()>,
630               spawn_timing_func: fn(*const (), LocalBoxFuture<'static, O>, usize) -> Result<()>,
631               timeout_func: fn(*const (), usize) -> LocalBoxFuture<'static, ()>) -> Self {
632        LocalAsyncRuntime {
633            inner,
634            get_id_func,
635            spawn_func,
636            spawn_local_func: spawn_func,
637            spawn_timing_func,
638            timeout_func,
639        }
640    }
641
642    /// 获取本地运行时的id
643    #[inline]
644    pub fn get_id(&self) -> usize {
645        (self.get_id_func)(self.inner)
646    }
647
648    /// 派发一个指定的异步任务到本地线程绑定的异步运行时
649    #[inline]
650    pub fn spawn<F>(&self, future: F) -> Result<()>
651        where F: Future<Output = O> + 'static {
652        (self.spawn_func)(self.inner, async move {
653            future.await
654        }.boxed_local())
655    }
656
657    /// 派发一个指定的异步任务到本地线程绑定的异步运行时
658    #[inline]
659    pub fn spawn_local<F>(&self, future: F) -> Result<()>
660    where F: Future<Output = O> + 'static {
661        (self.spawn_local_func)(self.inner, async move {
662            future.await
663        }.boxed_local())
664    }
665
666    /// 定时派发一个指定的异步任务到本地线程绑定的异步运行时
667    #[inline]
668    pub fn sapwn_timing_func<F>(&self, future: F, timeout: usize) -> Result<()>
669        where F: Future<Output = O> + 'static {
670        (self.spawn_timing_func)(self.inner,
671                                 async move {
672                                     future.await
673                                 }.boxed_local(),
674                                 timeout)
675    }
676
677    /// 挂起本地线程绑定的异步运行时的当前任务，等待指定的时间后唤醒当前任务
678    #[inline]
679    pub fn timeout(&self, timeout: usize) -> LocalBoxFuture<'static, ()> {
680        (self.timeout_func)(self.inner, timeout)
681    }
682}
683
684///
685/// 获取本地线程绑定的顺序执行任务的异步运行时
686/// 注意：O如果与本地线程绑定的运行时的O不相同，则无法获取本地线程绑定的运行时
687///
688pub fn local_serial_async_runtime<O: Default + 'static>() -> Option<Arc<LocalAsyncRuntime<O>>> {
689    match PI_ASYNC_LOCAL_THREAD_ASYNC_RUNTIME.try_with(move |ptr| {
690        let raw = ptr.load(Ordering::Relaxed) as *const LocalAsyncRuntime<O>;
691        unsafe {
692            if raw.is_null() {
693                //本地线程未绑定异步运行时
694                None
695            } else {
696                //本地线程已绑定异步运行时
697                let shared: Arc<LocalAsyncRuntime<O>> = unsafe { Arc::from_raw(raw) };
698                let result = shared.clone();
699                Arc::into_raw(shared); //避免提前释放
700                Some(result)
701            }
702        }
703    }) {
704        Err(_) => None, //本地线程没有绑定异步运行时
705        Ok(rt) => rt,
706    }
707}
708
709///
710/// 派发任务到本地线程绑定的异步运行时，如果本地线程没有异步运行时，则返回错误
711/// 注意：F::Output如果与本地线程绑定的运行时的O不相同，则无法执行指定任务
712///
713pub fn spawn_local<O, F>(future: F) -> Result<()>
714    where O: Default + 'static,
715          F: Future<Output = O> + 'static {
716    if let Some(rt) = local_serial_async_runtime::<O>() {
717        rt.spawn(future)
718    } else {
719        Err(Error::new(ErrorKind::Other, format!("Spawn task to local thread failed, reason: runtime not exist")))
720    }
721}
722
723///
724/// 获取本地线程绑定的异步运行时
725/// 注意：O如果与本地线程绑定的运行时的O不相同，则无法获取本地线程绑定的运行时
726///
727pub fn local_async_runtime<O: Default + 'static>() -> Option<Arc<LocalAsyncRuntime<O>>> {
728    match PI_ASYNC_LOCAL_THREAD_ASYNC_RUNTIME.try_with(move |ptr| {
729        let raw = ptr.load(Ordering::Relaxed) as *const LocalAsyncRuntime<O>;
730        unsafe {
731            if raw.is_null() {
732                //本地线程未绑定异步运行时
733                None
734            } else {
735                //本地线程已绑定异步运行时
736                let shared: Arc<LocalAsyncRuntime<O>> = unsafe { Arc::from_raw(raw) };
737                let result = shared.clone();
738                Arc::into_raw(shared); //避免提前释放
739                Some(result)
740            }
741        }
742    }) {
743        Err(_) => None, //本地线程没有绑定异步运行时
744        Ok(rt) => rt,
745    }
746}
747
748///
749/// 同步非阻塞的异步值，只允许被同步非阻塞的设置一次值
750///
751pub struct AsyncValue<V: 'static>(Arc<InnerAsyncValue<V>>);
752
753unsafe impl<V: 'static> Send for AsyncValue<V> {}
754unsafe impl<V: 'static> Sync for AsyncValue<V> {}
755
756impl<V: 'static> Clone for AsyncValue<V> {
757    fn clone(&self) -> Self {
758        AsyncValue(self.0.clone())
759    }
760}
761
762impl<V: Send + 'static> Debug for AsyncValue<V> {
763    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
764        write!(f,
765               "AsyncValue[status = {}]",
766               self.0.status.load(Ordering::Acquire))
767    }
768}
769
770impl<V: 'static> Future for AsyncValue<V> {
771    type Output = V;
772
773    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
774        let mut spin_len = 1;
775        while self.0.status.load(Ordering::Acquire) == 2 {
776            //还未完成设置值，则自旋等待
777            spin_len = spin(spin_len);
778        }
779
780        if self.0.status.load(Ordering::Acquire) == 3 {
781            if let Some(value) = unsafe { (*(&self).0.value.get()).take() } {
782                //异步值已就绪
783                return Poll::Ready(value);
784            }
785        }
786
787        unsafe {
788            *self.0.waker.get() = Some(cx.waker().clone()); //设置异步值的唤醒器
789        }
790
791        let mut spin_len = 1;
792        loop {
793            match self.0.status.compare_exchange(0,
794                                                 1, Ordering::Acquire,
795                                                 Ordering::Relaxed) {
796                Err(2) => {
797                    //异步值准备设置值，则稍后重试
798                    spin_len = spin(spin_len);
799                    continue;
800                },
801                Err(3) => {
802                    //异步值已就绪
803                    let value = unsafe { (*(&self).0.value.get()).take().unwrap() };
804                    return Poll::Ready(value);
805                },
806                Err(_) => {
807                    unimplemented!();
808                },
809                Ok(_) => {
810                    //异步值等待设置后唤醒
811                    return Poll::Pending;
812                },
813            }
814        }
815    }
816}
817
818/*
819* 同步非阻塞的异步值同步方法
820*/
821impl<V: 'static> AsyncValue<V> {
822    /// 构建异步值，默认值为未就绪
823    pub fn new() -> Self {
824        let inner = InnerAsyncValue {
825            value: UnsafeCell::new(None),
826            waker: UnsafeCell::new(None),
827            status: AtomicU8::new(0),
828        };
829
830        AsyncValue(Arc::new(inner))
831    }
832
833    /// 判断异步值是否已完成设置
834    pub fn is_complete(&self) -> bool {
835        self
836            .0
837            .status
838            .load(Ordering::Relaxed) == 3
839    }
840
841    /// 设置异步值
842    pub fn set(self, value: V) {
843        loop {
844            match self.0.status.compare_exchange(1,
845                                                 2,
846                                                 Ordering::Acquire,
847                                                 Ordering::Relaxed) {
848                Err(0) => {
849                    match self.0.status.compare_exchange(0,
850                                                         2,
851                                                         Ordering::Acquire,
852                                                         Ordering::Relaxed) {
853                        Err(1) => {
854                            //异步值的唤醒器已就绪，则继续尝试获取锁
855                            continue;
856                        },
857                        Err(_) => {
858                            //异步值正在设置或已完成设置，则立即返回
859                            return;
860                        },
861                        Ok(_) => {
862                            //异步值的唤醒器未就绪且获取到锁，则设置异步值后将状态设置为已完成设置，并立即返回
863                            unsafe { *self.0.value.get() = Some(value); }
864                            self.0.status.store(3, Ordering::Release);
865                            return;
866                        }
867                    }
868                },
869                Err(_) => {
870                    //异步值正在设置或已完成设置，则立即返回
871                    return;
872                },
873                Ok(_) => {
874                    //异步值的唤醒器已就绪且获取到锁，则立即退出自旋
875                    break;
876                }
877            }
878        }
879
880        //已锁且获取到锁，则设置异步值，将状态设置为已完成设置，并立即唤醒异步值
881        unsafe { *self.0.value.get() = Some(value); }
882        self.0.status.store(3, Ordering::Release);
883        let waker = unsafe { (*self.0.waker.get()).take().unwrap() };
884        waker.wake();
885    }
886}
887
888// 同步非阻塞的内部异步值，只允许被同步非阻塞的设置一次值
889pub struct InnerAsyncValue<V: 'static> {
890    value:  UnsafeCell<Option<V>>,      //值
891    waker:  UnsafeCell<Option<Waker>>,  //唤醒器
892    status: AtomicU8,                   //状态
893}
894
895///
896/// 异步非阻塞可变值的守护者
897///
898pub struct AsyncVariableGuard<'a, V: 'static> {
899    value:  &'a UnsafeCell<Option<V>>,      //值
900    waker:  &'a UnsafeCell<Option<Waker>>,  //唤醒器
901    status: &'a AtomicU8,                   //值状态
902}
903
904unsafe impl<V: 'static> Send for AsyncVariableGuard<'_, V> {}
905
906impl<V: 'static> Drop for AsyncVariableGuard<'_, V> {
907    fn drop(&mut self) {
908        //当前异步可变值已锁定，则解除锁定
909        //当前异步可变值的状态为2或6，表示当前异步可变值的唤醒器未就绪并已锁定，或当前异步可变值不需要唤醒并已完成所有修改
910        //当前异步可变值的状态为3或7，表示当前异步可变值的唤醒器已就绪并已锁定，或当前异步可变值已唤醒并已完成所有修改
911        self.status.fetch_sub(2, Ordering::Relaxed);
912    }
913}
914
915impl<V: 'static> Deref for AsyncVariableGuard<'_, V> {
916    type Target = Option<V>;
917
918    fn deref(&self) -> &Self::Target {
919        unsafe {
920            &*self.value.get()
921        }
922    }
923}
924
925impl<V: 'static> DerefMut for AsyncVariableGuard<'_, V> {
926    fn deref_mut(&mut self) -> &mut Self::Target {
927        unsafe {
928            &mut *self.value.get()
929        }
930    }
931}
932
933impl<V: 'static> AsyncVariableGuard<'_, V> {
934    /// 完成异步可变值的修改
935    pub fn finish(self) {
936        //设置异步可变值的状态为已完成修改
937        if self.status.fetch_add(4, Ordering::Relaxed) == 3 {
938            if let Some(waker) = unsafe { (&mut *self.waker.get()).take() } {
939                //当前异步可变值需要唤醒，则立即唤醒异步可变值
940                waker.wake();
941            }
942        }
943    }
944}
945
946///
947/// 异步非阻塞可变值，在完成前允许被同步非阻塞的修改多次
948///
949pub struct AsyncVariable<V: 'static>(Arc<InnerAsyncVariable<V>>);
950
951unsafe impl<V: 'static> Send for AsyncVariable<V> {}
952unsafe impl<V: 'static> Sync for AsyncVariable<V> {}
953
954impl<V: 'static> Clone for AsyncVariable<V> {
955    fn clone(&self) -> Self {
956        AsyncVariable(self.0.clone())
957    }
958}
959
960impl<V: 'static> Future for AsyncVariable<V> {
961    type Output = V;
962
963    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
964        unsafe {
965            *self.0.waker.get() = Some(cx.waker().clone()); //设置异步可变值的唤醒器准备就绪
966        }
967
968        let mut spin_len = 1;
969        loop {
970            match self.0.status.compare_exchange(0,
971                                                 1,
972                                                 Ordering::Acquire,
973                                                 Ordering::Relaxed) {
974                Err(current) if current & 4 != 0 => {
975                    //异步可变值已完成所有修改，则立即返回
976                    unsafe {
977                        let _ = (&mut *self.0.waker.get()).take(); //释放异步可变值的唤醒器
978                        return Poll::Ready((&mut *(&self).0.value.get()).take().unwrap());
979                    }
980                },
981                Err(_) => {
982                    //还未完成值修改，则自旋等待
983                    spin_len = spin(spin_len);
984                },
985                Ok(_) => {
986                    //异步可变值已挂起
987                    return Poll::Pending;
988                },
989            }
990        }
991    }
992}
993
994impl<V: 'static> AsyncVariable<V> {
995    /// 构建异步可变值，默认值为未就绪
996    pub fn new() -> Self {
997        let inner = InnerAsyncVariable {
998            value: UnsafeCell::new(None),
999            waker: UnsafeCell::new(None),
1000            status: AtomicU8::new(0),
1001        };
1002
1003        AsyncVariable(Arc::new(inner))
1004    }
1005
1006    /// 判断异步可变值是否已完成设置
1007    pub fn is_complete(&self) -> bool {
1008        self
1009            .0
1010            .status
1011            .load(Ordering::Acquire) & 4 != 0
1012    }
1013
1014    /// 锁住待修改的异步可变值，并返回当前异步可变值的守护者，如果异步可变值已完成修改则返回空
1015    pub fn lock(&self) -> Option<AsyncVariableGuard<V>> {
1016        let mut spin_len = 1;
1017        loop {
1018            match self
1019                .0
1020                .status
1021                .compare_exchange(1,
1022                                  3,
1023                                  Ordering::Acquire,
1024                                  Ordering::Relaxed) {
1025                Err(0) => {
1026                    //异步可变值还未就绪，则自旋等待
1027                    match self
1028                        .0
1029                        .status
1030                        .compare_exchange(0,
1031                                          2,
1032                                          Ordering::Acquire,
1033                                          Ordering::Relaxed) {
1034                        Err(1) => {
1035                            //异步可变值已就绪，则继续尝试获取锁
1036                            continue;
1037                        },
1038                        Err(2) => {
1039                            //异步可变值的唤醒器未就绪且已锁，但未获取到锁，则自旋等待
1040                            spin_len = spin(spin_len);
1041                        },
1042                        Err(3) => {
1043                            //异步可变值的唤醒器已就绪且已锁，但未获取到锁，则自旋等待
1044                            spin_len = spin(spin_len);
1045                        },
1046                        Err(_) => {
1047                            //已完成，则返回空
1048                            return None;
1049                        },
1050                        Ok(_) => {
1051                            //异步可变值的唤醒器未就绪且获取到锁，则返回异步可变值的守护者
1052                            let guard = AsyncVariableGuard {
1053                                value: &self.0.value,
1054                                waker: &self.0.waker,
1055                                status: &self.0.status,
1056                            };
1057
1058                            return Some(guard)
1059                        },
1060                    }
1061                },
1062                Err(2) => {
1063                    //异步可变值的唤醒器未就绪且已锁，但未获取到锁，则自旋等待
1064                    spin_len = spin(spin_len);
1065                },
1066                Err(3) => {
1067                    //异步可变值的唤醒器已就绪且已锁，但未获取到锁，则自旋等待
1068                    spin_len = spin(spin_len);
1069                },
1070                Err(_) => {
1071                    //已完成，则返回空
1072                    return None;
1073                }
1074                Ok(_) => {
1075                    //异步可变值的唤醒器已就绪且获取到锁，则返回异步可变值的守护者
1076                    let guard = AsyncVariableGuard {
1077                        value: &self.0.value,
1078                        waker: &self.0.waker,
1079                        status: &self.0.status,
1080                    };
1081
1082                    return Some(guard)
1083                },
1084            }
1085        }
1086    }
1087}
1088
1089// 内部异步非阻塞可变值，在完成前允许被同步非阻塞的修改多次
1090pub struct InnerAsyncVariable<V: 'static> {
1091    value:  UnsafeCell<Option<V>>,      //值
1092    waker:  UnsafeCell<Option<Waker>>,  //唤醒器
1093    status: AtomicU8,                   //状态
1094}
1095
1096///
1097/// 等待异步任务运行的结果
1098///
1099pub struct AsyncWaitResult<V: 'static>(pub Arc<RefCell<Option<Result<V>>>>);
1100
1101unsafe impl<V: 'static> Send for AsyncWaitResult<V> {}
1102unsafe impl<V: 'static> Sync for AsyncWaitResult<V> {}
1103
1104impl<V: 'static> Clone for AsyncWaitResult<V> {
1105    fn clone(&self) -> Self {
1106        AsyncWaitResult(self.0.clone())
1107    }
1108}
1109
1110///
1111/// 等待异步任务运行的结果集
1112///
1113pub struct AsyncWaitResults<V: 'static>(pub Arc<RefCell<Option<Vec<Result<V>>>>>);
1114
1115unsafe impl<V: 'static> Send for AsyncWaitResults<V> {}
1116unsafe impl<V: 'static> Sync for AsyncWaitResults<V> {}
1117
1118impl<V: 'static> Clone for AsyncWaitResults<V> {
1119    fn clone(&self) -> Self {
1120        AsyncWaitResults(self.0.clone())
1121    }
1122}
1123
1124///
1125/// 异步定时器任务
1126///
1127pub enum AsyncTimingTask<
1128    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O>,
1129    O: Default + 'static = (),
1130> {
1131    Pended(TaskId),                     //已挂起的定时任务
1132    WaitRun(Arc<AsyncTask<P, O>>),      //等待执行的定时任务
1133    TimeoutWake(Arc<TimeoutWaiter>),    //等待timeout到期的唤醒句柄
1134}
1135
1136///
1137/// 异步任务本地定时器
1138///
1139pub struct AsyncTaskTimer<
1140    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O>,
1141    O: Default + 'static = (),
1142> {
1143    producor:   Sender<(usize, AsyncTimingTask<P, O>)>,                     //定时任务生产者
1144    consumer:   Receiver<(usize, AsyncTimingTask<P, O>)>,                   //定时任务消费者
1145    timer:      Arc<RefCell<Timer<AsyncTimingTask<P, O>, 1000, 60, 3>>>,    //定时器
1146    clock:      Clock,                                                      //定时器时钟
1147    now:        QInstant,                                                   //当前时间
1148}
1149
1150unsafe impl<
1151    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O>,
1152    O: Default + 'static,
1153> Send for AsyncTaskTimer<P, O> {}
1154unsafe impl<
1155    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O>,
1156    O: Default + 'static,
1157> Sync for AsyncTaskTimer<P, O> {}
1158
1159impl<
1160    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O>,
1161    O: Default + 'static,
1162> AsyncTaskTimer<P, O> {
1163    /// 构建异步任务本地定时器
1164    pub fn new() -> Self {
1165        let (producor, consumer) = unbounded();
1166        let clock = Clock::new();
1167        let now = clock.recent();
1168
1169        AsyncTaskTimer {
1170            producor,
1171            consumer,
1172            timer: Arc::new(RefCell::new(Timer::<AsyncTimingTask<P, O>, 1000, 60, 3>::default())),
1173            clock,
1174            now,
1175        }
1176    }
1177
1178    /// 获取定时任务生产者
1179    #[inline]
1180    pub fn get_producor(&self) -> &Sender<(usize, AsyncTimingTask<P, O>)> {
1181        &self.producor
1182    }
1183
1184    /// 获取剩余未到期的定时器任务数量
1185    #[inline]
1186    pub fn len(&self) -> usize {
1187        let timer = self.timer.as_ref().borrow();
1188        timer.add_count() - timer.remove_count()
1189    }
1190
1191    /// 设置定时器
1192    pub fn set_timer(&self, task: AsyncTimingTask<P, O>, timeout: usize) -> usize {
1193        let current_time = self
1194            .clock
1195            .recent()
1196            .duration_since(self.now)
1197            .as_millis() as u64;
1198        self
1199            .timer
1200            .borrow_mut()
1201            .push_time(current_time + timeout as u64, task)
1202            .data()
1203            .as_ffi() as usize
1204    }
1205
1206    /// 取消定时器
1207    pub fn cancel_timer(&self, timer_ref: usize) -> Option<AsyncTimingTask<P, O>> {
1208        if let Some(item) =self
1209            .timer
1210            .borrow_mut()
1211            .cancel(KeyData::from_ffi(timer_ref as u64).into()) {
1212            Some(item)
1213        } else {
1214            None
1215        }
1216    }
1217
1218    /// 消费所有定时任务，返回定时任务数量
1219    pub fn consume(&self) -> usize {
1220        let mut len = 0;
1221        let timer_tasks = self.consumer.try_iter().collect::<Vec<(usize, AsyncTimingTask<P, O>)>>();
1222        for (timeout, task) in timer_tasks {
1223            self.set_timer(task, timeout);
1224            len += 1;
1225        }
1226
1227        len
1228    }
1229
1230    /// 判断当前时间是否有可以弹出的任务，如果有可以弹出的任务，则返回当前时间，否则返回空
1231    pub fn is_require_pop(&self) -> Option<u64> {
1232        let current_time = self
1233            .clock
1234            .recent()
1235            .duration_since(self.now)
1236            .as_millis() as u64;
1237        if self.timer.borrow_mut().is_ok(current_time) {
1238            Some(current_time)
1239        } else {
1240            None
1241        }
1242    }
1243
1244    /// 从定时器中弹出指定时间的一个到期任务
1245    pub fn pop(&self, current_time: u64) -> Option<(usize, AsyncTimingTask<P, O>)> {
1246        if let Some((key, item)) = self.timer.borrow_mut().pop_kv(current_time) {
1247            Some((key.data().as_ffi() as usize, item))
1248        } else {
1249            None
1250        }
1251    }
1252}
1253
1254///
1255/// 等待指定超时
1256///
1257pub struct AsyncWaitTimeout<
1258    RT: AsyncRuntime<O>,
1259    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O>,
1260    O: Default + 'static = (),
1261> {
1262    rt:         RT,                                     //当前运行时
1263    producor:   Sender<(usize, AsyncTimingTask<P, O>)>, //超时请求生产者
1264    timeout:    usize,                                  //超时时长，单位ms
1265    registered: AtomicBool,                             //是否已注册到定时器
1266    waiter:     Arc<TimeoutWaiter>,                     //timeout专用等待句柄
1267}
1268
1269unsafe impl<
1270    RT: AsyncRuntime<O>,
1271    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O>,
1272    O: Default + 'static,
1273> Send for AsyncWaitTimeout<RT, P, O> {}
1274unsafe impl<
1275    RT: AsyncRuntime<O>,
1276    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O>,
1277    O: Default + 'static,
1278> Sync for AsyncWaitTimeout<RT, P, O> {}
1279
1280impl<
1281    RT: AsyncRuntime<O>,
1282    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O>,
1283    O: Default + 'static,
1284> Future for AsyncWaitTimeout<RT, P, O> {
1285    type Output = ();
1286
1287    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
1288        if self.waiter.is_fired() {
1289            //已到期，则返回
1290            return Poll::Ready(());
1291        }
1292
1293        self.waiter.register(cx.waker());
1294
1295        if !self.registered.swap(true, Ordering::AcqRel) {
1296            //发送超时请求，并返回
1297            let _ = self
1298                .producor
1299                .send((self.timeout, AsyncTimingTask::TimeoutWake(self.waiter.clone())));
1300        }
1301
1302        if self.waiter.is_fired() {
1303            Poll::Ready(())
1304        } else {
1305            Poll::Pending
1306        }
1307    }
1308}
1309
1310impl<
1311    RT: AsyncRuntime<O>,
1312    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O>,
1313    O: Default + 'static,
1314> Drop for AsyncWaitTimeout<RT, P, O> {
1315    fn drop(&mut self) {
1316        self.waiter.clear_waker();
1317    }
1318}
1319
1320impl<
1321    RT: AsyncRuntime<O>,
1322    P: AsyncTaskPoolExt<O> + AsyncTaskPool<O>,
1323    O: Default + 'static,
1324> AsyncWaitTimeout<RT, P, O> {
1325    /// 构建等待指定超时任务的方法
1326    pub fn new(rt: RT,
1327               producor: Sender<(usize, AsyncTimingTask<P, O>)>,
1328               timeout: usize) -> Self {
1329        AsyncWaitTimeout {
1330            rt,
1331            producor,
1332            timeout,
1333            registered: AtomicBool::new(false), //设置初始值
1334            waiter: Arc::new(TimeoutWaiter::new()),
1335        }
1336    }
1337}
1338
1339///
1340/// 等待异步任务执行完成
1341///
1342pub struct AsyncWait<V: 'static>(AsyncWaitAny<V>);
1343
1344unsafe impl<V: 'static> Send for AsyncWait<V> {}
1345unsafe impl<V: 'static> Sync for AsyncWait<V> {}
1346
1347/*
1348* 等待异步任务执行完成同步方法
1349*/
1350impl<V: 'static> AsyncWait<V> {
1351    /// 构建等待异步任务执行完成
1352    pub(crate) fn new(inner: AsyncWaitAny<V>) -> Self {
1353        AsyncWait(inner)
1354    }
1355
1356    /// 派发指定超时时间的指定任务到指定的运行时，并返回派发是否成功
1357    pub fn spawn<RT, O, F>(&self,
1358                           rt: RT,
1359                           timeout: Option<usize>,
1360                           future: F) -> Result<()>
1361        where RT: AsyncRuntime<O>,
1362              O: Default + 'static,
1363              F: Future<Output = Result<V>> + 'static {
1364        self.0.spawn(rt.clone(), future)?;
1365
1366        if let Some(timeout) = timeout {
1367            //设置了超时时间
1368            let rt_copy = rt.clone();
1369            self.0.spawn(rt, async move {
1370                rt_copy.timeout(timeout).await;
1371
1372                //返回超时错误
1373                Err(Error::new(ErrorKind::TimedOut, format!("Time out")))
1374            })
1375        } else {
1376            //未设置超时时间
1377            Ok(())
1378        }
1379    }
1380
1381    /// 派发指定超时时间的指定任务到本地运行时，并返回派发是否成功
1382    pub fn spawn_local<O, F>(&self,
1383                             timeout: Option<usize>,
1384                             future: F) -> Result<()>
1385        where O: Default + 'static,
1386              F: Future<Output = Result<V>> + 'static {
1387        if let Some(rt) = local_serial_async_runtime::<O>() {
1388            //当前线程有绑定运行时
1389            self.0.spawn_local(future)?;
1390
1391            if let Some(timeout) = timeout {
1392                //设置了超时时间
1393                let rt_copy = rt.clone();
1394                self.0.spawn_local(async move {
1395                    rt_copy.timeout(timeout).await;
1396
1397                    //返回超时错误
1398                    Err(Error::new(ErrorKind::TimedOut, format!("Time out")))
1399                })
1400            } else {
1401                //未设置超时时间
1402                Ok(())
1403            }
1404        } else {
1405            //当前线程未绑定运行时
1406            Err(Error::new(ErrorKind::Other, format!("Spawn wait task failed, reason: local async runtime not exist")))
1407        }
1408    }
1409}
1410
1411/*
1412* 等待异步任务执行完成异步方法
1413*/
1414impl<V: 'static> AsyncWait<V> {
1415    /// 异步等待已派发任务的结果
1416    pub async fn wait_result(self) -> Result<V> {
1417        self.0.wait_result().await
1418    }
1419}
1420
1421///
1422/// 等待任意异步任务执行完成
1423///
1424pub struct AsyncWaitAny<V: 'static> {
1425    capacity:       usize,                      //派发任务的容量
1426    producor:       AsyncSender<Result<V>>,     //异步返回值生成器
1427    consumer:       AsyncReceiver<Result<V>>,   //异步返回值接收器
1428}
1429
1430unsafe impl<V: 'static> Send for AsyncWaitAny<V> {}
1431unsafe impl<V: 'static> Sync for AsyncWaitAny<V> {}
1432
1433/*
1434* 等待任意异步任务执行完成同步方法
1435*/
1436impl<V: 'static> AsyncWaitAny<V> {
1437    /// 构建等待任意异步任务执行完成
1438    pub(crate) fn new(capacity: usize,
1439                      producor: AsyncSender<Result<V>>,
1440                      consumer: AsyncReceiver<Result<V>>) -> Self {
1441        AsyncWaitAny {
1442            capacity,
1443            producor,
1444            consumer,
1445        }
1446    }
1447
1448    /// 派发指定任务到指定的运行时，并返回派发是否成功
1449    pub fn spawn<RT, O, F>(&self,
1450                           rt: RT,
1451                           future: F) -> Result<()>
1452        where RT: AsyncRuntime<O>,
1453              O: Default + 'static,
1454              F: Future<Output = Result<V>> + 'static {
1455        let producor = self.producor.clone();
1456        rt.spawn_by_id(rt.alloc::<O>(), async move {
1457                    let value = future.await;
1458                    producor.into_send_async(value).await;
1459
1460                    //返回异步任务的默认值
1461                    Default::default()
1462                })
1463    }
1464
1465    /// 派发指定任务到本地运行时，并返回派发是否成功
1466    pub fn spawn_local<F>(&self,
1467                          future: F) -> Result<()>
1468        where F: Future<Output = Result<V>> + 'static {
1469        if let Some(rt) = local_serial_async_runtime() {
1470            //本地线程有绑定运行时
1471            let producor = self.producor.clone();
1472            rt.spawn(async move {
1473                let value = future.await;
1474                producor.into_send_async(value).await;
1475            })
1476        } else {
1477            //本地线程未绑定运行时
1478            Err(Error::new(ErrorKind::Other, format!("Spawn wait any task failed, reason: local async runtime not exist")))
1479        }
1480    }
1481}
1482
1483/*
1484* 等待任意异步任务执行完成异步方法
1485*/
1486impl<V: 'static> AsyncWaitAny<V> {
1487    /// 异步等待任意已派发任务的结果
1488    pub async fn wait_result(self) -> Result<V> {
1489        match self.consumer.recv_async().await {
1490            Err(e) => {
1491                //接收错误，则立即返回
1492                Err(Error::new(ErrorKind::Other, format!("Wait any result failed, reason: {:?}", e)))
1493            },
1494            Ok(result) => {
1495                //接收成功，则立即返回
1496                result
1497            },
1498        }
1499    }
1500}
1501
1502///
1503/// 等待任意异步任务执行完成
1504///
1505pub struct AsyncWaitAnyCallback<V: 'static> {
1506    capacity:   usize,                      //派发任务的容量
1507    producor:   AsyncSender<Result<V>>,     //异步返回值生成器
1508    consumer:   AsyncReceiver<Result<V>>,   //异步返回值接收器
1509}
1510
1511unsafe impl<V: 'static> Send for AsyncWaitAnyCallback<V> {}
1512unsafe impl<V: 'static> Sync for AsyncWaitAnyCallback<V> {}
1513
1514/*
1515* 等待任意异步任务执行完成同步方法
1516*/
1517impl<V: 'static> AsyncWaitAnyCallback<V> {
1518    /// 构建等待任意异步任务执行完成
1519    pub(crate) fn new(capacity: usize,
1520                      producor: AsyncSender<Result<V>>,
1521                      consumer: AsyncReceiver<Result<V>>) -> Self {
1522        AsyncWaitAnyCallback {
1523            capacity,
1524            producor,
1525            consumer,
1526        }
1527    }
1528
1529    /// 派发指定任务到指定的运行时，并返回派发是否成功
1530    pub fn spawn<RT, O, F>(&self,
1531                           rt: RT,
1532                           future: F) -> Result<()>
1533        where RT: AsyncRuntime<O>,
1534              O: Default + 'static,
1535              F: Future<Output = Result<V>> + 'static {
1536        let producor = self.producor.clone();
1537        rt.spawn_by_id(rt.alloc::<O>(), async move {
1538                    let value = future.await;
1539                    producor.into_send_async(value).await;
1540
1541                    //返回异步任务的默认值
1542                    Default::default()
1543                })
1544    }
1545
1546    /// 派发指定任务到本地运行时，并返回派发是否成功
1547    pub fn spawn_local<F>(&self,
1548                          future: F) -> Result<()>
1549        where F: Future<Output = Result<V>> + 'static {
1550        if let Some(rt) = local_serial_async_runtime() {
1551            //当前线程有绑定运行时
1552            let producor = self.producor.clone();
1553            rt.spawn(async move {
1554                let value = future.await;
1555                producor.into_send_async(value).await;
1556            })
1557        } else {
1558            //当前线程未绑定运行时
1559            Err(Error::new(ErrorKind::Other, format!("Spawn wait any task failed by callback, reason: current async runtime not exist")))
1560        }
1561    }
1562}
1563
1564/*
1565* 等待任意异步任务执行完成异步方法
1566*/
1567impl<V: 'static> AsyncWaitAnyCallback<V> {
1568    /// 异步等待满足用户回调需求的已派发任务的结果
1569    pub async fn wait_result(mut self,
1570                             callback: impl Fn(&Result<V>) -> bool + 'static) -> Result<V> {
1571        let checker = create_checker(self.capacity, callback);
1572        loop {
1573            match self.consumer.recv_async().await {
1574                Err(e) => {
1575                    //接收错误，则立即返回
1576                    return Err(Error::new(ErrorKind::Other, format!("Wait any result failed by callback, reason: {:?}", e)));
1577                },
1578                Ok(result) => {
1579                    //接收成功，则检查是否立即返回
1580                    if checker(&result) {
1581                        //检查通过，则立即唤醒等待的任务，否则等待其它任务唤醒
1582                        return result;
1583                    }
1584                },
1585            }
1586        }
1587    }
1588}
1589
1590// 根据用户提供的回调，生成检查器
1591fn create_checker<V, F>(len: usize,
1592                        callback: F) -> Arc<dyn Fn(&Result<V>) -> bool + 'static>
1593    where V: 'static,
1594          F: Fn(&Result<V>) -> bool + 'static {
1595    let mut check_counter = AtomicUsize::new(len); //初始化检查计数器
1596    Arc::new(move |result| {
1597        if check_counter.fetch_sub(1, Ordering::SeqCst) == 1 {
1598            //最后一个任务的检查，则忽略用户回调，并立即返回成功
1599            true
1600        } else {
1601            //不是最后一个任务的检查，则调用用户回调，并根据用户回调确定是否成功
1602            callback(result)
1603        }
1604    })
1605}
1606
1607///
1608/// 异步映射归并
1609///
1610pub struct AsyncMapReduce<V: 'static> {
1611    count:          usize,                              //派发的任务数量
1612    capacity:       usize,                              //派发任务的容量
1613    producor:       AsyncSender<(usize, Result<V>)>,    //异步返回值生成器
1614    consumer:       AsyncReceiver<(usize, Result<V>)>,  //异步返回值接收器
1615}
1616
1617unsafe impl<V: 'static> Send for AsyncMapReduce<V> {}
1618
1619/*
1620* 异步映射归并同步方法
1621*/
1622impl<V: 'static> AsyncMapReduce<V> {
1623    /// 构建异步映射归并
1624    pub(crate) fn new(count: usize,
1625                      capacity: usize,
1626                      producor: AsyncSender<(usize, Result<V>)>,
1627                      consumer: AsyncReceiver<(usize, Result<V>)>) -> Self {
1628        AsyncMapReduce {
1629            count,
1630            capacity,
1631            producor,
1632            consumer,
1633        }
1634    }
1635
1636    /// 映射指定任务到指定的运行时，并返回任务序号
1637    pub fn map<RT, O, F>(&mut self, rt: RT, future: F) -> Result<usize>
1638        where RT: AsyncRuntime<O>,
1639              O: Default + 'static,
1640              F: Future<Output = Result<V>> + 'static {
1641        if self.count >= self.capacity {
1642            //已派发任务已达可派发任务的限制，则返回错误
1643            return Err(Error::new(ErrorKind::Other, format!("Map task to runtime failed, capacity: {}, reason: out of capacity", self.capacity)));
1644        }
1645
1646        let index = self.count;
1647        let producor = self.producor.clone();
1648        rt.spawn(async move {
1649                    let value = future.await;
1650                    producor.into_send_async((index, value)).await;
1651
1652                    //返回异步任务的默认值
1653                    Default::default()
1654                })?;
1655
1656        self.count += 1; //派发任务成功，则计数
1657        Ok(index)
1658    }
1659}
1660
1661/*
1662* 异步映射归并异步方法
1663*/
1664impl<V: 'static> AsyncMapReduce<V> {
1665    /// 归并所有派发的任务
1666    pub async fn reduce(self, order: bool) -> Result<Vec<Result<V>>> {
1667        let mut count = self.count;
1668        let mut results = Vec::with_capacity(count);
1669        while count > 0 {
1670            match self.consumer.recv_async().await {
1671                Err(e) => {
1672                    //接收错误，则立即返回
1673                    return Err(Error::new(ErrorKind::Other, format!("Reduce result failed, reason: {:?}", e)));
1674                },
1675                Ok((index, result)) => {
1676                    //接收成功，则继续
1677                    results.push((index, result));
1678                    count -= 1;
1679                },
1680            }
1681        }
1682
1683        if order {
1684            //需要对结果集进行排序
1685            results.sort_by_key(|(key, _value)| {
1686                key.clone()
1687            });
1688        }
1689        let (_, values) = results
1690            .into_iter()
1691            .unzip::<usize, Result<V>, Vec<usize>, Vec<Result<V>>>();
1692
1693        Ok(values)
1694    }
1695}
1696
1697///
1698/// 派发一个工作线程
1699/// 返回线程的句柄，可以通过句柄关闭线程
1700/// 线程在没有任务可以执行时会休眠，当派发任务或唤醒任务时会自动唤醒线程
1701///
1702pub fn spawn_worker_thread<F0, F1>(thread_name: &str,
1703                                   thread_stack_size: usize,
1704                                   thread_handler: Arc<AtomicBool>,
1705                                   thread_waker: Arc<(AtomicBool, Mutex<()>, Condvar)>, //用于唤醒运行时所在线程的条件变量
1706                                   sleep_timeout: u64,                                  //休眠超时时长，单位毫秒
1707                                   loop_interval: Option<u64>,                          //工作者线程循环的间隔时长，None为无间隔，单位毫秒
1708                                   loop_func: F0,
1709                                   get_queue_len: F1) -> Arc<AtomicBool>
1710    where F0: Fn() -> (bool, Duration) + Send + 'static,
1711          F1: Fn() -> usize + Send + 'static {
1712    let thread_status_copy = thread_handler.clone();
1713
1714    thread::Builder::new()
1715        .name(thread_name.to_string())
1716        .stack_size(thread_stack_size)
1717        .spawn(move || {
1718            let mut sleep_count = 0;
1719
1720            while thread_handler.load(Ordering::Relaxed) {
1721                let (is_no_task, run_time) = loop_func();
1722
1723                if is_no_task {
1724                    //当前没有任务
1725                    if sleep_count > 1 {
1726                        //当前没有任务连续达到2次，则休眠线程
1727                        sleep_count = 0; //重置休眠计数
1728                        let (is_sleep, lock, condvar) = &*thread_waker;
1729                        if get_queue_len() > 0 {
1730                            //当前有任务，则继续工作
1731                            continue;
1732                        }
1733
1734                        {
1735                            let _locked = lock.lock();
1736                            if !is_sleep.load(Ordering::Acquire) {
1737                                //发布休眠状态，外部唤醒端会在同一把锁内确认后再notify
1738                                is_sleep.store(true, Ordering::Release);
1739                            }
1740                        }
1741
1742                        if get_queue_len() > 0 {
1743                            //发布休眠后再次检查任务，避免外部唤醒落在发布窗口内
1744                            is_sleep.store(false, Ordering::Release);
1745                            continue;
1746                        }
1747
1748                        let mut locked = lock.lock();
1749                        if is_sleep.load(Ordering::Acquire) {
1750                            let _ = condvar.wait_for(
1751                                &mut locked,
1752                                Duration::from_millis(sleep_timeout),
1753                            );
1754                        }
1755                        is_sleep.store(false, Ordering::Release);
1756
1757                        continue; //唤醒后立即尝试执行任务
1758                    }
1759
1760                    sleep_count += 1; //休眠计数
1761                    if let Some(interval) = &loop_interval {
1762                        //设置了循环间隔时长
1763                        if let Some(remaining_interval) = Duration::from_millis(*interval).checked_sub(run_time){
1764                            //本次运行少于循环间隔，则休眠剩余的循环间隔，并继续执行任务
1765                            thread::sleep(remaining_interval);
1766                        }
1767                    }
1768                } else {
1769                    //当前有任务
1770                    sleep_count = 0; //重置休眠计数
1771                    if let Some(interval) = &loop_interval {
1772                        //设置了循环间隔时长
1773                        if let Some(remaining_interval) = Duration::from_millis(*interval).checked_sub(run_time){
1774                            //本次运行少于循环间隔，则休眠剩余的循环间隔，并继续执行任务
1775                            thread::sleep(remaining_interval);
1776                        }
1777                    }
1778                }
1779            }
1780    });
1781
1782    thread_status_copy
1783}
1784
1785/// 唤醒工作者所在线程，如果线程当前正在运行，则忽略
1786pub fn wakeup_worker_thread<O, P>(worker_waker: &Arc<(AtomicBool, Mutex<()>, Condvar)>,
1787                                  rt: &SingleTaskRuntime<O, P>)
1788    where O: Default + 'static,
1789          P: AsyncTaskPoolExt<O> + AsyncTaskPool<O, Pool = P> {
1790    //检查工作者所在线程是否需要唤醒
1791    if worker_waker.0.load(Ordering::Relaxed) && rt.len() > 0 {
1792        let _ = wake_thread_waker(worker_waker);
1793    }
1794}
pi_async_rt/rt/serial.rs

pi_async_rt/rt/
serial.rs
