use crate::event::{BoxedScheduler, Scheduler};
use crate::thread::{self, JoinHandle, Semaphore};
use crate::Result;
use core::marker::PhantomData;
use core::mem::ManuallyDrop;
use hipool::{Allocator, Boxed, MemPool};
pub trait Runnable {
fn active(&mut self, ctx: &mut ThreadContext) -> Result<()>;
}
impl<T: Runnable + Send, A: Allocator> Runnable for Boxed<'_, T, A> {
fn active(&mut self, ctx: &mut ThreadContext) -> Result<()> {
Runnable::active(&mut **self, ctx)
}
}
type Pool = &'static MemPool;
#[repr(C)]
pub struct ThreadContext {
sched: BoxedScheduler,
}
impl ThreadContext {
pub fn sched(&mut self) -> &mut Scheduler {
&mut self.sched
}
pub fn stop(&mut self) {
self.sched.stop();
}
}
#[repr(C)]
pub(crate) struct Thread<T> {
ctx: ThreadContext,
body: T,
}
unsafe impl<T: Send> Send for Thread<T> {}
impl<T: Runnable + Send + 'static> Thread<T> {
pub(crate) fn new_in(pool: Pool, body: T) -> Result<ThreadProxy> {
let sched = Scheduler::new_in(pool)?;
let mut this = Boxed::new_in(
pool,
Self {
ctx: ThreadContext { sched },
body,
},
)?
.into();
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<T: Runnable + Send> Thread<T> {
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();
}
}