#![allow(dead_code)]

use std::fmt::{Debug, Display};

use std::io::Write as IOWrite;

use super::ast::*;

use super::token_parsing::*;

const INDENT: usize = 2;

const PREFIX_EMPTY: &'static str = "    ";

const PREFIX_ROOT_ID: &'static str = "Root";

const PREFIX_PRAGMA_ID: &'static str = "Prag";

const PREFIX_IMPORT_ID: &'static str = "Imp ";

const PREFIX_TYPE_ANNOT_ID: &'static str = "TyAn";

const PREFIX_VARIABLE_ID: &'static str = "Var ";

const PREFIX_DEFINITION_ID: &'static str = "Def ";

const PREFIX_STRUCT_ID: &'static str = "DefS";

const PREFIX_ENUM_ID: &'static str = "DefE";

const PREFIX_UNION_ID: &'static str = "DefU";

const PREFIX_COMPONENT_ID: &'static str = "DefC";

const PREFIX_FUNCTION_ID: &'static str = "DefF";

const PREFIX_STMT_ID: &'static str = "Stmt";

const PREFIX_BLOCK_STMT_ID: &'static str = "SBl ";

const PREFIX_ENDBLOCK_STMT_ID: &'static str = "SEBl";

const PREFIX_LOCAL_STMT_ID: &'static str = "SLoc";

const PREFIX_MEM_STMT_ID: &'static str = "SMem";

const PREFIX_CHANNEL_STMT_ID: &'static str = "SCha";

const PREFIX_SKIP_STMT_ID: &'static str = "SSki";

const PREFIX_LABELED_STMT_ID: &'static str = "SLab";

const PREFIX_IF_STMT_ID: &'static str = "SIf ";

const PREFIX_ENDIF_STMT_ID: &'static str = "SEIf";

const PREFIX_WHILE_STMT_ID: &'static str = "SWhi";

const PREFIX_ENDWHILE_STMT_ID: &'static str = "SEWh";

const PREFIX_BREAK_STMT_ID: &'static str = "SBre";

const PREFIX_CONTINUE_STMT_ID: &'static str = "SCon";

const PREFIX_SYNC_STMT_ID: &'static str = "SSyn";

const PREFIX_ENDSYNC_STMT_ID: &'static str = "SESy";

const PREFIX_RETURN_STMT_ID: &'static str = "SRet";

const PREFIX_ASSERT_STMT_ID: &'static str = "SAsr";

const PREFIX_GOTO_STMT_ID: &'static str = "SGot";

const PREFIX_NEW_STMT_ID: &'static str = "SNew";

const PREFIX_PUT_STMT_ID: &'static str = "SPut";

const PREFIX_EXPR_STMT_ID: &'static str = "SExp";

const PREFIX_ASSIGNMENT_EXPR_ID: &'static str = "EAsi";

const PREFIX_BINDING_EXPR_ID: &'static str = "EBnd";

const PREFIX_CONDITIONAL_EXPR_ID: &'static str = "ECnd";

const PREFIX_BINARY_EXPR_ID: &'static str = "EBin";

const PREFIX_UNARY_EXPR_ID: &'static str = "EUna";

const PREFIX_INDEXING_EXPR_ID: &'static str = "EIdx";

const PREFIX_SLICING_EXPR_ID: &'static str = "ESli";

const PREFIX_SELECT_EXPR_ID: &'static str = "ESel";

const PREFIX_LITERAL_EXPR_ID: &'static str = "ELit";

const PREFIX_CAST_EXPR_ID: &'static str = "ECas";

const PREFIX_CALL_EXPR_ID: &'static str = "ECll";

const PREFIX_VARIABLE_EXPR_ID: &'static str = "EVar";

struct KV<'a> {

    buffer: &'a mut String,

    prefix: Option<(&'static str, i32)>,

    indent: usize,

    temp_key: &'a mut String,

    temp_val: &'a mut String,

}

impl<'a> KV<'a> {

    fn new(buffer: &'a mut String, temp_key: &'a mut String, temp_val: &'a mut String, indent: usize) -> Self {

        temp_key.clear();

        temp_val.clear();

        KV{

            buffer,

            prefix: None,

            indent,

            temp_key,

            temp_val

        }

    }

    fn with_id(mut self, prefix: &'static str, id: i32) -> Self {

        self.prefix = Some((prefix, id));

        self

    }

    fn with_s_key(self, key: &str) -> Self {

        self.temp_key.push_str(key);

        self

            },

            Statement::While(stmt) => {

                self.kv(indent).with_id(PREFIX_WHILE_STMT_ID, stmt.this.0.index)

                    .with_s_key("While");

                self.kv(indent2).with_s_key("EndWhile").with_disp_val(&stmt.end_while.0.index);

                self.kv(indent2).with_s_key("InSync")

                    .with_disp_val(&stmt.in_sync.0.index);

                self.kv(indent2).with_s_key("Condition");

                self.write_expr(heap, stmt.test, indent3);

                self.kv(indent2).with_s_key("Body");

                self.write_stmt(heap, stmt.body.upcast(), indent3);

            },

            Statement::EndWhile(stmt) => {

                self.kv(indent).with_id(PREFIX_ENDWHILE_STMT_ID, stmt.this.0.index)

                    .with_s_key("EndWhile");

                self.kv(indent2).with_s_key("StartWhile").with_disp_val(&stmt.start_while.0.index);

                self.kv(indent2).with_s_key("Next").with_disp_val(&stmt.next.index);

            },

            Statement::Break(stmt) => {

                self.kv(indent).with_id(PREFIX_BREAK_STMT_ID, stmt.this.0.index)

                    .with_s_key("Break");

                self.kv(indent2).with_s_key("Label")

                    .with_opt_identifier_val(stmt.label.as_ref());

                self.kv(indent2).with_s_key("Target")

                    .with_opt_disp_val(stmt.target.as_ref().map(|v| &v.0.index));

            },

            Statement::Continue(stmt) => {

                self.kv(indent).with_id(PREFIX_CONTINUE_STMT_ID, stmt.this.0.index)

                    .with_s_key("Continue");

                self.kv(indent2).with_s_key("Label")

                    .with_opt_identifier_val(stmt.label.as_ref());

                self.kv(indent2).with_s_key("Target")

                    .with_opt_disp_val(stmt.target.as_ref().map(|v| &v.0.index));

            },

            Statement::Synchronous(stmt) => {

                self.kv(indent).with_id(PREFIX_SYNC_STMT_ID, stmt.this.0.index)

                    .with_s_key("Synchronous");

                self.kv(indent2).with_s_key("EndSync").with_disp_val(&stmt.end_sync.0.index);

                self.kv(indent2).with_s_key("Body");

                self.write_stmt(heap, stmt.body.upcast(), indent3);

            },

            Statement::EndSynchronous(stmt) => {

                self.kv(indent).with_id(PREFIX_ENDSYNC_STMT_ID, stmt.this.0.index)

                    .with_s_key("EndSynchronous");

                self.kv(indent2).with_s_key("StartSync").with_disp_val(&stmt.start_sync.0.index);

                self.kv(indent2).with_s_key("Next").with_disp_val(&stmt.next.index);

            },

            Statement::Return(stmt) => {

                self.kv(indent).with_id(PREFIX_RETURN_STMT_ID, stmt.this.0.index)

                    .with_s_key("Return");

                self.kv(indent2).with_s_key("Expressions");

                for expr_id in &stmt.expressions {

                    self.write_expr(heap, *expr_id, indent3);

                }

            },

            Statement::Goto(stmt) => {

                self.kv(indent).with_id(PREFIX_GOTO_STMT_ID, stmt.this.0.index)

                    .with_s_key("Goto");

                self.kv(indent2).with_s_key("Label").with_identifier_val(&stmt.label);

                self.kv(indent2).with_s_key("Target")

                    .with_opt_disp_val(stmt.target.as_ref().map(|v| &v.0.index));

            },

            Statement::New(stmt) => {

                self.kv(indent).with_id(PREFIX_NEW_STMT_ID, stmt.this.0.index)

                    .with_s_key("New");

                self.kv(indent2).with_s_key("Expression");

                self.write_expr(heap, stmt.expression.upcast(), indent3);

                self.kv(indent2).with_s_key("Next").with_disp_val(&stmt.next.index);

            },

            Statement::Expression(stmt) => {

                self.kv(indent).with_id(PREFIX_EXPR_STMT_ID, stmt.this.0.index)

                    .with_s_key("ExpressionStatement");

                self.write_expr(heap, stmt.expression, indent2);

                self.kv(indent2).with_s_key("Next").with_disp_val(&stmt.next.index);

            }

        }

    }

    fn write_expr(&mut self, heap: &Heap, expr_id: ExpressionId, indent: usize) {

        let expr = &heap[expr_id];

        let indent2 = indent + 1;

        let indent3 = indent2 + 1;

        match expr {

            Expression::Assignment(expr) => {

                self.kv(indent).with_id(PREFIX_ASSIGNMENT_EXPR_ID, expr.this.0.index)

                    .with_s_key("AssignmentExpr");

                self.kv(indent2).with_s_key("Operation").with_debug_val(&expr.operation);

                self.kv(indent2).with_s_key("Left");

                self.write_expr(heap, expr.left, indent3);

                self.kv(indent2).with_s_key("Right");

                self.write_expr(heap, expr.right, indent3);

                self.kv(indent2).with_s_key("Parent")

                    .with_custom_val(|v| write_expression_parent(v, &expr.parent));

            },

pub struct BranchId {

    pub index: u32

}

impl BranchId {

    #[inline]

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

        return Self{ index: 0 };

    }

    #[inline]

    fn new(index: u32) -> Self {

        debug_assert!(index != 0);

        return Self{ index };

    }

    #[inline]

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

        return self.index != 0;

    }

}

#[derive(Debug, PartialEq, Eq)]

pub(crate) enum SpeculativeState {

    // Non-synchronous variants

    RunningNonSync,         // regular execution of code

    Error,                  // encountered a runtime error

    Finished,               // finished executing connector's code

    // Synchronous variants

    RunningInSync,          // running within a sync block

    HaltedAtBranchPoint,    // at a branching point (at a `get` call)

    ReachedSyncEnd,         // reached end of sync block, branch represents a local solution

    Inconsistent,           // branch can never represent a local solution, so halted

}

/// The execution state of a branch. This envelops the PDL code and the

/// execution state. And derived from that: if we're ready to keep running the

/// code, or if we're halted for some reason (e.g. waiting for a message).

pub(crate) struct Branch {

    pub id: BranchId,

    pub parent_id: BranchId,

    // Execution state

    pub code_state: ComponentState,

    pub sync_state: SpeculativeState,

    pub awaiting_port: PortIdLocal, // only valid if in "awaiting message" queue. TODO: Maybe put in enum

    pub next_in_queue: BranchId, // used by `ExecTree`/`BranchQueue`

    pub inbox: HashMap<PortIdLocal, ValueGroup>, // TODO: Remove, currently only valid in single-get/put mode

    pub prepared_channel: Option<(Value, Value)>, // TODO: Maybe remove?

}

impl BranchListItem for Branch {

    #[inline] fn get_id(&self) -> BranchId { return self.id; }

    #[inline] fn set_next_id(&mut self, id: BranchId) { self.next_in_queue = id; }

    #[inline] fn get_next_id(&self) -> BranchId { return self.next_in_queue; }

}

impl Branch {

    /// Creates a new non-speculative branch

    pub(crate) fn new_non_sync(component_state: ComponentState) -> Self {

        Branch {

            id: BranchId::new_invalid(),

            parent_id: BranchId::new_invalid(),

            code_state: component_state,

            sync_state: SpeculativeState::RunningNonSync,

            awaiting_port: PortIdLocal::new_invalid(),

            next_in_queue: BranchId::new_invalid(),

            inbox: HashMap::new(),

            prepared_channel: None,

        }

    }

    /// Constructs a sync branch. The provided branch is assumed to be the

    /// parent of the new branch within the execution tree.

    fn new_sync(new_index: u32, parent_branch: &Branch) -> Self {

        debug_assert!(parent_branch.prepared_channel.is_none());

        Branch {

            id: BranchId::new(new_index),

            parent_id: parent_branch.id,

            code_state: parent_branch.code_state.clone(),

            sync_state: SpeculativeState::RunningInSync,

            awaiting_port: parent_branch.awaiting_port,

            next_in_queue: BranchId::new_invalid(),

            inbox: parent_branch.inbox.clone(),

            prepared_channel: None,

        }

    }

    /// Inserts a message into the branch for retrieval by a corresponding

    /// `get(port)` call.

    pub(crate) fn insert_message(&mut self, target_port: PortIdLocal, contents: ValueGroup) {

        debug_assert!(target_port.is_valid());

        debug_assert!(self.awaiting_port == target_port);

        self.awaiting_port = PortIdLocal::new_invalid();

        self.inbox.insert(target_port, contents);

    }

}

/// Queue of branches. Just a little helper.

#[derive(Copy, Clone)]

struct BranchQueue {

    first: BranchId,

    last: BranchId,

}

impl BranchQueue {

    #[inline]

    fn new() -> Self {

        Self{

            first: BranchId::new_invalid(),

            last: BranchId::new_invalid()

        }

    }

    #[inline]

    fn is_empty(&self) -> bool {

        debug_assert!(self.first.is_valid() == self.last.is_valid());

        return !self.first.is_valid();

    }

}

const NUM_QUEUES: usize = 3;

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

pub(crate) enum QueueKind {

    Runnable,

    AwaitingMessage,

    FinishedSync,

}

impl QueueKind {

    fn as_index(&self) -> usize {

        return match self {

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

use crate::PortId;

use crate::common::ComponentState;

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

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

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, Eq, PartialEq)]

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

}

pub(crate) struct ConnectorPDL {

    tree: ExecTree,

    consensus: Consensus,

    last_finished_handled: Option<BranchId>,

}

struct ConnectorRunContext<'a> {

    branch_id: BranchId,

    consensus: &'a Consensus,

    received: &'a HashMap<PortIdLocal, ValueGroup>,

    scheduler: SchedulerCtx<'a>,

    prepared_channel: Option<(Value, Value)>,

}

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

    fn did_put(&mut self, port: PortId) -> bool {

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

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

        return annotation.registered_id.is_some();

    }

    fn get(&mut self, port: PortId) -> Option<ValueGroup> {

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

        match self.received.get(&port_id) {

            Some(data) => Some(data.clone()),

            None => None,

        }

    }

    fn fires(&mut self, port: PortId) -> Option<Value> {

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

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

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

    }

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

        return self.prepared_channel.take();

    }

}

impl Connector for ConnectorPDL {

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

        self.handle_new_messages(comp_ctx);

        if self.tree.is_in_sync() {

            // Run in sync mode

            let scheduling = self.run_in_sync_mode(sched_ctx, comp_ctx);

            // Handle any new finished branches

            let mut iter_id = self.last_finished_handled.or(self.tree.get_queue_first(QueueKind::FinishedSync));

            while let Some(branch_id) = iter_id {

                iter_id = self.tree.get_queue_next(branch_id);

                self.last_finished_handled = Some(branch_id);

                if let Some(solution_branch_id) = self.consensus.handle_new_finished_sync_branch(branch_id, comp_ctx) {

                    // Actually found a solution

                    self.collapse_sync_to_solution_branch(solution_branch_id, comp_ctx);

                    return ConnectorScheduling::Immediate;

                }

                self.last_finished_handled = Some(branch_id);

            }

            return scheduling;

        } else {

            let scheduling = self.run_in_deterministic_mode(sched_ctx, comp_ctx);

            return scheduling;

        }

    }

}

impl ConnectorPDL {

    pub fn new(initial: ComponentState) -> 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),

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

            receiving_branch.insert_message(message.data_header.target_port, 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) {

            self.collapse_sync_to_solution_branch(solution_branch_id, 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,

            received: &branch.inbox,

            scheduler: sched_ctx,

            prepared_channel: branch.prepared_channel.take(),

        };

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

        // Handle the returned result. Note that this match statement contains

        // explicit returns in case the run result requires that the component's

        // code is ran again immediately

        match run_result {

            RunResult::BranchInconsistent => {

                // Branch became inconsistent

                branch.sync_state = SpeculativeState::Inconsistent;

            },

            RunResult::BranchMissingPortState(port_id) => {

                // Branch called `fires()` on a port that has not been used yet.

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

                // Create two forks, one that assumes the port will fire, and

                // one that assumes the port remains silent

                branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

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

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

                self.consensus.notify_of_new_branch(branch_id, firing_branch_id);

                let _result = self.consensus.notify_of_speculative_mapping(firing_branch_id, port_id, true);

                debug_assert_eq!(_result, Consistency::Valid);

                self.consensus.notify_of_new_branch(branch_id, silent_branch_id);

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

                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;

            },

            RunResult::BranchMissingPortValue(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);

                let consistency = self.consensus.notify_of_speculative_mapping(branch_id, port_id, true);

                if consistency == Consistency::Valid {

                    // `get()` is valid, so mark the branch as awaiting a message

                    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.insert_message(port_id, message.content.as_message().unwrap().clone());

                            self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

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

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

                            any_message_received = true;

                        }

                    }

                    if any_message_received {

                        return ConnectorScheduling::Immediate;

                    }

                } else {

                    branch.sync_state = SpeculativeState::Inconsistent;

                }

            }

            RunResult::BranchAtSyncEnd => {

                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;

                }

            },

            RunResult::BranchPut(port_id, content) => {

                // Branch is attempting to send data

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

                let consistency = self.consensus.notify_of_speculative_mapping(branch_id, port_id, true);

                if consistency == Consistency::Valid {

                    // `put()` is valid.

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

                    comp_ctx.submit_message(Message::Data(DataMessage {

                        sync_header, data_header,

                        content: DataContent::Message(content),

                    }));

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

                    return ConnectorScheduling::Immediate;

                } else {

                    branch.sync_state = SpeculativeState::Inconsistent;

                }

            },

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

            received: &branch.inbox,

            scheduler: sched_ctx,

            prepared_channel: branch.prepared_channel.take(),

        };

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

        match run_result {

            RunResult::ComponentTerminated => {

                branch.sync_state = SpeculativeState::Finished;

                return ConnectorScheduling::Exit;

            },

            RunResult::ComponentAtSyncStart => {

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

                return ConnectorScheduling::Immediate;

            },

            RunResult::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_state = ComponentState {

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

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

                comp_ctx.workspace_ports.clear();

                return ConnectorScheduling::Later;

            },

            RunResult::NewChannel => {

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

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

                branch.prepared_channel = Some((

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

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

                ));

                comp_ctx.push_port(putter);

            ApplicationSyncAction::Get(channel.getter_id),

        ]).expect("start sync round");

        let result = api.wait().expect("finish sync round");

        assert!(result.len() == 1 && result[0].values.len() == 1);

        if let Value::UInt32(gotten) = result[0].values[0] {

            assert_eq!(gotten, 1337 + loop_idx);

        } else {

            assert!(false);

        }

    }

}

#[test]

fn test_putting_to_component() {

    const CODE: &'static str = "

    primitive loop_receiver(in<u32> numbers, u32 cur, u32 last) {

        while (cur < last) {

            sync {

                auto number = get(numbers);

                assert(number == cur);

                cur += 1;

            }

        }

    }

    ";

    let pd = ProtocolDescription::parse(CODE.as_bytes()).unwrap();

    let rt = Runtime::new(NUM_THREADS, pd);

    let mut api = rt.create_interface();

    let channel = api.create_channel().unwrap();

    api.create_connector("", "loop_receiver", ValueGroup::new_stack(vec![

        Value::Input(PortId::new(channel.getter_id.index)),

        Value::UInt32(42),

        Value::UInt32(42 + NUM_LOOPS)

    ])).unwrap();

    for loop_idx in 0..NUM_LOOPS {

        api.perform_sync_round(vec![

            ApplicationSyncAction::Put(channel.putter_id, ValueGroup::new_stack(vec![Value::UInt32(42 + loop_idx)])),

        ]).expect("start sync round");

        // Note: if we finish a round, then it must have succeeded :)

        api.wait().expect("finish sync round");

    }

}

mod network_shapes;

mod api_component;

use super::*;

use crate::{PortId, ProtocolDescription};

use crate::common::Id;

use crate::protocol::eval::*;

use crate::runtime2::native::{ApplicationSyncAction};

//

// Generic testing constants, use when appropriate to simplify stress-testing

pub(crate) const NUM_THREADS: u32 = 3;     // number of threads in runtime

pub(crate) const NUM_INSTANCES: u32 = 5;   // number of test instances constructed

pub(crate) const NUM_LOOPS: u32 = 5;       // number of loops within a single test (not used by all tests)

fn create_runtime(pdl: &str) -> Runtime {

    let protocol = ProtocolDescription::parse(pdl.as_bytes()).expect("parse pdl");

    let runtime = Runtime::new(NUM_THREADS, protocol);

    return runtime;

}

fn run_test_in_runtime<F: Fn(&mut ApplicationInterface)>(pdl: &str, constructor: F) {

    let protocol = ProtocolDescription::parse(pdl.as_bytes())

        .expect("parse PDL");

    let runtime = Runtime::new(NUM_THREADS, protocol);

    let mut api = runtime.create_interface();

    for _ in 0..NUM_INSTANCES {

        constructor(&mut api);

    }

}

pub(crate) struct TestTimer {

    name: &'static str,

    started: std::time::Instant

}

impl TestTimer {

    pub(crate) fn new(name: &'static str) -> Self {

        Self{ name, started: std::time::Instant::now() }

    }

}

impl Drop for TestTimer {

    fn drop(&mut self) {

        let delta = std::time::Instant::now() - self.started;

        let nanos = (delta.as_secs_f64() * 1_000_000.0) as u64;

        let millis = nanos / 1000;