}

    }

}

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 {

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

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

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

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

    }

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

                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 {

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

                    },

                }

            }

        }

    }

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

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

                            }

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;

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

}

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

    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]

        }

    }

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

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

            return scheduling;

        }

    }

}

impl ConnectorPDL {

        Self{

            tree: ExecTree::new(initial),

            consensus: Consensus::new(),

            last_finished_handled: None,

        }

    }

        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;

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

                }

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

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

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

                ));

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

        }

        ctx.notify_sync_end(&[]);

        self.last_finished_handled = None;

    }

}

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

        }

            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

                    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