let peer_comp_id = transmit_port_info.peer_comp_id;

        let peer_port_id = transmit_port_info.peer_port_id;

        // Note: we checked earlier that we are currently in sync mode. Now we

        // will check if we've already used the port we're about to transmit.

        if !transmit_port_info.last_instruction.is_none() {

            return Err((

                sending_port_instruction,

                String::from("Cannot transmit one of the ports in this message, as it is used in this sync round")

            ));

        }

        if transmit_port_info.state.is_set(PortStateFlag::Transmitted) {

            return Err((

                sending_port_instruction,

                String::from("Cannot transmit one of the ports in this message, as that port is already transmitted")

            ));

        }

        // Set the flag for transmission

        transmit_port_info.state.set(PortStateFlag::Transmitted);

        // Block the peer of the port

        let message = control.create_port_transfer_message(unblock_put_control_id, comp_ctx.id, peer_port_id);

        let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);

        let peer_info = comp_ctx.get_peer(peer_handle);

        peer_info.handle.send_message_logged(sched_ctx, message, true);

    }

    // We've set up the protocol, once all the PPC's are blocked we are supposed

    // to transfer the message to the recipient. So store it temporarily

    exec_state.mode = CompMode::BlockedPutPorts;

    exec_state.mode_port = sending_port_id;

    exec_state.mode_value = value;

    return Ok(());

}

/// Performs the transmission of a data message that contains ports. These were

/// all stored in the component's execution state by the

/// `prepare_send_message_with_ports` function.

fn perform_send_message_with_ports(

    exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, consensus: &mut Consensus,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

) {

    debug_assert_eq!(exec_state.mode, CompMode::BlockedPutPorts);

        annotated_message.ports.push(TransmittedPort{

            locations: port_locations,

            messages: transmit_messages,

            peer_comp: transmit_port_info.peer_comp_id,

            peer_port: transmit_port_info.peer_port_id,

            kind: transmit_port_info.kind,

            state: transmit_port_state

        })

    }

    // And finally, send the message to the peer

    let peer_info = comp_ctx.get_peer(peer_handle);

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

}

/// Handles an `Ack` for the control layer.

fn default_handle_ack(

    exec_state: &mut CompExecState, control: &mut ControlLayer, control_id: ControlId,

    sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

) {

    // Since an `Ack` may cause another one, handle them in a loop

    let mut to_ack = control_id;

    loop {

        let (action, new_to_ack) = 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_logged(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::UnblockPutWithPorts => {

                // Send the message (containing ports) to the recipient

                perform_send_message_with_ports(exec_state, sched_ctx, comp_ctx, consensus, inbox_main, inbox_backup);

                exec_state.mode = CompMode::Sync;

                exec_state.mode_port = PortId::new_invalid();

            },

            AckAction::None => {}

        }

        match new_to_ack {

            Some(new_to_ack) => to_ack = new_to_ack,

            None => break,

        }

    }

}

/// Little helper for sending the most common kind of `Ack`

fn default_send_ack(

    causer_of_ack_id: ControlId, peer_handle: LocalPeerHandle,

    sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx

) {

    let peer_info = comp_ctx.get_peer(peer_handle);

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

        id: causer_of_ack_id,

        sender_comp_id: comp_ctx.id,

        target_port_id: None,

    pub inbox_main: InboxMain,

    pub inbox_backup: InboxBackup,

}

impl Component for CompPDL {

    fn on_creation(&mut self, _id: CompId, _sched_ctx: &SchedulerCtx) {

        // Intentionally empty

    }

    fn on_shutdown(&mut self, _sched_ctx: &SchedulerCtx) {

        // Intentionally empty

    }

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

        let port_handle = comp_ctx.get_port_handle(message.data_header.target_port);

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

        } else {

            self.inbox_backup.push(message);

        }

    }

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

        if let Some(new_target) = self.control.should_reroute(&mut message) {

            let mut target = sched_ctx.runtime.get_component_public(new_target); // TODO: @NoDirectHandle

            target.send_message_logged(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) => {

                if let Err(location_and_message) = component::default_handle_control_message(

                    &mut self.exec_state, &mut self.control, &mut self.consensus,

                    message, sched_ctx, comp_ctx, &mut self.inbox_main, &mut self.inbox_backup

                ) {

                    self.handle_generic_component_error(sched_ctx, location_and_message);

                }

            },

            Message::Sync(message) => {

                self.handle_incoming_sync_message(sched_ctx, comp_ctx, message);

            },

            Message::Poll => {

                unreachable!(); // because we never register at the polling thread

            }

        }

    }

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

        use EvalContinuation as EC;

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

use crate::runtime2::communication::*;

use crate::runtime2::component::*;

use super::component_context::*;

#[derive(Debug, PartialEq, Eq, Clone, Copy)]

impl ControlId {

    /// Like other invalid IDs, this one doesn't care any significance, but is

    /// just set at u32::MAX to hopefully bring out bugs sooner.

    pub(crate) fn new_invalid() -> Self {

        return ControlId(u32::MAX);

    }

}

struct ControlEntry {

    id: ControlId,

    ack_countdown: u32,

    content: ControlContent,

}

enum ControlContent {

    PeerChange(ContentPeerChange),

    ScheduleComponent(CompId),

    ClosedPort(PortId),

    UnblockPutWithPorts

}

struct ContentPeerChange {

    source_port: PortId,

        return !self.entries.is_empty();

    }

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

    // Port transfer

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

    /// Adds an entry that, when completely ack'd, will schedule a component.

    pub(crate) fn add_schedule_entry(&mut self, to_schedule_id: CompId) -> ControlId {

        let entry_id = self.take_id();

        self.entries.push(ControlEntry{

            id: entry_id,

            ack_countdown: 0, // incremented by calls to `add_reroute_entry`

            content: ControlContent::ScheduleComponent(to_schedule_id),

        });

        return entry_id;

    }

    /// Adds an entry that returns the similarly named Ack action

    pub(crate) fn add_unblock_put_with_ports_entry(&mut self) -> ControlId {

        let entry_id = self.take_id();

        self.entries.push(ControlEntry{

            id: entry_id,

            content: ControlContent::UnblockPutWithPorts,

        });

        return entry_id;

    }

    /// Adds a rerouting entry (used to ensure all messages will end up at a

    /// newly created component). Used when creating a new component.

    pub(crate) fn add_reroute_entry(

        &mut self, creator_comp_id: CompId,

        source_port_id: PortId, source_comp_id: CompId,

        old_target_port_id: PortId, new_target_port_id: PortId, new_comp_id: CompId,

        schedule_entry_id: ControlId,

    ) -> Message {

        let entry_id = self.take_id();

        self.entries.push(ControlEntry{

            id: entry_id,

            ack_countdown: 1,

            content: ControlContent::PeerChange(ContentPeerChange{

                source_port: source_port_id,

                source_comp: source_comp_id,

                old_target_port: old_target_port_id,

                new_target_port: new_target_port_id,

                new_target_comp: new_comp_id,

use super::*;

#[test]

    compile_and_create_component("

    primitive port_sender(out<in<u32>> tx) {

        channel a -> b;

        sync put(tx, b);

    }

    primitive port_receiver(in<in<u32>> rx) {

        sync auto a = get(rx);

    }

    composite constructor() {

        channel a -> b;

        new port_sender(a);

        new port_receiver(b);

    }

    ", "constructor", no_args());

}