FuncFires,

    FuncCreate,

    FuncLength,

    FuncAssert,

    FuncPrint,

    // Buitlin functions, not available to user

    FuncSelectStart,

    FuncSelectRegisterCasePort,

    FuncSelectWait,

    // Builtin components, available to user

    CompRandomU32, // TODO: Remove, temporary thing

    CompTcpClient,

    CompTcpListener,

}

impl ProcedureSource {

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

        match self {

            ProcedureSource::FuncUserDefined | ProcedureSource::CompUserDefined => false,

            _ => true,

        }

    }

}

/// Generic storage for functions and components.

// Note that we will have function definitions for builtin functions as well. In

// that case the span, the identifier span and the body are all invalid.

#[derive(Debug)]

pub struct ProcedureDefinition {

    pub this: ProcedureDefinitionId,

    pub defined_in: RootId,

    // Symbol scanning

    pub kind: ProcedureKind,

    pub identifier: Identifier,

    pub poly_vars: Vec<Identifier>,

    // Parser

    pub source: ProcedureSource,

    pub return_type: Option<ParserType>, // present on functions, not components

    pub parameters: Vec<VariableId>,

    pub scope: ScopeId,

    pub body: BlockStatementId,

    // Monomorphization of typed procedures

    pub monomorphs: Vec<ProcedureDefinitionMonomorph>,

}

impl ProcedureDefinition {

    pub(crate) fn new_empty(

        this: ProcedureDefinitionId, defined_in: RootId,

        kind: ProcedureKind, identifier: Identifier, poly_vars: Vec<Identifier>

    ) -> Self {

        Self {

            this, defined_in,

            kind, identifier, poly_vars,

            source: ProcedureSource::FuncUserDefined,

            return_type: None,

            parameters: Vec::new(),

            scope: ScopeId::new_invalid(),

            body: BlockStatementId::new_invalid(),

            monomorphs: Vec::new(),

        }

    }

}

#[derive(Debug, Clone)]

pub enum Statement {

    Block(BlockStatement),

    EndBlock(EndBlockStatement),

    Local(LocalStatement),

    Labeled(LabeledStatement),

    If(IfStatement),

    EndIf(EndIfStatement),

    While(WhileStatement),

    EndWhile(EndWhileStatement),

    Break(BreakStatement),

    Continue(ContinueStatement),

    Synchronous(SynchronousStatement),

    EndSynchronous(EndSynchronousStatement),

    Fork(ForkStatement),

    EndFork(EndForkStatement),

    Select(SelectStatement),

    EndSelect(EndSelectStatement),

    Return(ReturnStatement),

    Goto(GotoStatement),

    New(NewStatement),

    Expression(ExpressionStatement),

}

impl Statement {

    pub fn as_new(&self) -> &NewStatement {

        match self {

            Statement::New(result) => result,

            _ => panic!("Unable to cast `Statement` to `NewStatement`"),

        }

    }

        match self {

            Statement::EndBlock(_)

            | Statement::EndIf(_)

            | Statement::EndWhile(_)

            | Statement::EndSynchronous(_)

            | Statement::EndFork(_)

        }

    }

    pub fn link_next(&mut self, next: StatementId) {

        match self {

            Statement::Block(stmt) => stmt.next = next,

            Statement::EndBlock(stmt) => stmt.next = next,

            Statement::Local(stmt) => match stmt {

                LocalStatement::Channel(stmt) => stmt.next = next,

                LocalStatement::Memory(stmt) => stmt.next = next,

            },

            Statement::EndIf(stmt) => stmt.next = next,

            Statement::EndWhile(stmt) => stmt.next = next,

            Statement::EndSynchronous(stmt) => stmt.next = next,

            Statement::EndFork(stmt) => stmt.next = next,

            Statement::EndSelect(stmt) => stmt.next = next,

            Statement::New(stmt) => stmt.next = next,

            Statement::Expression(stmt) => stmt.next = next,

            Statement::Return(_)

            | Statement::Break(_)

            | Statement::Continue(_)

            | Statement::Synchronous(_)

            | Statement::Fork(_)

            | Statement::Select(_)

            | Statement::Goto(_)

            | Statement::While(_)

            | Statement::Labeled(_)

            | Statement::If(_) => unreachable!(),

        }

    }

}

#[derive(Debug, Clone)]

pub struct BlockStatement {

    pub this: BlockStatementId,

    // Phase 1: parser

    pub span: InputSpan, // of the complete block

    pub statements: Vec<StatementId>,

    pub end_block: EndBlockStatementId,

    // Phase 2: linker

    pub scope: ScopeId,

    pub next: StatementId,

}

#[derive(Debug, Clone)]

pub struct EndBlockStatement {

    pub this: EndBlockStatementId,

    // Parser

    pub start_block: BlockStatementId,

    // Validation/Linking

    pub next: StatementId,

}

#[derive(Debug, Clone)]

pub enum LocalStatement {

    Memory(MemoryStatement),

    Channel(ChannelStatement),

}

impl LocalStatement {

    pub fn this(&self) -> LocalStatementId {

        match self {

            LocalStatement::Memory(stmt) => stmt.this.upcast(),

            LocalStatement::Channel(stmt) => stmt.this.upcast(),

        }

    }

    pub fn span(&self) -> InputSpan {

        match self {

            LocalStatement::Channel(v) => v.span,

            LocalStatement::Memory(v) => v.span,

        }

    }

}

#[derive(Debug, Clone)]

pub struct MemoryStatement {

    pub this: MemoryStatementId,

    // Phase 1: parser

    pub span: InputSpan,

    pub variable: VariableId,

    pub initial_expr: AssignmentExpressionId,

    // Phase 2: linker

    pub next: StatementId,

}

/// ChannelStatement is the declaration of an input and output port associated

/// with the same channel. Note that the polarity of the ports are from the

/// point of view of the component. So an output port is something that a

/// component uses to send data over (i.e. it is the "input end" of the

/// channel), and vice versa.

#[derive(Debug, Clone)]

pub struct ChannelStatement {

    pub this: ChannelStatementId,

    // Phase 1: parser

    pub span: InputSpan, // of the "channel" keyword

    pub from: VariableId, // output

    pub to: VariableId,   // input

    // Phase 2: linker

use std::fmt;

use crate::protocol::{

    ast::*,

    Module,

    input_source::{ErrorStatement, StatementKind}

};

use super::executor::*;

/// Represents a stack frame recorded in an error

#[derive(Debug)]

pub struct EvalFrame {

    pub module_name: String,

    pub procedure: String, // function or component

    pub is_func: bool,

}

impl fmt::Display for EvalFrame {

    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

        let func_or_comp = if self.is_func {

            "function "

        } else {

            "component"

        };

        if self.module_name.is_empty() {

        } else {

        }

    }

}

/// Represents an error that ocurred during evaluation. Contains error

/// statements just like in parsing errors. Additionally may display the current

/// execution state.

#[derive(Debug)]

pub struct EvalError {

    pub(crate) statements: Vec<ErrorStatement>,

    pub(crate) frames: Vec<EvalFrame>,

}

impl EvalError {

    pub(crate) fn new_error_at_expr(prompt: &Prompt, modules: &[Module], heap: &Heap, expr_id: ExpressionId, msg: String) -> EvalError {

        // Create frames

        debug_assert!(!prompt.frames.is_empty());

        let mut frames = Vec::with_capacity(prompt.frames.len());

        let mut last_module_source = &modules[0].source;

        for frame in prompt.frames.iter() {

            let definition = &heap[frame.definition];

            let statement = &heap[frame.position];

            // Lookup module name, if it has one

            let module = modules.iter().find(|m| m.root_id == definition.defined_in).unwrap();

            let module_name = if let Some(name) = &module.name {

                name.as_str().to_string()

            } else {

                String::new()

            };

            last_module_source = &module.source;

            frames.push(EvalFrame{

                module_name,

                procedure: definition.identifier.value.as_str().to_string(),

                is_func: definition.kind == ProcedureKind::Function,

            });

        }

        let expr = &heap[expr_id];

        let statements = vec![

            ErrorStatement::from_source_at_span(StatementKind::Error, last_module_source, expr.full_span(), msg)

        ];

        EvalError{ statements, frames }

    }

}

impl fmt::Display for EvalError {

    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

        // Display error statement(s)

        self.statements[0].fmt(f)?;

        for statement in self.statements.iter().skip(1) {

            writeln!(f)?;

            statement.fmt(f)?;

        }

        // Display stack trace

        writeln!(f)?;

        writeln!(f, " +-  Stack trace:")?;

        for frame in self.frames.iter().rev() {

            write!(f, " | ")?;

            frame.fmt(f)?;

            writeln!(f)?;

        }

        Ok(())

    }

}

    fn update_solution(&mut self) {

        if self.matched_channels == self.solution.channel_mapping.len() {

            if self.solution.decision != SyncRoundDecision::Failure {

                self.solution.decision = SyncRoundDecision::Solution;

            }

        }

    }

}

/// Tracking consensus state

pub struct Consensus {

    // General state of consensus manager

    mapping_counter: u32,

    mode: Mode,

    // State associated with sync round

    round_index: u32,

    highest_id: CompId,

    ports: Vec<PortAnnotation>,

    // State associated with arriving at a solution and being a (temporary)

    // leader in the consensus round

    solution: SolutionCombiner,

}

impl Consensus {

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

        return Self{

            round_index: 0,

            highest_id: CompId::new_invalid(),

            ports: Vec::new(),

            mapping_counter: 0,

            mode: Mode::NonSync,

            solution: SolutionCombiner::new(),

        }

    }

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

    // Managing sync state

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

    /// Notifies the consensus management that the PDL code has reached the

    /// start of a sync block.

    pub(crate) fn notify_sync_start(&mut self, comp_ctx: &CompCtx) {

        debug_assert_eq!(self.mode, Mode::NonSync);

        self.highest_id = comp_ctx.id;

        self.mapping_counter = 0;

        self.mode = Mode::SyncBusy;

        // Make the internally stored port annotation array consistent with the

        // ports that the component currently owns. They should match by index

        // (i.e. annotation at index `i` corresponds to port `i` in `comp_ctx`).

        let mut needs_setting_ports = false;

        if comp_ctx.num_ports() != self.ports.len() {

            needs_setting_ports = true;

        } else {

            for (idx, port) in comp_ctx.iter_ports().enumerate() {

                let comp_port_id = port.self_id;

                let cons_port_id = self.ports[idx].self_port_id;

                if comp_port_id != cons_port_id {

                    needs_setting_ports = true;

                    break;

                }

            }

        }

        if needs_setting_ports {

            // Reset all ports

            self.ports.clear();

            self.ports.reserve(comp_ctx.num_ports());

            for port in comp_ctx.iter_ports() {

                self.ports.push(PortAnnotation::new(comp_ctx.id, port.self_id, port.kind));

            }

        } else {

            // Make sure that we consider all peers as undiscovered again

            for annotation in self.ports.iter_mut() {

                annotation.peer_discovered = false;

            }

        }

    }

    /// Notifies the consensus management that the PDL code has reached the end

    /// of a sync block. A local solution will be submitted, after which we wait

    /// until the participants in the round (hopefully) reach a conclusion.

    pub(crate) fn notify_sync_end_success(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx) -> SyncRoundDecision {

        debug_assert_eq!(self.mode, Mode::SyncBusy);

        self.mode = Mode::SyncAwaitingSolution;

        let local_solution = self.generate_local_solution(comp_ctx, false);

        let decision = self.handle_local_solution(sched_ctx, comp_ctx, comp_ctx.id, local_solution, false);

        return decision;

    }

    /// Notifies the consensus management that the component has encountered a

    /// critical error during the synchronous round. Hence we should report that

    /// we've failed and wait until all the participants have been notified of

    /// the error.

    pub(crate) fn notify_sync_end_failure(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx) -> SyncRoundDecision {

        self.mode = Mode::SyncAwaitingSolution;

        let local_solution = self.generate_local_solution(comp_ctx, true);

        let decision = self.handle_local_solution(sched_ctx, comp_ctx, comp_ctx.id, local_solution, true);

        return decision;

    }

    /// Notifies that a decision has been reached. Note that the caller should

    /// still take the appropriate actions based on the decision it is supplying

    /// to the consensus layer.

    pub(crate) fn notify_sync_decision(&mut self, _decision: SyncRoundDecision) {

        // Reset everything for the next round

        debug_assert_eq!(self.mode, Mode::SyncAwaitingSolution);

        self.mode = Mode::NonSync;

        self.round_index = self.round_index.wrapping_add(1);

        for port in self.ports.iter_mut() {

            port.mapping = None;

        }

        self.solution.clear();

    }

    pub(crate) fn notify_of_new_port(&mut self, _expected_index: usize, port_handle: LocalPortHandle, comp_ctx: &CompCtx) {

        debug_assert_eq!(_expected_index, self.ports.len());

        let port_info = comp_ctx.get_port(port_handle);

        self.ports.push(PortAnnotation{

            self_comp_id: comp_ctx.id,

            self_port_id: port_info.self_id,

            peer_comp_id: port_info.peer_comp_id,

            peer_port_id: port_info.peer_port_id,

            peer_discovered: false,

            mapping: None,

            kind: port_info.kind,

        });

    }

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

    // Handling inbound and outbound messages

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

    /// Prepares a set of values to be sent of a channel.

    pub(crate) fn annotate_data_message(&mut self, comp_ctx: &CompCtx, port_info: &Port, content: ValueGroup) -> DataMessage {

        debug_assert_eq!(self.mode, Mode::SyncBusy); // can only send between sync start and sync end

        debug_assert!(self.ports.iter().any(|v| v.self_port_id == port_info.self_id));

        let data_header = self.create_data_header_and_update_mapping(port_info);

        let sync_header = self.create_sync_header(comp_ctx);

        return DataMessage{

            data_header, sync_header, content,

            ports: Vec::new()

        };

    }

    /// Handles the arrival of a new data message (needs to be called for every

    /// new data message, even though it might not end up being received). This

    /// is used to determine peers of `get`ter ports.

    // TODO: The use of this function is rather ugly. Find a more robust

    //  scheme about owners of `get`ter ports not knowing about their peers.

    pub(crate) fn handle_incoming_data_message(&mut self, comp_ctx: &CompCtx, message: &DataMessage) {

        let target_handle = comp_ctx.get_port_handle(message.data_header.target_port);

        let target_index = comp_ctx.get_port_index(target_handle);

        let annotation = &mut self.ports[target_index];

        debug_assert!(

            !annotation.peer_discovered || (

                annotation.peer_comp_id == message.sync_header.sending_id &&

                annotation.peer_port_id == message.data_header.source_port

            )

        );

        annotation.peer_comp_id = message.sync_header.sending_id;

        annotation.peer_port_id = message.data_header.source_port;

        annotation.peer_discovered = true;

    }

    /// Checks if the data message can be received (due to port annotations), if

    /// it can then `true` is returned and the caller is responsible for handing

    /// the message of to the PDL code. Otherwise the message cannot be

    /// received.

    pub(crate) fn try_receive_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: &DataMessage) -> bool {

        debug_assert_eq!(self.mode, Mode::SyncBusy);

        debug_assert!(self.ports.iter().any(|v| v.self_port_id == message.data_header.target_port));

        // Make sure the expected mapping matches the currently stored mapping

        for (peer_port_kind, expected_annotation) in &message.data_header.expected_mapping {

            // Determine our annotation, in order to do so we need to find the

            // port matching the peer ports

            let mut self_annotation = None;

            let mut self_annotation_found = false;

            match peer_port_kind {

                PortAnnotationKind::Putter(peer_port) => {

                    for self_port in &self.ports {

                        if self_port.kind == PortKind::Getter &&

                            self_port.peer_discovered &&

                            self_port.peer_comp_id == peer_port.self_comp_id &&

                            self_port.peer_port_id == peer_port.self_port_id

                        {

use crate::protocol::*;

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

use crate::runtime2::runtime::*;

use crate::runtime2::component::{CompCtx, CompPDL};

mod messaging;

mod error_handling;

mod transfer_ports;

mod internet;

const LOG_LEVEL: LogLevel = LogLevel::Debug;

const NUM_THREADS: u32 = 1;

pub(crate) fn compile_and_create_component(source: &str, routine_name: &str, args: ValueGroup) {

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

        .expect("successful compilation");

        .expect("successful runtime startup");

    create_component(&runtime, "", routine_name, args);

}

pub(crate) fn create_component(rt: &Runtime, module_name: &str, routine_name: &str, args: ValueGroup) {

    let prompt = rt.inner.protocol.new_component(

        module_name.as_bytes(), routine_name.as_bytes(), args

    ).expect("create prompt");

    let reserved = rt.inner.start_create_component();

    let ctx = CompCtx::new(&reserved);

    let component = Box::new(CompPDL::new(prompt, 0));

    let (key, _) = rt.inner.finish_create_component(reserved, component, ctx, false);

    rt.inner.enqueue_work(key);

}

pub(crate) fn no_args() -> ValueGroup { ValueGroup::new_stack(Vec::new()) }

#[test]

fn test_component_creation() {

    let pd = ProtocolDescription::parse(b"

    comp nothing_at_all() {

        s32 a = 5;

        auto b = 5 + a;

    }

    ").expect("compilation");

    let rt = Runtime::new(1, LOG_LEVEL, pd).unwrap();

    for _i in 0..20 {

        create_component(&rt, "", "nothing_at_all", no_args());

    }

}

#[test]

fn test_simple_select() {

    let pd = ProtocolDescription::parse(b"

    func infinite_assert<T>(T val, T expected) -> () {

        while (val != expected) { print(\"nope!\"); }

        return ();

    }

    comp receiver(in<u32> in_a, in<u32> in_b, u32 num_sends) {

        auto num_from_a = 0;

        auto num_from_b = 0;

        while (num_from_a + num_from_b < 2 * num_sends) {

            sync select {

                auto v = get(in_a) -> {

                    print(\"got something from A\");

                    auto _ = infinite_assert(v, num_from_a);

                    num_from_a += 1;

                }

                auto v = get(in_b) -> {

                    print(\"got something from B\");

                    auto _ = infinite_assert(v, num_from_b);

                    num_from_b += 1;

                }

            }

        }

    }

    comp sender(out<u32> tx, u32 num_sends) {

        auto index = 0;

        while (index < num_sends) {

            sync {

                put(tx, index);

                index += 1;

            }

        }

    }

    comp constructor() {

        auto num_sends = 1;

        channel tx_a -> rx_a;

        channel tx_b -> rx_b;

        new sender(tx_a, num_sends);

        new receiver(rx_a, rx_b, num_sends);

        new sender(tx_b, num_sends);

    }

    ").expect("compilation");

    let rt = Runtime::new(3, LOG_LEVEL, pd).unwrap();

    create_component(&rt, "", "constructor", no_args());

}

#[test]

fn test_unguarded_select() {

    let pd = ProtocolDescription::parse(b"

    comp constructor_outside_select() {

        u32 index = 0;

        while (index < 5) {

            sync select { auto v = () -> print(\"hello\"); }

            index += 1;

        }

    }

    comp constructor_inside_select() {

        u32 index = 0;

        while (index < 5) {