unreachable!("executor calling 'get' on value {:?}", value)

                                    };

                                        Some(result) => {

                                            // We have the result. Merge the `ValueGroup` with the

                                            // stack/heap storage.

                                    let msg_value = cur_frame.expr_values.pop_front().unwrap();

                                    let deref_msg_value = self.store.maybe_read_ref(&msg_value).clone();

                                        // We're fine, deallocate in case the expression value stack

                                        // held an owned value

                                        self.store.drop_value(msg_value.get_heap_pos());

                                    } else {

                                        cur_frame.expr_values.push_front(msg_value);

                                        cur_frame.expr_values.push_front(port_value);

                                        cur_frame.expr_stack.push_back(ExprInstruction::EvalExpr(expr_id));

                    LocalStatement::Channel(stmt) => {

                        // Need to create a new channel by requesting it from

                        // the runtime.

                            None => {

                                // No channel is pending. So request one

                                Ok(EvalContinuation::NewChannel)

                    cur_frame.position = stmt.left_body.upcast();

                } else {

                    // Need to fork

                        // Runtime has created a fork

                        if go_left {

                            cur_frame.position = stmt.left_body.upcast();

// TODO: @temp Should just become a concrete thing that is passed in

pub trait RunContext {

    fn fires(&mut self, port: PortId) -> Option<Value>; // None if not yet branched

}

#[derive(Debug)]

    // Can only occur inside sync blocks

    BranchInconsistent, // branch has inconsistent behaviour

    BranchMissingPortState(PortId), // branch doesn't know about port firing

    BranchAtSyncEnd,

    BranchFork,

    BranchPut(PortId, ValueGroup),

                    EC::NewFork =>

                        return RR::BranchFork,

                    EC::BlockFires(port_id) => return RR::BranchMissingPortState(port_id),

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

                        let value_group = ValueGroup::from_store(&self.prompt.store, &[value]);

                        return RR::BranchPut(port_id, value_group);

}

impl RunContext for EvalContext<'_> {

        match self {

            EvalContext::None => unreachable!(),

            EvalContext::Nonsync(_) => unreachable!(),

        }

    }

        match self {

            EvalContext::None => unreachable!(),

            EvalContext::Nonsync(_) => unreachable!(),

        }

    }

        match self {

            EvalContext::None => unreachable!(),

            EvalContext::Nonsync(context) => {

        }

    }

        // Never actually used in the old runtime

        return None;

    }

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

}

impl BranchListItem for Branch {

            sync_state: SpeculativeState::RunningNonSync,

            awaiting_port: PortIdLocal::new_invalid(),

            next_in_queue: BranchId::new_invalid(),

        }

    }

        //     (parent_branch.sync_state == SpeculativeState::RunningNonSync && !parent_branch.parent_id.is_valid()) ||

        //     (parent_branch.sync_state == SpeculativeState::HaltedAtBranchPoint)

        // ); // forking from non-sync, or forking from a branching point

        Branch {

            id: BranchId::new(new_index),

            sync_state: SpeculativeState::RunningInSync,

            awaiting_port: parent_branch.awaiting_port,

            next_in_queue: BranchId::new_invalid(),

        }

    }

}

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

        branch.sync_state = SpeculativeState::RunningNonSync;

        debug_assert!(!branch.awaiting_port.is_valid());

        branch.next_in_queue = BranchId::new_invalid();

        // Clear out all the queues

        for queue_idx in 0..NUM_QUEUES {

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

    last_finished_handled: Option<BranchId>,

}

struct ConnectorRunContext<'a> {

    branch_id: BranchId,

    consensus: &'a Consensus,

}

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

    }

    }

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

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

    }

    }

    }

}

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

            // And prepare the branch for running

        let mut run_context = ConnectorRunContext{

            branch_id,

            consensus: &self.consensus,

        };

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

                return ConnectorScheduling::Immediate;

            },

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

                    }

                }

            }

            RunResult::BranchAtSyncEnd => {

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

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

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

            }

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

                // Branch is attempting to send data

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

            },

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

        }

        let mut run_context = ConnectorRunContext{

            branch_id: branch.id,

            consensus: &self.consensus,

        };

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

            RunResult::NewChannel => {

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

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

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

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

                ));

use crate::collections::VecSet;

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

use super::ConnectorId;

use super::branch::BranchId;

struct BranchAnnotation {

    port_mapping: Vec<PortAnnotation>,

}

#[derive(Debug)]

pub(crate) struct LocalSolution {

    component: ConnectorId,

    final_branch_id: BranchId,

}

#[derive(Debug, Clone)]

pub(crate) struct GlobalSolution {

    component_branches: Vec<(ConnectorId, BranchId)>,

}

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

    // --- State that is cleared after each round

    // Local component's state

    highest_connector_id: ConnectorId,

    // Gathered state from communication

    encountered_ports: VecSet<PortIdLocal>, // to determine if we should send "port remains silent" messages.

    solution_combiner: SolutionCombiner,

        return Self {

            highest_connector_id: ConnectorId::new_invalid(),

            branch_annotations: Vec::new(),

            encountered_ports: VecSet::new(),

            solution_combiner: SolutionCombiner::new(),

            peers: Vec::new(),

                    expected_firing: None,

                })

                .collect(),

        });

        self.highest_connector_id = ctx.id;

        // 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_branch_annotations = BranchAnnotation{

            port_mapping: parent_branch_annotations.port_mapping.clone(),

        };

        self.branch_annotations.push(new_branch_annotations);

    }

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

                        expected_mapping: source_mapping.clone(),

                        sending_port: port.port_id,

                        target_port: peer_port_id,

                    },

                    content: DataContent::SilentPortNotification,

                }));

            target_mapping.push((

                channel_id,

            ));

        }

        // Construct data header

        // TODO: Handle multiple firings. Right now we just assign the current

        //  branch to the `None` value because we know we can only send once.

        let port_info = ctx.get_port_by_id(source_port_id).unwrap();

        let data_header = DataHeader{

            expected_mapping: branch.port_mapping.clone(),

            sending_port: port_info.self_id,

            target_port: port_info.peer_id,

        };

        // Update port mapping

        for mapping in &mut branch.port_mapping {

            if mapping.port_id == source_port_id {

                mapping.expected_firing = Some(true);

            }

        }

        self.encountered_ports.push(source_port_id);

        return (self.create_sync_header(ctx), data_header);

#[derive(Debug)]

struct MatchedLocalSolution {

    final_branch_id: BranchId,

    matches: Vec<ComponentMatches>,

}

#[derive(Debug, Copy, Clone)]

pub(crate) struct PortAnnotation {

    pub port_id: PortIdLocal,

    pub expected_firing: Option<bool>,

}

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

#[derive(Debug, Clone)]

pub(crate) struct SyncHeader {

    pub expected_mapping: Vec<PortAnnotation>,

    pub sending_port: PortIdLocal,

    pub target_port: PortIdLocal,

}

// TODO: Very much on the fence about this. On one hand I thought making it a

            match &cur_instruction {

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

                    let port_id = *port_id;

                },

                ApplicationSyncAction::Get(port_id) => {

                    let port_id = *port_id;

                        }

                    }

                }

            }

    // TODO: Remove, this is debugging stuff

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

    }

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

    }

}

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

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

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

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

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