let _result = self.consensus.notify_of_speculative_mapping(silent_branch_id, port_id, false, comp_ctx);

                debug_assert_eq!(_result, Consistency::Valid);

                // Somewhat important: we push the firing one first, such that

                // that branch is ran again immediately.

                self.tree.push_into_queue(QueueKind::Runnable, firing_branch_id);

                self.tree.push_into_queue(QueueKind::Runnable, silent_branch_id);

                return ConnectorScheduling::Immediate;

            },

            EvalContinuation::BlockGet(port_id) => {

                // Branch performed a `get()` on a port that does not have a

                // received message on that port.

                let port_id = PortIdLocal::new(port_id.0.u32_suffix);

                branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

                branch.awaiting_port = port_id;

                self.tree.push_into_queue(QueueKind::AwaitingMessage, branch_id);

                // Note: we only know that a branch is waiting on a message when

                // it reaches the `get` call. But we might have already received

                // a message that targets this branch, so check now.

                let mut any_message_received = false;

                for message in comp_ctx.get_read_data_messages(port_id) {

                    if self.consensus.branch_can_receive(branch_id, &message) {

                        // This branch can receive the message, so we do the

                        // fork-and-receive dance

                        let receiving_branch_id = self.tree.fork_branch(branch_id);

                        let branch = &mut self.tree[receiving_branch_id];

                        branch.awaiting_port = PortIdLocal::new_invalid();

                        branch.prepared = PreparedStatement::PerformedGet(message.content.clone());

                        self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

                        self.consensus.notify_of_received_message(receiving_branch_id, &message, comp_ctx);

                        self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);

                        any_message_received = true;

                    }

                }

                if any_message_received {

                    return ConnectorScheduling::Immediate;

                }

            }

            EvalContinuation::SyncBlockEnd => {

                let consistency = self.consensus.notify_of_finished_branch(branch_id);

                if consistency == Consistency::Valid {

                    branch.sync_state = SpeculativeState::ReachedSyncEnd;

                    self.tree.push_into_queue(QueueKind::FinishedSync, branch_id);

                } else {

                    branch.sync_state = SpeculativeState::Inconsistent;

                }

            },

            EvalContinuation::NewFork => {

                // Like the `NewChannel` result. This means we're setting up

                // a branch and putting a marker inside the RunContext for the

                // next time we run the PDL code

                let left_id = branch_id;

                let right_id = self.tree.fork_branch(left_id);

                self.consensus.notify_of_new_branch(left_id, right_id);

                self.tree.push_into_queue(QueueKind::Runnable, left_id);

                self.tree.push_into_queue(QueueKind::Runnable, right_id);

                let left_branch = &mut self.tree[left_id];

                left_branch.prepared = PreparedStatement::ForkedExecution(true);

                let right_branch = &mut self.tree[right_id];

                right_branch.prepared = PreparedStatement::ForkedExecution(false);

            }

            EvalContinuation::Put(port_id, content) => {

                // Branch is attempting to send data

                let port_id = PortIdLocal::new(port_id.0.u32_suffix);

                let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);

                let message = DataMessage{ sync_header, data_header, content };

                match comp_ctx.submit_message(Message::Data(message)) {

                    Ok(_) => {

                        // Message is underway

                        branch.prepared = PreparedStatement::PerformedPut;

                        self.tree.push_into_queue(QueueKind::Runnable, branch_id);

                        return ConnectorScheduling::Immediate;

                    },

                    Err(_) => {

                        // We don't own the port

                        let pd = &sched_ctx.runtime.protocol_description;

                        let eval_error = branch.code_state.new_error_at_expr(

                            &pd.modules, &pd.heap,

                            String::from("attempted to 'put' on port that is no longer owned")

                        );

                        self.eval_error = Some(eval_error);

                        self.mode = Mode::SyncError;

                        println!("DEBUGERINO: Notify of fatal branch");

                        if let Some(conclusion) = self.consensus.notify_of_fatal_branch(branch_id, comp_ctx) {

                            println!("DEBUGERINO: Actually got {:?}", conclusion);

                            return self.enter_non_sync_mode(conclusion, comp_ctx);

                        }

                    }

                }

            },

            _ => unreachable!("unexpected run result {:?} in sync mode", run_result),

        }

        // If here then the run result did not require a particular action. We

        // return whether we have more active branches to run or not.

        if self.tree.queue_is_empty(QueueKind::Runnable) {

            return ConnectorScheduling::NotNow;

        } else {

            return ConnectorScheduling::Later;

        }

    }

    pub fn run_in_deterministic_mode(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {

        debug_assert!(!self.tree.is_in_sync() && !self.consensus.is_in_sync());

        let branch = self.tree.base_branch_mut();

        debug_assert!(branch.sync_state == SpeculativeState::RunningNonSync);

        let mut run_context = ConnectorRunContext{

            branch_id: branch.id,

            consensus: &self.consensus,

            prepared: branch.prepared.take(),

        };

        let run_result = Self::run_prompt(&mut branch.code_state, &sched_ctx.runtime.protocol_description, &mut run_context);

        if let Err(eval_error) = run_result {

            comp_ctx.push_error(eval_error);

            return ConnectorScheduling::Exit

        }

        let run_result = run_result.unwrap();

        match run_result {

            EvalContinuation::ComponentTerminated => {

                branch.sync_state = SpeculativeState::Finished;

                return ConnectorScheduling::Exit;

            },

            EvalContinuation::SyncBlockStart => {

                comp_ctx.notify_sync_start();

                let sync_branch_id = self.tree.start_sync();

                debug_assert!(self.last_finished_handled.is_none());

                self.consensus.start_sync(comp_ctx);

                self.consensus.notify_of_new_branch(BranchId::new_invalid(), sync_branch_id);

                self.tree.push_into_queue(QueueKind::Runnable, sync_branch_id);

                self.mode = Mode::Sync;

                return ConnectorScheduling::Immediate;

            },

            EvalContinuation::NewComponent(definition_id, monomorph_idx, arguments) => {

                // Note: we're relinquishing ownership of ports. But because

                // we are in non-sync mode the scheduler will handle and check

                // port ownership transfer.

                debug_assert!(comp_ctx.workspace_ports.is_empty());

                find_ports_in_value_group(&arguments, &mut comp_ctx.workspace_ports);

                let new_prompt = Prompt::new(

                    &sched_ctx.runtime.protocol_description.types,

                    &sched_ctx.runtime.protocol_description.heap,

                    definition_id, monomorph_idx, arguments

                );

                let new_component = ConnectorPDL::new(new_prompt);

                comp_ctx.push_component(new_component, comp_ctx.workspace_ports.clone());

                comp_ctx.workspace_ports.clear();

                return ConnectorScheduling::Later;

            },

            EvalContinuation::NewChannel => {

                let (getter, putter) = sched_ctx.runtime.create_channel(comp_ctx.id);

                debug_assert!(getter.kind == PortKind::Getter && putter.kind == PortKind::Putter);

                branch.prepared = PreparedStatement::CreatedChannel((

                    Value::Output(PortId::new(putter.self_id.index)),

                    Value::Input(PortId::new(getter.self_id.index)),

                ));

                comp_ctx.push_port(putter);

                comp_ctx.push_port(getter);

                return ConnectorScheduling::Immediate;

            },

            _ => unreachable!("unexpected run result '{:?}' while running in non-sync mode", run_result),

        }

    }

    /// Helper that moves the component's state back into non-sync mode, using

    /// the provided solution branch ID as the branch that should be comitted to

    /// memory. If this function returns false, then the component is supposed

    /// to exit.

    fn enter_non_sync_mode(&mut self, conclusion: RoundConclusion, ctx: &mut ComponentCtx) -> ConnectorScheduling {

        debug_assert!(self.mode == Mode::Sync || self.mode == Mode::SyncError);

        // Depending on local state decide what to do

        let final_branch_id = match conclusion {

            RoundConclusion::Success(branch_id) => Some(branch_id),

            RoundConclusion::Failure => None,

        };

        if let Some(solution_branch_id) = final_branch_id {

            let mut fake_vec = Vec::new();

            self.tree.end_sync(solution_branch_id);

            self.consensus.end_sync(solution_branch_id, &mut fake_vec);

            debug_assert!(fake_vec.is_empty());

            ctx.notify_sync_end(&[]);

            self.last_finished_handled = None;

            self.eval_error = None; // in case we came from the SyncError mode

            self.mode = Mode::NonSync;

            return ConnectorScheduling::Immediate;

        } else {

            // No final branch, because we're supposed to exit!

            self.last_finished_handled = None;

            self.mode = Mode::Error;

            if let Some(eval_error) = self.eval_error.take() {

                ctx.push_error(eval_error);

            }

            return ConnectorScheduling::Exit;

        }

    }

    /// Runs the prompt repeatedly until some kind of execution-blocking

    /// condition appears.

    #[inline]

    fn run_prompt(prompt: &mut Prompt, pd: &ProtocolDescription, ctx: &mut ConnectorRunContext) -> Result<EvalContinuation, EvalError> {

        loop {

            let result = prompt.step(&pd.types, &pd.heap, &pd.modules, ctx);

            if let Ok(EvalContinuation::Stepping) = result {

                continue;

            }

            return result;

        }

    }

}

                        scheduled.ctx.ports.push(port_change.port);

                    } else {

                        let index = scheduled.ctx.ports

                            .iter()

                            .position(|v| v.self_id == port_change.port.self_id)

                            .unwrap();

                        scheduled.ctx.ports.remove(index);

                    }

                }

            }

        }

        // Finally, check if we just entered or just left a sync region

        if scheduled.ctx.changed_in_sync {

            if scheduled.ctx.is_in_sync {

                // Just entered sync region

            } else {

                // Just left sync region. So prepare inbox for the next sync

                // round

                scheduled.ctx.inbox.clear_read_messages();

            }

            scheduled.ctx.changed_in_sync = false; // reset flag

        }

    }

    fn try_go_to_sleep(&self, connector_key: ConnectorKey, connector: &mut ScheduledConnector) {

        debug_assert_eq!(connector_key.index, connector.ctx.id.index);

        debug_assert_eq!(connector.public.sleeping.load(Ordering::Acquire), false);

        // This is the running connector, and only the running connector may

        // decide it wants to sleep again.

        connector.public.sleeping.store(true, Ordering::Release);

        // But due to reordering we might have received messages from peers who

        // did not consider us sleeping. If so, then we wake ourselves again.

        if !connector.public.inbox.is_empty() {

            // Try to wake ourselves up (needed because someone might be trying

            // the exact same atomic compare-and-swap at this point in time)

            let should_wake_up_again = connector.public.sleeping

                .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)

                .is_ok();

            if should_wake_up_again {

                self.runtime.push_work(connector_key)

            }

        }

    }

    // TODO: Remove, this is debugging stuff

    fn debug(&self, message: &str) {

        println!("DEBUG [thrd:{:02} conn:  ]: {}", self.scheduler_id, message);

    }

    fn debug_conn(&self, conn: ConnectorId, message: &str) {

        println!("DEBUG [thrd:{:02} conn:{:02}]: {}", self.scheduler_id, conn.index, message);

    }

}

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

// ComponentCtx

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

enum ComponentStateChange {

    CreatedComponent(ConnectorPDL, Vec<PortIdLocal>),

    CreatedPort(Port),

    ChangedPort(ComponentPortChange),

}

#[derive(Clone)]

pub(crate) struct ComponentPortChange {

    pub is_acquired: bool, // otherwise: released

    pub port: Port,

}

/// The component context (better name may be invented). This was created

/// because part of the component's state is managed by the scheduler, and part

/// of it by the component itself. When the component starts a sync block or

/// exits a sync block the partially managed state by both component and

/// scheduler need to be exchanged.

pub(crate) struct ComponentCtx {

    // Mostly managed by the scheduler

    pub(crate) id: ConnectorId,

    ports: Vec<Port>,

    inbox: Inbox,

    // Submitted by the component

    is_in_sync: bool,

    changed_in_sync: bool,

    outbox: VecDeque<Message>,

    state_changes: VecDeque<ComponentStateChange>,

    // Workspaces that may be used by components to (generally) prevent

    // allocations. Be a good scout and leave it empty after you've used it.

    // TODO: Move to scheduler ctx, this is the wrong place

    pub workspace_ports: Vec<PortIdLocal>,

    pub workspace_branches: Vec<BranchId>,

}

impl ComponentCtx {

    pub(crate) fn new_empty() -> Self {

        return Self{

            id: ConnectorId::new_invalid(),

            ports: Vec::new(),

            inbox: Inbox::new(),

            is_in_sync: false,

            changed_in_sync: false,

            outbox: VecDeque::new(),

            state_changes: VecDeque::new(),

            workspace_ports: Vec::new(),

            workspace_branches: Vec::new(),

        };

    }

    /// Notify the runtime that the component has created a new component. May

    /// only be called outside of a sync block.

    pub(crate) fn push_component(&mut self, component: ConnectorPDL, initial_ports: Vec<PortIdLocal>) {

        debug_assert!(!self.is_in_sync);

        self.state_changes.push_back(ComponentStateChange::CreatedComponent(component, initial_ports));

    }

    /// Notify the runtime that the component has created a new port. May only

    /// be called outside of a sync block (for ports received during a sync

    /// block, pass them when calling `notify_sync_end`).

    pub(crate) fn push_port(&mut self, port: Port) {

        debug_assert!(!self.is_in_sync);

        self.state_changes.push_back(ComponentStateChange::CreatedPort(port))

    }

    /// Notify the runtime of an error. Note that this will not perform any

    /// special action beyond printing the error. The component is responsible

    /// for waiting until it is appropriate to shut down (i.e. being outside

    /// of a sync region) and returning the `Exit` scheduling code.

    pub(crate) fn push_error(&mut self, error: EvalError) {

        println!("ERROR: Component ({}) encountered a critical error:\n{}", self.id.index, error);

    }

    #[inline]

    pub(crate) fn get_ports(&self) -> &[Port] {

        return self.ports.as_slice();

    }

    pub(crate) fn get_port_by_id(&self, id: PortIdLocal) -> Option<&Port> {

        return self.ports.iter().find(|v| v.self_id == id);

    }

    pub(crate) fn get_port_by_channel_id(&self, id: ChannelId) -> Option<&Port> {

        return self.ports.iter().find(|v| v.channel_id == id);

    }

    fn get_port_mut_by_id(&mut self, id: PortIdLocal) -> Option<&mut Port> {

        return self.ports.iter_mut().find(|v| v.self_id == id);

    }

    /// Notify that component will enter a sync block. Note that after calling

    /// this function you must allow the scheduler to pick up the changes in the

    /// context by exiting your code-executing loop, and to continue executing

    /// code the next time the scheduler picks up the component.

    pub(crate) fn notify_sync_start(&mut self) {

        debug_assert!(!self.is_in_sync);

        self.is_in_sync = true;

        self.changed_in_sync = true;

    }

    #[inline]

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

        return self.is_in_sync;

    }

    /// Submit a message for the scheduler to send to the appropriate receiver.

    /// May only be called inside of a sync block.

    pub(crate) fn submit_message(&mut self, contents: Message) -> Result<(), ()> {

        debug_assert!(self.is_in_sync);

        if let Some(port_id) = contents.source_port() {

            let port_info = self.get_port_by_id(port_id);

            let is_valid = match port_info {

                Some(port_info) => {

                    port_info.state == PortState::Open

                },

                None => false,

            };

            if !is_valid {

                // We don't own the port

                println!(" ****** DEBUG ****** : Sending through closed port!!! {}", port_id.index);

                return Err(());

            }

        }

        self.outbox.push_back(contents);

        return Ok(());

    }

    /// Notify that component just finished a sync block. Like

    /// `notify_sync_start`: drop out of the `Component::Run` function.

    pub(crate) fn notify_sync_end(&mut self, changed_ports: &[ComponentPortChange]) {

        debug_assert!(self.is_in_sync);

        self.is_in_sync = false;

        self.changed_in_sync = true;

        self.state_changes.reserve(changed_ports.len());

        for changed_port in changed_ports {

            self.state_changes.push_back(ComponentStateChange::ChangedPort(changed_port.clone()));

        }

    }

    /// Retrieves messages matching a particular port and branch id. But only

    /// those messages that have been previously received with

    /// `read_next_message`.

    pub(crate) fn get_read_data_messages(&self, match_port_id: PortIdLocal) -> MessagesIter {

        return self.inbox.get_read_data_messages(match_port_id);

    }

    pub(crate) fn get_next_message_ticket(&mut self) -> Option<MessageTicket> {

        if !self.is_in_sync { return None; }

        return self.inbox.get_next_message_ticket();

    }

    #[inline]

    pub(crate) fn get_next_message_ticket_even_if_not_in_sync(&mut self) -> Option<MessageTicket> {

        return self.inbox.get_next_message_ticket();

    }

    #[inline]

    pub(crate) fn read_message_using_ticket(&self, ticket: MessageTicket) -> &Message {

        return self.inbox.read_message_using_ticket(ticket);

    }

    #[inline]

    pub(crate) fn take_message_using_ticket(&mut self, ticket: MessageTicket) -> Message {

        return self.inbox.take_message_using_ticket(ticket)

    }

    /// Puts back a message back into the inbox. The reason being that the

    /// message is actually part of the next sync round. This will

    pub(crate) fn put_back_message(&mut self, message: Message) {

        self.inbox.put_back_message(message);

    }

}

pub(crate) struct MessagesIter<'a> {

    messages: &'a [Message],

    next_index: usize,

    max_index: usize,

    match_port_id: PortIdLocal,

}

impl<'a> Iterator for MessagesIter<'a> {

    type Item = &'a DataMessage;

    fn next(&mut self) -> Option<Self::Item> {

        // Loop until match is found or at end of messages

        while self.next_index < self.max_index {

            let message = &self.messages[self.next_index];

            if let Message::Data(message) = &message {

                if message.data_header.target_port == self.match_port_id {

                    // Found a match

                    self.next_index += 1;

                    return Some(message);

                }

            } else {

                // Unreachable because:

                //  1. We only iterate over messages that were previously retrieved by `read_next_message`.

                //  2. Inbox does not contain control/ping messages.

                //  3. If `read_next_message` encounters anything else than a data message, it is removed from the inbox.

                unreachable!();

            }

            self.next_index += 1;

        }

        // No more messages

        return None;

    }

}

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

// Private Inbox

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

/// A structure that contains inbox messages. Some messages are left inside and

/// continuously re-read. Others are taken out, but may potentially be put back

/// for later reading. Later reading in this case implies that they are put back

/// for reading in the next sync round.

struct Inbox {

    temp_m: Vec<Message>,

    temp_d: Vec<Message>,

    messages: RawVec<Message>,

    next_delay_idx: u32,

    start_read_idx: u32,

    next_read_idx: u32,

    last_read_idx: u32,

    generation: u32,

}

#[derive(Clone, Copy)]

pub(crate) struct MessageTicket {

    index: u32,

    generation: u32,

}

impl Inbox {

    fn new() -> Self {

        return Inbox {

            temp_m: Vec::new(), temp_d: Vec::new(),

            messages: RawVec::new(),

            next_delay_idx: 0,

            start_read_idx: 0,

            next_read_idx: 0,

            last_read_idx: 0,

            generation: 0,

        }

    }

    fn insert_new(&mut self, message: Message) {

        assert!(self.messages.len() < u32::MAX as usize); // TODO: @Size

        self.temp_m.push(message);

        return;

        self.messages.push(message);

    }

    fn get_next_message_ticket(&mut self) -> Option<MessageTicket> {

        if self.next_read_idx as usize >= self.temp_m.len() { return None };

        let idx = self.next_read_idx;

        self.generation += 1;

        self.next_read_idx += 1;

        return Some(MessageTicket{ index: idx, generation: self.generation });

        let cur_read_idx = self.next_read_idx as usize;

        if cur_read_idx >= self.messages.len() {

            return None;

        }

        self.generation += 1;

        self.next_read_idx += 1;

        return Some(MessageTicket{

            index: cur_read_idx as u32,

            generation: self.generation

        });

    }

    fn read_message_using_ticket(&self, ticket: MessageTicket) -> &Message {

        debug_assert_eq!(self.generation, ticket.generation);

        return &self.temp_m[ticket.index as usize];

        return unsafe{ &*self.messages.get(ticket.index as usize) }

    }

    fn take_message_using_ticket(&mut self, ticket: MessageTicket) -> Message {

        debug_assert_eq!(self.generation, ticket.generation);

        debug_assert!(ticket.index < self.next_read_idx);

        self.next_read_idx -= 1;

        return self.temp_m.remove(ticket.index as usize);

        unsafe {

            let take_idx = ticket.index as usize;

            let val = std::ptr::read(self.messages.get(take_idx));

            // Move messages to the right, clearing up space in the

            // front.

            let num_move_right = take_idx - self.start_read_idx as usize;

            self.messages.move_range(

                self.start_read_idx as usize,

                self.start_read_idx as usize + 1,

                num_move_right

            );

            self.start_read_idx += 1;

            return val;

        }

    }

    fn put_back_message(&mut self, message: Message) {

        // We have space in front of the array because we've taken out a message

        // before.

        self.temp_d.push(message);

        return;

        debug_assert!(self.next_delay_idx < self.start_read_idx);

        unsafe {

            // Write to front of the array

            std::ptr::write(self.messages.get_mut(self.next_delay_idx as usize), message);

            self.next_delay_idx += 1;

        }

    }

    fn get_read_data_messages(&self, match_port_id: PortIdLocal) -> MessagesIter {