self.serialize_expression(heap, *arg_expr_id);

                }

            },

            Expression::Variable(_expr) => {

                // No subexpressions

            }

        }

    }

}

pub type EvalResult = Result<EvalContinuation, EvalError>;

#[derive(Debug)]

pub enum EvalContinuation {

    // Returned in both sync and non-sync modes

    Stepping,

    // Returned only in sync mode

    BranchInconsistent,

    SyncBlockEnd,

    NewFork,

    BlockFires(PortId),

    BlockGet(ExpressionId, PortId),

    Put(ExpressionId, PortId, ValueGroup),

    SelectStart(u32, u32), // (num_cases, num_ports_total)

    SelectWait, // wait until select can continue

    // Returned only in non-sync mode

    ComponentTerminated,

    SyncBlockStart,

    NewComponent(ProcedureDefinitionId, TypeId, ValueGroup),

    NewChannel,

}

// Note: cloning is fine, methinks. cloning all values and the heap regions then

// we end up with valid "pointers" to heap regions.

#[derive(Debug, Clone)]

pub struct Prompt {

    pub(crate) frames: Vec<Frame>,

    pub(crate) store: Store,

}

impl Prompt {

    pub fn new(types: &TypeTable, heap: &Heap, def: ProcedureDefinitionId, type_id: TypeId, args: ValueGroup) -> Self {

        let mut prompt = Self{

            frames: Vec::new(),

            store: Store::new(),

        };

        // Maybe do typechecking in the future?

                                    let mut message = String::with_capacity(elements.len());

                                    for element in elements {

                                        message.push(element.as_char());

                                    }

                                    // Drop the heap-allocated value from the

                                    // store

                                    if is_literal_string {

                                        self.store.drop_heap_pos(value_heap_pos);

                                    }

                                    println!("{}", message);

                                },

                                Method::SelectStart => {

                                    let num_cases = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_uint32();

                                    let num_ports = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_uint32();

                                    return Ok(EvalContinuation::SelectStart(num_cases, num_ports));

                                },

                                Method::SelectRegisterCasePort => {

                                    let case_index = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_uint32();

                                    let port_index = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_uint32();

                                    let port_value = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_port_id();

                                },

                                Method::SelectWait => {

                                    match ctx.performed_select_wait() {

                                        Some(select_index) => {

                                            cur_frame.expr_values.push_back(Value::UInt32(select_index));

                                        },

                                        None => {

                                            cur_frame.expr_stack.push_back(ExprInstruction::EvalExpr(expr.this.upcast()));

                                            return Ok(EvalContinuation::SelectWait)

                                        },

                                    }

                                },

                                Method::ComponentRandomU32 | Method::ComponentTcpClient => {

                                    debug_assert_eq!(heap[expr.procedure].parameters.len(), cur_frame.expr_values.len());

                                    debug_assert_eq!(heap[cur_frame.position].as_new().expression, expr.this);

                                },

                                Method::UserComponent => {

                                    // This is actually handled by the evaluation

                                    // of the statement.

                                    debug_assert_eq!(heap[expr.procedure].parameters.len(), cur_frame.expr_values.len());

                                    debug_assert_eq!(heap[cur_frame.position].as_new().expression, expr.this);

                                },

                                Method::UserFunction => {

                                    // Push a new frame. Note that all expressions have

            let variable_decl_stmt_id = create_ast_variable_declaration_stmt(ctx, self.current_procedure_id, variable_id, port_type_ref, port_expr_id);

            // Replace the original port expression in the call with a reference

            // to the replacement variable

            let variable_expr_id = create_ast_variable_expr(ctx, self.current_procedure_id, variable_id, port_type_ref);

            let call_expr = &mut ctx.heap[get_call_expr_id];

            call_expr.arguments[0] = variable_expr_id.upcast();

            transformed_stmts.push(variable_decl_stmt_id.upcast().upcast());

            locals.push((variable_id, port_type_ref));

        }

        // Insert runtime calls that facilitate the semantics of the select

        // block.

        // Create the call that indicates the start of the select block

        {

            let num_cases_expression_id = create_ast_literal_integer_expr(ctx, self.current_procedure_id, total_num_cases as u64, ctx.arch.uint32_type_id);

            let num_ports_expression_id = create_ast_literal_integer_expr(ctx, self.current_procedure_id, total_num_ports as u64, ctx.arch.uint32_type_id);

            let arguments = vec![

                num_cases_expression_id.upcast(),

                num_ports_expression_id.upcast()

            ];

            let call_statement_id = create_ast_expression_stmt(ctx, call_expression_id.upcast());

            transformed_stmts.push(call_statement_id.upcast());

        }

        // Create calls for each select case that will register the ports that

        // we are waiting on at the runtime.

        {

            let mut total_port_index = 0;

            for case_index in 0..total_num_cases {

                let case = &ctx.heap[id].cases[case_index];

                let case_num_ports = case.involved_ports.len();

                for case_port_index in 0..case_num_ports {

                    // Arguments to runtime call

                    let (port_variable_id, port_variable_type) = locals[total_port_index]; // so far this variable contains the temporary variables for the port expressions

                    let case_index_expr_id = create_ast_literal_integer_expr(ctx, self.current_procedure_id, case_index as u64, ctx.arch.uint32_type_id);

                    let port_index_expr_id = create_ast_literal_integer_expr(ctx, self.current_procedure_id, case_port_index as u64, ctx.arch.uint32_type_id);

                    let port_variable_expr_id = create_ast_variable_expr(ctx, self.current_procedure_id, port_variable_id, port_variable_type);

                    let runtime_call_arguments = vec![

                        case_index_expr_id.upcast(),

                        port_index_expr_id.upcast(),

                        port_variable_expr_id.upcast()

                    ];

                    // Create runtime call, then store it

                    let runtime_call_stmt_id = create_ast_expression_stmt(ctx, runtime_call_expr_id.upcast());

                    transformed_stmts.push(runtime_call_stmt_id.upcast());

                    total_port_index += 1;

                }

            }

        }

        // Create the variable that will hold the result of a completed select

        // block. Then create the runtime call that will produce this result

        let select_variable_id = create_ast_variable(ctx, outer_scope_id);

        let select_variable_type = TypeIdReference::DirectTypeId(ctx.arch.uint32_type_id);

        locals.push((select_variable_id, select_variable_type));

        {

            let variable_stmt_id = create_ast_variable_declaration_stmt(ctx, self.current_procedure_id, select_variable_id, select_variable_type, runtime_call_expr_id.upcast());

            transformed_stmts.push(variable_stmt_id.upcast().upcast());

        }

        call_id_section.forget();

        expr_id_section.forget();

        // Now we transform each of the select block case's guard and code into

        // a chained if-else statement.

        let relative_pos = transformed_stmts.len() as i32;

        if total_num_cases > 0 {

            let (if_stmt_id, end_if_stmt_id, scope_id) = transform_select_case_code(ctx, self.current_procedure_id, id, 0, select_variable_id, select_variable_type);

            link_existing_child_to_new_parent_scope(ctx, &mut self.scope_buffer, outer_scope_id, scope_id, relative_pos);

            let first_end_if_stmt = &mut ctx.heap[end_if_stmt_id];

            first_end_if_stmt.next = outer_end_block_id.upcast();

            let mut last_if_stmt_id = if_stmt_id;

            let mut last_end_if_stmt_id = end_if_stmt_id;

            let mut last_parent_scope_id = outer_scope_id;

            let mut last_relative_pos = transformed_stmts.len() as i32 + 1;

            transformed_stmts.push(last_if_stmt_id.upcast());

            for case_index in 1..total_num_cases {

                let (if_stmt_id, end_if_stmt_id, scope_id) = transform_select_case_code(ctx, self.current_procedure_id, id, case_index, select_variable_id, select_variable_type);

            full_span: InputSpan::new(),

        },

        identifier: Identifier::new_empty(InputSpan::new()),

        relative_pos_in_parent: -1,

        unique_id_in_scope: -1,

    });

    let scope = &mut ctx.heap[scope_id];

    scope.variables.push(variable_id);

    return variable_id;

}

fn create_ast_variable_expr(ctx: &mut Ctx, containing_procedure_id: ProcedureDefinitionId, variable_id: VariableId, variable_type_id: TypeIdReference) -> VariableExpressionId {

    let variable_type_index = add_new_procedure_expression_type(ctx, containing_procedure_id, variable_type_id);

    return ctx.heap.alloc_variable_expression(|this| VariableExpression{

        this,

        identifier: Identifier::new_empty(InputSpan::new()),

        declaration: Some(variable_id),

        used_as_binding_target: false,

        parent: ExpressionParent::None,

        type_index: variable_type_index,

    });

}

    let call_type_id = match method {

        Method::SelectStart => ctx.arch.void_type_id,

        Method::SelectRegisterCasePort => ctx.arch.void_type_id,

        Method::SelectWait => ctx.arch.uint32_type_id, // TODO: Not pretty, this. Pretty error prone

        _ => unreachable!(), // if this goes of, add the appropriate method here.

    };

    let expression_ids = buffer.start_section_initialized(&arguments);

    let call_type_index = add_new_procedure_expression_type(ctx, containing_procedure_id, TypeIdReference::DirectTypeId(call_type_id));

    let call_expression_id = ctx.heap.alloc_call_expression(|this| CallExpression{

        func_span: InputSpan::new(),

        this,

        parser_type: ParserType{

            elements: Vec::new(),

            full_span: InputSpan::new(),

        },

        method,

        arguments,

        procedure: ProcedureDefinitionId::new_invalid(),

        parent: ExpressionParent::None,

        type_index: call_type_index,

    });

    for argument_index in 0..expression_ids.len() {

        let argument_id = expression_ids[argument_index];

        let argument_expr = &mut ctx.heap[argument_id];

        *argument_expr.parent_mut() = ExpressionParent::Expression(call_expression_id.upcast(), argument_index as u32);

    }

    return call_expression_id;

}

fn create_ast_literal_integer_expr(ctx: &mut Ctx, containing_procedure_id: ProcedureDefinitionId, unsigned_value: u64, type_id: TypeId) -> LiteralExpressionId {

    let literal_type_index = add_new_procedure_expression_type(ctx, containing_procedure_id, TypeIdReference::DirectTypeId(type_id));

    return ctx.heap.alloc_literal_expression(|this| LiteralExpression{

        this,

    match message.content {

        ControlMessageContent::Ack => {

            default_handle_ack(control, message.id, sched_ctx, comp_ctx);

        },

        ControlMessageContent::BlockPort(port_id) => {

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

            // blocked.

            let port_handle = comp_ctx.get_port_handle(port_id);

            let port_info = comp_ctx.get_port(port_handle);

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

            if port_info.state == PortState::Open {

                // only when open: we don't do this when closed, and we we don't do this if we're blocked due to peer changes

                comp_ctx.set_port_state(port_handle, PortState::BlockedDueToFullBuffers);

            }

        },

        ControlMessageContent::ClosePort(content) => {

            // Request to close the port. We immediately comply and remove

            // the component handle as well

            let port_handle = comp_ctx.get_port_handle(content.port_to_close);

            // We're closing the port, so we will always update the peer of the

            // port (in case of error messages)

            let port_info = comp_ctx.get_port_mut(port_handle);

            port_info.peer_comp_id = message.sender_comp_id;

            let port_info = comp_ctx.get_port(port_handle);

            let peer_comp_id = port_info.peer_comp_id;

            let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);

            // One exception to sending an `Ack` is if we just closed the

            // port ourselves, meaning that the `ClosePort` messages got

            // sent to one another.

            if let Some(control_id) = control.has_close_port_entry(port_handle, comp_ctx) {

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

                // the channel at the same time.

                default_handle_ack(control, control_id, sched_ctx, comp_ctx);

            } else {

                // Respond to the message

                let last_instruction = port_info.last_instruction;

                let port_was_used = last_instruction != PortInstruction::None;

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

                comp_ctx.remove_peer(sched_ctx, port_handle, peer_comp_id, false); // do not remove if closed

                comp_ctx.set_port_state(port_handle, PortState::Closed); // now set to closed

                // Make sure that we've not reached an error condition. Note

                // that if this condition is not met, then we don't error out

                // now, but we may error out in the next sync block when we

                // try to `put`/`get` on the port. This condition makes sure

                // that if we have a successful sync round, followed by the peer

                // closing the port, that we don't consider the sync round to

                // have failed by mistake.

                if content.closed_in_sync_round && exec_state.mode.is_in_sync_block() && port_was_used {

                    return Err((

                        last_instruction,

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

                    ));

                }

            }

        },

        ControlMessageContent::UnblockPort(port_id) => {

            // We were previously blocked (or already closed)

            let port_handle = comp_ctx.get_port_handle(port_id);

            let port_info = comp_ctx.get_port(port_handle);

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

            if port_info.state == PortState::BlockedDueToFullBuffers {

                default_handle_unblock_put(exec_state, consensus, port_handle, sched_ctx, comp_ctx);

            }

        },

        ControlMessageContent::PortPeerChangedBlock(port_id) => {

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

            // temporarily block the port (while our original recipient is

            // potentially rerouting some of the in-flight messages) and

            // Ack. Then we wait for the `unblock` call.

            debug_assert_eq!(message.target_port_id, Some(port_id));

            let port_handle = comp_ctx.get_port_handle(port_id);

    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.log("Component busy exiting, still has `Ack`s remaining");

        return CompScheduling::Sleep;

    } else {

        sched_ctx.log("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(

    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.log(&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.

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

use crate::runtime2::scheduler::*;

use crate::runtime2::runtime::*;

use crate::runtime2::communication::*;

use crate::protocol::ExpressionId;

/// Helper struct to remember when the last operation on the port took place.

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

pub enum PortInstruction {

    None,

    NoSource,

    SourceLocation(ExpressionId),

}

#[derive(Debug)]

pub struct Port {

    pub self_id: PortId,

    pub peer_comp_id: CompId, // eventually consistent

    pub peer_port_id: PortId, // eventually consistent

    pub kind: PortKind,

    pub state: PortState,

    #[cfg(debug_assertions)] pub(crate) associated_with_peer: bool,

}

pub struct Peer {

    pub id: CompId,

    pub num_associated_ports: u32,

    pub(crate) handle: CompHandle,

}

/// Port and peer management structure. Will keep a local reference counter to

/// the ports associate with peers, additionally manages the atomic reference

/// counter associated with the peers' component handles.

pub struct CompCtx {

    pub id: CompId,

    ports: Vec<Port>,

    peers: Vec<Peer>,

    port_id_counter: u32,

}

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

pub struct LocalPortHandle(PortId);

#[derive(Copy, Clone)]

pub struct LocalPeerHandle(CompId);

    /// Creates a new component context based on a reserved entry in the

    /// component store. This reservation is used such that we already know our

    /// assigned ID.

    pub(crate) fn new(reservation: &CompReserved) -> Self {

        return Self{

            id: reservation.id(),

            ports: Vec::new(),

            peers: Vec::new(),

            port_id_counter: 0,

        }

    }

    /// Creates a new channel that is fully owned by the component associated

    /// with this context.

    pub(crate) fn create_channel(&mut self) -> Channel {

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

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

        self.ports.push(Port{

            self_id: putter_id,

            peer_port_id: getter_id,

            kind: PortKind::Putter,

            state: PortState::Open,

            peer_comp_id: self.id,

            last_instruction: PortInstruction::None,

            #[cfg(debug_assertions)] associated_with_peer: false,

        });

        self.ports.push(Port{

            self_id: getter_id,

            peer_port_id: putter_id,

            kind: PortKind::Getter,

            state: PortState::Open,

            peer_comp_id: self.id,

            last_instruction: PortInstruction::None,

            #[cfg(debug_assertions)] associated_with_peer: false,

        });

        return Channel{ putter_id, getter_id };

    }

    /// Adds a new port. Make sure to call `add_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,

            #[cfg(debug_assertions)] associated_with_peer: false,

        });

        return LocalPortHandle(self_id);

    }

    /// Removes a port. Make sure you called `remove_peer` first.

    pub(crate) fn remove_port(&mut self, port_handle: LocalPortHandle) -> Port {

        let port_index = self.must_get_port_index(port_handle);

        let port = self.ports.remove(port_index);

        dbg_code!(assert!(!port.associated_with_peer));

        return port;

    }

    /// Adds a new peer. This must be called for every port, no matter the

    /// component the channel is connected to. If a `CompHandle` is supplied,

    /// then it will be used to add the peer. Otherwise it will be retrieved

    /// from the runtime using its ID.

    pub(crate) fn add_peer(&mut self, port_handle: LocalPortHandle, sched_ctx: &SchedulerCtx, peer_comp_id: CompId, handle: Option<&CompHandle>) {

        let self_id = self.id;

        let port = self.get_port_mut(port_handle);

        debug_assert_eq!(port.peer_comp_id, peer_comp_id);

        dbg_code!(assert!(!port.associated_with_peer));

        if !Self::requires_peer_reference(port, self_id, false) {

            return;

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

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

                self.handle_sync_end(sched_ctx, comp_ctx);

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

                let port_handle = comp_ctx.get_port_handle(port_id);

                let port_index = comp_ctx.get_port_index(port_handle);

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

                    // Check if we can actually receive the message

                    if self.consensus.try_receive_data_message(sched_ctx, comp_ctx, message) {

                        // Message was received. Make sure any blocked peers and

                        // pending messages are handled.

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

                        let receive_result = component::default_handle_received_data_message(

                            port_id, PortInstruction::SourceLocation(expr_id),

                            &mut self.inbox_main[port_index], &mut self.inbox_backup,

                            comp_ctx, sched_ctx, &mut self.control

                        );

                        if let Err(location_and_message) = receive_result {

                            self.handle_generic_component_error(sched_ctx, location_and_message);

                            return CompScheduling::Immediate

                        } else {

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

                            return CompScheduling::Immediate;

                        }

                    } else {

                        return CompScheduling::Sleep;

                    }

                } else {

                    // We need to wait

                    self.exec_state.set_as_blocked_get(port_id);

                    return CompScheduling::Sleep;

                }

            },

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

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

                sched_ctx.log(&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, 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;

            },

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

                let port_id = port_id_from_eval(port_id);

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

                }

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

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

                self.handle_sync_start(sched_ctx, comp_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);

        for message in self.inbox_main.iter() {

            if let Some(message) = message {

                self.consensus.handle_incoming_data_message(comp_ctx, message);

            }

        }

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

        self.exec_state.mode = CompMode::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_success(sched_ctx, comp_ctx);

        self.exec_state.mode = CompMode::SyncEnd;

    }

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

        sched_ctx.log(&format!("Component exiting (reason: {:?}", self.exec_state.exit_reason));

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

        self.exec_state.mode = CompMode::BusyExit;

        let exit_inside_sync = self.exec_state.exit_reason.is_in_sync();

        // Doing this by index, then retrieving the handle is a bit rediculous,

        // but Rust is being Rust with its 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 == PortState::Closed {

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

                continue;

            }

            // Mark as closed

            let port_id = port.self_id;

            port.state = PortState::Closed;

            // Notify peer of closing

            let port_handle = comp_ctx.get_port_handle(port_id);

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