my-ecs 0.1.2

use super::{
    comm::{
        self, CommandSender, GroupSignal, ParkingReceiver, ParkingSender, SubComm, WORKER_ID_GEN,
    },
    par::FnContext,
    task::{AsyncTask, ParTask, SysTask, Task},
};
use crate::{
    cb_deque::{Injector, Steal},
    ecs::{
        cache::{CacheItem, RefreshCacheStorage},
        cmd::CommandObject,
        sys::system::{RawSystemCycleIter, SystemData, SystemGroup, SystemId},
        wait::WaitQueues,
        worker::{Message, PanicMessage, Work, WorkerId},
    },
    utils::ds::{
        Array, ListPos, ManagedConstPtr, ManagedMutPtr, NonNullExt, SetValueList, UnsafeFuture,
        WakeSend,
    },
    FxBuildHasher, MAX_GROUP,
};
use my_utils::{ds::SignalSlot, Or};
use std::{
    any::Any,
    cell::{Cell, UnsafeCell},
    collections::HashMap,
    hash::BuildHasher,
    marker::PhantomPinned,
    mem,
    ops::{Deref, IndexMut},
    pin::Pin,
    ptr::NonNull,
    sync::{
        atomic::{AtomicU32, Ordering},
        Arc,
    },
    thread::{self, Thread, ThreadId},
    time::Duration,
};

#[derive(Debug)]
pub(crate) struct Scheduler<W: Work + 'static> {
    wgroups: Array<WorkGroup<W>, MAX_GROUP>,

    waits: WaitQueues,

    /// System run record.
    record: ScheduleRecord,

    /// A list holding pending tasks due to data dependency.
    nor_pendings: Array<Pending, MAX_GROUP>,

    /// A list holding pending tasks due to data dependency.
    dedi_pendings: Array<Pending, MAX_GROUP>,

    /// Ready dedicated tasks.
    dedi_tx: ParkingSender<Task>,
    dedi_rx: ParkingReceiver<Task>,

    /// Channel sending messages to main worker.
    tx_msg: ParkingSender<Message>,

    /// Channel receiving messages from sub workers.
    rx_msg: ParkingReceiver<Message>,

    /// Channel sending commands to main worker.
    tx_cmd: CommandSender,

    /// Main worker id.
    wid: WorkerId,

    /// To avoid frequent creation.
    waker: MainWaker,

    dedi_fut_cnt: Arc<AtomicU32>,
}

impl<W: Work + 'static> Scheduler<W> {
    pub(crate) fn new(mut workers: Vec<W>, groups: &[usize], tx_cmd: CommandSender) -> Self {
        assert_eq!(workers.len(), groups.iter().sum::<usize>());

        let num_groups = groups.len();
        let pending_limit: usize = workers.len();

        let (nor_pendings, dedi_pendings) =
            (0..num_groups).fold((Array::new(), Array::new()), |(mut nor, mut dedi), _| {
                nor.push(Pending::new(pending_limit));
                dedi.push(Pending::new(pending_limit));
                (nor, dedi)
            });

        let (tx_msg, rx_msg) = comm::parking_channel(thread::current());

        let wgroups = (0..num_groups).fold(Array::new(), |mut wgs, i| {
            // Splits off left piece from entire worker array.
            let mut left = workers.split_off(groups[i]); // right
            mem::swap(&mut workers, &mut left); // right -> left

            // Creates sub context group and initializes it.
            let mut wg = WorkGroup::new(i as u16, left, &tx_msg, &tx_cmd);
            wg.initialize(&tx_msg, &rx_msg);
            wgs.push(wg);

            wgs
        });

        let (dedi_tx, dedi_rx) = comm::parking_channel(thread::current());
        let waker = MainWaker::new(dedi_tx.clone());

        let id = WORKER_ID_GEN.with(Cell::get);
        WORKER_ID_GEN.with(|gen| gen.set(id + 1));
        let wid = WorkerId::new(
            id,
            WorkerId::dummy().group_index(),
            WorkerId::dummy().worker_index(),
        );

        // Main worker may have multiple ECS instances, so that setting thread local variable is not
        // possible.
        // WORKER_ID.set(wid);

        Self {
            wgroups,
            waits: WaitQueues::new(),
            record: ScheduleRecord::new(),
            nor_pendings,
            dedi_pendings,
            dedi_tx,
            dedi_rx,
            tx_msg,
            rx_msg,
            tx_cmd,
            wid,
            waker,
            dedi_fut_cnt: Arc::new(AtomicU32::new(0)),
        }
    }

    pub(crate) fn num_groups(&self) -> usize {
        self.wgroups.len()
    }

    pub(crate) fn num_workers(&self) -> usize {
        self.wgroups.iter().map(WorkGroup::len).sum()
    }

    /// Returns number of open sub workers.
    pub(crate) fn is_work_groups_exhausted(&self) -> bool {
        self.wgroups.iter().all(|wg| wg.is_exhausted())
    }

    pub(crate) fn dedicated_future_count(&self) -> u32 {
        self.dedi_fut_cnt.load(Ordering::Relaxed)
    }

    pub(crate) fn wait_exhausted(&self) {
        for wg in self.wgroups.iter() {
            wg.wait_exhausted();
        }
    }

    pub(crate) fn wait_receiving_dedicated_task(&self) {
        self.dedi_rx.wait_timeout(Duration::MAX);
    }

    /// Destroys this scheduler and then returns internal worker array.
    //
    // Scheduler cannot exist without workers, so consumes itself.
    pub(crate) fn take_workers(mut self) -> Vec<W> {
        self.wgroups.iter_mut().fold(Vec::new(), |mut acc, wgroup| {
            acc.append(&mut wgroup.take_workers());
            acc
        })
    }

    pub(crate) fn get_wait_queues_mut(&mut self) -> &mut WaitQueues {
        &mut self.waits
    }

    pub(crate) fn get_dedicated_send_task_queue(&self) -> &ParkingSender<Task> {
        &self.dedi_tx
    }

    pub(crate) fn get_send_message_queue(&self) -> &ParkingSender<Message> {
        &self.tx_msg
    }

    pub(crate) fn get_dedicated_future_count(&self) -> &Arc<AtomicU32> {
        &self.dedi_fut_cnt
    }

    fn work_one(&mut self) {
        if let Ok(task) = self.dedi_rx.try_recv() {
            // NOTE: Panics can occur here.
            // TODO: Handle panic?
            match task {
                Task::System(task) => self.work_for_system_task(task),
                Task::Parallel(task) => self.work_for_parallel_task(task),
                Task::Async(task) => self.work_for_async_task(task),
            }
        }
    }

    fn work_for_system_task(&self, task: SysTask) {
        let sid = task.sid();

        let resp = match task.execute(self.wid) {
            Ok(_) => Message::Fin(self.wid, sid),
            Err(payload) => Message::Panic(PanicMessage {
                wid: self.wid,
                sid,
                payload,
                unrecoverable: false,
            }),
        };

        // Even if the main worker, it needs to send Fin message to release data dependency.
        self.tx_msg.send(resp).unwrap();
    }

    fn work_for_parallel_task(&self, task: ParTask) {
        task.execute(self.wid, FnContext::NOT_MIGRATED);
    }

    fn work_for_async_task(&self, task: AsyncTask) {
        // Sets waker if needed.
        unsafe {
            if !task.will_wake(&self.waker) {
                task.set_waker(self.waker.clone());
            }
        }

        // Executes.
        let on_ready = |ready| {
            // Decreases future count.
            self.dedi_fut_cnt.fetch_sub(1, Ordering::Relaxed);

            // Sends the ready future as a command.
            let cmd = CommandObject::Future(ready);
            self.tx_cmd.send_or_cancel(cmd);
        };
        task.execute(self.wid, on_ready);
    }

    pub(crate) fn execute_all<T, S>(&mut self, sgroups: &mut T, cache: &mut RefreshCacheStorage<S>)
    where
        T: IndexMut<usize, Output = SystemGroup>,
        S: BuildHasher + Default,
    {
        // # Procedures
        // 1. Opens sub workers through `WorkGroup`s.
        // 2. Runs each group.
        // 3. Cleans up.

        // Preparation.
        let num_groups = self.wgroups.len();
        let mut lives: Array<bool, MAX_GROUP> = Array::new();
        let mut cycles: Array<RawSystemCycleIter, MAX_GROUP> = Array::new();
        for i in 0..num_groups {
            lives.push(sgroups[i].len_active() > 0);
            cycles.push(sgroups[i].get_active_mut().iter_begin().into_raw());
        }
        let tickables = lives.clone();
        let mut panicked = Vec::new();

        // 1. Opens work groups.
        for (i, _) in lives.iter().enumerate().filter(|(_, live)| **live) {
            self.wgroups[i].open();
        }

        // 2. Runs each group. This procedure follows the order below.
        // (1) Tries to pull system tasks as many as possible.
        // (2) If there are dedicated tasks, performs just one task. Because we want to make sub
        //     workers busy rather than doing something.
        // (3) Waits for messages from work groups.
        loop {
            // (1) Pulls system tasks from cycles and inject them into work groups.
            for (i, live) in lives.iter_mut().enumerate().filter(|(_, live)| **live) {
                let pull_end = self.pull_many(&mut cycles[i], cache) == PullRes::Empty;
                let no_pending = !self.has_pending(i);
                if pull_end && no_pending {
                    self.wgroups[i].close();
                    *live = false;
                }
            }

            // (2) If we have dedicated task, performs just one task.
            self.work_one();
            if !self.dedi_rx.is_empty() {
                continue;
            }

            // (3) If any cycle is not empty, waits for messages.
            if lives.iter().any(|&live| live) {
                self.wait(&mut cycles, cache, &mut panicked);
            } else {
                self.consume_messages(cache, &mut panicked);
                break;
            }
        }

        // 3. Gets cleaned up.
        // - Marks panicked systems with poison.
        // - Increases tick.
        // - Resets run record.
        while let Some((sid, payload)) = panicked.pop() {
            sgroups[sid.group_index() as usize]
                .poison(&sid, payload)
                .unwrap();
        }
        for i in 0..num_groups {
            if tickables[i] {
                sgroups[i].tick();
            }
        }
        self.record.clear();

        #[cfg(feature = "check")]
        self.validate_clean();
    }

    fn pull_many<'s, S: BuildHasher + Default>(
        &mut self,
        cycle: &mut RawSystemCycleIter,
        cache: &mut RefreshCacheStorage<'s, S>,
    ) -> PullRes {
        loop {
            match self.pull_one(cycle, cache) {
                PullRes::Empty => return PullRes::Empty,
                PullRes::Success => {}
                PullRes::PendingFull => return PullRes::PendingFull,
            }
        }
    }

    fn pull_one<'s, S: BuildHasher + Default>(
        &mut self,
        cycle: &mut RawSystemCycleIter,
        cache: &mut RefreshCacheStorage<'s, S>,
    ) -> PullRes {
        // There are four possible cases here.
        // 1. No tasks to pull      : Does nothing.
        // 2. Data dependency? No   : Appends it onto (unbounded) ready queue.
        // 3. Data dependency? Yes  : Appends it onto pending queue.
        // 4. Pending queue is full : Does nothing.

        // 1. No tasks.
        if cycle.position().is_end() {
            return PullRes::Empty;
        }

        // Takes out one system from the cycle list.
        let sdata = unsafe { cycle.get().unwrap() };
        let sid = sdata.id();
        let gi = sid.group_index() as usize;
        let (pending, target) = if sdata.flags().is_dedi() {
            let ptr = unsafe { NonNull::new_unchecked(&mut self.dedi_tx as *mut _) };
            (&mut self.dedi_pendings[gi], Or::B(ptr))
        } else {
            let ptr = unsafe { NonNull::new_unchecked(&mut self.wgroups[gi] as *mut _) };
            (&mut self.nor_pendings[gi], Or::A(ptr))
        };

        // 2. No data dependency : Ready queue.
        if let Some(cache) = Helper::update_task(&mut self.waits, sdata, cache) {
            self.record.insert(sid, RunResult::Injected);
            Helper::move_ready_system(target, sdata, cache);
            unsafe { cycle.next() };
            PullRes::Success
        }
        // 3. Data dependency detected : Pending queue.
        else if pending.push(cycle.position()) {
            self.record.insert(sid, RunResult::Injected);
            unsafe { cycle.next() };
            PullRes::Success
        }
        // 4. Pending queue is full : Ignore.
        else {
            PullRes::PendingFull
        }
    }

    fn wait<'s, S: BuildHasher + Default + 's>(
        &mut self,
        cycles: &mut Array<RawSystemCycleIter, MAX_GROUP>,
        cache: &mut RefreshCacheStorage<'_, S>,
        panicked: &mut Vec<(SystemId, Box<dyn Any + Send>)>,
    ) {
        // Consumes buffered messages as many as possible.
        if let Ok(msg) = self.rx_msg.recv_timeout(Duration::MAX) {
            self.handle_message(msg, cache, panicked);
        }
        while let Ok(msg) = self.rx_msg.try_recv() {
            self.handle_message(msg, cache, panicked);
        }

        self.pending_to_ready(cycles, cache);
    }

    fn consume_messages<S: BuildHasher>(
        &mut self,
        cache: &mut RefreshCacheStorage<'_, S>,
        panicked: &mut Vec<(SystemId, Box<dyn Any + Send>)>,
    ) {
        while self.record.num_injected() > self.record.num_completed() {
            if let Ok(msg) = self.rx_msg.recv_timeout(Duration::MAX) {
                self.handle_message(msg, cache, panicked);
            }
        }
    }

    fn handle_message<S: BuildHasher>(
        &mut self,
        msg: Message,
        cache: &mut RefreshCacheStorage<'_, S>,
        panicked: &mut Vec<(SystemId, Box<dyn Any + Send>)>,
    ) {
        match msg {
            Message::Fin(_wid, sid) => {
                self.record.insert(sid, RunResult::Finished);
                let cache = cache.get(&sid).unwrap();
                self.waits.dequeue(&cache.get_wait_indices());
            }
            Message::Aborted(_wid, sid) => {
                self.record.insert(sid, RunResult::Aborted);
                let cache = cache.get(&sid).unwrap();
                self.waits.dequeue(&cache.get_wait_indices());
            }
            Message::Panic(msg) => {
                self.record.insert(msg.sid, RunResult::Panicked);
                self.panic_helper(cache, panicked, msg);
            }
            Message::Handle(..) => unreachable!(),
        };
    }

    fn pending_to_ready<'s, S: BuildHasher + Default + 's>(
        &mut self,
        cycles: &mut Array<RawSystemCycleIter, MAX_GROUP>,
        cache: &mut RefreshCacheStorage<'_, S>,
    ) {
        #[allow(clippy::needless_range_loop)] // indexing more than one.
        let num_groups = self.wgroups.len();
        for i in 0..num_groups {
            unsafe {
                // For normal tasks.
                let target = NonNull::new_unchecked(&mut self.wgroups[i] as *mut _);
                Helper::pending_to_ready::<W, S>(
                    Or::A(target),
                    &mut self.nor_pendings[i],
                    &mut self.waits,
                    &mut cycles[i],
                    cache,
                );

                // For dedi tasks.
                let target = &mut self.dedi_tx as *mut _;
                let target = NonNull::new_unchecked(target);
                Helper::pending_to_ready::<W, S>(
                    Or::B(target),
                    &mut self.dedi_pendings[i],
                    &mut self.waits,
                    &mut cycles[i],
                    cache,
                );
            }
        }
    }

    fn panic_helper<S: BuildHasher>(
        &mut self,
        cache: &mut RefreshCacheStorage<S>,
        panicked: &mut Vec<(SystemId, Box<dyn Any + Send>)>,
        msg: PanicMessage,
    ) {
        if msg.unrecoverable {
            panic!("unrecoverable");
        }

        let cache = {
            #[cfg(not(target_arch = "wasm32"))]
            {
                cache.get(&msg.sid).unwrap()
            }

            // In web, buffer is not released by [`BufferCleaner`] because wasm-bindgen make it
            // abort instead of unwinding stack. So, we need to release it manually.
            #[cfg(target_arch = "wasm32")]
            {
                let mut cache = cache.get_mut(&msg.sid).unwrap();
                let buf = cache.get_request_buffer_mut();
                buf.clear();
                cache
            }
        };

        self.waits.dequeue(&cache.get_wait_indices());
        panicked.push((msg.sid, msg.payload));

        #[cfg(target_arch = "wasm32")]
        {
            // The worker must be closed without notification. Re-open it with `Search` state so
            // that it can empty its local queue.
            debug_assert_eq!(msg.sid.group_index(), msg.wid.group_index());
            let gi = msg.wid.group_index() as usize;
            let wi = msg.wid.worker_index() as usize;

            self.wgroups[gi].insert_search(wi);
        }
    }

    fn has_pending(&self, index: usize) -> bool {
        !self.nor_pendings[index].is_empty() || !self.dedi_pendings[index].is_empty()
    }

    fn clear(&mut self) {
        // Discards remaining dedicated futures.
        const WAIT_MILLIS: u64 = 10;
        const TOTAL_WAIT_MILLIS: u64 = if cfg!(test) || cfg!(debug_assertions) {
            500
        } else {
            5000
        };

        for _ in 0..TOTAL_WAIT_MILLIS / WAIT_MILLIS {
            if self.dedicated_future_count() == 0 {
                break;
            }

            // Discards remaining tasks.
            while let Ok(task) = self.dedi_rx.try_recv() {
                match task {
                    Task::System(_task) => { /* Do we need to do something here? */ }
                    Task::Parallel(_task) => { /* Do we need to do something here? */ }
                    Task::Async(task) => {
                        // Safety: Uncompleted future task is aborted and deallocated in here only.
                        unsafe { task.destroy() };
                        self.dedi_fut_cnt.fetch_sub(1, Ordering::Relaxed);
                    }
                }
            }

            thread::park_timeout(Duration::from_millis(WAIT_MILLIS));
        }

        let remain = self.dedicated_future_count();
        if remain > 0 {
            eprintln!("failed to remove {remain} remaining async tasks for {TOTAL_WAIT_MILLIS} ms");
        }
    }

    #[cfg(feature = "check")]
    fn validate_clean(&self) {
        // Validates if there's no waiting requests. It takes O(n).
        assert!(self.waits.is_all_queue_empty());

        // Validates if system record is clean.
        assert!(self.record.is_empty());

        // Validates if there's no pending tasks.
        let num_groups = self.wgroups.len();
        for i in 0..num_groups {
            assert!(!self.has_pending(i));
        }

        // Validates if there's no uncompleted dedicated tasks. System tasks and parallel tasks on
        // the main worker must have been completed, while async runners are free to send async
        // tasks to the dedicated queue at any time even when the scheduler is not running.
        for task in self.dedi_rx.buffer().iter() {
            if matches!(task, Task::System(_) | Task::Parallel(_)) {
                panic!("expected empty dedicated queue, but found: {task:?}");
            }
        }

        // Remaining dedicated futures are not failure not yet. They will be checked and deallocated
        // when the scheduler is dropped.
        let remain_dedi_fut_num = self.dedicated_future_count();
        if remain_dedi_fut_num > 0 {
            println!("found incomplete dedicated futures"); // Use log?
        }

        // Validates if message channel is empty.
        match self.rx_msg.try_recv() {
            Err(crossbeam_channel::TryRecvError::Empty) => {}
            Ok(msg) => panic!("unexpected remaining msg in channel: {msg:?}"),
            Err(err) => panic!("unexpected error from channel: {err:?}"),
        }
    }
}

impl<W: Work + 'static> Drop for Scheduler<W> {
    fn drop(&mut self) {
        self.clear();
    }
}

struct Helper;

impl Helper {
    /// Determines if the system(task) is runnable which means there's no data dependency at the
    /// moment. If it's runnable, the system's cached buffer is updated and then returned.
    fn update_task<'a, S: BuildHasher + Default>(
        waits: &mut WaitQueues,
        sdata: &mut SystemData,
        cache: &'a mut RefreshCacheStorage<S>,
    ) -> Option<&'a mut CacheItem> {
        let sid = sdata.id();
        let mut cache = cache.get_mut(&sid).unwrap();
        let (wait, retry) = cache.get_wait_retry_indices_mut();
        if waits.enqueue(&wait, retry) {
            // Before we send the data, we need to update(re-borrow) it.
            drop(wait);
            Some(cache.refresh())
        } else {
            None
        }
    }

    fn pending_to_ready<W, S>(
        target: Or<NonNull<WorkGroup<W>>, NonNull<ParkingSender<Task>>>,
        pending: &mut Pending,
        waits: &mut WaitQueues,
        cycle: &mut RawSystemCycleIter,
        cache: &mut RefreshCacheStorage<'_, S>,
    ) where
        W: Work + 'static,
        S: BuildHasher + Default,
    {
        let mut cur = pending.first_position();

        while let Some((next, &cycle_pos)) = pending.iter_next(cur) {
            let sdata = unsafe { cycle.get_at(cycle_pos).unwrap() };
            if let Some(cache) = Self::update_task(waits, sdata, cache) {
                pending.remove(&cycle_pos);
                Self::move_ready_system(target, sdata, cache);
            }
            cur = next;
        }
    }

    fn move_ready_system<W: Work + 'static>(
        target: Or<NonNull<WorkGroup<W>>, NonNull<ParkingSender<Task>>>,
        sdata: &mut SystemData,
        cache: &mut CacheItem,
    ) {
        let sid = sdata.id();

        // Safety:
        // - `invoker` and `buf` is safe because `Scheduler` guarantees that those pointers will be
        //   accessed in a sub worker only.
        // - `target` is safe because `Scheduler` guarantees that the pointer has exclusive
        //   authority.
        unsafe {
            let mut invoker = sdata.task_ptr();
            let buf = ManagedMutPtr::new(cache.request_buffer_ptr());

            // There are three branches here.
            // 1. Private system task   : Main worker handles it.
            // 2. Normal system task    : Appends it onto `wgroup`'s injector.
            // 3. Dedicated system task : Appends it onto `dedi` queue.

            if sdata.flags().is_private() {
                invoker.invoke_private(sid, buf);
            } else {
                let task = Task::System(SysTask::new(invoker, buf, sid));
                match target {
                    Or::A(wgroup) => wgroup.as_ref().inject_task(task),
                    Or::B(dedi) => dedi.as_ref().send(task).unwrap(),
                }
            }
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PullRes {
    Empty,
    Success,
    PendingFull,
}

#[derive(Debug)]
pub(crate) struct ScheduleRecord {
    record: HashMap<SystemId, RunResult, FxBuildHasher>,
    injected: usize,
    finished: usize,
    panicked: usize,
    aborted: usize,
}

impl ScheduleRecord {
    fn new() -> Self {
        Self {
            record: HashMap::with_hasher(FxBuildHasher::default()),
            injected: 0,
            finished: 0,
            panicked: 0,
            aborted: 0,
        }
    }

    #[allow(unused)]
    pub(crate) fn len(&self) -> usize {
        self.record.len()
    }

    #[allow(unused)]
    pub(crate) fn is_empty(&self) -> bool {
        self.len() == 0
    }

    pub(crate) fn clear(&mut self) {
        self.record.clear();
        self.injected = 0;
        self.finished = 0;
        self.panicked = 0;
        self.aborted = 0;
    }

    pub(crate) fn num_injected(&self) -> usize {
        self.injected
    }

    pub(crate) fn num_completed(&self) -> usize {
        self.finished + self.panicked
    }

    fn insert(&mut self, sid: SystemId, state: RunResult) {
        match state {
            RunResult::Injected => self.injected += 1,
            RunResult::Finished => self.finished += 1,
            RunResult::Panicked => self.panicked += 1,
            RunResult::Aborted => self.aborted += 1,
        }
        self.record.insert(sid, state);
    }
}

impl Default for ScheduleRecord {
    fn default() -> Self {
        Self::new()
    }
}

#[derive(Debug)]
pub(crate) enum RunResult {
    Injected,
    Finished,
    Panicked,
    Aborted,
}

#[derive(Debug)]
struct Pending {
    /// Pending system positions in a system cycle.
    list: SetValueList<ListPos, FxBuildHasher>,

    /// Maximum number of pending systems.
    limit: usize,
}

impl Pending {
    fn new(limit: usize) -> Self {
        let dummy = ListPos::end();
        Self {
            list: SetValueList::with_hasher(dummy, FxBuildHasher::default),
            limit,
        }
    }

    fn is_empty(&self) -> bool {
        self.list.is_empty()
    }

    fn push(&mut self, pos: ListPos) -> bool {
        if self.list.len() < self.limit {
            self.list.push_back(pos);
            true
        } else {
            false
        }
    }

    fn remove(&mut self, pos: &ListPos) {
        self.list.remove(pos);
    }
}

// Do not implement `DerefMut` to prevent pushing over the `limit`.
impl Deref for Pending {
    type Target = SetValueList<ListPos, FxBuildHasher>;

    fn deref(&self) -> &Self::Target {
        &self.list
    }
}

#[derive(Debug)]
struct WorkGroup<W: Work + 'static> {
    /// Sub workers belonging to the group.
    workers: Vec<W>,

    /// Context for each sub worker.
    sub_cxs: Vec<Pin<Box<SubContext>>>,

    /// Signal to wait or be woken by others in the same group.
    signal: Arc<GroupSignal>,

    /// Queue that shared over sub workers of the group.
    injector: Arc<Injector<Task>>,
}

impl<W> WorkGroup<W>
where
    W: Work + 'static,
{
    fn new(
        group_index: u16,
        workers: Vec<W>,
        tx_msg: &ParkingSender<Message>,
        tx_cmd: &CommandSender,
    ) -> Self {
        // Creates global queue.
        let injector = Arc::new(Injector::new());

        // This signal will be replaced at `initialize` with sub worker's handles.
        let dummy_signal = Arc::new(GroupSignal::new(Vec::new()));

        let comms = SubComm::with_len(
            group_index,
            &injector,
            &dummy_signal,
            tx_msg,
            tx_cmd,
            workers.len(),
        );

        // Sub contexts.
        let sub_cxs = comms
            .into_iter()
            .map(|comm| {
                Box::pin(SubContext {
                    guide: sub::SubStateGuide::new(),
                    // Puts in dummy handle for now. Sub workers will overwrite their handles at
                    // their first execution.
                    handle: UnsafeCell::new(thread::current()),
                    comm,
                    need_close: Cell::new(false),
                    _pin: PhantomPinned,
                })
            })
            .collect();

        Self {
            workers,
            sub_cxs,
            signal: dummy_signal,
            injector,
        }
    }

    fn initialize(&mut self, tx_msg: &ParkingSender<Message>, rx_msg: &ParkingReceiver<Message>) {
        // Handshake: Sub workers will set their thread handles on their very first open.
        for i in 0..self.len() {
            self.unpark_one(i);
        }

        // Handshake: Sub workers will send a message as a signal of successful setting their thread
        // handle.
        let mut remain = self.len();
        while remain > 0 {
            if let Ok(msg) = rx_msg.recv_timeout(Duration::MAX) {
                debug_assert!(matches!(msg, Message::Handle(_)));
                remain -= 1;
            }
        }

        // * TEMPORARY work-around
        // * Sub workers may still be accessing their `SubContext`.
        // - We received all messages from sub workers though, sub workers may still not finish
        //   their sending process via `tx_msg`. They may be in the middle of cleanup.
        // - In other words, sub workers may be accessing `ParkingSender` inside `SubContext`.
        //
        // * We're going to replace `GroupSignal` inside the `SubContext` here.
        //
        // * `ParkingSender` and `GroupSignal` locates in different memory locations though, they
        //   are members of `SubContext`.
        //
        // * Therefore, making `&mut SubContext` is technically UB. Miri might detect this from time
        //   to time.
        //
        // * So, we need to make sure that no one accesses `SubContext`.
        // - This may avoid the problem, but still dangerous.
        tx_msg.send(Message::Handle(WorkerId::dummy())).unwrap();
        while let Ok(msg) = rx_msg.recv_timeout(Duration::MAX) {
            if matches!(msg, Message::Handle(wid) if wid == WorkerId::dummy()) {
                break;
            } else {
                tx_msg.send(msg).unwrap();
            }
        }

        // Sub workers all sets their thread handles successfully. We can now gather their handles
        // and make a valid group signal.
        let slots = self
            .sub_cxs
            .iter()
            .map(|sub_cx| {
                let thread = unsafe { (*sub_cx.handle.get()).clone() };
                SignalSlot::awake_slot(thread)
            })
            .collect();
        let group_signal = GroupSignal::new(slots);
        self.signal = Arc::new(group_signal);

        for sub_cx in self.sub_cxs.iter_mut() {
            sub_cx.as_mut().set_signal(Arc::clone(&self.signal));
        }
    }

    fn open(&mut self) {
        for i in 0..self.len() {
            if self.sub_cxs[i].guide.push_open() {
                self.unpark_one(i);
            }
        }
    }

    fn close(&mut self) {
        for i in 0..self.len() {
            self.sub_cxs[i].guide.push_close();
        }
        self.signal.sub().notify_all();
    }

    fn len(&self) -> usize {
        debug_assert_eq!(self.workers.len(), self.sub_cxs.len());

        self.sub_cxs.len()
    }

    fn take_workers(&mut self) -> Vec<W> {
        self.destroy();
        self.sub_cxs.clear();
        mem::take(&mut self.workers)
    }

    /// Waits until the work group gets closed.
    fn wait_exhausted(&self) {
        while !self.is_exhausted() {
            self.signal.wait_open_count(0);
        }
    }

    /// Determines whether the work group is exhausted or not.
    ///
    /// `exhausted` here means that the work group has closed and cannot be woken up without
    /// intervention from outside.
    fn is_exhausted(&self) -> bool {
        // If guidance queue is empty, it means that sub workers cannot open themselves again.
        let is_guide_empty = self.sub_cxs.iter().all(|cx| cx.guide.is_empty());

        // Queue is empty, Also there's no open sub workers, they are completely in closed state.
        let is_all_closed = self.signal.open_count() == 0;

        is_guide_empty && is_all_closed
    }

    fn inject_task(&self, task: Task) {
        debug_assert!(
            !self.workers.is_empty(),
            "no workers for a non-dedicated task"
        );

        self.injector.push(task);
        self.signal.sub().notify_one();
    }

    #[cfg(debug_assertions)]
    fn validate_clean(&self) {
        // Validates sub contexts.
        for cx in self.sub_cxs.iter() {
            cx.validate_clean();
        }

        // Validates signal.
        assert_eq!(self.signal.open_count(), 0);
        assert_eq!(self.signal.work_count(), 0);
        assert_eq!(self.signal.future_count(), 0);

        // Validates if injector is empty.
        assert!(self.injector.is_empty());
    }

    fn insert_reset(&mut self, index: usize) {
        self.sub_cxs[index].guide.push_reset();
        self.unpark_one(index);
    }

    /// Do not call this method unless panic is detected.
    //
    // Allow dead_code?
    // - This will be compiled for wasm32 target.
    // - For keeping consistency with other methods.
    #[allow(dead_code)]
    fn insert_search(&mut self, index: usize) {
        // Search is only pushed when a sub worker has panicked. It means that the worker was open
        // state, and in turn, the queue has one space at least because 'open' must be popped from
        // it.
        let must_true = self.sub_cxs[index].guide.push_search();
        debug_assert!(must_true);
        self.unpark_one(index);
    }

    fn unpark_one(&mut self, index: usize) {
        unsafe {
            // Safety: Unparking changes nothing.
            let ptr = self.sub_cxs[index].as_mut().get_unchecked_mut() as *mut SubContext;
            let ptr = NonNullExt::new_unchecked(ptr);
            let ptr = ManagedConstPtr::new(ptr);

            let must_true = self.workers[index].unpark(ptr);
            assert!(must_true);
        }
    }

    fn destroy(&mut self) {
        // Aborts remaining tasks and waits for sub workers to be completely closed.
        self.signal.set_abort(true);
        while !self.is_exhausted() {
            self.signal.sub().notify_all();
            self.signal.wait_open_count(0);
        }
        self.signal.set_abort(false);

        // Resets thread local variables on sub worker contexts.
        for i in 0..self.len() {
            self.insert_reset(i);
        }
        self.signal.wait_open_count(self.len() as u32);

        // Closes again.
        self.close();
        self.signal.wait_open_count(0);

        #[cfg(debug_assertions)]
        self.validate_clean();
    }
}

impl<W> Drop for WorkGroup<W>
where
    W: Work + 'static,
{
    fn drop(&mut self) {
        self.destroy();
    }
}

thread_local! {
    /// Thread local pointer to [`SubContext`].
    pub(crate) static SUB_CONTEXT: Cell<NonNullExt<SubContext>> = const {
        Cell::new(NonNullExt::dangling())
    };

    /// Thread local worker id.
    pub(crate) static WORKER_ID: Cell<WorkerId> = const {
        Cell::new(WorkerId::dummy())
    };
}

/// Context for a sub worker.
///
/// TODO: Make this Send and send to a sub worker. Sub worker may return this when it's dropped.
#[derive(Debug)]
pub struct SubContext {
    guide: sub::SubStateGuide,

    /// Thread handle.
    handle: UnsafeCell<Thread>,

    comm: SubComm,

    need_close: Cell<bool>,

    _pin: PhantomPinned,
}

impl SubContext {
    /// Use this function in your [`Worker::unpark`] implementation.
    ///
    /// [`Worker::unpark`]: crate::prelude::Worker::unpark
    #[rustfmt::skip]
    pub fn execute(ptr: ManagedConstPtr<Self>) {
        // Hand-shake for the first time.
        if ptr.comm.maybe_uninit_worker_id() != WORKER_ID.with(Cell::get) {
            Self::set_handle(ptr);
            return;
        }

        // * Deals with managed pointer for internal debugging.
        // (1) WASM32    : Drops managed pointer for unexpected panics.
        // (2) Otherwise : Drops managed pointer later.
        let this = {
            #[cfg(target_arch = "wasm32")]      { ptr.into_ref() }
            #[cfg(not(target_arch = "wasm32"))] { &*ptr }
        };

        // ** Notifies start of accessing `SubContext`.
        // Main worker cannot release `SubContext` while we're accessing it.
        this.comm.signal().add_open_count(1);

        // Runs state machine.
        let mut cur = this.guide.pop();
        let mut steal = Steal::Empty;
        while let Some(next) = this.execute_by_state(cur, &mut steal) {
            cur = next;
        }

        // * Deals with managed pointer for internal debugging.
        // (2) Drops it here, before we notify end of execution.
        #[cfg(not(target_arch = "wasm32"))]
        let this = ptr.into_ref();

        // ** Notifies end of accessing `SubContext`.
        // Main worker is now free to release `SubContext`. In web, however, this procedure must
        // occur in somewhere during panic. See `web_panic_hook` for more details.
        this.comm.signal().sub_open_count(1);
    }

    pub(crate) fn get_comm(&self) -> &SubComm {
        &self.comm
    }

    fn set_handle(ptr: ManagedConstPtr<Self>) {
        let handle = unsafe { &mut *ptr.handle.get() };
        *handle = thread::current();
        SUB_CONTEXT.with(|cx| cx.set(ptr.as_nonnullext()));
        WORKER_ID.with(|worker_id| worker_id.set(ptr.comm.maybe_uninit_worker_id()));
        ptr.comm.send_message(Message::Handle(ptr.comm.worker_id()));
    }

    #[inline]
    fn execute_by_state(&self, cur: SubState, steal: &mut Steal<Task>) -> Option<SubState> {
        match cur {
            SubState::Wait => {
                self.comm.wait();
                if self.comm.signal().is_abort() {
                    Some(SubState::Abort)
                } else {
                    Some(SubState::Search)
                }
            }
            SubState::Search => {
                *steal = self.comm.search();
                if steal.is_success() {
                    Some(SubState::Work)
                } else if self.need_close.take() {
                    Some(SubState::Close)
                } else if self.can_close() {
                    // Because search() and can_close() are not atomic, we have to search once
                    // again.
                    self.need_close.set(true);
                    Some(SubState::Search)
                } else {
                    Some(SubState::Wait)
                }
            }
            SubState::Work => {
                self.work(steal);
                Some(SubState::Search)
            }
            SubState::Abort => {
                self.abort();
                Some(SubState::Search)
            }
            SubState::Close => {
                self.comm.signal().sub().notify_all();
                None
            }
            SubState::Reset => {
                SUB_CONTEXT.with(|cx| cx.set(NonNullExt::dangling()));
                WORKER_ID.with(|worker_id| worker_id.set(WorkerId::dummy()));
                Some(SubState::Search)
            }
        }
    }

    #[inline]
    fn can_close(&self) -> bool {
        // Currently there is no any remaining task, but this sub worker must wait for other
        // siblings for future task. Imagine that a sibling is working on a system task, which
        // produces a future task. If the future task should be run in parallel, this sub worker
        // must join it.
        let fut_cnt = self.comm.signal().future_count();
        let work_cnt = self.comm.signal().work_count();
        if fut_cnt > 0 || work_cnt > 0 {
            return false;
        }

        // It seems likely that siblings and this sub worker are completely free to exit execution.
        // If main worker sent exit signal, we can exit.
        if self.guide.need_close() {
            return true;
        }

        // Main worker has not sent exit signal. Waits for new tasks.
        false
    }

    pub(super) fn work(&self, steal: &mut Steal<Task>) {
        // * Notifies start of working state.
        self.comm.signal().add_work_count(1);

        // Works for tasks as much as possible.
        // NOTE: Panics can occur here.
        loop {
            match mem::replace(steal, Steal::Empty) {
                Steal::Success(cur) => match cur {
                    Task::System(task) => self.work_for_system_task(task),
                    Task::Parallel(task) => self.work_for_parallel_task(task),
                    Task::Async(task) => self.work_for_async_task(task),
                },
                Steal::Empty => break,
                Steal::Retry => {}
            }
            *steal = self.comm.pop();
        }

        // * Notifies end of working state.
        // In web, however, this procedure must occur in somewhere during panic. See
        // `web_panic_hook` for more details.
        self.comm.signal().sub_work_count(1);
    }

    fn work_for_system_task(&self, task: SysTask) {
        let wid = self.comm.worker_id();
        let sid = task.sid();

        let resp = match task.execute(wid) {
            Ok(_) => Message::Fin(self.comm.worker_id(), sid),
            Err(payload) => Message::Panic(PanicMessage {
                wid: self.comm.worker_id(),
                sid,
                payload,
                unrecoverable: false,
            }),
        };

        self.comm.send_message(resp);
    }

    fn work_for_parallel_task(&self, task: ParTask) {
        let wid = self.comm.worker_id();

        task.execute(wid, FnContext::MIGRATED);
    }

    fn work_for_async_task(&self, task: AsyncTask) {
        // Sets waker if needed.
        let waker = UnsafeWaker::new(self as *const SubContext); // Cheap
        unsafe {
            if !task.will_wake(&waker) {
                task.set_waker(waker);
            }
        }

        // Executes.
        let wid = self.comm.worker_id();
        let on_ready = |ready| {
            // Decreases running future count.
            self.comm.signal().sub_future_count(1);

            // Sends the ready future as a command.
            let cmd = CommandObject::Future(ready);
            self.comm.send_command_or_cancel(cmd);
        };
        task.execute(wid, on_ready);
    }

    /// Cancels out remaining tasks.
    //
    // NOTE: Future tasks can be cancelled out at the next await points. So that if any future
    // executor, or runtime, doesn't call poll() on future tasks again, this method will block
    // infinitely.
    fn abort(&self) {
        loop {
            match self.comm.pop() {
                Steal::Success(task) => self.abort_task(task),
                Steal::Empty => {
                    if self.comm.signal().future_count() == 0 {
                        // Before escaping the loop, notifies all other sub workers, so that they
                        // can escape their loops as well.
                        self.comm.signal().sub().notify_all();
                        break;
                    }

                    // This worker may be supposed to be woken by a future task.
                    self.comm.wait();
                }
                Steal::Retry => {}
            }
        }
    }

    fn abort_task(&self, task: Task) {
        match task {
            // To abort system task, drops it and sends Aborted to main worker.
            Task::System(task) => {
                let wid = self.comm.worker_id();
                let sid = task.sid();
                self.comm.send_message(Message::Aborted(wid, sid));
            }
            // Parallel task cannot be aborted. It should be aborted at system task level.
            Task::Parallel(task) => {
                self.work_for_parallel_task(task);
            }
            // To abort async task, destroys it and reduces running future count.
            Task::Async(task) => {
                // Safety: Uncompleted future task is aborted and deallocated in here only.
                unsafe { task.destroy() };
                self.comm.signal().sub_future_count(1);
            }
        }
    }

    fn set_signal(self: Pin<&mut Self>, signal: Arc<GroupSignal>) {
        // Safety: We do not move self.
        unsafe {
            let this = self.get_unchecked_mut();
            this.comm.set_signal(signal);
        }
    }

    #[cfg(debug_assertions)]
    fn validate_clean(&self) {
        // Validates if guide is empty.
        let mut v = Vec::new();
        while !self.guide.is_empty() {
            v.push(self.guide.pop());
        }
        if !v.is_empty() {
            panic!("guide is not empry: {v:?}");
        }

        // Validates comm.
        match self.get_comm().search() {
            Steal::Empty => {}
            _ => panic!("validation failed due to remaining task"),
        }
    }
}

// This module helps us not to access fields of structs in this module directly.
mod sub {
    use super::SubState;
    use crate::utils::ds::ArrayDeque;
    use std::sync::{
        atomic::{AtomicU32, Ordering},
        Mutex,
    };

    /// Guidance for [`SubContext`] on what state it should start with or need for closing. You can
    /// push or pop some states onto this struct. Allowed states are as follows.
    ///
    /// - [`SubState::Wait`]   : Normal open request.
    /// - [`SubState::Close`]  : Request for closing.
    /// - [`SubState::Reset`]  : Request for resetting thread local variables.
    /// - [`SubState::Search`] : It has panicked, start with search state.
    ///
    /// States can be buffered at most a fixed size(4 for now).
    ///
    /// [`SubContext`]: super::SubContext
    //
    // Why we need buffering?
    // - Main worker and sub workers are not tightly synchronized.
    //   * Main worker doesn't care of in which states sub workers are.
    //   * Main worker just notifies that there will be no new system tasks to sub workers by
    //     sending Close request to sub workers and waits for completion of system tasks only, not
    //     for future tasks.
    // - We don't want that sub workers are in open states too long.
    //   * Worker may have other roles, and we may need to hand over the control for them. To do
    //     that, we need to reach close state from time to time.
    #[derive(Debug)]
    pub(super) struct SubStateGuide {
        /// SPSC queue holding state requests for a [`SubContext`].
        /// - Producer: Main worker.
        /// - Consumer: A sub worker.
        ///
        /// [`SubContext`]: super::SubContext
        queue: Mutex<ArrayDeque<SubState, 8>>,

        /// Close request counter.
        //
        // Helps us not to lock `queue` too frequently.
        close: AtomicU32,

        /// Whether open or reset request has been pushed.
        //
        // Prevents from appending 'not paired close' onto the queue.
        #[cfg(debug_assertions)]
        open: std::cell::Cell<bool>,
    }

    impl SubStateGuide {
        pub(super) fn new() -> Self {
            Self {
                queue: Mutex::new(ArrayDeque::new()),
                close: AtomicU32::new(0),
                #[cfg(debug_assertions)]
                open: std::cell::Cell::new(false),
            }
        }

        pub(super) fn is_empty(&self) -> bool {
            let queue = self.queue.lock().unwrap();
            queue.is_empty()
        }

        /// If queue has enough room for open-close request pair, then pushes an open request,
        /// [`SubState::Wait`], then returns true.
        ///
        /// However, queue was full and open request was not actually pushed, then the last
        /// open-close pair in the queue turns into single open request by popping close request,
        /// then returns false.
        ///
        /// You have to unpark worker if and only if true is returned.
        ///
        /// Push operation is allowed for main worker only.
        pub(super) fn push_open(&self) -> bool {
            #[cfg(debug_assertions)]
            {
                // Open or reset cannot follow another open or reset.
                assert!(!self.open.get());
                self.open.set(true);
            }

            // If an open-close pair cannot be appended in a row, unchains the pair by popping close
            // request.
            let mut queue = self.queue.lock().unwrap();
            if queue.capacity() - queue.len() < 2 {
                debug_assert_eq!(queue[queue.len() - 1], SubState::Close);
                debug_assert_eq!(queue[queue.len() - 2], SubState::Wait);

                queue.pop_back();
                return false;
            }

            // We checked the room, so that it must succeed.
            queue.push_back(SubState::Wait);
            true
        }

        /// Pushes a close request onto queue. You must call this method after [`Self::push_open`].
        ///
        /// Push operation is allowed for main worker only.
        pub(super) fn push_close(&self) {
            #[cfg(debug_assertions)]
            {
                // Close must follow open or reset request.
                assert!(self.open.get());
                self.open.set(false);
            }

            let mut queue = self.queue.lock().unwrap();
            let must_true = queue.push_back(SubState::Close);
            debug_assert!(must_true);
            self.close.fetch_add(1, Ordering::Release);
        }

        pub(super) fn push_reset(&self) {
            #[cfg(debug_assertions)]
            {
                // Open or reset cannot follow another open or reset.
                assert!(!self.open.get());
                self.open.set(true);
            }

            let mut queue = self.queue.lock().unwrap();
            let must_true = queue.push_back(SubState::Reset);
            debug_assert!(must_true);
        }

        /// Push operation is allowed for main worker only.
        //
        // Allow dead_code?
        // - This will be compiled for wasm32 target.
        // - For keeping consistency with other methods.
        #[allow(dead_code)]
        pub(super) fn push_search(&self) -> bool {
            let mut queue = self.queue.lock().unwrap();
            queue.push_front(SubState::Search)
        }

        /// Pop operation is allowed for sub worker only.
        pub(super) fn pop(&self) -> SubState {
            let mut queue = self.queue.lock().unwrap();
            queue.pop_front().unwrap()
        }

        /// Pop operation is allowed for sub worker only.
        pub(super) fn need_close(&self) -> bool {
            if self.close.load(Ordering::Acquire) == 0 {
                return false;
            }

            let mut queue = self.queue.lock().unwrap();
            if queue.front() != Some(&SubState::Close) {
                return false;
            }

            self.close.fetch_sub(1, Ordering::Relaxed);
            queue.pop_front();
            true
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SubState {
    Wait,
    Search,
    Work,
    Abort,
    Close,
    Reset,
}

#[derive(Debug, Clone)]
pub(crate) struct MainWaker {
    tx_dedi: ParkingSender<Task>,
    tid: ThreadId,
}

impl MainWaker {
    pub(crate) fn new(tx_dedi: ParkingSender<Task>) -> Self {
        Self {
            tx_dedi,
            tid: thread::current().id(),
        }
    }
}

impl WakeSend for MainWaker {
    fn wake_send(&self, handle: UnsafeFuture) {
        let task = Task::Async(AsyncTask(handle));

        // The scheduler, who is owner of the opposite reciver, may be destroyed without waiting for
        // remaining futures. Ignores transmission failure in that case. Anyway, clients destroyed
        // it for some reason, then we cannot make progress any longer.
        if self.tx_dedi.send(task).is_err() {
            // Safety: We're not copying the handle somewhere else. Therefore, it's safe to destroy
            // the handle.
            unsafe { handle.destroy() };
        }
    }
}

impl PartialEq for MainWaker {
    fn eq(&self, other: &Self) -> bool {
        self.tid == other.tid
    }
}

#[derive(Debug, Clone, Copy, PartialEq)]
#[repr(transparent)]
pub(crate) struct UnsafeWaker {
    cx: *const SubContext,
}

impl UnsafeWaker {
    pub(crate) const fn new(cx: *const SubContext) -> Self {
        Self { cx }
    }
}

unsafe impl Send for UnsafeWaker {}
unsafe impl Sync for UnsafeWaker {}

impl WakeSend for UnsafeWaker {
    fn wake_send(&self, handle: UnsafeFuture) {
        // Safety: Scheduler holds sub contexts while it is executing.
        let comm = unsafe { self.cx.as_ref().unwrap_unchecked().get_comm() };

        // Pushes the future handle onto local future queue.
        comm.push_future_task(handle);

        // Tries to wake up the worker which called poll() on the future. If it is already awaken,
        // wakes another worker.
        comm.wake_self();
    }
}