use crate::runtime2::*;

use super::CompCtx;

pub enum CompScheduling {

    Immediate,

    Requeue,

    Sleep,

    Exit,

}

pub(crate) trait Component {

}

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,

        }

            let candidate_index = self.random.get_u64() as usize % self.candidates_workspace.len();

            return SelectDecision::Case(self.candidates_workspace[candidate_index] as u32);

        }

    }

}

pub(crate) struct CompPDL {

    pub mode: Mode,

    pub mode_port: PortId, // when blocked on a port

    pub mode_value: ValueGroup, // when blocked on a put

    select: SelectState,

    pub prompt: Prompt,

    pub control: ControlLayer,

    pub consensus: Consensus,

    pub sync_counter: u32,

    pub exec_ctx: ExecCtx,

    // TODO: Temporary field, simulates future plans of having one storage place

    //  reserved per port.

    // Should be same length as the number of ports. Corresponding indices imply

    // message is intended for that port.

    pub inbox_main: InboxMain,

    pub inbox_backup: Vec<DataMessage>,

}

        }

    }

        sched_ctx.log(&format!("handling message: {:#?}", message));

        if let Some(new_target) = self.control.should_reroute(&mut message) {

            let mut target = sched_ctx.runtime.get_component_public(new_target);

            target.send_message(sched_ctx, message, false); // not waking up: we schedule once we've received all PortPeerChanged Acks

            let _should_remove = target.decrement_users();

            debug_assert!(_should_remove.is_none());

            return;

        }

        match message {

            Message::Data(message) => {

                self.handle_incoming_data_message(sched_ctx, comp_ctx, message);

            },

            Message::Control(message) => {

                self.handle_incoming_control_message(sched_ctx, comp_ctx, message);

            },

            Message::Sync(message) => {

                self.handle_incoming_sync_message(sched_ctx, comp_ctx, message);

            }

        }

    }

        use EvalContinuation as EC;

        sched_ctx.log(&format!("Running component (mode: {:?})", self.mode));

        // Depending on the mode don't do anything at all, take some special

        // actions, or fall through and run the PDL code.

        match self.mode {

            Mode::NonSync | Mode::Sync | Mode::BlockedSelect => {

                // continue and run PDL code

            },

            Mode::SyncEnd | Mode::BlockedGet | Mode::BlockedPut => {

                return Ok(CompScheduling::Sleep);

            }

            Mode::StartExit => {

                self.handle_component_exit(sched_ctx, comp_ctx);

                return Ok(CompScheduling::Immediate);

            },

            Mode::BusyExit => {

                if self.control.has_acks_remaining() {

                    return Ok(CompScheduling::Sleep);

                } else {

                    self.mode = Mode::Exit;

                    return Ok(CompScheduling::Exit);

                }

                return Ok(CompScheduling::Immediate);

            },

            EC::NewComponent(definition_id, type_id, arguments) => {

                debug_assert_eq!(self.mode, Mode::NonSync);

                self.create_component_and_transfer_ports(

                    sched_ctx, comp_ctx,

                    definition_id, type_id, arguments

                );

                return Ok(CompScheduling::Requeue);

            },

            EC::NewChannel => {

                debug_assert_eq!(self.mode, Mode::NonSync);

                debug_assert!(self.exec_ctx.stmt.is_none());

                let channel = comp_ctx.create_channel();

                self.exec_ctx.stmt = ExecStmt::CreatedChannel((

                    Value::Output(port_id_to_eval(channel.putter_id)),

                    Value::Input(port_id_to_eval(channel.getter_id))

                ));

                self.inbox_main.push(None);

                self.inbox_main.push(None);

                return Ok(CompScheduling::Immediate);

            }

        }

    }

    fn execute_prompt(&mut self, sched_ctx: &SchedulerCtx) -> EvalResult {

        let mut step_result = EvalContinuation::Stepping;

        while let EvalContinuation::Stepping = step_result {

            step_result = self.prompt.step(

                &sched_ctx.runtime.protocol.types, &sched_ctx.runtime.protocol.heap,

                &sched_ctx.runtime.protocol.modules, &mut self.exec_ctx,

            )?;

        }

        return Ok(step_result)

    }

    fn handle_sync_start(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {

        sched_ctx.log("Component starting sync mode");

        self.consensus.notify_sync_start(comp_ctx);

        for message in self.inbox_main.iter() {

            if let Some(message) = message {

                self.consensus.handle_new_data_message(comp_ctx, message);

            }

        }

        debug_assert_eq!(self.mode, Mode::NonSync);

        self.mode = Mode::Sync;

    }

                    }

                }

            } else {

                // Peer is a different component. We'll deal with sending the

                // appropriate messages later

                let peer_handle = creator_ctx.get_peer_handle(created_port_info.peer_comp_id);

                let peer_info = creator_ctx.get_peer(peer_handle);

                created_ctx.add_peer(pair.created_handle, sched_ctx, peer_info.id, Some(&peer_info.handle));

                created_component_has_remote_peers = true;

            }

        }

        // We'll now actually turn our reservation for a new component into an

        // actual component. Note that we initialize it as "not sleeping" as

        // its initial scheduling might be performed based on `Ack`s in response

        // to message exchanges between remote peers.

        let prompt = Prompt::new(

            &sched_ctx.runtime.protocol.types, &sched_ctx.runtime.protocol.heap,

            definition_id, type_id, arguments,

        );

        let component = CompPDL::new(prompt, port_id_pairs.len());

        let (created_key, component) = sched_ctx.runtime.finish_create_pdl_component(

            reservation, component, created_ctx, false,

        );

        // Now modify the creator's ports: remove every transferred port and

        for pair in port_id_pairs.iter() {

            // Remove peer if appropriate

            let creator_port_info = creator_ctx.get_port(pair.creator_handle);

            let creator_port_index = creator_ctx.get_port_index(pair.creator_handle);

            let creator_peer_comp_id = creator_port_info.peer_comp_id;

            creator_ctx.remove_peer(sched_ctx, pair.creator_handle, creator_peer_comp_id, false);

            creator_ctx.remove_port(pair.creator_handle);

            // Transfer any messages

            if let Some(mut message) = self.inbox_main.remove(creator_port_index) {

                message.data_header.target_port = pair.created_id;

            }

            let mut message_index = 0;

            while message_index < self.inbox_backup.len() {

                let message = &self.inbox_backup[message_index];

                if message.data_header.target_port == pair.creator_id {

                    // transfer message

                    let mut message = self.inbox_backup.remove(message_index);

                    message.data_header.target_port = pair.created_id;

                } else {

                    message_index += 1;

                }

            }

            // Handle potential channel between creator and created component

            if created_port_info.peer_comp_id == creator_ctx.id {

                let peer_port_handle = creator_ctx.get_port_handle(created_port_info.peer_port_id);

                let peer_port_info = creator_ctx.get_port_mut(peer_port_handle);

                peer_port_info.peer_port_id = created_port_info.self_id;

            }

        }

        if created_component_has_remote_peers {

            let schedule_entry_id = self.control.add_schedule_entry(created_ctx.id);

            for pair in port_id_pairs.iter() {

                let port_info = created_ctx.get_port(pair.created_handle);

                if port_info.peer_comp_id != creator_ctx.id && port_info.peer_comp_id != created_ctx.id {

                    let message = self.control.add_reroute_entry(

                        creator_ctx.id, port_info.peer_port_id, port_info.peer_comp_id,

                        pair.creator_id, pair.created_id, created_ctx.id,

                        schedule_entry_id

                    );

                    let peer_handle = created_ctx.get_peer_handle(port_info.peer_comp_id);

                    let peer_info = created_ctx.get_peer(peer_handle);

                    peer_info.handle.send_message(sched_ctx, message, true);

                }

            }

        } else {

            // Peer can be scheduled immediately

            sched_ctx.runtime.enqueue_work(created_key);

        }

    }

}

#[inline]

fn port_id_from_eval(port_id: EvalPortId) -> PortId {

    return PortId(port_id.id);

mod component_pdl;

mod component_context;

mod control_layer;

mod consensus;

mod component;

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;

pub(crate) use scheduler::SchedulerCtx;

use std::sync::{Arc, Mutex, Condvar};

use std::sync::atomic::{AtomicU32, AtomicBool, Ordering};

use std::collections::VecDeque;

use crate::protocol::*;

use super::communication::Message;

use super::store::{ComponentStore, ComponentReservation, QueueDynMpsc, QueueDynProducer};

use super::scheduler::*;

// -----------------------------------------------------------------------------

// Component

// -----------------------------------------------------------------------------

/// Key to a component. Type system somewhat ensures that there can only be one

/// of these. Only with a key one may retrieve privately-accessible memory for

/// a component. Practically just a generational index, like `CompId` is.

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

pub(crate) struct CompKey(pub u32);

impl CompKey {

    pub(crate) fn downgrade(&self) -> CompId {

        return CompId(self.0);

    }

}

/// Generational ID of a component

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

pub struct CompId(pub u32);

impl CompId {

    pub(crate) fn new_invalid() -> CompId {

        return CompId(u32::MAX);

    }

    /// Upgrade component ID to component key. Unsafe because the caller needs

    /// to make sure that only one component key can exist at a time (to ensure

    /// a component can only be scheduled/executed by one thread).

    pub(crate) unsafe fn upgrade(&self) -> CompKey {

        return CompKey(self.0);

    }

}

/// Handle to a component that is being created.

pub(crate) struct CompReserved {

    reservation: ComponentReservation,

}

impl CompReserved {

    pub(crate) fn id(&self) -> CompId {

        return CompId(self.reservation.index)

    }

}

pub(crate) struct RuntimeComp {

    pub public: CompPublic,

    pub ctx: CompCtx,

    pub inbox: QueueDynMpsc<Message>,

    pub exiting: bool,

}

/// Should contain everything that is accessible in a thread-safe manner

// TODO: Do something about the `num_handles` thing. This needs to be a bit more

//  "foolproof" to lighten the mental burden of using the `num_handles`

//  variable.

pub(crate) struct CompPublic {

    pub sleeping: AtomicBool,

    pub num_handles: AtomicU32, // manually modified (!)

    inbox: QueueDynProducer<Message>,

}

/// Handle to public part of a component. Would be nice if we could

/// automagically manage the `num_handles` counter. But when it reaches zero we

/// need to manually remove the handle from the runtime. So we just have debug

/// code to make sure this actually happens.

pub(crate) struct CompHandle {

    target: *const CompPublic,

    id: CompId, // TODO: @Remove after debugging

    #[cfg(debug_assertions)] decremented: bool,

}

    pub(crate) fn take_work(&self) -> Option<CompKey> {

        let mut lock = self.work_queue.lock().unwrap();

        while lock.is_empty() && self.active_elements.load(Ordering::Acquire) != 0 {

            lock = self.work_condvar.wait(lock).unwrap();

        }

        // We have work, or the schedulers should exit.

        return lock.pop_front();

    }

    pub(crate) fn enqueue_work(&self, key: CompKey) {

        let mut lock = self.work_queue.lock().unwrap();

        lock.push_back(key);

        self.work_condvar.notify_one();

    }

    // Creating/destroying components

    pub(crate) fn start_create_pdl_component(&self) -> CompReserved {

        self.increment_active_components();

        let reservation = self.components.reserve();

        return CompReserved{ reservation };

    }

        &self, reserved: CompReserved,

    ) -> (CompKey, &mut RuntimeComp) {

        let inbox_queue = QueueDynMpsc::new(16);

        let inbox_producer = inbox_queue.producer();

        let _id = reserved.id();

        context.id = reserved.id();

        let component = RuntimeComp {

            public: CompPublic{

                sleeping: AtomicBool::new(initially_sleeping),

                num_handles: AtomicU32::new(1), // the component itself acts like a handle

                inbox: inbox_producer,

            },

            ctx: context,

            inbox: inbox_queue,

            exiting: false,

        };

        let index = self.components.submit(reserved.reservation, component);

        debug_assert_eq!(index, _id.0);

        let component = self.components.get_mut(index);

        return (CompKey(index), component);

    }

    pub(crate) fn get_component(&self, key: CompKey) -> &mut RuntimeComp {

        let component = self.components.get_mut(key.0);

        return component;

    }

    pub(crate) fn get_component_public(&self, id: CompId) -> CompHandle {

        let component = self.components.get(id.0);

        return CompHandle::new(id, &component.public);

    }

    pub(crate) fn destroy_component(&self, key: CompKey) {

        dbg_code!({

    pub fn new(runtime: Arc<RuntimeInner>, scheduler_id: u32, debug_logging: bool) -> Self {

        return Scheduler{ runtime, scheduler_id, debug_logging }

    }

    pub fn run(&mut self) {

        let mut scheduler_ctx = SchedulerCtx::new(&*self.runtime, self.scheduler_id, self.debug_logging);

        'run_loop: loop {

            // Wait until we have something to do (or need to quit)

            let comp_key = self.runtime.take_work();

            if comp_key.is_none() {

                break 'run_loop;

            }

            let comp_key = comp_key.unwrap();

            let component = self.runtime.get_component(comp_key);

            scheduler_ctx.comp = comp_key.0;

            // Run the component until it no longer indicates that it needs to

            // be re-executed immediately.

            let mut new_scheduling = CompScheduling::Immediate;

            while let CompScheduling::Immediate = new_scheduling {

                while let Some(message) = component.inbox.pop() {

                }

            }

            // Handle the new scheduling

            match new_scheduling {

                CompScheduling::Immediate => unreachable!(),

                CompScheduling::Requeue => { self.runtime.enqueue_work(comp_key); },

                CompScheduling::Sleep => { self.mark_component_as_sleeping(comp_key, component); },

                CompScheduling::Exit => { self.mark_component_as_exiting(&scheduler_ctx, component); }

            }

        }

    }

    // local utilities

    /// Marks component as sleeping, if after marking itself as sleeping the

    /// inbox contains messages then the component will be immediately

    /// rescheduled. After calling this function the component should not be

    /// executed anymore.

    fn mark_component_as_sleeping(&self, key: CompKey, component: &mut RuntimeComp) {

        debug_assert_eq!(key.downgrade(), component.ctx.id); // make sure component matches key

        debug_assert_eq!(component.public.sleeping.load(Ordering::Acquire), false); // we're executing it, so it cannot be sleeping

        component.public.sleeping.store(true, Ordering::Release);

        if component.inbox.can_pop() {