// If exiting while inside sync mode, report to the leader of the current

    // round that we've failed.

    if exit_inside_sync {

        let decision = consensus.notify_sync_end_failure(sched_ctx, comp_ctx);

        default_handle_sync_decision(sched_ctx, exec_state, comp_ctx, decision, consensus);

    }

    // Iterating over ports by index to work around borrowing rules

    for port_index in 0..comp_ctx.num_ports() {

        let port = comp_ctx.get_port_by_index_mut(port_index);

        if port.state.is_closed() || port.close_at_sync_end {

            // Already closed, or in the process of being closed

            continue;

        }

        // Mark as closed

        let port_id = port.self_id;

        port.state.set(PortStateFlag::Closed);

        // Notify peer of closing

        let port_handle = comp_ctx.get_port_handle(port_id);

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

        let peer_info = comp_ctx.get_peer(peer);

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

    }

    return CompScheduling::Immediate; // to check if we can shut down immediately

}

/// Handles a component waiting until all peers are notified that it is quitting

/// (i.e. after calling `default_handle_start_exit`).

#[must_use]

pub(crate) fn default_handle_busy_exit(

    exec_state: &mut CompExecState, control: &ControlLayer,

    sched_ctx: &SchedulerCtx

) -> CompScheduling {

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

    if control.has_acks_remaining() {

        sched_ctx.info("Component busy exiting, still has `Ack`s remaining");

        return CompScheduling::Sleep;

    } else {

        sched_ctx.info("Component busy exiting, now shutting down");

        exec_state.mode = CompMode::Exit;

        return CompScheduling::Exit;

    }

}

/// Handles a potential synchronous round decision. If there was a decision then

/// the `Some(success)` value indicates whether the round succeeded or not.

/// Might also end up changing the `ExecState`.

///

/// Might be called in two cases:

/// 1. The component is in regular execution mode, at the end of a sync round,

///     and is waiting for a solution to the round.

/// 2. The component has encountered an error during a sync round and is

///     exiting, hence is waiting for a "Failure" message from the leader.

pub(crate) fn default_handle_sync_decision(

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

    decision: SyncRoundDecision, consensus: &mut Consensus

) -> Option<bool> {

    let success = match decision {

        SyncRoundDecision::None => return None,

        SyncRoundDecision::Solution => true,

        SyncRoundDecision::Failure => false,

    };

    debug_assert!(

        exec_state.mode == CompMode::SyncEnd || (

            exec_state.mode.is_busy_exiting() && exec_state.exit_reason.is_error()

        ) || (

            exec_state.mode.is_in_sync_block() && decision == SyncRoundDecision::Failure

        )

    );

    sched_ctx.info(&format!("Handling decision {:?} (in mode: {:?})", decision, exec_state.mode));

    consensus.notify_sync_decision(decision);

    if success {

        // We cannot get a success message if the component has encountered an

        // error.

        for port_index in 0..comp_ctx.num_ports() {

            let port_info = comp_ctx.get_port_by_index_mut(port_index);

            if port_info.close_at_sync_end {

                port_info.state.set(PortStateFlag::Closed);

            }

            port_info.state.clear(PortStateFlag::Received);

        }

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

        exec_state.mode = CompMode::NonSync;

        return Some(true);

    } else {

        // We may get failure both in all possible cases. But we should only

        // modify the execution state if we're not already in exit mode

        if !exec_state.mode.is_busy_exiting() {

            sched_ctx.error("failed synchronous round, initiating exit");

            exec_state.set_as_start_exit(ExitReason::ErrorNonSync);

        }

        return Some(false);

    }

}

/// Performs the default action of printing the provided error, and then putting

/// the component in the state where it will shut down. Only to be used for

/// builtin components: their error message construction is simpler (and more

/// common) as they don't have any source code.

pub(crate) fn default_handle_error_for_builtin(

    exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx,

    location_and_message: (PortInstruction, String)

) {

    let (_location, message) = location_and_message;

    sched_ctx.error(&message);

    let exit_reason = if exec_state.mode.is_in_sync_block() {

        ExitReason::ErrorInSync

    } else {

        ExitReason::ErrorNonSync

    };

    exec_state.set_as_start_exit(exit_reason);

}

#[inline]

pub(crate) fn default_handle_exit(_exec_state: &CompExecState) -> CompScheduling {

    debug_assert_eq!(_exec_state.mode, CompMode::Exit);

    return CompScheduling::Exit;

}

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

// Internal messaging/state utilities

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

/// Sends a message without any transmitted ports. Does not check if sending

/// is actually valid.

fn send_message_without_ports(

    sending_port_handle: LocalPortHandle, value: ValueGroup,

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

) {

    let port_info = comp_ctx.get_port(sending_port_handle);

    debug_assert!(port_info.state.can_send());

    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 = consensus.annotate_data_message(comp_ctx, port_info, value);

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

}

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

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

    debug_assert!(!transmit_ports.is_empty()); // requisite for calling this function

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

    for (_, port_id) in &transmit_ports {

        let transmit_port_handle = comp_ctx.get_port_handle(*port_id);

        let transmit_port_info = comp_ctx.get_port_mut(transmit_port_handle);

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

    // 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, transmit_port_id) in transmit_ports {

        let transmit_port_handle = comp_ctx.get_port_handle(transmit_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);

}

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

        content: ControlMessageContent::Ack

    }), true);

}

/// Handles the unblocking of a putter port. In case there is a pending message

/// on that port then it will be sent.

fn default_handle_recently_unblocked_port(

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

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

) {

    let port_info = comp_ctx.get_port_mut(port_handle);

    let port_id = port_info.self_id;

    debug_assert!(!port_info.state.is_blocked()); // should have been done by the caller

    if exec_state.is_blocked_on_put(port_id) {

        // Return to the regular execution mode

        exec_state.mode = CompMode::Sync;

        exec_state.mode_port = PortId::new_invalid();

        // 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 to_send = exec_state.mode_value.take();

        let to_send = consensus.annotate_data_message(comp_ctx, port_info, to_send);

        // Retrieve peer to send the message

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

        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; // because we're blocked on a `put`, we must've started in the sync state.

        exec_state.mode_port = PortId::new_invalid();

    }

}

#[inline]

pub(crate) fn port_id_from_eval(port_id: EvalPortId) -> PortId {

    return PortId(port_id.id);

}

#[inline]

pub(crate) fn port_id_to_eval(port_id: PortId) -> EvalPortId {

    return EvalPortId{ id: port_id.0 };

}

// TODO: Optimize double vec

type EncounteredPorts = Vec<(Vec<ValueId>, PortId)>;

/// Recursively goes through the value group, attempting to find ports.

/// Duplicates will only be added once.

pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut EncounteredPorts) {

    // Helper to check a value for a port and recurse if needed.

    fn find_port_in_value(group: &ValueGroup, value: &Value, value_location: ValueId, ports: &mut EncounteredPorts) {

        match value {

            Value::Input(port_id) | Value::Output(port_id) => {

                // This is an actual port

                let cur_port = PortId(port_id.id);

                for prev_port in ports.iter_mut() {

                    if prev_port.1 == cur_port {

                        // Already added

                        prev_port.0.push(value_location);

                        return;

                    }

                }

                ports.push((vec![value_location], cur_port));

            },

            Value::Array(heap_pos) |

            Value::Message(heap_pos) |

            Value::String(heap_pos) |

            Value::Struct(heap_pos) |

            Value::Union(_, heap_pos) => {

                // Reference to some dynamic thing which might contain ports,

                // so recurse

                let heap_region = &group.regions[*heap_pos as usize];

                for (value_index, embedded_value) in heap_region.iter().enumerate() {

                    let value_location = ValueId::Heap(*heap_pos, value_index as u32);

                    find_port_in_value(group, embedded_value, value_location, ports);

                }

            },

            _ => {}, // values we don't care about

        }

    }

    // Clear the ports, then scan all the available values

    ports.clear();

    for (value_index, value) in value_group.values.iter().enumerate() {

        find_port_in_value(value_group, value, ValueId::Stack(value_index as u32), ports);

    }

}

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

}

    fn performed_put(&mut self, _port: EvalPortId) -> bool {

        match self.stmt.take() {

            ExecStmt::None => return false,

            ExecStmt::PerformedPut => return true,

            _ => unreachable!(),

        }

    }

    fn performed_get(&mut self, _port: EvalPortId) -> Option<ValueGroup> {

        match self.stmt.take() {

            ExecStmt::None => return None,

            ExecStmt::PerformedGet(value) => return Some(value),

            _ => unreachable!(),

        }

    }

    fn fires(&mut self, _port: EvalPortId) -> Option<Value> {

        todo!("remove fires")

    }

    fn performed_fork(&mut self) -> Option<bool> {

        todo!("remove fork")

    }

    fn created_channel(&mut self) -> Option<(Value, Value)> {

        match self.stmt.take() {

            ExecStmt::None => return None,

            ExecStmt::CreatedChannel(ports) => return Some(ports),

            _ => unreachable!(),

        }

    }

    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!(),

        }

    }

}

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

}

impl SelectState {

    fn new() -> Self {

        return Self{

            cases: Vec::new(),

            next_case: 0,

            num_cases: 0,

            random: Random::new(),

            candidates_workspace: Vec::new(),

        }

    }

    fn handle_select_start(&mut self, num_cases: u32) {

        self.cases.clear();

        self.next_case = 0;

        self.num_cases = num_cases;

    }

    /// Register a port as belonging to a particular case. As for correctness of

    /// PDL code one cannot register the same port twice, this function might

    /// return an error

    fn register_select_case_port(&mut self, comp_ctx: &CompCtx, case_index: u32, _port_index: u32, port_id: PortId) -> Result<(), PortId> {

        // Retrieve case and port handle

        self.ensure_at_case(case_index);

        let cur_case = &mut self.cases[case_index as usize];

        let port_handle = comp_ctx.get_port_handle(port_id);

        debug_assert_eq!(cur_case.involved_ports.len(), _port_index as usize);

        // Make sure port wasn't added before, we disallow having the same port

        // in the same select guard twice.

        if cur_case.involved_ports.contains(&port_handle) {

            return Err(port_id);

        }

        cur_case.involved_ports.push(port_handle);

        return Ok(());

    }

    /// Notification that all ports have been registered and we should now wait

    /// until the appropriate messages have come in.

    fn handle_select_waiting_point(&mut self, inbox: &InboxMain, comp_ctx: &CompCtx) -> SelectDecision {

        if self.num_cases != self.next_case {

            // This happens when there are >=1 select cases written at the end

            // of the select block.

            self.ensure_at_case(self.num_cases - 1);

        }

        return self.has_decision(inbox, comp_ctx);

    }

    fn handle_updated_inbox(&mut self, inbox: &InboxMain, comp_ctx: &CompCtx) -> SelectDecision {

        return self.has_decision(inbox, comp_ctx);

    }

    /// Internal helper, pushes empty cases inbetween last case and provided new

    /// case index.

    fn ensure_at_case(&mut self, new_case_index: u32) {

        // Push an empty case for all intermediate cases that were not

        // registered with a port.

        debug_assert!(new_case_index >= self.next_case && new_case_index < self.num_cases);

        for _ in self.next_case..new_case_index + 1 {

            self.cases.push(SelectCase{ involved_ports: Vec::new() });

        }

        self.next_case = new_case_index + 1;

    }

    /// Checks if a decision can be reached

    fn has_decision(&mut self, inbox: &InboxMain, comp_ctx: &CompCtx) -> SelectDecision {

        self.candidates_workspace.clear();

        if self.cases.is_empty() {

            // If there are no cases then we can immediately reach a "bogus

            // decision".

            return SelectDecision::Case(0);

        }

        // Need to check for valid case

        'case_loop: for (case_index, case) in self.cases.iter().enumerate() {

            for port_handle in case.involved_ports.iter().copied() {

                let port_index = comp_ctx.get_port_index(port_handle);

                if inbox[port_index].is_none() {

                    // Condition not satisfied

                    continue 'case_loop;

                }

            }

            // If here then the case guard is satisfied

            self.candidates_workspace.push(case_index);

        }

        if self.candidates_workspace.is_empty() {

            return SelectDecision::None;

        } else {

            let candidate_index = self.random.get_u64() as usize % self.candidates_workspace.len();

            return SelectDecision::Case(self.candidates_workspace[candidate_index] as u32);

        }

    }

}

pub(crate) struct CompPDL {

    pub exec_state: CompExecState,

    select_state: SelectState,

    pub prompt: Prompt,

    pub control: ControlLayer,

    pub consensus: Consensus,

    pub sync_counter: u32,

    pub exec_ctx: ExecCtx,

    // TODO: Temporary field, simulates future plans of having one storage place

    //  reserved per port.

    // Should be same length as the number of ports. Corresponding indices imply

    // message is intended for that port.

    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;

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

        // Depending on the mode don't do anything at all, take some special

        // actions, or fall through and run the PDL code.

        match self.exec_state.mode {

            CompMode::NonSync | CompMode::Sync => {

                // continue and run PDL code

            },

            CompMode::SyncEnd | CompMode::BlockedGet | CompMode::BlockedPut |

            CompMode::BlockedSelect | CompMode::BlockedPutPorts => {

                return CompScheduling::Sleep;

            }

            CompMode::StartExit => return component::default_handle_start_exit(

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

            ),

            CompMode::BusyExit => return component::default_handle_busy_exit(

                &mut self.exec_state, &self.control, sched_ctx

            ),

            CompMode::Exit => return component::default_handle_exit(&self.exec_state),

        }

        let run_result = self.execute_prompt(&sched_ctx);

        if let Err(error) = run_result {

            self.handle_component_error(sched_ctx, CompError::Executor(error));

            return CompScheduling::Immediate;

        }

        let run_result = run_result.unwrap();

        match run_result {

            EC::Stepping => unreachable!(), // execute_prompt runs until this is no longer returned

            EC::BranchInconsistent | EC::NewFork | EC::BlockFires(_) => todo!("remove these"),

            // Results that can be returned in sync mode

            EC::SyncBlockEnd => {

                component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);

                return CompScheduling::Immediate;

            },

            EC::BlockGet(expr_id, port_id) => {

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

                debug_assert!(self.exec_ctx.stmt.is_none());

                let port_id = port_id_from_eval(port_id);

                match component::default_attempt_get(

                    &mut self.exec_state, port_id, PortInstruction::SourceLocation(expr_id),

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

                    &mut self.control, &mut self.consensus

                ) {

                    GetResult::Received(message) => {

                        self.exec_ctx.stmt = ExecStmt::PerformedGet(message.content);

                        return CompScheduling::Immediate;

                    },

                    GetResult::NoMessage => {

                        return CompScheduling::Sleep;

                    },

                    GetResult::Error(location_and_message) => {

                        self.handle_generic_component_error(sched_ctx, location_and_message);

                        return CompScheduling::Immediate;

                    }

                }

            },

            EC::Put(expr_id, port_id, value) => {

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

                sched_ctx.info(&format!("Putting value {:?}", value));

                // Send the message

                let target_port_id = port_id_from_eval(port_id);

                let send_result = component::default_send_data_message(

                    &mut self.exec_state, target_port_id,

                    PortInstruction::SourceLocation(expr_id), value,

                    sched_ctx, &mut self.consensus, &mut self.control, comp_ctx

                );

                if let Err(location_and_message) = send_result {

                    self.handle_generic_component_error(sched_ctx, location_and_message);

                    return CompScheduling::Immediate;

                } else {

                    // When `run` is called again (potentially after becoming

                    // unblocked) we need to instruct the executor that we performed

                    // the `put`

                    let scheduling = send_result.unwrap();

                    self.exec_ctx.stmt = ExecStmt::PerformedPut;

                    return scheduling;

                }

            },

            EC::SelectStart(num_cases, _num_ports) => {

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

                self.select_state.handle_select_start(num_cases);

                return CompScheduling::Requeue;

            },

            EC::SelectRegisterPort(expr_id, case_index, port_index, port_id) => {

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

                let port_id = port_id_from_eval(port_id);

                let port_handle = comp_ctx.get_port_handle(port_id);

                // Note: we register the "last_instruction" here already. This

                // way if we get a `ClosePort` message, the condition to fail

                // the synchronous round is satisfied.

                let port_info = comp_ctx.get_port_mut(port_handle);

                port_info.last_instruction = PortInstruction::SourceLocation(expr_id);

                let port_is_closed = port_info.state.is_closed();

                // Register port as part of select guard

                if let Err(_err) = self.select_state.register_select_case_port(comp_ctx, case_index, port_index, port_id) {

                    // Failure occurs if a port is used twice in the same guard

                    let protocol = &sched_ctx.runtime.protocol;

                    self.handle_component_error(sched_ctx, CompError::Executor(EvalError::new_error_at_expr(

                        &self.prompt, &protocol.modules, &protocol.heap, expr_id,

                        String::from("Cannot have the one port appear in the same guard twice")

                    )));

                } else if port_is_closed {

                    // Port is closed

                    let peer_id = comp_ctx.get_port(port_handle).peer_comp_id;

                    let protocol = &sched_ctx.runtime.protocol;

                    self.handle_component_error(sched_ctx, CompError::Executor(EvalError::new_error_at_expr(

                        &self.prompt, &protocol.modules, &protocol.heap, expr_id,

                        format!("Cannot register port, as the peer component (id:{}) has shut down", peer_id.0)

                    )));

                }

                return CompScheduling::Immediate;

            },

            EC::SelectWait => {

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

                let select_decision = self.select_state.handle_select_waiting_point(&self.inbox_main, comp_ctx);

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

                    // Reached a conclusion, so we can continue immediately

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

                    self.exec_state.mode = CompMode::Sync;

                    return CompScheduling::Immediate;

                } else {

                    // No decision yet

                    self.exec_state.mode = CompMode::BlockedSelect;

                    return CompScheduling::Sleep;

                }

            },

            // Results that can be returned outside of sync mode

            EC::ComponentTerminated => {

                self.exec_state.set_as_start_exit(ExitReason::Termination);

                return CompScheduling::Immediate;

            },

            EC::SyncBlockStart => {

                component::default_handle_sync_start(

                    &mut self.exec_state, &mut self.inbox_main, sched_ctx, comp_ctx, &mut self.consensus

                );

                return CompScheduling::Immediate;

            },

            EC::NewComponent(definition_id, type_id, arguments) => {

                debug_assert_eq!(self.exec_state.mode, CompMode::NonSync);

                self.create_component_and_transfer_ports(

                    sched_ctx, comp_ctx,

                    definition_id, type_id, arguments

                );

                return CompScheduling::Requeue;

            },

            EC::NewChannel => {

                debug_assert_eq!(self.exec_state.mode, CompMode::NonSync);

                debug_assert!(self.exec_ctx.stmt.is_none());

                let channel = comp_ctx.create_channel();

                self.exec_ctx.stmt = ExecStmt::CreatedChannel((

                    Value::Output(port_id_to_eval(channel.putter_id)),

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

                ));

                self.inbox_main.push(None);

                self.inbox_main.push(None);

                return CompScheduling::Immediate;

            }

        }

    }

}

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,

    source_comp: CompId,

    old_target_port: PortId,

    new_target_port: PortId,

    new_target_comp: CompId,

    schedule_entry_id: ControlId,

}