use super::{Attr, Config, GroupSender, Task, TaskQueue, TaskRef, Worker};
use crate::channel::mpsc;
use crate::thread::{self, JoinHandle, Mutex};
use crate::{Error, Result};
use core::cell::UnsafeCell;
use core::future::Future;
use core::result;
use core::sync::atomic::{AtomicBool, Ordering};
use hipool::{Arc, Boxed, BoxedMemPool, MemPool, NullAlloc, PoolAlloc};
pub(crate) type MpscSender = mpsc::Sender<'static, Signo, MemPool>;
pub(crate) type MpscReceiver = mpsc::Receiver<'static, Signo, MemPool>;
#[allow(non_camel_case_types)]
pub enum Signo {
SIG_STOP,
SIG_NAME(&'static str),
}
static GROUPS: GroupArray = GroupArray::new();
const NON_GROUP: Option<BoxedGroup> = None;
struct GroupArray {
groups: UnsafeCell<[Option<BoxedGroup>; 256]>,
lock: Mutex<()>,
}
unsafe impl Sync for GroupArray {}
impl GroupArray {
const fn new() -> Self {
Self {
groups: UnsafeCell::new([NON_GROUP; 256]),
lock: Mutex::new(()),
}
}
fn get(&self, id: u8) -> Option<&Group> {
let groups = unsafe { &*self.groups.get() };
groups[id as usize].as_deref()
}
fn active(&self, grp: BoxedGroup) -> Result<()> {
let id = grp.id() as usize;
let _ = self.lock.lock();
let groups = unsafe { &mut *self.groups.get() };
if groups[id].is_some() {
return Err(Error::new(libc::EEXIST));
}
groups[id].replace(grp);
Ok(())
}
unsafe fn reactive(&self, grp: BoxedGroup) -> Result<()> {
if let Some(worker) = Worker::current() {
if worker.group_id() == grp.id() {
return Err(Error::new(libc::EPERM));
}
}
let id = grp.id() as usize;
let _ = self.lock.lock();
let groups = unsafe { &mut *self.groups.get() };
if let Some(ref mut old_grp) = groups[id] {
old_grp.stop();
}
groups[id].replace(grp);
Ok(())
}
}
#[repr(C)]
pub(crate) struct WorkerProxy {
id: usize,
mpscs: ArcMpsc,
stopped: AtomicBool,
thread: Option<JoinHandle<()>>,
}
unsafe impl Sync for WorkerProxy {}
type BoxedWorkers = Boxed<'static, [Option<WorkerProxy>], MemPool>;
impl WorkerProxy {
fn stop(&mut self) {
if !self.stopped.swap(true, Ordering::Relaxed) {
if let Some(handle) = self.thread.take() {
self.signal(Signo::SIG_STOP);
let _ = handle.join();
}
}
}
fn signal(&self, sig: Signo) {
self.mpscs[self.id].send(sig);
}
fn try_signal(&self, sig: Signo) -> result::Result<(), Signo> {
self.mpscs[self.id].try_send(sig)
}
}
impl Drop for WorkerProxy {
fn drop(&mut self) {
self.stop();
}
}
type ArcQueue = Arc<'static, [TaskQueue], MemPool>;
type ArcMpsc = Arc<'static, [MpscSender], MemPool>;
#[allow(dead_code)]
#[repr(C)]
pub struct Group {
sender: GroupSender,
queues: ArcQueue,
mpscs: ArcMpsc,
workers: BoxedWorkers,
conf: Config,
pool: BoxedMemPool<PoolAlloc>,
}
unsafe impl Sync for Group {}
type BoxedGroup = Boxed<'static, Group, NullAlloc>;
type GroupRef = &'static Group;
impl Group {
pub(crate) fn new(conf: &Config) -> Result<BoxedGroup> {
let conf = Self::conf(conf);
let pool = MemPool::new_boxed(0)?.leak().0;
let boxed_pool = unsafe { MemPool::from_raw(pool) };
let nth = conf.nth;
let queues = Arc::new_slice_then_in(pool, nth, |_| Ok(TaskQueue::new()))?;
let mut channels = Channels::new(pool, nth, conf.qlen)?;
let mpscs = Arc::new_slice_then_in(pool, nth, |n| Ok(channels.mpsc_send(n)))?;
let mut workers = Boxed::new_slice_then_in::<Option<WorkerProxy>, _>(pool, nth, |_| Ok(None))?;
for n in 0..nth {
let id = conf.id;
let channel = channels.mpsc_recv(n);
let queue = queues.clone();
let thread = Worker::active(id, n as u16, channel, queue)?;
let proxy = WorkerProxy {
id: n,
mpscs: mpscs.clone(),
thread: Some(thread),
stopped: AtomicBool::new(false),
};
let _ = proxy.try_signal(Signo::SIG_NAME(conf.name));
workers[n].replace(proxy);
}
let group = Boxed::new_in(
pool,
Self {
sender: GroupSender::new(),
queues,
mpscs,
workers,
conf,
pool: boxed_pool,
},
)
.map(|boxed| boxed.leak_boxed())?;
Ok(group)
}
pub(crate) fn active(grp: BoxedGroup) -> Result<()> {
GROUPS.active(grp)
}
pub(crate) unsafe fn reactive(grp: BoxedGroup) -> Result<()> {
GROUPS.reactive(grp)
}
}
impl Group {
pub fn get(id: u8) -> GroupRef {
return GROUPS.get(id).expect("Runtime {id} don't be inited");
}
pub(crate) fn spawn<T: Future>(&self, future: T, attr: &Attr) -> Result<TaskRef> {
let mut task = Task::new(future, attr)?;
let queue = self.queues.as_ref();
if attr.hash == 0 {
self.sender.send(queue, task.clone());
} else {
let id = attr.hash % queue.len();
task.status.set_local(id as u16);
queue[id].push(task.clone());
}
Ok(task)
}
pub(crate) fn sched(&self, task: TaskRef, local: Option<u16>) {
let queue = self.queues.as_ref();
#[allow(clippy::unnecessary_unwrap)]
if local.is_none() {
self.sender.send(queue, task);
} else {
queue[local.unwrap() as usize].push(task);
}
}
pub(crate) fn id(&self) -> u8 {
self.conf.id
}
}
impl Group {
fn stop(&mut self) {
for worker in &mut *self.workers {
if let Some(worker) = worker.as_mut() {
worker.stop();
}
}
}
fn conf(conf: &Config) -> Config {
Config {
qlen: Self::get_qlen(conf.qlen),
nth: Self::get_nth(conf.nth),
..conf.clone()
}
}
fn get_qlen(qlen: usize) -> usize {
if qlen > 0 {
qlen
} else {
8
}
}
fn get_nth(nth: usize) -> usize {
if nth == 0 {
thread::get_cpu_count()
} else if nth >= ((u16::MAX as usize) >> 1) {
u16::MAX as usize >> 1
} else {
nth
}
}
}
struct Channels {
mpsc_send: Boxed<'static, [Option<MpscSender>], PoolAlloc>,
mpsc_recv: Boxed<'static, [Option<MpscReceiver>], PoolAlloc>,
}
impl Channels {
fn new(pool: &'static MemPool, nth: usize, qlen: usize) -> Result<Self> {
let mut mpsc_send =
Boxed::new_slice_then_in::<Option<MpscSender>, _>(&PoolAlloc, nth, |_| Ok(None))?;
let mut mpsc_recv =
Boxed::new_slice_then_in::<Option<MpscReceiver>, _>(&PoolAlloc, nth, |_| Ok(None))?;
for (send, recv) in mpsc_send.iter_mut().zip(mpsc_recv.iter_mut()) {
let (mpsc_send, mpsc_recv) = mpsc::channel_in(pool, qlen)?;
send.replace(mpsc_send);
recv.replace(mpsc_recv);
}
Ok(Self {
mpsc_send,
mpsc_recv,
})
}
fn mpsc_recv(&mut self, n: usize) -> MpscReceiver {
self.mpsc_recv[n].take().unwrap()
}
fn mpsc_send(&mut self, n: usize) -> MpscSender {
self.mpsc_send[n].as_ref().unwrap().clone()
}
}