hirun 0.1.13

A concurrent framework for asynchronous programming based on event-driven, non-blocking I/O mechanism
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

use crate::event::{SchedImpl, Scheduler};
use crate::thread::{self, JoinHandle, Semaphore};
use crate::Result;
use core::marker::PhantomData;
use core::mem::ManuallyDrop;
use hipool::{Allocator, Boxed, Pool};

pub trait Runnable {
    type Context;
    fn active(&mut self, ctx: &mut Self::Context) -> Result<()>;
}

impl<'a, T: Runnable<Context = ThreadContext<'a, A>> + Send, A: Allocator> Runnable
    for Boxed<'a, T, A>
{
    type Context = T::Context;
    fn active(&mut self, ctx: &mut Self::Context) -> Result<()> {
        Runnable::active(&mut **self, ctx)
    }
}

#[repr(C)]
pub struct ThreadContext<'a, A: Allocator> {
    sched: Boxed<'a, Scheduler, A>,
}

impl<'a, A: Allocator> ThreadContext<'a, A> {
    pub fn sched(&mut self) -> &mut Scheduler {
        self.sched.as_mut()
    }
    pub fn stop(&mut self) {
        self.sched.stop();
    }
}

#[repr(C)]
pub(crate) struct Thread<'a, T, A: Allocator> {
    ctx: ThreadContext<'a, A>,
    body: T,
}

unsafe impl<T: Send, A: Allocator + Pool> Send for Thread<'_, T, A> {}

impl<T: Runnable<Context = ThreadContext<'static, A>> + Send + 'static, A: Allocator + Pool>
    Thread<'static, T, A>
{
    pub(crate) fn new_in(pool: &'static A, body: T) -> Result<ThreadProxy> {
        let sched = SchedImpl::new_in(pool)?;
        let mut this = Boxed::new_in(
            pool,
            Self {
                ctx: ThreadContext { sched },
                body,
            },
        )?;
        let sem = Semaphore::new().unwrap();
        let sem_ref = unsafe { &*(&sem as *const Semaphore) };
        let handle = thread::spawn(move || match this.active() {
            Ok(_) => {
                sem_ref.post();
                this.ctx.sched.run();
            }
            Err(err) => panic!("Thread Body active failed, errno = {err}"),
        });
        sem_ref.wait();
        Ok(ThreadProxy {
            handle: ManuallyDrop::new(handle),
            mark: PhantomData,
        })
    }
}

impl<'a, T: Runnable<Context = ThreadContext<'a, A>> + Send, A: Allocator> Thread<'a, T, A> {
    fn active(&mut self) -> Result<()> {
        self.body.active(&mut self.ctx)
    }
}

pub struct ThreadProxy {
    handle: ManuallyDrop<JoinHandle<()>>,
    mark: PhantomData<*const ()>,
}

unsafe impl Send for ThreadProxy {}

impl ThreadProxy {
    pub(crate) fn join(&mut self) {
        let _ = unsafe { ManuallyDrop::take(&mut self.handle) }.join();
    }
}