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,

}

    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);

                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)

    }

        // 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);

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;

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);

/// 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 (!)

        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);

    }

                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); }

            }

        }

    }