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

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

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

}