use std::collections::VecDeque;

use super::value::*;

use super::store::*;

use super::error::*;

use crate::protocol::*;

use crate::protocol::ast::*;

use crate::protocol::type_table::*;

macro_rules! debug_enabled { () => { false }; }

macro_rules! debug_log {

    ($format:literal) => {

        enabled_debug_print!(false, "exec", $format);

    };

    ($format:literal, $($args:expr),*) => {

        enabled_debug_print!(false, "exec", $format, $($args),*);

    };

}

#[derive(Debug, Clone)]

pub(crate) enum ExprInstruction {

    EvalExpr(ExpressionId),

    PushValToFront,

}

#[derive(Debug, Clone)]

pub(crate) struct Frame {

    pub(crate) definition: ProcedureDefinitionId,

    pub(crate) monomorph_index: usize,

    pub(crate) position: StatementId,

    pub(crate) expr_stack: VecDeque<ExprInstruction>, // hack for expression evaluation, evaluated by popping from back

    pub(crate) expr_values: VecDeque<Value>, // hack for expression results, evaluated by popping from front/back

    pub(crate) max_stack_size: u32,

}

impl Frame {

    /// Creates a new execution frame. Does not modify the stack in any way.

        let definition = &heap[definition_id];

        let outer_scope_id = definition.scope;

        let first_statement_id = definition.body;

        // Another not-so-pretty thing that has to be replaced somewhere in the

        // future...

        fn determine_max_stack_size(heap: &Heap, scope_id: ScopeId, max_size: &mut u32) {

            let scope = &heap[scope_id];

            // Check current block

            let cur_size = scope.next_unique_id_in_scope as u32;

            if cur_size > *max_size { *max_size = cur_size; }

            // And child blocks

            for child_scope in &scope.nested {

                determine_max_stack_size(heap, *child_scope, max_size);

            }

        }

        let mut max_stack_size = 0;

        determine_max_stack_size(heap, outer_scope_id, &mut max_stack_size);

        Frame{

            definition: definition_id,

            monomorph_index: monomorph_index as usize,

            position: first_statement_id.upcast(),

            expr_stack: VecDeque::with_capacity(128),

            expr_values: VecDeque::with_capacity(128),

            max_stack_size,

        }

    }

    /// Prepares a single expression for execution. This involves walking the

    /// expression tree and putting them in the `expr_stack` such that

    /// continuously popping from its back will evaluate the expression. The

    /// results of each expression will be stored by pushing onto `expr_values`.

    pub fn prepare_single_expression(&mut self, heap: &Heap, expr_id: ExpressionId) {

        debug_assert!(self.expr_stack.is_empty());

        self.expr_values.clear(); // May not be empty if last expression result(s) were discarded

        self.serialize_expression(heap, expr_id);

    }

    /// Prepares multiple expressions for execution (i.e. evaluating all

    /// function arguments or all elements of an array/union literal). Per

    /// expression this works the same as `prepare_single_expression`. However

    /// after each expression is evaluated we insert a `PushValToFront`

    /// instruction

    pub fn prepare_multiple_expressions(&mut self, heap: &Heap, expr_ids: &[ExpressionId]) {

        debug_assert!(self.expr_stack.is_empty());

        self.expr_values.clear();

        for expr_id in expr_ids {

            self.expr_stack.push_back(ExprInstruction::PushValToFront);

            self.serialize_expression(heap, *expr_id);

        }

    }

    /// Performs depth-first serialization of expression tree. Let's not care

    /// about performance for a temporary runtime implementation

    fn serialize_expression(&mut self, heap: &Heap, id: ExpressionId) {

        self.expr_stack.push_back(ExprInstruction::EvalExpr(id));

        match &heap[id] {

            Expression::Assignment(expr) => {

                self.serialize_expression(heap, expr.left);

                self.serialize_expression(heap, expr.right);

            },

            Expression::Binding(expr) => {

                self.serialize_expression(heap, expr.bound_to);

                self.serialize_expression(heap, expr.bound_from);

            },

        // Create calls for each select case that will register the ports that

        // we are waiting on at the runtime.

        {

            let mut total_port_index = 0;

            for case_index in 0..total_num_cases {

                let case = &ctx.heap[id].cases[case_index];

                let case_num_ports = case.involved_ports.len();

                for case_port_index in 0..case_num_ports {

                    // Arguments to runtime call

                    let (port_variable_id, port_variable_type) = locals[total_port_index]; // so far this variable contains the temporary variables for the port expressions

                    let case_index_expr_id = create_ast_literal_integer_expr(ctx, self.current_procedure_id, case_index as u64, ctx.arch.uint32_type_id);

                    let port_index_expr_id = create_ast_literal_integer_expr(ctx, self.current_procedure_id, case_port_index as u64, ctx.arch.uint32_type_id);

                    let port_variable_expr_id = create_ast_variable_expr(ctx, self.current_procedure_id, port_variable_id, port_variable_type);

                    let runtime_call_arguments = vec![

                        case_index_expr_id.upcast(),

                        port_index_expr_id.upcast(),

                        port_variable_expr_id.upcast()

                    ];

                    // Create runtime call, then store it

                    let runtime_call_expr_id = create_ast_call_expr(ctx, self.current_procedure_id, Method::SelectRegisterCasePort, &mut self.expression_buffer, runtime_call_arguments);

                    let runtime_call_stmt_id = create_ast_expression_stmt(ctx, runtime_call_expr_id.upcast());

                    transformed_stmts.push(runtime_call_stmt_id.upcast());

                    total_port_index += 1;

                }

            }

        }

        // Create the variable that will hold the result of a completed select

        // block. Then create the runtime call that will produce this result

        let select_variable_id = create_ast_variable(ctx, outer_scope_id);

        let select_variable_type = TypeIdReference::DirectTypeId(ctx.arch.uint32_type_id);

        locals.push((select_variable_id, select_variable_type));

        {

            let runtime_call_expr_id = create_ast_call_expr(ctx, self.current_procedure_id, Method::SelectWait, &mut self.expression_buffer, Vec::new());

            let variable_stmt_id = create_ast_variable_declaration_stmt(ctx, self.current_procedure_id, select_variable_id, select_variable_type, runtime_call_expr_id.upcast());

            transformed_stmts.push(variable_stmt_id.upcast().upcast());

        }

        call_id_section.forget();

        expr_id_section.forget();

        // Now we transform each of the select block case's guard and code into

        // a chained if-else statement.

        if total_num_cases > 0 {

            let (if_stmt_id, end_if_stmt_id, scope_id) = transform_select_case_code(ctx, self.current_procedure_id, id, 0, select_variable_id, select_variable_type);

            link_existing_child_to_new_parent_scope(ctx, &mut self.scope_buffer, outer_scope_id, scope_id, relative_pos);

            let first_end_if_stmt = &mut ctx.heap[end_if_stmt_id];

            first_end_if_stmt.next = outer_end_block_id.upcast();

            let mut last_if_stmt_id = if_stmt_id;

            let mut last_end_if_stmt_id = end_if_stmt_id;

            let mut last_parent_scope_id = outer_scope_id;

            let mut last_relative_pos = transformed_stmts.len() as i32 + 1;

            transformed_stmts.push(last_if_stmt_id.upcast());

            for case_index in 1..total_num_cases {

                let (if_stmt_id, end_if_stmt_id, scope_id) = transform_select_case_code(ctx, self.current_procedure_id, id, case_index, select_variable_id, select_variable_type);

                let false_case_scope_id = ctx.heap.alloc_scope(|this| Scope::new(this, ScopeAssociation::If(last_if_stmt_id, false)));

                link_existing_child_to_new_parent_scope(ctx, &mut self.scope_buffer, false_case_scope_id, scope_id, 0);

                link_new_child_to_existing_parent_scope(ctx, &mut self.scope_buffer, last_parent_scope_id, false_case_scope_id, last_relative_pos);

                set_ast_if_statement_false_body(ctx, last_if_stmt_id, last_end_if_stmt_id, IfStatementCase{ body: if_stmt_id.upcast(), scope: false_case_scope_id });

                let end_if_stmt = &mut ctx.heap[end_if_stmt_id];

                end_if_stmt.next = last_end_if_stmt_id.upcast();

                last_if_stmt_id = if_stmt_id;

                last_end_if_stmt_id = end_if_stmt_id;

                last_parent_scope_id = false_case_scope_id;

                last_relative_pos = 0;

            }

        }

        // Final steps: set the statements of the replacement block statement,

        // link all of those statements together, and update the scopes.

        let first_stmt_id = transformed_stmts[0];

        let mut last_stmt_id = transformed_stmts[0];

        for stmt_id in transformed_stmts.iter().skip(1).copied() {

            set_ast_statement_next(ctx, last_stmt_id, stmt_id);

            last_stmt_id = stmt_id;

        }

        if total_num_cases == 0 {

            // If we don't have any cases, then we didn't connect the statements

            // up to the end of the outer block, so do that here

            set_ast_statement_next(ctx, last_stmt_id, outer_end_block_id.upcast());

        }

        let outer_block_stmt = &mut ctx.heap[outer_block_id];

        outer_block_stmt.next = first_stmt_id;

        outer_block_stmt.statements = transformed_stmts;

            let arg_expr_id = section[arg_expr_idx];

            self.expr_parent = ExpressionParent::Expression(upcast_id, arg_expr_idx as u32);

            self.visit_expr(ctx, arg_expr_id)?;

        }

        section.forget();

        self.expr_parent = old_expr_parent;

        Ok(())

    }

    fn visit_variable_expr(&mut self, ctx: &mut Ctx, id: VariableExpressionId) -> VisitorResult {

        let var_expr = &ctx.heap[id];

        // Check if declaration was already resolved (this occurs for the

        // variable expr that is on the LHS of the assignment expr that is

        // associated with a variable declaration)

        let mut variable_id = var_expr.declaration;

        let mut is_binding_target = false;

        // Otherwise try to find it

        if variable_id.is_none() {

            variable_id = self.find_variable(ctx, self.relative_pos_in_parent, &var_expr.identifier);

        }

        // Otherwise try to see if is a variable introduced by a binding expr

        let variable_id = if let Some(variable_id) = variable_id {

            variable_id

        } else {

            if self.in_binding_expr.is_invalid() || !self.in_binding_expr_lhs {

                return Err(ParseError::new_error_str_at_span(

                    &ctx.module().source, var_expr.identifier.span, "unresolved variable"

                ));

            }

            // This is a binding variable, but it may only appear in very

            // specific locations.

            let is_valid_binding = match self.expr_parent {

                ExpressionParent::Expression(expr_id, idx) => {

                    match &ctx.heap[expr_id] {

                        Expression::Binding(_binding_expr) => {

                            // Nested binding is disallowed, and because of

                            // the check above we know we're directly at the

                            // LHS of the binding expression

                            debug_assert_eq!(_binding_expr.this, self.in_binding_expr);

                            debug_assert_eq!(idx, 0);

                            true

                        }

                            // Only struct, unions, tuples and arrays can

                            // have subexpressions, so we're always fine

                            dbg_code!({

                                    Literal::Struct(_) | Literal::Union(_) | Literal::Array(_) | Literal::Tuple(_) => {},

                                    _ => unreachable!(),

                                }

                            });

                            true

                        },

                        _ => false,

                    }

                },

                _ => {

                    false

                }

            };

            if !is_valid_binding {

                let binding_expr = &ctx.heap[self.in_binding_expr];

                return Err(ParseError::new_error_str_at_span(

                    &ctx.module().source, var_expr.identifier.span,

                    "illegal location for binding variable: binding variables may only be nested under a binding expression, or a struct, union or array literal"

                ).with_info_at_span(

                    &ctx.module().source, binding_expr.operator_span, format!(

                        "'{}' was interpreted as a binding variable because the variable is not declared and it is nested under this binding expression",

                        var_expr.identifier.value.as_str()

                    )

                ));

            }

            // By now we know that this is a valid binding expression. Given

            // that a binding expression must be nested under an if/while

            // statement, we now add the variable to the scope associated with

            // that statement.

            let bound_identifier = var_expr.identifier.clone();

            let bound_variable_id = ctx.heap.alloc_variable(|this| Variable {

                this,

                kind: VariableKind::Binding,

                parser_type: ParserType {

                    elements: vec![ParserTypeElement {

                        element_span: bound_identifier.span,

                        variant: ParserTypeVariant::Inferred

                    }],

                    full_span: bound_identifier.span

                },

                identifier: bound_identifier,

                relative_pos_in_parent: 0,

                unique_id_in_scope: -1,

            });

 * The second one is where we compute byte sizes, alignment and offsets.

 *

 * The basic algorithm for type resolving and computing byte sizes is to

 * recursively try to lay out each member type of a particular type. This is

 * done in a stack-like fashion, where each embedded type pushes a breadcrumb

 * unto the stack. We may discover a cycle in embedded types (we call this a

 * "type loop"). After which the type table attempts to break the type loop by

 * making specific types heap-allocated. Upon doing so we know their size

 * because their stack-size is now based on pointers. Hence breaking the type

 * loop required for computing the byte size of types.

 *

 * The reason for these type shenanigans is because PDL is a value-based

 * language, but we would still like to be able to express recursively defined

 * types like trees or linked lists. Hence we need to insert pointers somewhere

 * to break these cycles.

 *

 * We will insert these pointers into the variants of unions. However note that

 * we can only compute the stack size of a union until we've looked at *all*

 * variants. Hence we perform an initial pass where we detect type loops, a

 * second pass where we compute the stack sizes of everything, and a third pass

 * where we actually compute the size of the heap allocations for unions.

 *

 * As a final bit of global documentation: non-polymorphic types will always

 * have one "monomorph" entry. This contains the non-polymorphic type's memory

 * layout.

 */

// Programmer note: deduplication of types is currently disabled, see the

// @Deduplication key. Tests might fail when it is re-enabled.

use std::collections::HashMap;

use std::hash::{Hash, Hasher};

use crate::protocol::ast::*;

use crate::protocol::parser::symbol_table::SymbolScope;

use crate::protocol::input_source::ParseError;

use crate::protocol::parser::*;

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

// Defined Types

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

/// Struct wrapping around a potentially polymorphic type. If the type does not

/// have any polymorphic arguments then it will not have any monomorphs and

/// `is_polymorph` will be set to `false`. A type with polymorphic arguments

/// only has `is_polymorph` set to `true` if the polymorphic arguments actually

/// appear in the types associated types (function return argument, struct

/// field, enum variant, etc.). Otherwise the polymorphic argument is just a

/// marker and does not influence the bytesize of the type.

pub struct DefinedType {

    pub(crate) ast_root: RootId,

    pub(crate) ast_definition: DefinitionId,

    pub(crate) definition: DefinedTypeVariant,

    pub(crate) poly_vars: Vec<PolymorphicVariable>,

    pub(crate) is_polymorph: bool,

}

pub enum DefinedTypeVariant {

    Enum(EnumType),

    Union(UnionType),

    Struct(StructType),

    Procedure(ProcedureType),

}

impl DefinedTypeVariant {

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

        use DefinedTypeVariant as DTV;

        match self {

            DTV::Struct(_) | DTV::Enum(_) | DTV::Union(_) => return true,

            DTV::Procedure(_) => return false,

        }

    }

    pub(crate) fn as_struct(&self) -> &StructType {

        match self {

            DefinedTypeVariant::Struct(v) => v,

            _ => unreachable!()

        }

    }

    pub(crate) fn as_enum(&self) -> &EnumType {

        match self {

            DefinedTypeVariant::Enum(v) => v,

            _ => unreachable!()

        }

    }

    pub(crate) fn as_union(&self) -> &UnionType {

        match self {

            DefinedTypeVariant::Union(v) => v,

            _ => unreachable!()

        }

    }

}

pub struct PolymorphicVariable {

                );

                parse_error = parse_error.with_info_at_span(module, span, message);

                error_counter += 1;

            }

            parse_error

        }

        for type_loop in &self.type_loops {

            let mut can_be_broken = false;

            debug_assert!(!type_loop.members.is_empty());

            for entry in &type_loop.members {

                if entry.is_union {

                    let mono_type = self.mono_types[entry.type_id.0 as usize].variant.as_union();

                    debug_assert!(!mono_type.variants.is_empty()); // otherwise it couldn't be part of the type loop

                    let has_stack_variant = mono_type.variants.iter().any(|variant| !variant.lives_on_heap);

                    if has_stack_variant {

                        can_be_broken = true;

                        break;

                    }

                }

            }

            if !can_be_broken {

                // Construct a type loop error

                return Err(construct_type_loop_error(&self.mono_types, type_loop, modules, heap));

            }

        }

        // If here, then all type loops have been resolved and we can lay out

        // all of the members

        self.type_loops.clear();

        return Ok(());

    }

    /// Checks if the specified type needs to be resolved (i.e. we need to push

    /// a breadcrumb), is already resolved (i.e. we can continue with the next

    /// member of the currently considered type) or is in the process of being

    /// resolved (i.e. we're in a type loop). Because of borrowing rules we

    /// don't do any modifications of internal types here. Hence: if we

    /// return `PushBreadcrumb` then call `handle_new_breadcrumb_for_type_loops`

    /// to take care of storing the appropriate types.

    fn check_member_for_type_loops(

        breadcrumbs: &[TypeLoopBreadcrumb], definition_map: &DefinitionMap, mono_type_map: &MonoTypeMap,

        mono_key: &mut MonoSearchKey, concrete_type: &ConcreteType

    ) -> TypeLoopResult {

        // Depending on the type, lookup if the type has already been visited

        // (i.e. either already has its memory layed out, or is part of a type

        // loop because we've already visited the type)

        debug_assert!(!concrete_type.parts.is_empty());

        let definition_id = if let ConcreteTypePart::Instance(definition_id, _) = concrete_type.parts[0] {

            definition_id

        } else {

            DefinitionId::new_invalid()

        };

        Self::set_search_key_to_type(mono_key, definition_map, &concrete_type.parts);

        if let Some(type_id) = mono_type_map.get(mono_key).copied() {

            for (breadcrumb_idx, breadcrumb) in breadcrumbs.iter().enumerate() {

                if breadcrumb.type_id == type_id {

                    return TypeLoopResult::TypeLoop(breadcrumb_idx);

                }

            }

            return TypeLoopResult::TypeExists;

        }

        // Type is not yet known, so we need to insert it into the lookup and

        // push a new breadcrumb.

        return TypeLoopResult::PushBreadcrumb(definition_id, concrete_type.clone());

    }

    /// Handles the `PushBreadcrumb` result for a `check_member_for_type_loops`

    /// call. Will preallocate entries in the monomorphed type storage (with

    /// all memory properties zeroed).

    fn handle_new_breadcrumb_for_type_loops(&mut self, arch: &TargetArch, definition_id: DefinitionId, concrete_type: ConcreteType) {

        use DefinedTypeVariant as DTV;

        use ConcreteTypePart as CTP;

        let mut is_union = false;

        let type_id = match &concrete_type.parts[0] {

            // Builtin types

            CTP::Void | CTP::Message | CTP::Bool |

            CTP::UInt8 | CTP::UInt16 | CTP::UInt32 | CTP::UInt64 |

            CTP::SInt8 | CTP::SInt16 | CTP::SInt32 | CTP::SInt64 |

            CTP::Character | CTP::String |

            CTP::Array | CTP::Slice | CTP::Input | CTP::Output | CTP::Pointer => {

                // Insert the entry for the builtin type, we should be able to

                // immediately "steal" the size from the preinserted builtins.

                let base_type_id = match &concrete_type.parts[0] {

                    CTP::Void => arch.void_type_id,

                    CTP::Message => arch.message_type_id,

                    CTP::Bool => arch.bool_type_id,

                    } else {

                        let index = scheduled.ctx.ports

                            .iter()

                            .position(|v| v.self_id == port_change.port.self_id)

                            .unwrap();

                        scheduled.ctx.ports.remove(index);

                    }

                }

            }

        }

        // Finally, check if we just entered or just left a sync region

        if scheduled.ctx.changed_in_sync {

            if scheduled.ctx.is_in_sync {

                // Just entered sync region

            } else {

                // Just left sync region. So prepare inbox for the next sync

                // round

                scheduled.ctx.inbox.clear_read_messages();

            }

            scheduled.ctx.changed_in_sync = false; // reset flag

        }

    }

    fn try_go_to_sleep(&self, connector_key: ConnectorKey, connector: &mut ScheduledConnector) {

        debug_assert_eq!(connector_key.index, connector.ctx.id.index);

        debug_assert_eq!(connector.public.sleeping.load(Ordering::Acquire), false);

        // This is the running connector, and only the running connector may

        // decide it wants to sleep again.

        connector.public.sleeping.store(true, Ordering::Release);

        // But due to reordering we might have received messages from peers who

        // did not consider us sleeping. If so, then we wake ourselves again.

        if !connector.public.inbox.is_empty() {

            // Try to wake ourselves up (needed because someone might be trying

            // the exact same atomic compare-and-swap at this point in time)

            let should_wake_up_again = connector.public.sleeping

                .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)

                .is_ok();

            if should_wake_up_again {

                self.runtime.push_work(connector_key)

            }

        }

    }

        // println!("DEBUG [thrd:{:02} conn:  ]: {}", self.scheduler_id, message);

    }

        // println!("DEBUG [thrd:{:02} conn:{:02}]: {}", self.scheduler_id, conn.index, message);

    }

}

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

// ComponentCtx

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

enum ComponentStateChange {

    CreatedComponent(ConnectorPDL, Vec<PortIdLocal>),

    CreatedPort(Port),

    ChangedPort(ComponentPortChange),

}

#[derive(Clone)]

pub(crate) struct ComponentPortChange {

    pub is_acquired: bool, // otherwise: released

    pub port: Port,

}

/// The component context (better name may be invented). This was created

/// because part of the component's state is managed by the scheduler, and part

/// of it by the component itself. When the component starts a sync block or

/// exits a sync block the partially managed state by both component and

/// scheduler need to be exchanged.

pub(crate) struct ComponentCtx {

    // Mostly managed by the scheduler

    pub(crate) id: ConnectorId,

    ports: Vec<Port>,

    inbox: Inbox,

    // Submitted by the component

    is_in_sync: bool,

    changed_in_sync: bool,

    outbox: VecDeque<Message>,

    state_changes: VecDeque<ComponentStateChange>,

    // Workspaces that may be used by components to (generally) prevent

    // allocations. Be a good scout and leave it empty after you've used it.

    // TODO: Move to scheduler ctx, this is the wrong place

    pub workspace_ports: Vec<PortIdLocal>,

    pub workspace_branches: Vec<BranchId>,

}

impl ComponentCtx {

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

        return Self{

            id: ConnectorId::new_invalid(),

            ports: Vec::new(),

}

pub struct Peer {

    pub id: CompId,

    pub num_associated_ports: u32,

    pub(crate) handle: CompHandle,

}

/// Port and peer management structure. Will keep a local reference counter to

/// the ports associate with peers, additionally manages the atomic reference

/// counter associated with the peers' component handles.

pub struct CompCtx {

    pub id: CompId,

    ports: Vec<Port>,

    peers: Vec<Peer>,

    port_id_counter: u32,

}

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

pub struct LocalPortHandle(PortId);

#[derive(Copy, Clone)]

pub struct LocalPeerHandle(CompId);

impl CompCtx {

    /// Creates a new component context based on a reserved entry in the

    /// component store. This reservation is used such that we already know our

    /// assigned ID.

    pub(crate) fn new(reservation: &CompReserved) -> Self {

        return Self{

            id: reservation.id(),

            ports: Vec::new(),

            peers: Vec::new(),

            port_id_counter: 0,

        }

    }

    /// Creates a new channel that is fully owned by the component associated

    /// with this context.

    pub(crate) fn create_channel(&mut self) -> Channel {

        let putter_id = PortId(self.take_port_id());

        let getter_id = PortId(self.take_port_id());

        self.ports.push(Port{

            self_id: putter_id,

            peer_port_id: getter_id,

            kind: PortKind::Putter,

            state: PortState::Open,

            peer_comp_id: self.id,

        });

        self.ports.push(Port{

            self_id: getter_id,

            peer_port_id: putter_id,

            kind: PortKind::Getter,

            state: PortState::Open,

            peer_comp_id: self.id,

        });

        return Channel{ putter_id, getter_id };

    }

    /// Adds a new port. Make sure to call `add_peer` afterwards.

    pub(crate) fn add_port(&mut self, peer_comp_id: CompId, peer_port_id: PortId, kind: PortKind, state: PortState) -> LocalPortHandle {

        let self_id = PortId(self.take_port_id());

        self.ports.push(Port{

            self_id, peer_comp_id, peer_port_id, kind, state,

            #[cfg(debug_assertions)] associated_with_peer: false,

        });

        return LocalPortHandle(self_id);

    }

    /// Removes a port. Make sure you called `remove_peer` first.

    pub(crate) fn remove_port(&mut self, port_handle: LocalPortHandle) -> Port {

        let port_index = self.must_get_port_index(port_handle);

        let port = self.ports.remove(port_index);

        return port;

    }

    /// Adds a new peer. This must be called for every port, no matter the

    /// component the channel is connected to. If a `CompHandle` is supplied,

    /// then it will be used to add the peer. Otherwise it will be retrieved

    /// from the runtime using its ID.

    pub(crate) fn add_peer(&mut self, port_handle: LocalPortHandle, sched_ctx: &SchedulerCtx, peer_comp_id: CompId, handle: Option<&CompHandle>) {

        let self_id = self.id;

        let port = self.get_port_mut(port_handle);

        debug_assert_eq!(port.peer_comp_id, peer_comp_id);

        if !Self::requires_peer_reference(port, self_id, false) {

            return;

        }

        dbg_code!(port.associated_with_peer = true);

        match self.get_peer_index_by_id(peer_comp_id) {

            Some(peer_index) => {

                let peer = &mut self.peers[peer_index];

                peer.num_associated_ports += 1;

            },

            None => {

                let handle = match handle {

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

                    None => sched_ctx.runtime.get_component_public(peer_comp_id)

                };

                self.peers.push(Peer{

                    id: peer_comp_id,

                    num_associated_ports: 1,

                    handle,

                });

            }

        }

    }

    /// Removes a peer associated with a port.

    pub(crate) fn remove_peer(&mut self, sched_ctx: &SchedulerCtx, port_handle: LocalPortHandle, peer_id: CompId, also_remove_if_closed: bool) {

        let self_id = self.id;

        let port = self.get_port_mut(port_handle);

        debug_assert_eq!(port.peer_comp_id, peer_id);

        if !Self::requires_peer_reference(port, self_id, also_remove_if_closed) {

            return;

        }

        dbg_code!(port.associated_with_peer = false);

        let peer_index = self.get_peer_index_by_id(peer_id).unwrap();

        let peer = &mut self.peers[peer_index];

        peer.num_associated_ports -= 1;

        if peer.num_associated_ports == 0 {

            let mut peer = self.peers.remove(peer_index);

            if let Some(key) = peer.handle.decrement_users() {

                debug_assert_ne!(key.downgrade(), self.id); // should be upheld by the code that shuts down a component

                sched_ctx.runtime.destroy_component(key);

            }

        }

    }

    pub(crate) fn set_port_state(&mut self, port_handle: LocalPortHandle, new_state: PortState) {

        let port_info = self.get_port_mut(port_handle);

        debug_assert_ne!(port_info.state, PortState::Closed); // because then we do not expect to change the state

        port_info.state = new_state;

    }

    pub(crate) fn get_port_handle(&self, port_id: PortId) -> LocalPortHandle {

        return LocalPortHandle(port_id);

    }

    // should perhaps be revised, used in main inbox

    pub(crate) fn get_port_index(&self, port_handle: LocalPortHandle) -> usize {

        return self.must_get_port_index(port_handle);

    }

    pub(crate) fn get_peer_handle(&self, peer_id: CompId) -> LocalPeerHandle {

        return LocalPeerHandle(peer_id);

    }

    pub(crate) fn get_port(&self, port_handle: LocalPortHandle) -> &Port {

        let index = self.must_get_port_index(port_handle);

        return &self.ports[index];

    }

    pub(crate) fn get_port_mut(&mut self, port_handle: LocalPortHandle) -> &mut Port {

        let index = self.must_get_port_index(port_handle);

        return &mut self.ports[index];

    }

    pub(crate) fn get_port_by_index_mut(&mut self, index: usize) -> &mut Port {

        return &mut self.ports[index];

    }

    pub(crate) fn get_peer(&self, peer_handle: LocalPeerHandle) -> &Peer {

        let index = self.must_get_peer_index(peer_handle);

}

/// Should contain everything that is accessible in a thread-safe manner

// TODO: Do something about the `num_handles` thing. This needs to be a bit more

//  "foolproof" to lighten the mental burden of using the `num_handles`

//  variable.

pub(crate) struct CompPublic {

    pub sleeping: AtomicBool,

    pub num_handles: AtomicU32, // manually modified (!)

    inbox: QueueDynProducer<Message>,

}

/// Handle to public part of a component. Would be nice if we could

/// automagically manage the `num_handles` counter. But when it reaches zero we

/// need to manually remove the handle from the runtime. So we just have debug

/// code to make sure this actually happens.

pub(crate) struct CompHandle {

    target: *const CompPublic,

    id: CompId, // TODO: @Remove after debugging

    #[cfg(debug_assertions)] decremented: bool,

}

impl CompHandle {

    fn new(id: CompId, public: &CompPublic) -> CompHandle {

        let handle = CompHandle{

            target: public,

            id,

            #[cfg(debug_assertions)] decremented: false,

        };

        handle.increment_users();

        return handle;

    }

    pub(crate) fn send_message(&self, sched_ctx: &SchedulerCtx, message: Message, try_wake_up: bool) {

        sched_ctx.log(&format!("Sending message to [c:{:03}, wakeup:{}]: {:?}", self.id.0, try_wake_up, message));

        self.inbox.push(message);

        if try_wake_up {

            wake_up_if_sleeping(sched_ctx, self.id, self);

        }

    }

    fn increment_users(&self) {

        let old_count = self.num_handles.fetch_add(1, Ordering::AcqRel);

        debug_assert!(old_count > 0); // because we should never be able to retrieve a handle when the component is (being) destroyed

    }

    /// Returns the `CompKey` to the component if it should be destroyed

    pub(crate) fn decrement_users(&mut self) -> Option<CompKey> {

        let old_count = self.num_handles.fetch_sub(1, Ordering::AcqRel);

        let new_count = old_count - 1;

        dbg_code!(self.decremented = true);

        if new_count == 0 {

            return Some(unsafe{ self.id.upgrade() });