self.serialize_expression(heap, *arg_expr_id);

                }

            },

            Expression::Variable(_expr) => {

                // No subexpressions

            }

        }

    }

}

type EvalResult = Result<EvalContinuation, EvalError>;

pub enum EvalContinuation {

    Stepping,

    SyncBlockEnd,

    NewFork,

    BlockFires(PortId),

    BlockGet(PortId),

}

// Note: cloning is fine, methinks. cloning all values and the heap regions then

// we end up with valid "pointers" to heap regions.

#[derive(Debug, Clone)]

pub struct Prompt {

    pub(crate) frames: Vec<Frame>,

    pub(crate) store: Store,

}

impl Prompt {

    pub fn new(_types: &TypeTable, heap: &Heap, def: DefinitionId, monomorph_idx: i32, args: ValueGroup) -> Self {

            return EvalError::new_error_at_expr(

                prompt, modules, heap, expr_id,

                format!("index {} is out of bounds: array length is {}", idx, array_len)

            )

        }

        // Checking if we're at the end of execution

        let cur_frame = self.frames.last_mut().unwrap();

        if cur_frame.position.is_invalid() {

            if heap[cur_frame.definition].is_function() {

                todo!("End of function without return, return an evaluation error");

            }

        }

        debug_log!("Taking step in '{}'", heap[cur_frame.definition].identifier().value.as_str());

        // Execute all pending expressions

        while !cur_frame.expr_stack.is_empty() {

            let next = cur_frame.expr_stack.pop_back().unwrap();

            debug_log!("Expr stack: {:?}", next);

            match next {

                ExprInstruction::PushValToFront => {

                    cur_frame.expr_values.rotate_right(1);

                },

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

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

                                    if ctx.performed_put(port_id) {

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

                                        // held an owned value

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

                                    } else {

                                        // Prepare to execute again

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

                                    }

                                },

                                Method::Fires => {

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

                                    let port_value_deref = self.store.maybe_read_ref(&port_value).clone();

                                    let port_id = match port_value_deref {

                                        Value::Input(port_id) => port_id,

                                        Value::Output(port_id) => port_id,

                                        _ => unreachable!("executor calling 'fires' on value {:?}", port_value_deref),

                                    };

                                    };

                                    let len = self.store.heap_regions[heap_pos as usize].values.len();

                                    // TODO: @PtrInt

                                    cur_frame.expr_values.push_back(Value::UInt32(len as u32));

                                    self.store.drop_value(value_heap_pos);

                                },

                                Method::Assert => {

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

                                    let value = self.store.maybe_read_ref(&value).clone();

                                    if !value.as_bool() {

                                    }

                                },

                                Method::Print => {

                                    // Convert the runtime-variant of a string

                                    // into an actual string.

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

                                    let value_heap_pos = value.as_string();

                                    let elements = &self.store.heap_regions[value_heap_pos as usize].values;

                                    let mut message = String::with_capacity(elements.len());

                                    for element in elements {

                                        message.push(element.as_char());

                self.frames.pop();

                // Clean up our section of the stack

                self.store.clear_stack(0);

                self.store.stack.truncate(self.store.cur_stack_boundary + 1);

                let prev_stack_idx = self.store.stack.pop().unwrap().as_stack_boundary();

                // TODO: Temporary hack for testing, remove at some point

                if self.frames.is_empty() {

                    debug_assert!(prev_stack_idx == -1);

                    debug_assert!(self.store.stack.len() == 0);

                    self.store.stack.push(return_value);

                }

                debug_assert!(prev_stack_idx >= 0);

                // Return to original state of stack frame

                self.store.cur_stack_boundary = prev_stack_idx as usize;

                let cur_frame = self.frames.last_mut().unwrap();

                cur_frame.expr_values.push_back(return_value);

                // We just returned to the previous frame, which might be in

                // the middle of evaluating expressions for a particular

                // statement. So we don't want to enter the code below.

                return Ok(EvalContinuation::Stepping);

        let module_root = self.lookup_module_root(module_name).unwrap();

        let root = &self.heap[module_root];

        let def = root.get_definition_ident(&self.heap, identifier).unwrap();

        // TODO: Check for polymorph

        ComponentState { prompt: Prompt::new(&self.types, &self.heap, def, 0, ValueGroup::new_stack(args)) }

    }

    // TODO: Ofcourse, rename this at some point, perhaps even remove it in its

    //  entirety. Find some way to interface with the parameter's types.

    pub(crate) fn new_component_v2(

        &self, module_name: &[u8], identifier: &[u8], arguments: ValueGroup

        // Find the module in which the definition can be found

        let module_root = self.lookup_module_root(module_name);

        if module_root.is_none() {

            return Err(ComponentCreationError::ModuleDoesntExist);

        }

        let module_root = module_root.unwrap();

        let root = &self.heap[module_root];

        let definition_id = root.get_definition_ident(&self.heap, identifier);

        if definition_id.is_none() {

            return Err(ComponentCreationError::DefinitionDoesntExist);

        }

        // - for each argument try to make sure the types match

        for arg_idx in 0..arguments.values.len() {

            let expected_type = &expr_data.arg_types[arg_idx];

            let provided_value = &arguments.values[arg_idx];

            if !self.verify_same_type(expected_type, 0, &arguments, provided_value) {

                return Err(ComponentCreationError::InvalidArgumentType(arg_idx));

            }

        }

        // By now we're sure that all of the arguments are correct. So create

        // the connector.

    }

    fn lookup_module_root(&self, module_name: &[u8]) -> Option<RootId> {

        for module in self.modules.iter() {

            match &module.name {

                Some(name) => if name.as_bytes() == module_name {

                    return Some(module.root_id);

                },

                None => if module_name.is_empty() {

                    return Some(module.root_id);

                }

            }

        loop {

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

            match step_result {

                Err(reason) => {

                    // TODO: @temp

                    println!("Evaluation error:\n{}", reason);

                    todo!("proper error handling/bubbling up");

                },

                Ok(continuation) => match continuation {

                    // TODO: Probably want to remove this translation

                    EC::Stepping => continue,

                    EC::SyncBlockStart => return RR::ComponentAtSyncStart,

                    EC::SyncBlockEnd => return RR::BranchAtSyncEnd,

                    EC::NewComponent(definition_id, monomorph_idx, args) =>

                        return RR::NewComponent(definition_id, monomorph_idx, args),

                    EC::NewChannel =>

                        return RR::NewChannel,

                    EC::NewFork =>

                        return RR::BranchFork,

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

                    EC::BlockGet(port_id) => return RR::BranchGet(port_id),

                        return RR::BranchPut(port_id, value_group);

                    },

                }

            }

        }

    }

}

// TODO: @remove the old stuff

impl ComponentState {

    pub(crate) fn nonsync_run<'a: 'b, 'b>(

        &'a mut self,

        pd: &'a ProtocolDescription,

    ) -> NonsyncBlocker {

        let mut context = EvalContext::Nonsync(context);

        loop {

            let result = self.prompt.step(&pd.types, &pd.heap, &pd.modules, &mut context);

            match result {

                Err(err) => {

                    println!("Evaluation error:\n{}", err);

                    panic!("proper error handling when component fails");

                },

                Ok(cont) => match cont {

                    EvalContinuation::Stepping => continue,

                    EvalContinuation::SyncBlockStart => return NonsyncBlocker::SyncBlockStart,

                    // Not possible to end sync block if never entered one

                    EvalContinuation::SyncBlockEnd => unreachable!(),

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

                        // Look up definition (TODO for now, assume it is a definition)

                        let mut moved_ports = HashSet::new();

                        for arg in args.values.iter() {

                            match arg {

                                Value::Output(port) => {

                                    moved_ports.insert(*port);

                                }

                                Value::Input(port) => {

        pd: &'a ProtocolDescription,

    ) -> SyncBlocker {

        let mut context = EvalContext::Sync(context);

        loop {

            let result = self.prompt.step(&pd.types, &pd.heap, &pd.modules, &mut context);

            match result {

                Err(err) => {

                    println!("Evaluation error:\n{}", err);

                    panic!("proper error handling when component fails");

                },

                Ok(cont) => match cont {

                    EvalContinuation::Stepping => continue,

                    // First need to exit synchronous block before definition may end

                    // No nested synchronous blocks

                    EvalContinuation::SyncBlockStart => unreachable!(),

                    EvalContinuation::SyncBlockEnd => return SyncBlocker::SyncBlockEnd,

                    // Not possible to create component in sync block

                    EvalContinuation::NewComponent(_, _, _) => unreachable!(),

                    EvalContinuation::NewChannel => unreachable!(),

                    EvalContinuation::NewFork => unreachable!(),

                    EvalContinuation::BlockFires(port) => {

                        return SyncBlocker::CouldntCheckFiring(port);

                    },

                    EvalContinuation::BlockGet(port) => {

                        return SyncBlocker::CouldntReadMsg(port);

                    },

                    EvalContinuation::Put(port, message) => {

                        let payload;

                            Value::Null => {

                                return SyncBlocker::Inconsistent;

                            },

                            Value::Message(heap_pos) => {

                                // Create a copy of the payload

                                let mut bytes = Vec::with_capacity(values.len());

                                for value in values {

                                    bytes.push(value.as_uint8());

                                }

                                payload = Payload(Arc::new(bytes));

                            }

                            _ => unreachable!(),

                        }

                        return SyncBlocker::PutMsg(port, payload);

                    }

                },

            }

use std::collections::HashMap;

use std::ops::{Index, IndexMut};

use super::port::PortIdLocal;

// To share some logic between the FakeTree and ExecTree implementation

trait BranchListItem {

    fn get_id(&self) -> BranchId;

    fn set_next_id(&mut self, id: BranchId);

    fn get_next_id(&self) -> BranchId;

}

/// Generic branch ID. A component will always have one branch: the

/// non-speculative branch. This branch has ID 0. Hence in a speculative context

            return replacement;

        }

    }

}

/// 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 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 prepared: PreparedStatement,

}

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

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

            prepared: PreparedStatement::None,

        }

    }

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

/// contains branches that ended up in a particular execution state. The lists

/// are described by the various `BranchQueue` instances and the `next_in_queue`

/// field in each branch.

pub(crate) struct ExecTree {

    // All branches. the `parent_id` field in each branch implies the shape of

    // the tree. Branches are index stable throughout a sync round.

    pub branches: Vec<Branch>,

    queues: [BranchQueue; NUM_QUEUES]

}

impl ExecTree {

    /// Constructs a new execution tree with a single non-sync branch.

        return Self {

            branches: vec![Branch::new_non_sync(component)],

            queues: [BranchQueue::new(); 3]

        }

    }

    // --- Generic branch (queue) management

    /// Returns if tree is in speculative mode

    pub fn is_in_sync(&self) -> bool {

        return self.branches.len() != 1;

    }

//     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::common::ComponentState;

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

        }

    }

}

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

}

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

struct ConnectorRunContext<'a> {

    branch_id: BranchId,

    consensus: &'a Consensus,

                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 {

        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 {

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

        };

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

                // Branch became inconsistent

                branch.sync_state = SpeculativeState::Inconsistent;

            },

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

            },

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

                        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;

                }

            }

                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;

                }

            },

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

            }

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

                comp_ctx.submit_message(Message::Data(DataMessage {

                    sync_header, data_header,

                    content: DataContent::Message(content),

                }));

                branch.prepared = PreparedStatement::PerformedPut;

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

                return ConnectorScheduling::Immediate;

            },

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

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

        let branch = self.tree.base_branch_mut();

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

        let mut run_context = ConnectorRunContext{

            branch_id: branch.id,

            consensus: &self.consensus,

            prepared: branch.prepared.take(),

        };

        match run_result {

                branch.sync_state = SpeculativeState::Finished;

                return ConnectorScheduling::Exit;

            },

                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;

            },

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

                    &sched_ctx.runtime.protocol_description.types,

                    &sched_ctx.runtime.protocol_description.heap,

                    definition_id, monomorph_idx, arguments

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

                comp_ctx.workspace_ports.clear();

                return ConnectorScheduling::Later;

            },

                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;

            },

        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;

    }

}

        for initial_port in &initial_ports {

            if !self.owned_ports.iter().any(|(_, v)| v == initial_port) {

                return Err(ComponentCreationError::UnownedPort);

            }

        }

        // We own all ports, so remove them on this side

        for initial_port in &initial_ports {

            let position = self.owned_ports.iter().position(|(_, v)| v == initial_port).unwrap();

            self.owned_ports.remove(position);

        }

        // Put on job queue

        {

            let mut queue = self.job_queue.lock().unwrap();

            queue.push_back(ApplicationJob::NewConnector(connector, initial_ports));

        }

        self.wake_up_connector_with_ping();

        return Ok(());

    }

            }

            // We have something to do

            let connector_key = connector_key.unwrap();

            let connector_id = connector_key.downcast();

            self.debug_conn(connector_id, &format!(" ... Got work, running {}", connector_key.index));

            let scheduled = self.runtime.get_component_private(&connector_key);

            // Keep running until we should no longer immediately schedule the

            // connector.

            let mut cur_schedule = ConnectorScheduling::Immediate;

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

                            );

                            self.debug_conn(connector_id, &format!("Sending message [ exit ] \n --- {:?}", message));

                            self.runtime.send_message(port.peer_connector, Message::Control(message));

                        }

                    }

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

                        self.runtime.destroy_component(connector_key);

                        continue 'thread_loop;

                    }

                    self.try_go_to_sleep(connector_key, scheduled);

                }

            }

        }

    }

    /// Receiving messages from the public inbox and handling them or storing

    /// them in the component's private inbox

    fn handle_inbox_messages(&mut self, scheduled: &mut ScheduledConnector) {

        let connector_id = scheduled.ctx.id;

        while let Some(message) = scheduled.public.inbox.take_message() {

            // Check if the message has to be rerouted because we have moved the