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

                    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;

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

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

        }

    }

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

    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

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

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

        }

        return None;

    }

}

impl Clone for CompHandle {

    fn clone(&self) -> Self {

        self.increment_users();

        return CompHandle{

            target: self.target,

            id: self.id,

            #[cfg(debug_assertions)] decremented: false,

        };

    }

}

impl std::ops::Deref for CompHandle {

    type Target = CompPublic;

    fn deref(&self) -> &Self::Target {

        return unsafe{ &*self.target };

    }

}

impl Drop for CompHandle {

    fn drop(&mut self) {

    }

}

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

// Runtime

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

pub struct Runtime {

    pub(crate) inner: Arc<RuntimeInner>,

    threads: Vec<std::thread::JoinHandle<()>>,

}

impl Runtime {

    // TODO: debug_logging should be removed at some point

    pub fn new(num_threads: u32, debug_logging: bool, protocol_description: ProtocolDescription) -> Runtime {

        assert!(num_threads > 0, "need a thread to perform work");

        let runtime_inner = Arc::new(RuntimeInner {

            protocol: protocol_description,

            components: ComponentStore::new(128),

            work_queue: Mutex::new(VecDeque::with_capacity(128)),

            work_condvar: Condvar::new(),

            active_elements: AtomicU32::new(1),

        });

        let mut runtime = Runtime {

}

pub(crate) struct SchedulerCtx<'a> {

    pub runtime: &'a RuntimeInner,

    pub id: u32,

    pub comp: u32,

    pub logging_enabled: bool,

}

impl<'a> SchedulerCtx<'a> {

    pub fn new(runtime: &'a RuntimeInner, id: u32, logging_enabled: bool) -> Self {

        return Self {

            runtime,

            id,

            comp: 0,

            logging_enabled,

        }

    }

    pub(crate) fn log(&self, text: &str) {

        if self.logging_enabled {

            println!("[s:{:02}, c:{:03}] {}", self.id, self.comp, text);

        }

    }

}

impl Scheduler {

    // public interface to thread

    pub fn new(runtime: Arc<RuntimeInner>, scheduler_id: u32, debug_logging: bool) -> Self {

        return Scheduler{ runtime, scheduler_id, debug_logging }

    }

    pub fn run(&mut self) {

        let mut scheduler_ctx = SchedulerCtx::new(&*self.runtime, self.scheduler_id, self.debug_logging);

        'run_loop: loop {

            // Wait until we have something to do (or need to quit)

            let comp_key = self.runtime.take_work();

            if comp_key.is_none() {

                break 'run_loop;

            }

            let comp_key = comp_key.unwrap();

            let component = self.runtime.get_component(comp_key);

            scheduler_ctx.comp = comp_key.0;

            // Run the component until it no longer indicates that it needs to

    index_mask: usize,

}

type InnerShared<'a, T> = UnfairSeLockSharedGuard<'a, Inner<T>>;

/// Reservation of a slot in the component store. Corresponds to the case where

/// an index has been taken from the freelist, but the element has not yet been

/// initialized

pub struct ComponentReservation {

    pub(crate) index: u32,

    #[cfg(debug_assertions)] submitted: bool,

}

impl ComponentReservation {

    fn new(index: u32) -> Self {

        return Self{

            index,

            #[cfg(debug_assertions)] submitted: false,

        }

    }

}

impl Drop for ComponentReservation {

    fn drop(&mut self) {

    }

}

impl<T: Sized> ComponentStore<T> {

    pub fn new(initial_size: usize) -> Self {

        Self::assert_valid_size(initial_size);

        // Fill initial freelist and preallocate data array

        let mut initial_freelist = Vec::with_capacity(initial_size);

        for idx in 0..initial_size {

            initial_freelist.push(idx as u32)

        }

        let mut initial_data = Vec::new();

        initial_data.resize(initial_size, ptr::null_mut());

        // Return initial store

        return Self{

            inner: UnfairSeLock::new(Inner{

                freelist: initial_freelist,

                data: initial_data,

                size: initial_size,

                compare_mask: 2*initial_size - 1,

                index_mask: initial_size - 1,

    pub fn get_mut(&self, index: u32) -> &mut T {

        let lock = self.inner.lock_shared();

        let value = lock.data[index as usize];

        unsafe {

            debug_assert!(!value.is_null());

            return &mut *value;

        }

    }

    #[inline]

    fn pop_freelist_index<'a>(&'a self, mut shared_lock: InnerShared<'a, T>) -> (InnerShared<'a, T>, u32) {

        'attempt_read: loop {

            // Load indices and check for reallocation condition

            let current_size = shared_lock.size;

            let mut read_index = self.read_head.load(Ordering::Relaxed);

            let limit_index = self.limit_head.load(Ordering::Acquire);

            if read_index == limit_index {

                shared_lock = self.reallocate(current_size, shared_lock);

                continue 'attempt_read;

            }

            loop {

                let preemptive_read = shared_lock.freelist[read_index & shared_lock.index_mask];

                    read_index, (read_index + 1) & shared_lock.compare_mask,

                    Ordering::AcqRel, Ordering::Acquire

                ) {

                    // We need to try again

                    continue 'attempt_read;

                }

                // If here then we performed the read

                return (shared_lock, preemptive_read);

            }

        }

    }

    #[inline]

    fn initialize_at_index(read_lock: InnerShared<T>, index: u32, value: T) {

        let mut target_ptr = read_lock.data[index as usize];

        unsafe {

            if target_ptr.is_null() {

                let layout = Layout::for_value(&value);

                target_ptr = std::alloc::alloc(layout).cast();

                let rewrite: *mut *mut T = transmute(read_lock.data.as_ptr());

                *rewrite.add(index as usize) = target_ptr;

            }

            std::ptr::write(target_ptr, value);

        }

    }