BlockedGet, // blocked because we need to receive a message on a particular port

    BlockedPut, // component is blocked because the port is blocked

    BlockedSelect, // waiting on message to complete the select statement

    NewComponentBlocked, // waiting until ports are in the appropriate state to create a new component

    StartExit, // temporary state: if encountered then we start the shutdown process.

    BusyExit, // temporary state: waiting for Acks for all the closed ports, potentially waiting for sync round to finish

        match self {

            Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect |

            NonSync | NewComponentBlocked | StartExit | BusyExit | Exit => false,

        }

    }

        match self {

            NonSync | Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect |

                NewComponentBlocked => false,

            StartExit | BusyExit => true,

            Exit => false,

    }

    pub(crate) fn set_as_blocked_put_with_ports(&mut self, port: PortId, value: ValueGroup) {

        self.mode_port = port;

        self.mode_value = value;

    }

            self.mode_port == port;

    }

    pub(crate) fn is_blocked_on_create_component(&self) -> bool {

        return self.mode == CompMode::NewComponentBlocked;

    }

            format!("Cannot send on this port, as the peer (id:{}) has shut down", port_info.peer_comp_id.0)

        ))

    } else if !ports.is_empty() {

            transmitting_port_id, port_instruction, value, exec_state,

            comp_ctx, sched_ctx, control

        )?;

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

    match message.content {

        ControlMessageContent::Ack => {

        },

        ControlMessageContent::BlockPort => {

            // One of our messages was accepted, but the port should be

                // The two components (sender and this component) are closing

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

                // content of the `ClosePort` message.

            } else {

                // Respond to the message

                let port_info = comp_ctx.get_port(port_handle);

            default_handle_recently_unblocked_port(

                exec_state, control, consensus, port_handle, sched_ctx,

                comp_ctx, inbox_main, inbox_backup

        },

        ControlMessageContent::PortPeerChangedBlock => {

            // The peer of our port has just changed. So we are asked to

            default_handle_recently_unblocked_port(

                exec_state, control, consensus, port_handle, 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.

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

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

    control: &mut ControlLayer

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

    debug_assert_eq!(exec_state.mode, CompMode::Sync); // busy in sync, trying to send

    let sending_port_handle = comp_ctx.get_port_handle(sending_port_id);

    let sending_port_info = comp_ctx.get_port_mut(sending_port_handle);

    sending_port_info.last_instruction = sending_port_instruction;

    let mut transmit_ports = Vec::new();

    find_ports_in_value_group(&value, &mut transmit_ports);

    // Set up the final Ack that triggers us to send our final message

    let unblock_put_control_id = control.add_unblock_put_with_ports_entry();

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

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

    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. Port must be ready to send!

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

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

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

    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

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

    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;

                // Send the message (containing ports) stored in the component

                // execution state to the recipient

                println!("DEBUG: Unblocking put with ports");

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

                default_handle_recently_unblocked_port(

                    exec_state, control, consensus, port_handle, sched_ctx,

                    comp_ctx, inbox_main, inbox_backup

            },

            AckAction::None => {}

        }

            None => break,

        }

    }

}

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

    exec_state: &mut CompExecState, control: &mut ControlLayer, consensus: &mut Consensus,

    port_handle: LocalPortHandle, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

    let port_info = comp_ctx.get_port_mut(port_handle);

    let port_id = port_info.self_id;

    if port_info.state.is_blocked() {

        // Port is still blocked. We wait until the next control message where

        // we unblock the port.

    }

    if exec_state.is_blocked_on_put_without_ports(port_id) {

        // Annotate the message that we're going to send

        let port_info = comp_ctx.get_port(port_handle); // for immutable access

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

        let peer_info = comp_ctx.get_peer(peer_handle);

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

        exec_state.mode = CompMode::Sync;

        exec_state.mode_port = PortId::new_invalid();

    } else if exec_state.is_blocked_on_create_component() {

        let mut ports = Vec::new();

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

            );

        }

    }

}

#[inline]

        match self.exec_state.mode {

            CompMode::BlockedSelect |

            CompMode::NewComponentBlocked => {

                // Not possible: we never enter this state

                unreachable!();

                // continue and run PDL code

            },

            CompMode::SyncEnd | CompMode::BlockedGet |

            CompMode::NewComponentBlocked => {

                return CompScheduling::Sleep;

            }

        sched_ctx.info(&format!("Running component ComponentRandomU32 (mode: {:?})", self.exec_state.mode));

        match self.exec_state.mode {

            CompMode::NewComponentBlocked => {

                // impossible for this component, no input ports and no select

                // blocks