pub fn is_blocked(&self) -> bool {

        match self {

            PortState::BlockedDueToPeerChange | PortState::BlockedDueToFullBuffers => true,

            PortState::Open | PortState::Closed => false,

        }

    }

}

pub struct Channel {

    pub putter_id: PortId,

    pub getter_id: PortId,

}

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

// Data messages

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

#[derive(Debug)]

pub struct DataMessage {

    pub data_header: MessageDataHeader,

    pub sync_header: MessageSyncHeader,

    pub content: ValueGroup,

}

pub enum PortAnnotationKind {

    Getter(PortAnnotationGetter),

    Putter(PortAnnotationPutter),

}

pub struct PortAnnotationGetter {

    pub self_comp_id: CompId,

    pub self_port_id: PortId,

    pub peer_comp_id: CompId,

    pub peer_port_id: PortId,

}

pub struct PortAnnotationPutter {

    pub self_comp_id: CompId,

    pub self_port_id: PortId,

}

#[derive(Debug)]

pub struct MessageDataHeader {

    pub expected_mapping: Vec<(PortAnnotationKind, Option<u32>)>,

    pub new_mapping: u32,

    pub source_port: PortId,

    pub target_port: PortId,

}

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

// Sync messages

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

#[derive(Debug)]

pub struct SyncMessage {

    pub sync_header: MessageSyncHeader,

    pub content: SyncMessageContent,

}

#[derive(Debug)]

    }

    fn performed_select_wait(&mut self) -> Option<u32> {

        match self.stmt.take() {

            ExecStmt::None => return None,

            ExecStmt::PerformedSelectWait(selected_case) => Some(selected_case),

            _v => unreachable!(),

        }

    }

}

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

pub(crate) enum Mode {

    NonSync, // not in sync mode

    Sync, // in sync mode, can interact with other components

    SyncEnd, // awaiting a solution, i.e. encountered the end of the sync block

    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

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

    BusyExit, // temporary state: waiting for Acks for all the closed ports

    Exit, // exiting: shutdown process started, now waiting until the reference count drops to 0

}

struct SelectCase {

    involved_ports: Vec<LocalPortHandle>,

}

// TODO: @Optimize, flatten cases into single array, have index-pointers to next case

struct SelectState {

    cases: Vec<SelectCase>,

    next_case: u32,

    num_cases: u32,

    random: Random,

    candidates_workspace: Vec<usize>,

}

enum SelectDecision {

    None,

    Case(u32), // contains case index, should be passed along to PDL code

}

type InboxMain = Vec<Option<DataMessage>>;

impl SelectState {

    fn new() -> Self {

        return Self{

            cases: Vec::new(),

                    Value::Input(port_id_to_eval(channel.getter_id))

                ));

                self.inbox_main.push(None);

                self.inbox_main.push(None);

                return Ok(CompScheduling::Immediate);

            }

        }

    }

    fn execute_prompt(&mut self, sched_ctx: &SchedulerCtx) -> EvalResult {

        let mut step_result = EvalContinuation::Stepping;

        while let EvalContinuation::Stepping = step_result {

            step_result = self.prompt.step(

                &sched_ctx.runtime.protocol.types, &sched_ctx.runtime.protocol.heap,

                &sched_ctx.runtime.protocol.modules, &mut self.exec_ctx,

            )?;

        }

        return Ok(step_result)

    }

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

        sched_ctx.log(&format!("Handling sync decision: {:?} (in mode {:?})", decision, self.mode));

        let is_success = match decision {

            SyncRoundDecision::None => {

                // No decision yet

                return;

            // Notify peer of closing

            let port_handle = comp_ctx.get_port_handle(port_id);

            let (peer, message) = self.control.initiate_port_closing(port_handle, comp_ctx);

            let peer_info = comp_ctx.get_peer(peer);

            peer_info.handle.send_message(sched_ctx, Message::Control(message), true);

        }

    }

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

    // Handling messages

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

    fn send_data_message_and_wake_up(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, source_port_handle: LocalPortHandle, value: ValueGroup) {

        let port_info = comp_ctx.get_port(source_port_handle);

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

        let peer_info = comp_ctx.get_peer(peer_handle);

        let annotated_message = self.consensus.annotate_data_message(comp_ctx, port_info, value);

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

    }

    /// Handles a message that came in through the public inbox. This function

    /// will handle putting it in the correct place, and potentially blocking

    /// the port in case too many messages are being received.

    fn handle_incoming_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: DataMessage) {

        // Check if we can insert it directly into the storage associated with

        // the port

        let target_port_id = message.data_header.target_port;

        let port_handle = comp_ctx.get_port_handle(target_port_id);

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

            // After direct insertion, check if this component's execution is 

            // blocked on receiving a message on that port

            debug_assert!(!comp_ctx.get_port(port_handle).state.is_blocked()); // because we could insert directly

            if self.mode == Mode::BlockedGet && self.mode_port == target_port_id {

                // We were indeed blocked

                self.mode = Mode::Sync;

                self.mode_port = PortId::new_invalid();

            } else if self.mode == Mode::BlockedSelect {

                let select_decision = self.select.handle_updated_inbox(&self.inbox_main, comp_ctx);

                if let SelectDecision::Case(case_index) = select_decision {

                    self.exec_ctx.stmt = ExecStmt::PerformedSelectWait(case_index);

                    self.mode = Mode::Sync;

                }

            }

            return;

        for port in self.ports.iter_mut() {

            port.mapping = None;

        }

        self.solution.clear();

    }

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

    // Handling inbound and outbound messages

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

    /// Prepares a set of values to be sent of a channel.

    pub(crate) fn annotate_data_message(&mut self, comp_ctx: &CompCtx, port_info: &Port, content: ValueGroup) -> DataMessage {

        debug_assert_eq!(self.mode, Mode::SyncBusy); // can only send between sync start and sync end

        debug_assert!(self.ports.iter().any(|v| v.self_port_id == port_info.self_id));

        let data_header = self.create_data_header_and_update_mapping(port_info);

        let sync_header = self.create_sync_header(comp_ctx);

        return DataMessage{ data_header, sync_header, content };

    }

    /// Handles the arrival of a new data message (needs to be called for every

    /// new data message, even though it might not end up being received). This

    /// is used to determine peers of `get`ter ports.

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

        let target_index = comp_ctx.get_port_index(target_handle);

        let annotation = &mut self.ports[target_index];

        debug_assert!(

            !annotation.peer_discovered || (

                annotation.peer_comp_id == message.sync_header.sending_id &&

                annotation.peer_port_id == message.data_header.source_port

            )

        );

        annotation.peer_comp_id = message.sync_header.sending_id;

    }

    /// Checks if the data message can be received (due to port annotations), if

    /// it can then `true` is returned and the caller is responsible for handing

    /// the message of to the PDL code. Otherwise the message cannot be

    /// received.

    pub(crate) fn try_receive_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: &DataMessage) -> bool {

        debug_assert_eq!(self.mode, Mode::SyncBusy);

        debug_assert!(self.ports.iter().any(|v| v.self_port_id == message.data_header.target_port));

        // Make sure the expected mapping matches the currently stored mapping

                return false;

            }

        }

        // Expected mapping matches current mapping, so we will receive the message

        self.set_annotation(message.sync_header.sending_id, &message.data_header);

        // Handle the sync header embedded within the data message

        self.handle_sync_header(sched_ctx, comp_ctx, &message.sync_header);

        return true;

    }

    /// Receives the sync message and updates the consensus state appropriately.

    pub(crate) fn receive_sync_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: SyncMessage) -> SyncRoundDecision {

        // Whatever happens: handle the sync header (possibly changing the

        // currently registered leader)

        self.handle_sync_header(sched_ctx, comp_ctx, &message.sync_header);

        match message.content {

            SyncMessageContent::NotificationOfLeader => {

                return SyncRoundDecision::None;

            },

            SyncMessageContent::LocalSolution(solution_generator_id, local_solution) => {

                if !performed_communication {

                    continue;

                }

                let message = SyncMessage{

                    sync_header: self.create_sync_header(comp_ctx),

                    content: SyncMessageContent::NotificationOfLeader,

                };

                peer.handle.send_message(sched_ctx, Message::Sync(message), true);

            }

            self.forward_partial_solution(sched_ctx, comp_ctx);

        } else if header.highest_id.0 < self.highest_id.0 {

            // Sender has a lower ID, so notify it of our higher one

            let message = SyncMessage{

                sync_header: self.create_sync_header(comp_ctx),

                content: SyncMessageContent::NotificationOfLeader,

            };

            let peer_handle = comp_ctx.get_peer_handle(header.sending_id);

            let peer_info = comp_ctx.get_peer(peer_handle);

            peer_info.handle.send_message(sched_ctx, Message::Sync(message), true);

        } // else: exactly equal

    }

    fn set_annotation(&mut self, source_comp_id: CompId, data_header: &MessageDataHeader) {

        for annotation in self.ports.iter_mut() {

            if annotation.self_port_id == data_header.target_port {

                annotation.mapping = Some(data_header.new_mapping);

            }

        }

    }

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

    // Leader-related methods

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

    fn forward_partial_solution(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {

        debug_assert_ne!(self.highest_id, comp_ctx.id); // not leader

        // Make sure that we have something to send

        if !self.solution.has_contributions() {

            return;

        }

        // Swap the container with the partial solution and then send it along

        let partial_solution = self.solution.take_partial_solution();

        self.send_to_leader(sched_ctx, comp_ctx, Message::Sync(SyncMessage{

            sync_header: self.create_sync_header(comp_ctx),

            content: SyncMessageContent::PartialSolution(partial_solution),

        }));

    }

    }

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

        debug_assert_ne!(self.highest_id, comp_ctx.id); // we're not the leader

        let mut leader_info = sched_ctx.runtime.get_component_public(self.highest_id);

        leader_info.send_message(sched_ctx, message, true);

        let should_remove = leader_info.decrement_users();

        if let Some(key) = should_remove {

            sched_ctx.runtime.destroy_component(key);

        }

    }

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

    // Creating message headers

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

    fn create_data_header_and_update_mapping(&mut self, port_info: &Port) -> MessageDataHeader {

        let mut expected_mapping = Vec::with_capacity(self.ports.len());

        let mut port_index = usize::MAX;

        for (index, port) in self.ports.iter().enumerate() {

            if port.self_port_id == port_info.self_id {

                port_index = index; // remember for later updating

            }

            let annotation_kind = match port.kind {

                        self_comp_id: port.self_comp_id,

                        self_port_id: port.self_port_id

                        self_comp_id: port.self_comp_id,

                        self_port_id: port.self_port_id,

                    })

                }

        }

        let new_mapping = self.take_mapping();

        self.ports[port_index].mapping = Some(new_mapping);

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

        return MessageDataHeader{

            expected_mapping,

            new_mapping,

            source_port: port_info.self_id,

            target_port: port_info.peer_port_id,

        };

    }

    #[inline]

    fn create_sync_header(&self, comp_ctx: &CompCtx) -> MessageSyncHeader {

        return MessageSyncHeader{

            sync_round: self.round_index,

            sending_id: comp_ctx.id,

            highest_id: self.highest_id,

        };

    }

    #[inline]

    fn take_mapping(&mut self) -> u32 {