// - 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::{PortId, ProtocolDescription};

use crate::common::ComponentState;

use crate::protocol::eval::{EvalContinuation, EvalError, Prompt, Value, ValueGroup};

use crate::protocol::{RunContext, RunResult};

use super::branch::{BranchId, ExecTree, QueueKind, SpeculativeState, PreparedStatement};

use super::consensus::{Consensus, Consistency, RoundConclusion, find_ports_in_value_group};

use super::native::Connector;

use super::port::{PortKind, PortIdLocal};

use super::scheduler::{ComponentCtx, SchedulerCtx, MessageTicket};

pub(crate) struct ConnectorPublic {

    pub inbox: PublicInbox,

    pub sleeping: AtomicBool,

}

impl ConnectorPublic {

    pub fn new(initialize_as_sleeping: bool) -> Self {

        ConnectorPublic{

    /// the provided solution branch ID as the branch that should be comitted to

    /// memory. If this function returns false, then the component is supposed

    /// to exit.

    fn enter_non_sync_mode(&mut self, conclusion: RoundConclusion, ctx: &mut ComponentCtx) -> ConnectorScheduling {

        debug_assert!(self.mode == Mode::Sync || self.mode == Mode::SyncError);

        // Depending on local state decide what to do

        let final_branch_id = match conclusion {

            RoundConclusion::Success(branch_id) => Some(branch_id),

            RoundConclusion::Failure => None,

        };

        if let Some(solution_branch_id) = final_branch_id {

            let mut fake_vec = Vec::new();

            self.tree.end_sync(solution_branch_id);

            self.consensus.end_sync(solution_branch_id, &mut fake_vec);

            debug_assert!(fake_vec.is_empty());

            ctx.notify_sync_end(&[]);

            self.last_finished_handled = None;

            self.eval_error = None; // in case we came from the SyncError mode

            self.mode = Mode::NonSync;

            return ConnectorScheduling::Immediate;

        } else {

            // No final branch, because we're supposed to exit!

            self.last_finished_handled = None;

            self.mode = Mode::Error;

            if let Some(eval_error) = self.eval_error.take() {

                ctx.push_error(eval_error);

            }

            return ConnectorScheduling::Exit;

        }

    }

    /// Runs the prompt repeatedly until some kind of execution-blocking

    /// condition appears.

    #[inline]

    fn run_prompt(prompt: &mut Prompt, pd: &ProtocolDescription, ctx: &mut ConnectorRunContext) -> Result<EvalContinuation, EvalError> {

        loop {

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

            if let Ok(EvalContinuation::Stepping) = result {

                continue;

            Message::SyncControl(_) => return None,

            Message::Control(message) => {

                match &message.content {

                    ControlContent::PortPeerChanged(port_id, _) => return Some(*port_id),

                    ControlContent::CloseChannel(port_id) => return Some(*port_id),

                    ControlContent::Ping => return None,

                    ControlContent::Ack => return None,

                }

            }

        }

    }

    pub(crate) fn as_data(&self) -> &DataMessage {

        match self {

            Message::Data(v) => v,

            _ => unreachable!(),

        }

    }

}

/// The public inbox of a connector. The thread running the connector that owns

/// this inbox may retrieved from it. Non-owning threads may only put new

/// messages inside of it.

// TODO: @Optimize, lazy concurrency. Probably ringbuffer with read/write heads.

impl ConnectorStore {

    fn with_capacity(capacity: usize) -> Self {

        Self {

            entries: RawVec::with_capacity(capacity),

            free: Vec::with_capacity(capacity),

        }

    }

    /// Retrieves public part of connector - accessible by many threads at once.

    fn get_public(&self, id: ConnectorId) -> &'static ConnectorPublic {

        unsafe {

            let entry = self.entries.get(id.index as usize);

            debug_assert!(!entry.is_null());

            return &(**entry).connector.public;

        }

    }

    /// Retrieves private part of connector - accessible by one thread at a

    /// time.

    fn get_private(&self, key: &ConnectorKey) -> &'static mut ScheduledConnector {

        unsafe {

            let entry = self.entries.get_mut(key.index as usize);

            debug_assert!(!entry.is_null());

            return &mut (**entry).connector;

        }

    }

    /// Creates a new connector. Caller should ensure ports are set up correctly

    /// and the connector is queued for execution if needed.

    fn create(&mut self, connector: ConnectorVariant, initially_sleeping: bool) -> ConnectorKey {

        let mut connector = ScheduledConnector {

            connector,

            ctx: ComponentCtx::new_empty(),

            public: ConnectorPublic::new(initially_sleeping),

            router: ControlMessageHandler::new(),

                    self.try_go_to_sleep(connector_key, scheduled);

                },

                ConnectorScheduling::Exit => {

                    // Prepare for exit. Set the shutdown flag and broadcast

                    // messages to notify peers of closing channels

                    scheduled.shutting_down = true;

                    for port in &scheduled.ctx.ports {

                        if port.state != PortState::Closed {

                            let message = scheduled.router.prepare_closing_channel(

                                port.self_id, port.peer_id,

                                connector_id

                            );

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

                        }

                    }

                    // Any messages still in the public inbox should be handled

                    scheduled.ctx.inbox.clear_read_messages();

                    while let Some(ticket) = scheduled.ctx.get_next_message_ticket_even_if_not_in_sync() {

                        let message = scheduled.ctx.take_message_using_ticket(ticket);

                        self.handle_message_while_shutting_down(message, scheduled);

                    }

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

        let connector_id = scheduled.ctx.id;

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

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

            // target port to another component.

            self.debug_conn(connector_id, &format!("Handling message\n --- {:#?}", message));

            if let Some(target_port) = message.target_port() {

                if let Some(other_component_id) = scheduled.router.should_reroute(target_port) {

                    self.debug_conn(connector_id, " ... Rerouting the message");

                    self.runtime.send_message(other_component_id, message);

                    continue;

                }

            }

            // If here, then we should handle the message

            self.debug_conn(connector_id, " ... Handling the message");

            if let Message::Control(message) = &message {

                match message.content {

                    ControlContent::PortPeerChanged(port_id, new_target_connector_id) => {

                        // Need to change port target

                        let port = scheduled.ctx.get_port_mut_by_id(port_id).unwrap();

                        port.peer_connector = new_target_connector_id;

                        // Note: for simplicity we program the scheduler to always finish

                        // running a connector with an empty outbox. If this ever changes

                        // then accepting the "port peer changed" message implies we need

                        // to change the recipient of the message in the outbox.

                        debug_assert!(scheduled.ctx.outbox.is_empty());

                        // And respond with an Ack

                        let ack_message = Message::Control(ControlMessage {

                            id: message.id,

                            sending_component_id: connector_id,

                            content: ControlContent::Ack,

                        });

                        self.runtime.send_message(message.sending_component_id, ack_message);

                    },

                    ControlContent::CloseChannel(port_id) => {

                        // Mark the port as being closed

                        let port = scheduled.ctx.get_port_mut_by_id(port_id).unwrap();

                        port.state = PortState::Closed;

                        // Send an Ack

                        let ack_message = Message::Control(ControlMessage {

                            id: message.id,

                            sending_component_id: connector_id,

                            content: ControlContent::Ack,

                        });

                        self.runtime.send_message(message.sending_component_id, ack_message);

                    },

                    ControlContent::Ack => {

                        if let Some((target_component, new_control_message)) = scheduled.router.handle_ack(connector_id, message.id) {

                            self.runtime.send_message(target_component, new_control_message);

                        };

                    },

                    ControlContent::Ping => {},

                }

            } else {

                // Not a control message

                if scheduled.shutting_down {

                    // Since we're shutting down, we just want to respond with a

                    // message saying the message did not arrive.

                    debug_assert!(scheduled.ctx.inbox.get_next_message_ticket().is_none()); // public inbox should be completely cleared

                    self.handle_message_while_shutting_down(message, scheduled);

            Message::SyncComp(_) => None,

            Message::SyncPort(msg) => Some((msg.target_port, msg.sync_header.sync_round)),

            Message::SyncControl(_) => None,

            Message::Control(_) => None,

        };

        if let Some((target_port, sync_round)) = target_port_and_round_number {

            // This message is aimed at a port, but we're shutting down, so

            // notify the peer that its was not received properly.

            // (also: since we're shutting down, we're not in sync mode and

            // the context contains the definitive set of owned ports)

            let port = scheduled.ctx.get_port_by_id(target_port).unwrap();

                let message = SyncControlMessage {

                    in_response_to_sync_round: sync_round,

                    target_component_id: port.peer_connector,

                    content: SyncControlContent::ChannelIsClosed(port.peer_id),

                };

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

            }

        }

    /// Handles changes to the context that were made by the component. This is

    /// the way (due to Rust's borrowing rules) that we bubble up changes in the

    /// component's state that the scheduler needs to know about (e.g. a message

    /// that the component wants to send, a port that has been added).

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

        let connector_id = scheduled.ctx.id;

        // Handling any messages that were sent

        while let Some(message) = scheduled.ctx.outbox.pop_front() {

            let target_component_id = match &message {

                Message::Data(content) => {

                    // Data messages are always sent to a particular port, and

                    // may end up being rerouted.

                    let port_desc = scheduled.ctx.get_port_by_id(content.data_header.sending_port).unwrap();

                    debug_assert_eq!(port_desc.peer_id, content.data_header.target_port);

                    if port_desc.state == PortState::Closed {

                        todo!("handle sending over a closed port")

                    }

                    port_desc.peer_connector

                    let port_desc = scheduled.ctx.get_port_by_id(content.source_port).unwrap();

                    debug_assert_eq!(port_desc.peer_id, content.target_port);

                    if port_desc.state == PortState::Closed {

                        todo!("handle sending over a closed port")

                    }

                    port_desc.peer_connector

                },

                Message::SyncControl(_) => unreachable!("component sending 'SyncControl' messages directly"),

                Message::Control(_) => unreachable!("component sending 'Control' messages directly"),

            };

            self.runtime.send_message(target_component_id, message);

        }

        while let Some(state_change) = scheduled.ctx.state_changes.pop_front() {

            match state_change {

                ComponentStateChange::CreatedComponent(component, initial_ports) => {

                    // Creating a new component. The creator needs to relinquish

                    // ownership of the ports that are given to the new

                    // component. All data messages that were intended for that

                    // port also needs to be transferred.

                    let new_key = self.runtime.create_pdl_component(component, false);

                    let new_connector = self.runtime.get_component_private(&new_key);

                        new_connector.ctx.ports.push(port.clone());

                        // Notify the peer that the port has changed, but only

                        // if the port wasn't already closed (otherwise the peer

                        // is gone).

                        if port.state == PortState::Open {

                            let reroute_message = scheduled.router.prepare_reroute(

                                port.self_id, port.peer_id, scheduled.ctx.id,

                                port.peer_connector, new_connector.ctx.id,

                                &mut new_connector.router

                            );

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

                        }

                    }

                    // Schedule new connector to run

                    self.runtime.push_work(new_key);

                },

                ComponentStateChange::CreatedPort(port) => {

                    scheduled.ctx.ports.push(port);

                },

                ComponentStateChange::ChangedPort(port_change) => {

                    if port_change.is_acquired {

            };

            if !is_valid {

                // We don't own the port

                println!(" ****** DEBUG ****** : Sending through closed port!!! {}", port_id.index);

                return Err(());

            }

        }

        self.outbox.push_back(contents);

        return Ok(());

    }

    /// Notify that component just finished a sync block. Like

    /// `notify_sync_start`: drop out of the `Component::Run` function.

    pub(crate) fn notify_sync_end(&mut self, changed_ports: &[ComponentPortChange]) {

        debug_assert!(self.is_in_sync);

        self.is_in_sync = false;

        self.changed_in_sync = true;

        self.state_changes.reserve(changed_ports.len());

        for changed_port in changed_ports {

            self.state_changes.push_back(ComponentStateChange::ChangedPort(changed_port.clone()));

        }

/// A structure that contains inbox messages. Some messages are left inside and

/// continuously re-read. Others are taken out, but may potentially be put back

/// for later reading. Later reading in this case implies that they are put back

/// for reading in the next sync round.

struct Inbox {

    temp_m: Vec<Message>,

    temp_d: Vec<Message>,

    messages: RawVec<Message>,

    next_delay_idx: u32,

    start_read_idx: u32,

    next_read_idx: u32,

    generation: u32,

}

#[derive(Clone, Copy)]

pub(crate) struct MessageTicket {

    index: u32,

    generation: u32,

}

impl Inbox {

    fn new() -> Self {

        return Inbox {

            temp_m: Vec::new(), temp_d: Vec::new(),

            messages: RawVec::new(),

            next_delay_idx: 0,

            start_read_idx: 0,

            next_read_idx: 0,

            generation: 0,

        }

    }

    fn insert_new(&mut self, message: Message) {

        assert!(self.messages.len() < u32::MAX as usize); // TODO: @Size

        self.temp_m.push(message);

        return;

        self.messages.push(message);

    }

    fn get_next_message_ticket(&mut self) -> Option<MessageTicket> {

mod api_component;

mod speculation;

mod data_transmission;

mod sync_failure;

use super::*;

use crate::{PortId, ProtocolDescription};

use crate::common::Id;

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

use crate::runtime2::native::{ApplicationSyncAction};

// Generic testing constants, use when appropriate to simplify stress-testing

pub(crate) const NUM_LOOPS: u32 = 15;

fn create_runtime(pdl: &str) -> Runtime {

    let protocol = ProtocolDescription::parse(pdl.as_bytes()).expect("parse pdl");

    let runtime = Runtime::new(NUM_THREADS, protocol);

    return runtime;

}

fn run_test_in_runtime<F: Fn(&mut ApplicationInterface)>(pdl: &str, constructor: F) {

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