pub locations: Vec<ValueId>, // within `content`

    pub messages: Vec<DataMessage>, // owned by previous component

    pub peer_comp: CompId,

    pub peer_port: PortId,

    pub kind: PortKind,

    pub state: PortState,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup, sched_ctx: &SchedulerCtx,

            let mut message = inbox_main[port_index].take().unwrap();

                target_port, target_port_instruction,

                &mut message, inbox_main, inbox_backup,

    targeted_port: PortId, port_instruction: PortInstruction, message: &mut DataMessage,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup,

    // If we received any ports, add them to the port tracking and inbox struct.

    // Then notify the peers that they can continue sending to this port, but

    // now at a new address.

    for received_port in &mut message.ports {

        // Transfer messages to main/backup inbox

        let new_inbox_index = inbox_main.len();

        if !received_port.messages.is_empty() {

            inbox_main.push(Some(received_port.messages.remove(0)));

        }

        inbox_backup.extend(received_port.messages.drain(..));

        // Create a new port locally

        let mut new_port_state = received_port.state;

        new_port_state.set(PortStateFlag::Received);

        let new_port_handle = comp_ctx.add_port(

            received_port.peer_comp, received_port.peer_port,

            received_port.kind, new_port_state

        );

        let new_port = comp_ctx.get_port(new_port_handle);

        comp_ctx.change_port_peer(sched_ctx, new_port_handle, Some(new_port.peer_comp_id));

        // Replace all references to the port in the received message

        for message_location in received_port.locations {

            let value = match message_location {

                ValueId::Heap(heap_pos, heap_index) => &mut message.content.regions[heap_pos as usize][heap_index as usize],

                ValueId::Stack(stack_index) => &mut message.content.values[stack_index as usize],

            };

            match value {

                Value::Input(_) => {

                    debug_assert_eq!(new_port.kind, PortKind::Getter);

                    *value = Value::Input(port_id_to_eval(new_port.self_id));

                },

                Value::Output(_) => {

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

                    *value = Value::Output(port_id_to_eval(new_port.self_id));

                },

                _ => unreachable!(),

            }

        }

        // Let the peer know that the port can now be used

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

        let peer_info = comp_ctx.get_peer(peer_handle);

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

            id: ControlId::new_invalid(),

            sender_comp_id: comp_ctx.id,

            target_port_id: Some(new_port.peer_port_id),

            content: ControlMessageContent::PortPeerChangedUnblock(new_port.id, comp_ctx.id)

        }), true);

    }

    message: ControlMessage, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

            default_handle_ack(exec_state, control, message.id, sched_ctx, comp_ctx, inbox_main, inbox_backup);

                default_handle_ack(exec_state, control, control_id, sched_ctx, comp_ctx, inbox_main, inbox_backup);

/// Prepares sending a message that contains ports. Only once a particular

/// protocol has completed (where we notify all the peers that the ports will

/// be transferred) will we actually send the message to the recipient.

fn prepare_send_message_with_ports(

    sending_port_id: PortId, sending_port_instruction: PortInstruction, value: ValueGroup,

    control: &mut ControlLayer

    let sending_port_handle = comp_ctx.get_port_handle(sending_port_id);

        let peer_comp_id = transmit_port_info.peer_comp_id;

        let peer_port_id = transmit_port_info.peer_port_id;

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

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

        // Set the flag for transmission

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

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

    // Find all ports again

    let mut transmit_ports = Vec::new();

    find_ports_in_value_group(&exec_state.mode_value, &mut transmit_ports);

    let port_handle = comp_ctx.get_port_handle(exec_state.mode_port);

    let port_info = comp_ctx.get_port(port_handle);

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

    // Annotate the data message

    let message_value = exec_state.mode_value.take();

    let mut annotated_message = consensus.annotate_data_message(comp_ctx, port_info, message_value);

    // And further enhance the message by adding data about the ports that are

    // being transferred

    for (port_locations, port_id) in transmit_ports {

        let transmit_port_handle = comp_ctx.get_port_handle(port_id);

        let transmit_port_info = comp_ctx.get_port(transmit_port_handle);

        let transmit_messages = take_port_messages(comp_ctx, transmit_port_id, inbox_main, inbox_backup);

        let mut transmit_port_state = transmit_port_info.state;

        debug_assert!(transmit_port_state.is_set(PortStateFlag::Transmitted));

        transmit_port_state.clear(PortStateFlag::Transmitted);

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

}

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

    sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, inbox_main: &mut InboxMain,

    inbox_backup: &mut InboxBackup

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

            },

/// Goes through the inbox of a component and takes out all the messages that

/// are targeted at a specific port

pub(crate) fn take_port_messages(

    comp_ctx: &CompCtx, port_id: PortId,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

) -> Vec<DataMessage> {

    let mut messages = Vec::new();

    let port_handle = comp_ctx.get_port_handle(port_id);

    let port_index = comp_ctx.get_port_index(port_handle);

    if let Some(message) = inbox_main[port_index].take() {

        messages.push(message);

    }

    let mut message_index = 0;

    while message_index < inbox_backup.len() {

        let message = &inbox_backup[message_index];

        if message.data_header.target_port == port_id {

            let message = inbox_backup.remove(message_index);

            messages.push(message);

        } else {

            message_index += 1;

        }

    }

    return messages;

}

    Received = 0x10, // Received, so cannot be used yet, only after the sync round

            !self.is_set(PortStateFlag::Transmitted) &&

            !self.is_set(PortStateFlag::Received);

            ("blocked_full_buffers", BlockedDueToFullBuffers),

            ("transmitted", Transmitted),

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

    // Port transfer

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

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

    /// Creates a PortPeerChanged message (and increments ack-counter on a

    /// pre-created control entry) that is used as a preliminary step before

    /// transferring a port over a channel.

    pub(crate) fn create_port_transfer_message(

        &mut self, associated_control_id: ControlId,

        sender_comp_id: CompId, target_port_id: PortId

    ) -> Message {

        let entry_index = self.get_entry_index_by_id(associated_control_id).unwrap();

        self.entries[entry_index].ack_countdown += 1;

        return Message::Control(ControlMessage{

            id: associated_control_id,

            sender_comp_id,

            target_port_id: Some(target_port_id),

            content: ControlMessageContent::PortPeerChangedBlock

        })

    }

pub(crate) use component_context::{CompCtx, PortInstruction, PortKind, PortState, PortStateFlag};