}

    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;

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

    }

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

}