return Ok(CompScheduling::Immediate);

                    } else {

                        todo!("handle sync failure due to message deadlock");

                        return Ok(CompScheduling::Sleep);

                    }

                } else {

                    // We need to wait

                    self.mode = Mode::BlockedGet;

                    self.mode_port = port_id;

                    return Ok(CompScheduling::Sleep);

                }

            },

            EC::Put(port_id, value) => {

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

                let port_id = port_id_from_eval(port_id);

                let port_handle = comp_ctx.get_port_handle(port_id);

                let port_info = comp_ctx.get_port(port_handle);

                if port_info.state.is_blocked() {

                    self.mode = Mode::BlockedPut;

                    self.mode_port = port_id;

                    self.mode_value = value;

                    self.exec_ctx.stmt = ExecStmt::PerformedPut; // prepare for when we become unblocked

                    return Ok(CompScheduling::Sleep);

                } else {

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

                    self.exec_ctx.stmt = ExecStmt::PerformedPut;

                    return Ok(CompScheduling::Immediate);

                }

            },

            // Results that can be returned outside of sync mode

            EC::ComponentTerminated => {

                self.mode = Mode::StartExit; // next call we'll take care of the exit

                return Ok(CompScheduling::Immediate);

            },

            EC::SyncBlockStart => {

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

                self.handle_sync_start(sched_ctx, comp_ctx);

                return Ok(CompScheduling::Immediate);

            },

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

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

                self.create_component_and_transfer_ports(

                    sched_ctx, comp_ctx,

                    definition_id, monomorph_idx, 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);

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

        self.mode = Mode::Sync;

    }

    /// Handles end of sync. The conclusion to the sync round might arise

    /// immediately (and be handled immediately), or might come later through

    /// messaging. In any case the component should be scheduled again

    /// immediately

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

        sched_ctx.log("Component ending sync mode (now waiting for solution)");

        let decision = self.consensus.notify_sync_end(sched_ctx, comp_ctx);

        self.mode = Mode::SyncEnd;

        self.handle_sync_decision(sched_ctx, comp_ctx, decision);

    }

    /// Handles decision from the consensus round. This will cause a change in

    /// the internal `Mode`, such that the next call to `run` can take the

    /// appropriate next steps.

    fn handle_sync_decision(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, decision: SyncRoundDecision) {

        let is_success = match decision {

            SyncRoundDecision::None => {

                // No decision yet

                return;

            },

            SyncRoundDecision::Solution => true,

            SyncRoundDecision::Failure => false,

        };

        // If here then we've reached a decision

        if is_success {

            self.mode = Mode::NonSync;

            self.consensus.notify_sync_decision(decision);

        } else {

            self.mode = Mode::StartExit;

        }

    }

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

        sched_ctx.log("Component exiting");

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

        self.mode = Mode::BusyExit;

        // Doing this by index, then retrieving the handle is a bit rediculous,

        // but Rust is being Rust with its borrowing rules.

        for port_index in 0..comp_ctx.num_ports() {

            let port = comp_ctx.get_port_by_index_mut(port_index);

            if port.state == PortState::Closed {

                // Already closed, or in the process of being closed

                continue;

            }

            // Mark as closed

            let port_id = port.self_id;

            port.state = PortState::Closed;

            // Notify peer of closing

            let port_handle = comp_ctx.get_port_handle(port_id);

            let (peer, message) = self.control.initiate_port_closing(port_handle, comp_ctx);

            let peer_info = comp_ctx.get_peer(peer);

            peer_info.handle.send_message(sched_ctx, Message::Control(message), true);

        }

    }

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

    // Handling messages

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

    fn send_data_message_and_wake_up(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, source_port_handle: LocalPortHandle, value: ValueGroup) {

        let port_info = comp_ctx.get_port(source_port_handle);

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

        let peer_info = comp_ctx.get_peer(peer_handle);

        let annotated_message = self.consensus.annotate_data_message(comp_ctx, port_info, value);

        peer_info.handle.send_message(sched_ctx, Message::Data(annotated_message), true);

    }

    /// Handles a message that came in through the public inbox. This function

    /// will handle putting it in the correct place, and potentially blocking

    /// the port in case too many messages are being received.

    fn handle_incoming_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: DataMessage) {

        // Check if we can insert it directly into the storage associated with

        // the port

        let target_port_id = message.data_header.target_port;

        let port_handle = comp_ctx.get_port_handle(target_port_id);

        let port_index = comp_ctx.get_port_index(port_handle);

        if self.inbox_main[port_index].is_none() {

            self.inbox_main[port_index] = Some(message);

            // After direct insertion, check if this component's execution is 

            // blocked on receiving a message on that port

            debug_assert!(!comp_ctx.get_port(port_handle).state.is_blocked()); // because we could insert directly

            if self.mode == Mode::BlockedGet && self.mode_port == target_port_id {

                // We were indeed blocked

                self.mode = Mode::Sync;

                self.mode_port = PortId::new_invalid();

            }

            return;

        }

        // The direct inbox is full, so the port will become (or was already) blocked

        let port_info = comp_ctx.get_port_mut(port_handle);

        debug_assert!(port_info.state == PortState::Open || port_info.state.is_blocked());

        if port_info.state == PortState::Open {

            comp_ctx.set_port_state(port_handle, PortState::BlockedDueToFullBuffers);

            let (peer_handle, message) =

                self.control.initiate_port_blocking(comp_ctx, port_handle);

            let peer = comp_ctx.get_peer(peer_handle);

            peer.handle.send_message(sched_ctx, Message::Control(message), true);

        }

        // But we still need to remember the message, so:

        self.inbox_backup.push(message);

    }

    /// Handles when a message has been handed off from the inbox to the PDL

    /// code. We check to see if there are more messages waiting and, if not,

    /// then we handle the case where the port might have been blocked

    /// previously.

    fn handle_received_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, port_handle: LocalPortHandle) {

        let port_index = comp_ctx.get_port_index(port_handle);

        debug_assert!(self.inbox_main[port_index].is_none()); // this function should be called after the message is taken out

        // Check for any more messages

        if port_info.state == PortState::BlockedDueToFullBuffers {

            comp_ctx.set_port_state(port_handle, PortState::Open);

            let (peer_handle, message) = self.control.cancel_port_blocking(comp_ctx, port_handle);

            let peer_info = comp_ctx.get_peer(peer_handle);

            peer_info.handle.send_message(sched_ctx, Message::Control(message), true);

        }

    }

    fn handle_incoming_control_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: ControlMessage) {

        // Little local utility to send an Ack

        fn send_control_ack_message(sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, causer_id: ControlId, peer_handle: LocalPeerHandle) {

            let peer_info = comp_ctx.get_peer(peer_handle);

            peer_info.handle.send_message(sched_ctx, Message::Control(ControlMessage{

                id: causer_id,

                sender_comp_id: comp_ctx.id,

                target_port_id: None,

                content: ControlMessageContent::Ack,

            }), true);

        }

        // Handle the content of the control message, and optionally Ack it

        match message.content {

            ControlMessageContent::Ack => {

                self.handle_ack(sched_ctx, comp_ctx, message.id);

            },

            ControlMessageContent::BlockPort(port_id) => {

                // On of our messages was accepted, but the port should be

                // blocked.

                let port_handle = comp_ctx.get_port_handle(port_id);

                let port_info = comp_ctx.get_port(port_handle);

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

                if port_info.state == PortState::Open {

                    // only when open: we don't do this when closed, and we we don't do this if we're blocked due to peer changes

                    comp_ctx.set_port_state(port_handle, PortState::BlockedDueToFullBuffers);

                }

            },

            ControlMessageContent::ClosePort(port_id) => {

                // Request to close the port. We immediately comply and remove

                // the component handle as well

                let port_handle = comp_ctx.get_port_handle(port_id);

                let peer_comp_id = comp_ctx.get_port(port_handle).peer_comp_id;

                let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);

                // One exception to sending an `Ack` is if we just closed the

                // port ourselves, meaning that the `ClosePort` messages got

                // sent to one another.

                if let Some(control_id) = self.control.has_close_port_entry(port_handle, comp_ctx) {

                    self.handle_ack(sched_ctx, comp_ctx, control_id);

                } else {

                    send_control_ack_message(sched_ctx, comp_ctx, message.id, peer_handle);

                    comp_ctx.remove_peer(sched_ctx, port_handle, peer_comp_id, false); // do not remove if closed

                    comp_ctx.set_port_state(port_handle, PortState::Closed); // now set to closed

                }

            },

            ControlMessageContent::UnblockPort(port_id) => {

                // We were previously blocked (or already closed)

                let port_handle = comp_ctx.get_port_handle(port_id);

                let port_info = comp_ctx.get_port(port_handle);

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

                if port_info.state == PortState::BlockedDueToFullBuffers {

                    self.handle_unblock_port_instruction(sched_ctx, comp_ctx, port_handle);

                }

            },

            ControlMessageContent::PortPeerChangedBlock(port_id) => {

                // The peer of our port has just changed. So we are asked to

                // temporarily block the port (while our original recipient is

                // potentially rerouting some of the in-flight messages) and

                // Ack. Then we wait for the `unblock` call.

                debug_assert_eq!(message.target_port_id, Some(port_id));

                let port_handle = comp_ctx.get_port_handle(port_id);

                comp_ctx.set_port_state(port_handle, PortState::BlockedDueToPeerChange);

                let port_info = comp_ctx.get_port(port_handle);

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

                send_control_ack_message(sched_ctx, comp_ctx, message.id, peer_handle);

            },

            ControlMessageContent::PortPeerChangedUnblock(new_port_id, new_comp_id) => {

                let port_handle = comp_ctx.get_port_handle(message.target_port_id.unwrap());

                let port_info = comp_ctx.get_port(port_handle);

                debug_assert!(port_info.state == PortState::BlockedDueToPeerChange);

                let old_peer_id = port_info.peer_comp_id;

                comp_ctx.remove_peer(sched_ctx, port_handle, old_peer_id, false);

                let port_info = comp_ctx.get_port_mut(port_handle);

                port_info.peer_comp_id = new_comp_id;

                port_info.peer_port_id = new_port_id;

                comp_ctx.add_peer(port_handle, sched_ctx, new_comp_id, None);

                self.handle_unblock_port_instruction(sched_ctx, comp_ctx, port_handle);

            }

        }

    }

    fn handle_incoming_sync_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: SyncMessage) {

        let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);

        self.handle_sync_decision(sched_ctx, comp_ctx, decision);

    }

    /// Little helper that notifies the control layer of an `Ack`, and takes the

    /// appropriate subsequent action

    fn handle_ack(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, control_id: ControlId) {

        let mut to_ack = control_id;

        loop {

            let (action, new_to_ack) = self.control.handle_ack(to_ack, sched_ctx, comp_ctx);

            match action {

                AckAction::SendMessage(target_comp, message) => {

                    // FIX @NoDirectHandle

                    let mut handle = sched_ctx.runtime.get_component_public(target_comp);

                    handle.send_message(sched_ctx, Message::Control(message), true);

                    let _should_remove = handle.decrement_users();

                    debug_assert!(_should_remove.is_none());

                },

                AckAction::ScheduleComponent(to_schedule) => {

                    // FIX @NoDirectHandle

                    let mut handle = sched_ctx.runtime.get_component_public(to_schedule);

                    // Note that the component is intentionally not

                    // sleeping, so we just wake it up

                    debug_assert!(!handle.sleeping.load(std::sync::atomic::Ordering::Acquire));

                    let key = unsafe{ to_schedule.upgrade() };

                    sched_ctx.runtime.enqueue_work(key);

                    let _should_remove = handle.decrement_users();

                    debug_assert!(_should_remove.is_none());

                },

                AckAction::None => {}

            }

            match new_to_ack {

                Some(new_to_ack) => to_ack = new_to_ack,

                None => break,

            }

        }

    }

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

    // Handling ports

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

    /// Unblocks a port, potentially continuing execution of the component, in

    /// response to a message that told us to unblock a previously blocked

    fn handle_unblock_port_instruction(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, port_handle: LocalPortHandle) {

        let port_info = comp_ctx.get_port_mut(port_handle);

        let port_id = port_info.self_id;

        debug_assert!(port_info.state.is_blocked());

        port_info.state = PortState::Open;

        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: DefinitionId, monomorph_index: i32, mut arguments: ValueGroup

    ) {

        struct PortPair{

            creator_handle: LocalPortHandle,

            creator_id: PortId,

            created_handle: LocalPortHandle,

            created_id: PortId,

        }

        let mut port_id_pairs = Vec::new();

        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 = ValueGroupIter::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;

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

    let (key, _) = rt.inner.finish_create_pdl_component(reserved, CompPDL::new(prompt, 0), ctx, false);

    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, 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) {

        u32 outside_index = 0;

        while (outside_index < outside_loops) {

            u32 inside_index = 0;

            sync while (inside_index < inside_loops) {

                put(o, inside_index);

                inside_index += 1;

            }

            outside_index += 1;

        }

    }

    primitive receiver(in<u32> i, u32 outside_loops, u32 inside_loops) {

        u32 outside_index = 0;

        while (outside_index < outside_loops) {

            u32 inside_index = 0;

            sync while (inside_index < inside_loops) {

                auto val = get(i);

                while (val != inside_index) {} // infinite loop if incorrect value is received

                inside_index += 1;

            }

            outside_index += 1;

        }

    }

    composite constructor() {

        channel o_orom -> i_orom;

        channel o_mrom -> i_mrom;

        channel o_ormm -> i_ormm;

        channel o_mrmm -> i_mrmm;

        // one round, one message per round

        new sender(o_orom, 1, 1);

        new receiver(i_orom, 1, 1);

        // multiple rounds, one message per round

        new sender(o_mrom, 5, 1);

        new receiver(i_mrom, 5, 1);

        // one round, multiple messages per round

        new sender(o_ormm, 1, 5);

        new receiver(i_ormm, 1, 5);

        // multiple rounds, multiple messages per round

        new sender(o_mrmm, 5, 5);

        new receiver(i_mrmm, 5, 5);

    }").expect("compilation");

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

}