match stmt {

                Statement::If(stmt) => cur_frame.prepare_single_expression(heap, stmt.test),

                Statement::While(stmt) => cur_frame.prepare_single_expression(heap, stmt.test),

                Statement::Return(stmt) => {

                    debug_assert_eq!(stmt.expressions.len(), 1); // TODO: @ReturnValues

                    cur_frame.prepare_single_expression(heap, stmt.expressions[0]);

                },

                Statement::New(stmt) => {

                    // Note that we will end up not evaluating the call itself.

                    // Rather we will evaluate its expressions and then

                    // instantiate the component upon reaching the "new" stmt.

                    let call_expr = &heap[stmt.expression];

                    cur_frame.prepare_multiple_expressions(heap, &call_expr.arguments);

                },

                Statement::Expression(stmt) => {

                    cur_frame.prepare_single_expression(heap, stmt.expression);

                }

                _ => {},

            }

        }

        return_value

    }

}

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

use crate::runtime2::branch::PreparedStatement;

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

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

use super::inbox::{DataMessage, DataContent, Message, SyncMessage, PublicInbox};

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

pub(crate) enum ConnectorScheduling {

    Immediate,          // Run again, immediately

    Later,              // Schedule for running, at some later point in time

    NotNow,             // Do not reschedule for running

    Exit,               // Connector has exited

    Error(EvalError),   // Connector has experienced a fatal error

}

pub(crate) struct ConnectorPDL {

    tree: ExecTree,

    consensus: Consensus,

    last_finished_handled: Option<BranchId>,

}

// TODO: Remove remaining fields once 'fires()' is removed from language.

struct ConnectorRunContext<'a> {

    branch_id: BranchId,

    consensus: &'a Consensus,

    prepared: PreparedStatement,

}

impl<'a> RunContext for ConnectorRunContext<'a>{

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

        return match self.prepared.take() {

            PreparedStatement::None => false,

            PreparedStatement::PerformedPut => true,

            taken => unreachable!("prepared statement is '{:?}' during 'performed_put()'", taken)

        };

    }

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

        return match self.prepared.take() {

            PreparedStatement::None => None,

        let annotation = self.consensus.get_annotation(self.branch_id, port_id);

        return annotation.expected_firing.map(|v| Value::Bool(v));

    }

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

        return match self.prepared.take() {

            PreparedStatement::None => None,

            PreparedStatement::CreatedChannel(ports) => Some(ports),

            taken => unreachable!("prepared statement is '{:?}' during 'created_channel)_'", taken),

        };

    }

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

        return match self.prepared.take() {

            PreparedStatement::None => None,

            PreparedStatement::ForkedExecution(path) => Some(path),

            taken => unreachable!("prepared statement is '{:?}' during 'performed_fork()'", taken),

        };

    }

}

impl Connector for ConnectorPDL {

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

        self.handle_new_messages(comp_ctx);

                }

        }

    }

}

impl ConnectorPDL {

    pub fn new(initial: Prompt) -> Self {

        Self{

            tree: ExecTree::new(initial),

            consensus: Consensus::new(),

            last_finished_handled: None,

        }

    }

    // --- Handling messages

    pub fn handle_new_messages(&mut self, ctx: &mut ComponentCtx) {

        while let Some(message) = ctx.read_next_message() {

            match message {

                Message::Data(message) => self.handle_new_data_message(message, ctx),

                Message::Sync(message) => self.handle_new_sync_message(message, ctx),

                Message::Control(_) => unreachable!("control message in component"),

            }

        }

    }

    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;

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

            receiving_branch.prepared = PreparedStatement::PerformedGet(message.content.as_message().unwrap().clone());

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

            // And prepare the branch for running

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

        }

    }

    pub fn handle_new_sync_message(&mut self, message: SyncMessage, ctx: &mut ComponentCtx) {

        if let Some(solution_branch_id) = self.consensus.handle_new_sync_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());

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

        if branch_id.is_none() {

            return ConnectorScheduling::NotNow;

        }

        // Retrieve the branch and run it

        let branch_id = branch_id.unwrap();

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

        let mut run_context = ConnectorRunContext{

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

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

                    sync_header, data_header,

                    content: DataContent::Message(content),

                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;

        }

    }

    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{

                );

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

        }

    }

        let mut fake_vec = Vec::new();

        self.tree.end_sync(solution_branch_id);

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

        for port in fake_vec {

            // TODO: Handle sent/received ports

            debug_assert!(ctx.get_port_by_id(port).is_some());

        }

        ctx.notify_sync_end(&[]);

        self.last_finished_handled = None;

    }

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

        }

    }

}

            cur_marker: BranchMarker::new_invalid(),

        });

        self.branch_markers.push(BranchId::new_invalid());

        self.highest_connector_id = ctx.id;

    }

    /// Notifies the consensus algorithm that a new branch has appeared. Must be

    /// called for each forked branch in the execution tree.

    pub fn notify_of_new_branch(&mut self, parent_branch_id: BranchId, new_branch_id: BranchId) {

        // If called correctly. Then each time we are notified the new branch's

        // index is the length in `branch_annotations`.

        debug_assert!(self.branch_annotations.len() == new_branch_id.index as usize);

        let parent_branch_annotations = &self.branch_annotations[parent_branch_id.index as usize];

        let new_marker = BranchMarker::new(self.branch_markers.len() as u32);

        let new_branch_annotations = BranchAnnotation{

            port_mapping: parent_branch_annotations.port_mapping.clone(),

            cur_marker: new_marker,

        };

        self.branch_annotations.push(new_branch_annotations);

        self.branch_markers.push(new_branch_id);

    }

    /// Notifies the consensus algorithm that a branch has reached the end of

    /// the sync block. A final check for consistency will be performed that the

    /// caller has to handle. Note that

    pub fn notify_of_finished_branch(&self, branch_id: BranchId) -> Consistency {

        debug_assert!(self.is_in_sync());

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

        for mapping in &branch.port_mapping {

            match mapping.expected_firing {

                Some(expected) => {

                    if expected != mapping.registered_id.is_some() {

                        // Inconsistent speculative state and actual state

                        debug_assert!(mapping.registered_id.is_none()); // because if we did fire on a silent port, we should've caught that earlier

                        return Consistency::Inconsistent;

                    }

                },

                None => {},

            }

        }

        return Consistency::Valid;

    }

    /// Notifies the consensus algorithm that a particular branch has assumed

    /// a speculative value for its port mapping.

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

    Sync(SyncMessage),

    Control(ControlMessage),

}

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

    }

    pub(crate) fn take_message(&self) -> Option<Message> {

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

        return lock.pop_front();

use std::collections::VecDeque;

use std::sync::Arc;

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

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

use super::port::{Port, PortState, PortIdLocal};

use super::native::Connector;

use super::branch::{BranchId};

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

use super::inbox::{Message, DataMessage, ControlMessage, ControlContent};

// Because it contains pointers we're going to do a copy by value on this one

#[derive(Clone, Copy)]

pub(crate) struct SchedulerCtx<'a> {

    pub(crate) runtime: &'a RuntimeInner

}

pub(crate) struct Scheduler {

    runtime: Arc<RuntimeInner>,

    scheduler_id: u32,

}

impl Scheduler {

    pub fn new(runtime: Arc<RuntimeInner>, scheduler_id: u32) -> Self {

        return Self{ runtime, scheduler_id };

    }

}

#[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_messages: Vec<Message>,

    inbox_len_read: usize,

    // 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_messages: Vec::new(),

            inbox_len_read: 0,

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

    }

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

    }

    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.

        debug_assert!(self.is_in_sync);

        self.outbox.push_back(contents);

    }

    /// 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 MessagesIter {

            messages: &self.inbox_messages,

            next_index: 0,

            max_index: self.inbox_len_read,