//     authoritative source during *sync* mode. They retrieve each other's

//     state during the transitions. Hence port data exists duplicated between

//     these two datastructures.

// - The execution tree is where executed branches reside. But the execution

//     tree is only aware of the tree shape itself (and keeps track of some

//     queues of branches that are in a particular state), and tends to store

//     the PDL program state. The consensus algorithm is also somewhat aware

//     of the execution tree, but only in terms of what is needed to complete

//     a sync round (for now, that means the port mapping in each branch).

//     Hence once more we have properties conceptually associated with branches

//     in two places.

// - TODO: Write about handling messages, consensus wrapping data

// - TODO: Write about way information is exchanged between PDL/component and scheduler through ctx

use std::collections::HashMap;

use std::sync::atomic::AtomicBool;

use crate::{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{

            inbox: PublicInbox::new(),

            sleeping: AtomicBool::new(initialize_as_sleeping),

        }

    }

}

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

enum Mode {

    NonSync,    // running non-sync code

    Sync,       // running sync code (in potentially multiple branches)

    SyncError,  // encountered an unrecoverable error in sync mode

    Error,      // encountered an error in non-sync mode (or finished handling the sync mode error).

                ));

                comp_ctx.push_port(putter);

                comp_ctx.push_port(getter);

                return ConnectorScheduling::Immediate;

            },

            _ => unreachable!("unexpected run result '{:?}' while running in non-sync mode", run_result),

        }

    }

    /// Helper that moves the component's state back into non-sync mode, using

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

            }

            return result;

        }

    }

}

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

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

        }

    }

    /// If the message is sent through a particular channel, then this function

    /// returns the target port through which the message was sent.

    pub(crate) fn target_port(&self) -> Option<PortIdLocal> {

        match self {

            Message::Data(message) => return Some(message.data_header.target_port),

            Message::SyncPort(message) => return Some(message.target_port),

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

            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.

//  Should behave as a MPSC queue.

pub struct PublicInbox {

    messages: Mutex<VecDeque<Message>>,

}

impl PublicInbox {

    pub fn new() -> Self {

        Self{

            messages: Mutex::new(VecDeque::new()),

        }

    }

    connector: ScheduledConnector,

    generation: std::sync::atomic::AtomicU32,

}

struct ConnectorStore {

    // Freelist storage of connectors. Storage should be pointer-stable as

    // someone might be mutating the vector while we're executing one of the

    // connectors.

    entries: RawVec<*mut StoreEntry>,

    free: Vec<usize>,

}

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

            shutting_down: false,

        };

        let index;

        let key;

        if self.free.is_empty() {

            // No free entries, allocate new entry

            index = self.entries.len();

            key = ConnectorKey{

                index: index as u32, generation: 0

            };

            // So enqueue it if requested, and otherwise put it in a sleeping

            // state.

            match cur_schedule {

                ConnectorScheduling::Immediate => unreachable!(),

                ConnectorScheduling::Later => {

                    // Simply queue it again later

                    self.runtime.push_work(connector_key);

                },

                ConnectorScheduling::NotNow => {

                    // Need to sleep, note that we are the only ones which are

                    // allows to set the sleeping state to `true`, and since

                    // we're running it must currently be `false`.

                    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 {

                        // All ports (if any) already closed

                        self.runtime.destroy_component(connector_key);

                        continue 'thread_loop;

                    }

                    self.try_go_to_sleep(connector_key, scheduled);

                },

            }

        }

    }

    /// Receiving messages from the public inbox and handling them or storing

    /// them in the component's private inbox

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

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

                } else {

                    scheduled.ctx.inbox.insert_new(message);

                }

            }

        }

    }

    fn handle_message_while_shutting_down(&mut self, message: Message, scheduled: &mut ScheduledConnector) {

        let target_port_and_round_number = match message {

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

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

        }

    }

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

                },

                Message::SyncComp(content) => {

                    // Sync messages are always sent to a particular component,

                    // the sender must make sure it actually wants to send to

                    // the specified component (and is not using an inconsistent

                    // component ID associated with a port).

                    content.target_component_id

                },

                Message::SyncPort(content) => {

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

                    for port_id in initial_ports {

                        // Transfer messages associated with the transferred port

                        scheduled.ctx.inbox.transfer_messages_for_port(port_id, &mut new_connector.ctx.inbox);

                        // Transfer the port itself

                        let port_index = scheduled.ctx.ports.iter()

                            .position(|v| v.self_id == port_id)

                            .unwrap();

                        let port = scheduled.ctx.ports.remove(port_index);

                        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 {

                        scheduled.ctx.ports.push(port_change.port);

                    } else {

                        let index = scheduled.ctx.ports

                            .iter()

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

                            .unwrap();

                        scheduled.ctx.ports.remove(index);

                    }

                }

            }

        }

    /// Submit a message for the scheduler to send to the appropriate receiver.

    /// May only be called inside of a sync block.

    pub(crate) fn submit_message(&mut self, contents: Message) -> Result<(), ()> {

        debug_assert!(self.is_in_sync);

        if let Some(port_id) = contents.source_port() {

            let port_info = self.get_port_by_id(port_id);

            let is_valid = match port_info {

                Some(port_info) => {

                    port_info.state == PortState::Open

                },

                None => false,

            };

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

        }

    }

    /// Retrieves messages matching a particular port and branch id. But only

    /// those messages that have been previously received with

    /// `read_next_message`.

    pub(crate) fn get_read_data_messages(&self, match_port_id: PortIdLocal) -> MessagesIter {

        return self.inbox.get_read_data_messages(match_port_id);

    }

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

        if !self.is_in_sync { return None; }

        return self.inbox.get_next_message_ticket();

            self.next_index += 1;

        }

        // No more messages

        return None;

    }

}

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

// Private Inbox

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

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

        if self.next_read_idx as usize >= self.temp_m.len() { return None };

        let idx = self.next_read_idx;

        self.generation += 1;

        self.next_read_idx += 1;

        return Some(MessageTicket{ index: idx, generation: self.generation });

        let cur_read_idx = self.next_read_idx as usize;

        if cur_read_idx >= self.messages.len() {

            return None;

        }

        self.generation += 1;

        self.next_read_idx += 1;

mod network_shapes;

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

        .expect("parse PDL");

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

    let mut api = runtime.create_interface();

    for _ in 0..NUM_INSTANCES {

        constructor(&mut api);

    }

}

pub(crate) struct TestTimer {

    name: &'static str,

    started: std::time::Instant