// So currently the code is organized as following:

// - The scheduler that is running the component is the authoritative source on

//     ports during *non-sync* mode. The consensus algorithm is the

//     authoritative source during *sync* mode. They retrieve each other's

//     state during the transitions. Hence port data exists duplicated between

//     these two datastructures.

// - The execution tree is where executed branches reside. But the execution

//     tree is only aware of the tree shape itself (and keeps track of some

//     queues of branches that are in a particular state), and tends to store

//     the PDL program state. The consensus algorithm is also somewhat aware

//     of the execution tree, but only in terms of what is needed to complete

//     a sync round (for now, that means the port mapping in each branch).

//     Hence once more we have properties conceptually associated with branches

//     in two places.

// - TODO: Write about handling messages, consensus wrapping data

// - TODO: Write about way information is exchanged between PDL/component and scheduler through ctx

use std::collections::HashMap;

use std::sync::atomic::AtomicBool;

use crate::{PortId, ProtocolDescription};

use crate::common::ComponentState;

use crate::protocol::eval::{EvalContinuation, EvalError, Prompt, Value, ValueGroup};

use crate::protocol::{RunContext, RunResult};

use super::native::Connector;

use super::port::{PortKind, PortIdLocal};

use super::scheduler::{ComponentCtx, SchedulerCtx};

pub(crate) struct ConnectorPublic {

    pub inbox: PublicInbox,

    pub sleeping: AtomicBool,

}

impl ConnectorPublic {

    pub fn new(initialize_as_sleeping: bool) -> Self {

        ConnectorPublic{

            inbox: PublicInbox::new(),

            sleeping: AtomicBool::new(initialize_as_sleeping),

        }

    }

}

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

enum Mode {

    NonSync,    // running non-sync code

    Sync,       // running sync code (in potentially multiple branches)

    SyncError,  // encountered an unrecoverable error in sync mode

    Error,      // encountered an error in non-sync mode (or finished handling the sync mode error).

    pub fn handle_new_data_message(&mut self, message: DataMessage, ctx: &mut ComponentCtx) {

        // Go through all branches that are awaiting new messages and see if

        // there is one that can receive this message.

        if !self.consensus.handle_new_data_message(&message, ctx) {

            // Old message, so drop it

            return;

        }

        let mut iter_id = self.tree.get_queue_first(QueueKind::AwaitingMessage);

        while let Some(branch_id) = iter_id {

            iter_id = self.tree.get_queue_next(branch_id);

            let branch = &self.tree[branch_id];

            if branch.awaiting_port != message.data_header.target_port { continue; }

            if !self.consensus.branch_can_receive(branch_id, &message) { continue; }

            // This branch can receive, so fork and given it the message

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

            self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

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

            debug_assert!(receiving_branch.awaiting_port == message.data_header.target_port);

            receiving_branch.awaiting_port = PortIdLocal::new_invalid();

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

            // And prepare the branch for running

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

        }

    }

    pub fn handle_new_sync_comp_message(&mut self, message: SyncCompMessage, ctx: &mut ComponentCtx) -> Option<ConnectorScheduling> {

        if let Some(round_conclusion) = self.consensus.handle_new_sync_comp_message(message, ctx) {

            return Some(self.enter_non_sync_mode(round_conclusion, ctx));

        }

        return None;

    }

    pub fn handle_new_sync_port_message(&mut self, message: SyncPortMessage, ctx: &mut ComponentCtx) {

        self.consensus.handle_new_sync_port_message(message, ctx);

    }

    // --- Running code

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

        // Check if we have any branch that needs running

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

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

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

                if let Err(_) = comp_ctx.submit_message(Message::Data(DataMessage {

                    sync_header, data_header,

                })) {

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

                }

                branch.prepared = PreparedStatement::PerformedPut;

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

                return ConnectorScheduling::Immediate;

            },

            _ => 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;

            RoundConclusion::Failure => if self.mode == Mode::SyncError {

                // We experienced an error, so exit now

                None

            } else {

                // We didn't experience an error, so retry

                // TODO: Decide what to do with sync errors

                Some(BranchId::new_invalid())

            }

        };

        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;

            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;

        }

    }

}

use crate::collections::VecSet;

use crate::protocol::eval::ValueGroup;

use super::ConnectorId;

use super::branch::BranchId;

use super::port::{ChannelId, PortIdLocal};

use super::inbox::{

    SyncCompMessage, SyncCompContent, SyncPortMessage, SyncPortContent, SyncHeader,

};

struct BranchAnnotation {

    channel_mapping: Vec<ChannelAnnotation>,

    cur_marker: BranchMarker,

}

#[derive(Debug)]

pub(crate) struct LocalSolution {

    component: ConnectorId,

    final_branch_id: BranchId,

    port_mapping: Vec<(ChannelId, BranchMarker)>,

}

#[derive(Debug, Clone)]

pub(crate) struct GlobalSolution {

    component_branches: Vec<(ConnectorId, BranchId)>,

    channel_mapping: Vec<(ChannelId, BranchMarker)>, // TODO: This can go, is debugging info

}

#[derive(Debug, PartialEq, Eq)]

pub enum RoundConclusion {

    Failure,

    Success(BranchId),

}

        unreachable!("notify_of_speculative_mapping called with unowned port");

    }

    /// Generates a new local solution from a finished branch. If the component

    /// is not the leader of the sync region then it will be sent to the

    /// appropriate component. If it is the leader then there is a chance that

    /// this solution completes a global solution. In that case the solution

    /// branch ID will be returned.

    pub(crate) fn handle_new_finished_sync_branch(&mut self, branch_id: BranchId, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {

        // Turn the port mapping into a local solution

        let source_mapping = &self.branch_annotations[branch_id.index as usize].channel_mapping;

        let mut target_mapping = Vec::with_capacity(source_mapping.len());

        for port in source_mapping {

            // Note: if the port is silent, and we've never communicated

            // over the port, then we need to do so now, to let the peer

            // component know about our sync leader state.

            let port_desc = ctx.get_port_by_channel_id(port.channel_id).unwrap();

            let self_port_id = port_desc.self_id;

            let peer_port_id = port_desc.peer_id;

            let channel_id = port_desc.channel_id;

            if !self.encountered_ports.contains(&self_port_id) {

                    sync_header: SyncHeader{

                        sending_component_id: ctx.id,

                        highest_component_id: self.highest_connector_id,

                        sync_round: self.sync_round

                    },

                }));

                self.encountered_ports.push(self_port_id);

            }

            target_mapping.push((

                channel_id,

                port.registered_id.unwrap_or(BranchMarker::new_invalid())

            ));

        }

        let local_solution = LocalSolution{

            component: ctx.id,

            final_branch_id: branch_id,

            port_mapping: target_mapping,

        };

        let maybe_conclusion = self.send_to_leader_or_handle_as_leader(SyncCompContent::LocalSolution(local_solution), ctx);

        return maybe_conclusion;

    }

    /// Notifies the consensus algorithm about the chosen branch to commit to

    /// memory (may be the invalid "start" branch)

    pub fn end_sync(&mut self, branch_id: BranchId, final_ports: &mut Vec<ComponentPortChange>) {

        debug_assert!(self.is_in_sync());

                return Some(RoundConclusion::Success(*branch_id));

            },

            SyncCompContent::GlobalFailure => {

                // Global failure of round, send Ack to leader

                debug_assert_ne!(self.highest_connector_id, ctx.id); // not the leader

                let _result = self.send_to_leader_or_handle_as_leader(SyncCompContent::AckFailure, ctx);

                debug_assert!(_result.is_none());

                return Some(RoundConclusion::Failure);

            },

            SyncCompContent::Notification => {

                // We were just interested in the sync header we handled above

                return None;

            }

        }

    }

    pub fn handle_new_sync_port_message(&mut self, message: SyncPortMessage, ctx: &mut ComponentCtx) {

        if !self.handle_received_sync_header(&message.sync_header, ctx) {

            return;

        }

        debug_assert!(self.is_in_sync());

        debug_assert!(ctx.get_port_by_id(message.target_port).is_some());

        match message.content {

            SyncPortContent::NotificationWave => {

                // Wave to discover everyone in the network, handling sync

                // header takes care of leader discovery, here we need to make

                // sure we propagate the wave

                if self.handled_wave {

                    return;

                }

                self.handled_wave = true;

                // Propagate wave to all peers except the one that has sent us

                // the wave.

                for mapping in &self.branch_annotations[0].channel_mapping {

                    let channel_id = mapping.channel_id;

                    let port_desc = ctx.get_port_by_channel_id(channel_id).unwrap();

                    if port_desc.self_id == message.target_port {

                        // Wave came from this port, no need to send one back

                        continue;

                    }

                    let message = SyncPortMessage{

                        sync_header: self.create_sync_header(ctx),

                        source_port: port_desc.self_id,

                        target_port: port_desc.peer_id,

                        content: SyncPortContent::NotificationWave,

                    };

                    ctx.submit_message(Message::SyncPort(message)).unwrap();

                }

            }

        }

    }

    pub fn notify_of_received_message(&mut self, branch_id: BranchId, message: &DataMessage, ctx: &ComponentCtx) {

        debug_assert!(self.branch_can_receive(branch_id, message));

        let target_port = ctx.get_port_by_id(message.data_header.target_port).unwrap();

        let branch = &mut self.branch_annotations[branch_id.index as usize];

        for mapping in &mut branch.channel_mapping {

            if mapping.channel_id == target_port.channel_id {

                // Found the port in which the message should be inserted

                mapping.registered_id = Some(message.data_header.new_mapping);

                // Check for sent ports

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

                if !self.workspace_ports.is_empty() {

                    todo!("handle received ports");

                    self.workspace_ports.clear();

                }

                return;

            }

        }

        // If here, then the branch didn't actually own the port? Means the

        // caller made a mistake

        unreachable!("incorrect notify_of_received_message");

    }

    /// Matches the mapping between the branch and the data message. If they

    /// match then the branch can receive the message.

    pub fn branch_can_receive(&self, branch_id: BranchId, message: &DataMessage) -> bool {

        if let Some(peer) = self.peers.iter().find(|v| v.id == message.sync_header.sending_component_id) {

            if message.sync_header.sync_round < peer.expected_sync_round {

                return false;

            }

        }

        let annotation = &self.branch_annotations[branch_id.index as usize];

        for expected in &message.data_header.expected_mapping {

            // If we own the port, then we have an entry in the

            // annotation, check if the current mapping matches

            for current in &annotation.channel_mapping {

                if expected.channel_id == current.channel_id {

                    if expected.registered_id != current.registered_id {

                        // IDs do not match, we cannot receive the

                        // message in this branch

                        return false;

                    }

                }

            }

        }

        return true;

    }

    // --- Internal helpers

    fn handle_received_sync_header(&mut self, sync_header: &SyncHeader, ctx: &mut ComponentCtx) -> bool {

        debug_assert!(sync_header.sending_component_id != ctx.id); // not sending to ourselves

        if !self.handle_peer(sync_header) {

            // We can drop this package

    #[inline]

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

        return Self{ marker: 0 }

    }

}

/// The header added by the synchronization algorithm to all.

#[derive(Debug, Clone)]

pub(crate) struct SyncHeader {

    pub sending_component_id: ConnectorId,

    pub highest_component_id: ConnectorId,

    pub sync_round: u32,

}

/// The header added to data messages

#[derive(Debug, Clone)]

pub(crate) struct DataHeader {

    pub expected_mapping: Vec<ChannelAnnotation>,

    pub sending_port: PortIdLocal,

    pub target_port: PortIdLocal,

    pub new_mapping: BranchMarker,

}

/// A data message is a message that is intended for the receiver's PDL code,

/// but will also be handled by the consensus algorithm

#[derive(Debug, Clone)]

pub(crate) struct DataMessage {

    pub sync_header: SyncHeader,

    pub data_header: DataHeader,

}

#[derive(Debug)]

pub(crate) enum SyncCompContent {

    LocalFailure, // notifying leader that component has failed (e.g. timeout, whatever)

    LocalSolution(LocalSolution), // sending a local solution to the leader

    PartialSolution(SolutionCombiner), // when new leader is detected, forward all local results

    GlobalSolution(GlobalSolution), // broadcasting to everyone

    GlobalFailure, // broadcasting to everyone

    AckFailure, // acknowledgement of failure to leader

    Notification, // just a notification (so purpose of message is to send the SyncHeader)

    Presence(ConnectorId, Vec<ChannelId>), // notifying leader of component presence (needed to ensure failing a round involves all components in a sync round)

}

/// A sync message is a message that is intended only for the consensus

/// algorithm. The message goes directly to a component.

#[derive(Debug)]

pub(crate) struct SyncCompMessage {

    pub sync_header: SyncHeader,

    pub target_component_id: ConnectorId,

    pub content: SyncCompContent,

}

#[derive(Debug)]

pub(crate) enum SyncPortContent {

    NotificationWave,

}

#[derive(Debug)]

pub(crate) struct SyncPortMessage {

    pub sync_header: SyncHeader,

    pub source_port: PortIdLocal,

    pub target_port: PortIdLocal,

    pub content: SyncPortContent,

}

/// A control message is a message intended for the scheduler that is executing

/// a component.

#[derive(Debug)]

pub(crate) struct ControlMessage {

    pub id: u32, // generic identifier, used to match request to response

    pub sending_component_id: ConnectorId,

    pub content: ControlContent,

}

#[derive(Debug)]

pub(crate) enum ControlContent {

    PortPeerChanged(PortIdLocal, ConnectorId),

    CloseChannel(PortIdLocal),

    Ack,

    Ping,

}

/// Combination of data message and control messages.

#[derive(Debug)]

pub(crate) enum Message {

    Data(DataMessage),

    SyncComp(SyncCompMessage),

    SyncPort(SyncPortMessage),

    Control(ControlMessage),

}

impl Message {

    /// If the message is sent through a particular channel, then this function

    /// returns the port through which the message was sent.

    pub(crate) fn source_port(&self) -> Option<PortIdLocal> {

        // Currently only data messages have a source port

        }

    }

}

/// The public inbox of a connector. The thread running the connector that owns

/// this inbox may retrieved from it. Non-owning threads may only put new

/// messages inside of it.

// TODO: @Optimize, lazy concurrency. Probably ringbuffer with read/write heads.

//  Should behave as a MPSC queue.

pub struct PublicInbox {

    messages: Mutex<VecDeque<Message>>,

}

impl PublicInbox {

    pub fn new() -> Self {

        Self{

            messages: Mutex::new(VecDeque::new()),

        }

    }

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

        let mut lock = self.messages.lock().unwrap();

        lock.push_back(message);

    }

use std::collections::VecDeque;

use std::sync::{Arc, Mutex, Condvar};

use std::sync::atomic::Ordering;

use std::collections::HashMap;

use crate::protocol::ComponentCreationError;

use crate::protocol::eval::ValueGroup;

use crate::runtime2::consensus::RoundConclusion;

use super::{ConnectorKey, ConnectorId, RuntimeInner};

use super::branch::{BranchId, FakeTree, QueueKind, SpeculativeState};

use super::scheduler::{SchedulerCtx, ComponentCtx};

use super::port::{Port, PortIdLocal, Channel, PortKind};

use super::consensus::{Consensus, Consistency, find_ports_in_value_group};

use super::connector::{ConnectorScheduling, ConnectorPDL};

use super::inbox::{

    SyncCompMessage, SyncPortMessage,

    ControlContent, ControlMessage

};

/// Generic connector interface from the scheduler's point of view.

pub(crate) trait Connector {

    /// Should run the connector's behaviour up until the next blocking point.

    /// One should generally request and handle new messages from the component

    /// context. Then perform any logic the component has to do, and in the

    /// process perhaps queue up some state changes using the same context.

    fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling;

}

pub(crate) struct FinishedSync {

    // In the order of the `get` calls

    success: bool,

    inbox: Vec<ValueGroup>,

}

type SyncDone = Arc<(Mutex<Option<FinishedSync>>, Condvar)>;

type JobQueue = Arc<Mutex<VecDeque<ApplicationJob>>>;

enum ApplicationJob {

    NewChannel((Port, Port)),

    pub(crate) fn handle_new_data_message(&mut self, message: DataMessage, ctx: &mut ComponentCtx) {

        // Go through all branches that are awaiting new messages and see if

        // there is one that can receive this message.

        if !self.consensus.handle_new_data_message(&message, ctx) {

            // Old message, so drop it

            return;

        }

        let mut iter_id = self.tree.get_queue_first(QueueKind::AwaitingMessage);

        while let Some(branch_id) = iter_id {

            iter_id = self.tree.get_queue_next(branch_id);

            let branch = &self.tree[branch_id];

            if branch.awaiting_port != message.data_header.target_port { continue; }

            if !self.consensus.branch_can_receive(branch_id, &message) { continue; }

            // This branch can receive, so fork and given it the message

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

            debug_assert!(receiving_branch_id.index as usize == self.branch_extra.len());

            self.branch_extra.push(self.branch_extra[branch_id.index as usize]); // copy instruction index

            self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

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

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

            // And prepare the branch for running

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

        }

    }

    pub(crate) fn handle_new_sync_comp_message(&mut self, message: SyncCompMessage, ctx: &mut ComponentCtx) {

        if let Some(conclusion) = self.consensus.handle_new_sync_comp_message(message, ctx) {

            self.collapse_sync_to_conclusion(conclusion, ctx);

        }

    }

    pub(crate) fn handle_new_sync_port_message(&mut self, message: SyncPortMessage, ctx: &mut ComponentCtx) {

        self.consensus.handle_new_sync_port_message(message, ctx);

    }

    fn run_in_sync_mode(&mut self, _sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {

        debug_assert!(self.is_in_sync);

        self.handle_new_messages(comp_ctx);

        let branch_id = self.tree.pop_from_queue(QueueKind::Runnable);

        if branch_id.is_none() {

        if instruction_idx >= self.sync_desc.len() {

            // Performed last instruction, so this branch is officially at the

            // end of the synchronous interaction.

            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;

            }

        } else {

            // We still have instructions to perform

            let cur_instruction = &self.sync_desc[instruction_idx];

            self.branch_extra[branch_id.index as usize] += 1;

            match &cur_instruction {

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

                    let port_id = *port_id;

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

                    let message = Message::Data(DataMessage {

                        sync_header,

                        data_header,

                    });

                    comp_ctx.submit_message(message);

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

                    return ConnectorScheduling::Immediate;

                },

                ApplicationSyncAction::Get(port_id) => {

                    let port_id = *port_id;

                    branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

                    branch.awaiting_port = port_id;

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

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

                            debug_assert!(receiving_branch_id.index as usize == self.branch_extra.len());

                            self.branch_extra.push(instruction_idx + 1);

                            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;

                    }

                }

            }

        }

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

            return ConnectorScheduling::NotNow;

        } else {

            return ConnectorScheduling::Later;

        }

    }

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

            // connector.

            let mut cur_schedule = ConnectorScheduling::Immediate;

            while let ConnectorScheduling::Immediate = cur_schedule {

                self.handle_inbox_messages(scheduled);

                // Run the main behaviour of the connector, depending on its

                // current state.

                if scheduled.shutting_down {

                    // Nothing to do. But we're stil waiting for all our pending

                    // control messages to be answered.

                    self.debug_conn(connector_id, &format!("Shutting down, {} Acks remaining", scheduled.router.num_pending_acks()));

                    if scheduled.router.num_pending_acks() == 0 {

                        // We're actually done, we can safely destroy the

                        // currently running connector

                        self.runtime.destroy_component(connector_key);

                        continue 'thread_loop;

                    } else {

                        cur_schedule = ConnectorScheduling::NotNow;

                    }

                } else {

                    self.debug_conn(connector_id, "Running ...");

                    let scheduler_ctx = SchedulerCtx{ runtime: &*self.runtime };

                    let new_schedule = scheduled.connector.run(scheduler_ctx, &mut scheduled.ctx);

                    self.debug_conn(connector_id, &format!("Finished running (new scheduling is {:?})", new_schedule));

                    // Handle all of the output from the current run: messages to

                    // send and connectors to instantiate.

                    self.handle_changes_in_context(scheduled);

                    cur_schedule = new_schedule;

                }

            }

            // If here then the connector does not require immediate execution.

            // So enqueue it if requested, and otherwise put it in a sleeping

            // state.

            match cur_schedule {

                ConnectorScheduling::Immediate => unreachable!(),

                ConnectorScheduling::Later => {

                    // Simply queue it again later

                    self.runtime.push_work(connector_key);

                },

                ConnectorScheduling::NotNow => {

                    // Need to sleep, note that we are the only ones which are

                    // allows to set the sleeping state to `true`, and since

                    // we're running it must currently be `false`.

                    self.try_go_to_sleep(connector_key, scheduled);

                },

                ConnectorScheduling::Exit => {