use crate::protocol::eval::*;

use crate::protocol::*;

use crate::runtime2::*;

use super::{CompCtx, CompPDL};

pub enum CompScheduling {

    Immediate,

    Requeue,

    Sleep,

    Exit,

}

/// Generic representation of a component (as viewed by a scheduler).

pub(crate) trait Component {

    /// Called if the component is created by another component and the messages

    /// are being transferred between the two.

    fn adopt_message(&mut self, comp_ctx: &mut CompCtx, message: DataMessage);

    /// Called if the component receives a new message. The component is

    /// responsible for deciding where that messages goes.

    fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message);

    /// Called if the component's routine should be executed. The return value

    /// can be used to indicate when the routine should be run again.

    fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> Result<CompScheduling, EvalError>;

}

/// Creates a new component based on its definition. Meaning that if it is a

/// user-defined component then we set up the PDL code state. Otherwise we

/// construct a custom component. This does NOT take care of port and message

/// management.

pub(crate) fn create_component(

    protocol: &ProtocolDescription,

    definition_id: ProcedureDefinitionId, type_id: TypeId,

    arguments: ValueGroup, num_ports: usize

) -> Box<dyn Component> {

    let definition = &protocol.heap[definition_id];

    debug_assert!(definition.kind == ProcedureKind::Primitive || definition.kind == ProcedureKind::Composite);

    if definition.source.is_builtin() {

        // Builtin component

    } else {

        // User-defined component

        let prompt = Prompt::new(

            &protocol.types, &protocol.heap,

            definition_id, type_id, arguments

        );

        let component = CompPDL::new(prompt, num_ports);

        return Box::new(component);

    }

}

use crate::protocol::eval::*;

use crate::runtime2::*;

use super::*;

/// TODO: Temporary component to figure out what to do with custom components.

///     This component sends random numbers between two u32 limits

}

    fn adopt_message(&mut self, _comp_ctx: &mut CompCtx, _message: DataMessage) {

        unreachable!("should not adopt messages");

    }

    fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message) {

        todo!()

    }

    fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> Result<CompScheduling, EvalError> {

        todo!()

    }

}

        if self.mode == Mode::BlockedPut && port_id == self.mode_port {

            // We were blocked on the port that just became unblocked, so

            // send the message.

            debug_assert_eq!(port_info.kind, PortKind::Putter);

            let mut replacement = ValueGroup::default();

            std::mem::swap(&mut replacement, &mut self.mode_value);

            self.send_data_message_and_wake_up(sched_ctx, comp_ctx, port_handle, replacement);

            self.mode = Mode::Sync;

            self.mode_port = PortId::new_invalid();

        }

    }

    fn create_component_and_transfer_ports(

        &mut self,

        sched_ctx: &SchedulerCtx, creator_ctx: &mut CompCtx,

        definition_id: ProcedureDefinitionId, type_id: TypeId, mut arguments: ValueGroup

    ) {

        struct PortPair{

            creator_handle: LocalPortHandle,

            creator_id: PortId,

            created_handle: LocalPortHandle,

            created_id: PortId,

        }

        let reservation = sched_ctx.runtime.start_create_pdl_component();

        let mut created_ctx = CompCtx::new(&reservation);

        // Take all the ports ID that are in the `args` (and currently belong to

        // the creator component) and translate them into new IDs that are

        // associated with the component we're about to create

        let mut arg_iter = ValueGroupPortIter::new(&mut arguments);

        while let Some(port_reference) = arg_iter.next() {

            // Create port entry for new component

            let creator_port_id = port_reference.id;

            let creator_port_handle = creator_ctx.get_port_handle(creator_port_id);

            let creator_port = creator_ctx.get_port(creator_port_handle);

            let created_port_handle = created_ctx.add_port(

                creator_port.peer_comp_id, creator_port.peer_port_id,

                creator_port.kind, creator_port.state

            );

            let created_port = created_ctx.get_port(created_port_handle);

            let created_port_id = created_port.self_id;

                creator_handle: creator_port_handle,

                creator_id: creator_port_id,

                created_handle: created_port_handle,

                created_id: created_port_id,

            // Modify value in arguments (bit dirty, but double vec in ValueGroup causes lifetime issues)

            let arg_value = if let Some(heap_pos) = port_reference.heap_pos {

                &mut arg_iter.group.regions[heap_pos][port_reference.index]

            } else {

                &mut arg_iter.group.values[port_reference.index]

            };

            match arg_value {

                Value::Input(id) => *id = port_id_to_eval(created_port_id),

                Value::Output(id) => *id = port_id_to_eval(created_port_id),

                _ => unreachable!(),

            }

        }

        // For each transferred port pair set their peer components to the

        // correct values. This will only change the values for the ports of

        // the new component.

        let mut created_component_has_remote_peers = false;

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

            let created_port_info = created_ctx.get_port_mut(pair.created_handle);

            if created_port_info.peer_comp_id == creator_ctx.id {

                // Port peer is owned by the creator as well

                    .iter()

                    .position(|v| v.creator_id == creator_port_info.peer_port_id);

                match created_peer_port_index {

                    Some(created_peer_port_index) => {

                        // Peer port moved to the new component as well. So

                        // adjust IDs appropriately.

                        created_port_info.peer_port_id = peer_pair.created_id;

                        created_port_info.peer_comp_id = reservation.id();

                        todo!("either add 'self peer', or remove that idea from Ctx altogether")

                    },

                    None => {

                        // Peer port remains with creator component.

                        created_port_info.peer_comp_id = creator_ctx.id;

                        created_ctx.add_peer(pair.created_handle, sched_ctx, creator_ctx.id, None);

                    }

                }

            } 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 (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

        // potentially remove the peer component.

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

                component.component.adopt_message(&mut component.ctx, message)

            }

            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;

                    component.component.adopt_message(&mut component.ctx, message);

                } else {

                    message_index += 1;

                }

            }

            // Handle potential channel between creator and created component

            let created_port_info = component.ctx.get_port(pair.created_handle);

            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_comp_id = component.ctx.id;

                peer_port_info.peer_port_id = created_port_info.self_id;

                creator_ctx.add_peer(peer_port_handle, sched_ctx, component.ctx.id, None);

            }

        }

        // By now all ports and messages have been transferred. If there are any

        // peers that need to be notified about this new component, then we

        // initiate the protocol that will notify everyone here.

        if created_component_has_remote_peers {

            let created_ctx = &component.ctx;

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

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

        }

    }

}

/// Recursively goes through the value group, attempting to find ports.

/// Duplicates will only be added once.

pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut Vec<PortId>) {

    // Helper to check a value for a port and recurse if needed.

    fn find_port_in_value(group: &ValueGroup, value: &Value, ports: &mut Vec<PortId>) {

        match value {

            Value::Input(port_id) | Value::Output(port_id) => {

                // This is an actual port

                let cur_port = PortId(port_id.id);

                for prev_port in ports.iter() {

                    if *prev_port == cur_port {

                        // Already added

                        return;

                    }

                }

                ports.push(cur_port);

            },

            Value::Array(heap_pos) |

            Value::Message(heap_pos) |

            Value::String(heap_pos) |

            Value::Struct(heap_pos) |

            Value::Union(_, heap_pos) => {

                // Reference to some dynamic thing which might contain ports,

            inner: runtime_inner,

            threads: Vec::with_capacity(num_threads as usize),

        };

        for thread_index in 0..num_threads {

            let mut scheduler = Scheduler::new(runtime.inner.clone(), thread_index, debug_logging);

            let thread_handle = std::thread::spawn(move || {

                scheduler.run();

            });

            runtime.threads.push(thread_handle);

        }

        return runtime;

    }

    pub fn create_component(&self, module_name: &[u8], routine_name: &[u8]) -> Result<(), ComponentCreationError> {

        use crate::protocol::eval::ValueGroup;

        let prompt = self.inner.protocol.new_component(

            module_name, routine_name,

            ValueGroup::new_stack(Vec::new())

        )?;

        let reserved = self.inner.start_create_pdl_component();

        let ctx = CompCtx::new(&reserved);

        self.inner.enqueue_work(key);

        return Ok(())

    }

}

impl Drop for Runtime {

    fn drop(&mut self) {

        self.inner.decrement_active_components();

        for handle in self.threads.drain(..) {

            handle.join().expect("join scheduler thread");

        }

    }

}

/// Memory that is maintained by "the runtime". In practice it is maintained by

/// multiple schedulers, and this serves as the common interface to that memory.

pub(crate) struct RuntimeInner {

    pub protocol: ProtocolDescription,

    components: ComponentStore<RuntimeComp>,

    work_queue: Mutex<VecDeque<CompKey>>,

    work_condvar: Condvar,

    active_elements: AtomicU32, // active components and APIs (i.e. component creators)

}

    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!({

use crate::protocol::*;

use crate::protocol::eval::*;

use crate::runtime2::runtime::*;

use crate::runtime2::component::{CompCtx, CompPDL};

fn create_component(rt: &Runtime, module_name: &str, routine_name: &str, args: ValueGroup) {

    let prompt = rt.inner.protocol.new_component(

        module_name.as_bytes(), routine_name.as_bytes(), args

    ).expect("create prompt");

    let reserved = rt.inner.start_create_pdl_component();

    let ctx = CompCtx::new(&reserved);

    rt.inner.enqueue_work(key);

}

fn no_args() -> ValueGroup { ValueGroup::new_stack(Vec::new()) }

#[test]

fn test_component_creation() {

    let pd = ProtocolDescription::parse(b"

    primitive nothing_at_all() {

        s32 a = 5;

        auto b = 5 + a;

    }

    ").expect("compilation");

    let rt = Runtime::new(1, true, pd);

    for _i in 0..20 {

        create_component(&rt, "", "nothing_at_all", no_args());

    }

}

#[test]

fn test_component_communication() {

    let pd = ProtocolDescription::parse(b"

    primitive sender(out<u32> o, u32 outside_loops, u32 inside_loops) {

            sync select {}

            index += 1;

        }

    }

    ").expect("compilation");

    let rt = Runtime::new(3, false, pd);

    create_component(&rt, "", "constructor", no_args());

}

#[test]

fn test_random_u32_temporary_thingo() {

    let pd = ProtocolDescription::parse(b"

    primitive random_taker(in<u32> generator) {

        sync {

            auto a = get(generator);

        }

    }

    composite constructor() {

        channel tx -> rx;

        new random_u32(tx, 1, 100);

        new random_taker(rx);

    }

    ").expect("compilation");

}