use crate::random::Random;

use crate::protocol::*;

use crate::protocol::ast::ProcedureDefinitionId;

use crate::protocol::eval::{

    PortId as EvalPortId, Prompt,

    ValueGroup, Value,

    EvalContinuation, EvalResult, EvalError

};

use crate::runtime2::scheduler::SchedulerCtx;

use crate::runtime2::communication::*;

use super::component_context::*;

use super::control_layer::*;

use super::consensus::Consensus;

pub enum ExecStmt {

    CreatedChannel((Value, Value)),

    PerformedPut,

    PerformedGet(ValueGroup),

    PerformedSelectWait(u32),

    None,

}

impl ExecStmt {

    fn take(&mut self) -> ExecStmt {

        let mut value = ExecStmt::None;

        std::mem::swap(self, &mut value);

        return value;

    }

    fn is_none(&self) -> bool {

        match self {

            ExecStmt::None => return true,

            _ => return false,

        }

    }

}

pub struct ExecCtx {

    stmt: ExecStmt,

}

impl RunContext for ExecCtx {

    fn performed_put(&mut self, _port: EvalPortId) -> bool {

        match self.stmt.take() {

            ExecStmt::None => return false,

            ExecStmt::PerformedPut => return true,

            _ => unreachable!(),

        }

    }

    fn performed_get(&mut self, _port: EvalPortId) -> Option<ValueGroup> {

        match self.stmt.take() {

            ExecStmt::None => return None,

            ExecStmt::PerformedGet(value) => return Some(value),

            _ => unreachable!(),

        }

    }

    fn fires(&mut self, _port: EvalPortId) -> Option<Value> {

        todo!("remove fires")

    }

    fn performed_fork(&mut self) -> Option<bool> {

        todo!("remove fork")

    }

    fn created_channel(&mut self) -> Option<(Value, Value)> {

        match self.stmt.take() {

            ExecStmt::None => return None,

            ExecStmt::CreatedChannel(ports) => return Some(ports),

            _ => unreachable!(),

        }

    }

    fn performed_select_wait(&mut self) -> Option<u32> {

        match self.stmt.take() {

            ExecStmt::None => return None,

            ExecStmt::PerformedSelectWait(selected_case) => Some(selected_case),

            _v => unreachable!(),

        }

    }

}

#[derive(Debug, Copy, Clone, PartialEq, Eq)]

pub(crate) enum Mode {

    NonSync, // not in sync mode

    Sync, // in sync mode, can interact with other components

    SyncEnd, // awaiting a solution, i.e. encountered the end of the sync block

    BlockedGet, // blocked because we need to receive a message on a particular port

    BlockedPut, // component is blocked because the port is blocked

    BlockedSelect, // waiting on message to complete the select statement

    StartExit, // temporary state: if encountered then we start the shutdown process

    BusyExit, // temporary state: waiting for Acks for all the closed ports

    Exit, // exiting: shutdown process started, now waiting until the reference count drops to 0

}

impl Mode {

    fn is_in_sync_block(&self) -> bool {

        use Mode::*;

        match self {

            Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect => true,

            NonSync | StartExit | BusyExit | Exit => false,

        }

    }

}

struct SelectCase {

    involved_ports: Vec<LocalPortHandle>,

}

// TODO: @Optimize, flatten cases into single array, have index-pointers to next case

struct SelectState {

    cases: Vec<SelectCase>,

    next_case: u32,

    num_cases: u32,

    random: Random,

    candidates_workspace: Vec<usize>,

}

enum SelectDecision {

    None,

    Case(u32), // contains case index, should be passed along to PDL code

}

type InboxMain = Vec<Option<DataMessage>>;

impl SelectState {

    fn new() -> Self {

        return Self{

            cases: Vec::new(),

            next_case: 0,

            num_cases: 0,

            random: Random::new(),

            candidates_workspace: Vec::new(),

        }

    }

    fn handle_select_start(&mut self, num_cases: u32) {

        self.cases.clear();

        self.next_case = 0;

        self.num_cases = num_cases;

    }

    /// Register a port as belonging to a particular case. As for correctness of

    /// PDL code one cannot register the same port twice, this function might

    /// return an error

    fn register_select_case_port(&mut self, comp_ctx: &CompCtx, case_index: u32, _port_index: u32, port_id: PortId) -> Result<(), PortId> {

        // Retrieve case and port handle

        self.ensure_at_case(case_index);

        let cur_case = &mut self.cases[case_index as usize];

        let port_handle = comp_ctx.get_port_handle(port_id);

        debug_assert_eq!(cur_case.involved_ports.len(), _port_index as usize);

        // Make sure port wasn't added before, we disallow having the same port

        // in the same select guard twice.

        if cur_case.involved_ports.contains(&port_handle) {

            return Err(port_id);

        }

        cur_case.involved_ports.push(port_handle);

        return Ok(());

    }

    /// Notification that all ports have been registered and we should now wait

    /// until the appropriate messages have come in.

    fn handle_select_waiting_point(&mut self, inbox: &InboxMain, comp_ctx: &CompCtx) -> SelectDecision {

        if self.num_cases != self.next_case {

            // This happens when there are >=1 select cases written at the end

            // of the select block.

            self.ensure_at_case(self.num_cases - 1);

        }

        return self.has_decision(inbox, comp_ctx);

    }

    fn handle_updated_inbox(&mut self, inbox: &InboxMain, comp_ctx: &CompCtx) -> SelectDecision {

        return self.has_decision(inbox, comp_ctx);

    }

    /// Internal helper, pushes empty cases inbetween last case and provided new

    /// case index.

    fn ensure_at_case(&mut self, new_case_index: u32) {

        // Push an empty case for all intermediate cases that were not

        // registered with a port.

        debug_assert!(new_case_index >= self.next_case && new_case_index < self.num_cases);

        for _ in self.next_case..new_case_index + 1 {

            self.cases.push(SelectCase{ involved_ports: Vec::new() });

        }

        self.next_case = new_case_index + 1;

    }

    /// Checks if a decision can be reached

    fn has_decision(&mut self, inbox: &InboxMain, comp_ctx: &CompCtx) -> SelectDecision {

        self.candidates_workspace.clear();

        if self.cases.is_empty() {

            // If there are no cases then we can immediately reach a "bogus

            // decision".

            return SelectDecision::Case(0);

        }

        // Need to check for valid case

        'case_loop: for (case_index, case) in self.cases.iter().enumerate() {

            for port_handle in case.involved_ports.iter().copied() {

                let port_index = comp_ctx.get_port_index(port_handle);

                if inbox[port_index].is_none() {

mod component_pdl;

mod component_context;

mod control_layer;

mod consensus;

pub(crate) use component_pdl::{CompPDL, CompScheduling};

pub(crate) use component_context::CompCtx;

pub(crate) use control_layer::{ControlId};

use super::scheduler::*;

use super::runtime::*;

/// If the component is sleeping, then that flag will be atomically set to

/// false. If we're the ones that made that happen then we add it to the work

/// queue.

pub(crate) fn wake_up_if_sleeping(sched_ctx: &SchedulerCtx, comp_id: CompId, handle: &CompHandle) {

    use std::sync::atomic::Ordering;

    let should_wake_up = handle.sleeping

        .compare_exchange(true, false, Ordering::AcqRel, Ordering::Acquire)

        .is_ok();

    if should_wake_up {

        let comp_key = unsafe{ comp_id.upgrade() };

        sched_ctx.runtime.enqueue_work(comp_key);

    }

}

mod store;

mod runtime;

mod component;

mod communication;

mod scheduler;

#[cfg(test)] mod tests;

pub use runtime::Runtime;