PortPeerChangedUnblock(PortId, CompId), // contains (new_port_id, new_component_id)

}

#[derive(Copy, Clone, Debug)]

pub struct ControlMessageClosePort {

    pub closed_in_sync_round: bool, // needed to ensure correct handling of errors

}

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

// Messages (generic)

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

) -> Result<CompScheduling, (PortInstruction, String)> {

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

    let port_handle = comp_ctx.get_port_handle(transmitting_port_id);

    let port_info = comp_ctx.get_port_mut(port_handle);

    port_info.last_instruction = port_instruction;

    let port_info = comp_ctx.get_port(port_handle);

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

    let mut ports = Vec::new();

    find_ports_in_value_group(&value, &mut ports);

            target_port_id: Some(new_port.peer_port_id),

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

        }), true);

    }

    // Modify last-known location where port instruction was retrieved

    debug_assert_ne!(port_info.last_instruction, PortInstruction::None); // set by caller

    debug_assert!(port_info.state.is_open()); // checked by caller

    // Check if there are any more messages in the backup buffer

    for message_index in 0..inbox_backup.len() {

        let message = &inbox_backup[message_index];

        if message.data_header.target_port == targeted_port {

            // One more message, place it in the slot

                // the channel at the same time. So we don't care about the

                // content of the `ClosePort` message.

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

            } else {

                // Respond to the message

                let port_info = comp_ctx.get_port(port_handle);

                let last_instruction = port_info.last_instruction;

                let port_has_had_message = port_info.received_message_for_sync;

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

                comp_ctx.change_port_peer(sched_ctx, port_handle, None);

                // Handle any possible error conditions (which boil down to: the

                // were actually received. It might also mean that e.g. the

                // component attempted a `get`, but there were no messages, so

                // now it is in the `BlockedGet` state.

                let port_was_used = last_instruction != PortInstruction::None;

                if exec_state.mode.is_in_sync_block() {

                        return Err((

                            last_instruction,

                            format!("Peer component (id:{}) shut down, so communication cannot (have) succeed(ed)", peer_comp_id.0)

                        ));

                    }

                } else {

    pub peer_comp_id: CompId, // eventually consistent

    pub peer_port_id: PortId, // eventually consistent

    // Generic operating state

    pub kind: PortKind,

    pub state: PortState,

    // State tracking for error detection and error handling

    pub last_instruction: PortInstruction, // used during sync round to detect port-closed-during-sync errors

    pub received_message_for_sync: bool, // used during sync round to detect port-closed-before-sync errors

    pub close_at_sync_end: bool, // set during sync round when receiving a port-closed-after-sync message

    pub(crate) associated_with_peer: bool,

}

        self.ports.push(Port{

            self_id: putter_id,

            peer_port_id: getter_id,

            kind: PortKind::Putter,

            state: PortState::new(),

            peer_comp_id: self.id,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            received_message_for_sync: false,

            associated_with_peer: false,

        });

        self.ports.push(Port{

            self_id: getter_id,

            peer_port_id: putter_id,

            kind: PortKind::Getter,

            state: PortState::new(),

            peer_comp_id: self.id,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            received_message_for_sync: false,

            associated_with_peer: false,

        });

    /// Adds a new port. Make sure to call `change_peer` afterwards.

    pub(crate) fn add_port(&mut self, peer_comp_id: CompId, peer_port_id: PortId, kind: PortKind, state: PortState) -> LocalPortHandle {

        let self_id = PortId(self.take_port_id());

        self.ports.push(Port{

            self_id, peer_comp_id, peer_port_id, kind, state,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            received_message_for_sync: false,

            associated_with_peer: false,

        });

        return LocalPortHandle(self_id);

            mapping_counter: 0,

            mode: Mode::NonSync,

            solution: SolutionCombiner::new(),

        }

    }

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

    // Managing sync state

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

    /// Notifies the consensus management that the PDL code has reached the

    /// start of a sync block.

            ControlMessage{

                id: entry_id,

                sender_comp_id: comp_ctx.id,

                target_port_id: Some(peer_port_id),

                content: ControlMessageContent::ClosePort(ControlMessageClosePort{

                    closed_in_sync_round: exit_inside_sync,

                }),

            }

        );

    }

    /// Generates the control message used to indicate to a peer that a port