//     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::native::Connector;

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

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

pub(crate) struct ConnectorPublic {

    pub inbox: PublicInbox,

    pub sleeping: AtomicBool,

}

impl ConnectorPublic {

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

        ConnectorPublic{

            let branch = &self.tree[branch_id];

            if branch.awaiting_port != message.data_header.target_port { continue; }

            if !self.consensus.branch_can_receive(branch_id, &message) { continue; }

            // This branch can receive, so fork and given it the message

            let receiving_branch_id = self.tree.fork_branch(branch_id);

            self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

            let receiving_branch = &mut self.tree[receiving_branch_id];

            debug_assert!(receiving_branch.awaiting_port == message.data_header.target_port);

            receiving_branch.awaiting_port = PortIdLocal::new_invalid();

            self.consensus.notify_of_received_message(receiving_branch_id, &message, ctx);

            // And prepare the branch for running

            self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);

        }

    }

    pub fn handle_new_sync_comp_message(&mut self, message: SyncCompMessage, ctx: &mut ComponentCtx) -> Option<ConnectorScheduling> {

        if let Some(round_conclusion) = self.consensus.handle_new_sync_comp_message(message, ctx) {

            return Some(self.enter_non_sync_mode(round_conclusion, ctx));

        }

                // Note: we only know that a branch is waiting on a message when

                // it reaches the `get` call. But we might have already received

                // a message that targets this branch, so check now.

                let mut any_message_received = false;

                for message in comp_ctx.get_read_data_messages(port_id) {

                    if self.consensus.branch_can_receive(branch_id, &message) {

                        // This branch can receive the message, so we do the

                        // fork-and-receive dance

                        let receiving_branch_id = self.tree.fork_branch(branch_id);

                        let branch = &mut self.tree[receiving_branch_id];

                        branch.awaiting_port = PortIdLocal::new_invalid();

                        self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

                        self.consensus.notify_of_received_message(receiving_branch_id, &message, comp_ctx);

                        self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);

                        any_message_received = true;

                    }

                }

                if any_message_received {

                    return ConnectorScheduling::Immediate;

                }

                let left_branch = &mut self.tree[left_id];

                left_branch.prepared = PreparedStatement::ForkedExecution(true);

                let right_branch = &mut self.tree[right_id];

                right_branch.prepared = PreparedStatement::ForkedExecution(false);

            }

            EvalContinuation::Put(port_id, content) => {

                // Branch is attempting to send data

                let port_id = PortIdLocal::new(port_id.0.u32_suffix);

                let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);

                if let Err(_) = comp_ctx.submit_message(Message::Data(DataMessage {

                    sync_header, data_header,

                })) {

                    // We don't own the port

                    let pd = &sched_ctx.runtime.protocol_description;

                    let eval_error = branch.code_state.new_error_at_expr(

                        &pd.modules, &pd.heap,

                        String::from("attempted to 'put' on port that is no longer owned")

                    );

                    self.eval_error = Some(eval_error);

                    self.mode = Mode::SyncError;

                }

                branch.prepared = PreparedStatement::PerformedPut;

            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;

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

use crate::collections::VecSet;

use crate::protocol::eval::ValueGroup;

use super::ConnectorId;

use super::branch::BranchId;

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

use super::inbox::{

    SyncCompMessage, SyncCompContent, SyncPortMessage, SyncPortContent, SyncHeader,

};

struct BranchAnnotation {

    channel_mapping: Vec<ChannelAnnotation>,

    cur_marker: BranchMarker,

}

#[derive(Debug)]

pub(crate) struct LocalSolution {

    component: ConnectorId,

    final_branch_id: BranchId,

    port_mapping: Vec<(ChannelId, BranchMarker)>,

}

        let mut target_mapping = Vec::with_capacity(source_mapping.len());

        for port in source_mapping {

            // Note: if the port is silent, and we've never communicated

            // over the port, then we need to do so now, to let the peer

            // component know about our sync leader state.

            let port_desc = ctx.get_port_by_channel_id(port.channel_id).unwrap();

            let self_port_id = port_desc.self_id;

            let peer_port_id = port_desc.peer_id;

            let channel_id = port_desc.channel_id;

            if !self.encountered_ports.contains(&self_port_id) {

                    sync_header: SyncHeader{

                        sending_component_id: ctx.id,

                        highest_component_id: self.highest_connector_id,

                        sync_round: self.sync_round

                    },

                }));

                self.encountered_ports.push(self_port_id);

            }

            target_mapping.push((

                channel_id,

                port.registered_id.unwrap_or(BranchMarker::new_invalid())

            ));

        }

        let local_solution = LocalSolution{

            component: ctx.id,

            }

        }

    }

    pub fn handle_new_sync_port_message(&mut self, message: SyncPortMessage, ctx: &mut ComponentCtx) {

        if !self.handle_received_sync_header(&message.sync_header, ctx) {

            return;

        }

        debug_assert!(self.is_in_sync());

        debug_assert!(ctx.get_port_by_id(message.target_port).is_some());

        match message.content {

            SyncPortContent::NotificationWave => {

                // Wave to discover everyone in the network, handling sync

                // header takes care of leader discovery, here we need to make

                // sure we propagate the wave

                if self.handled_wave {

                    return;

                }

                self.handled_wave = true;

                // Propagate wave to all peers except the one that has sent us

                // the wave.

    pub fn notify_of_received_message(&mut self, branch_id: BranchId, message: &DataMessage, ctx: &ComponentCtx) {

        debug_assert!(self.branch_can_receive(branch_id, message));

        let target_port = ctx.get_port_by_id(message.data_header.target_port).unwrap();

        let branch = &mut self.branch_annotations[branch_id.index as usize];

        for mapping in &mut branch.channel_mapping {

            if mapping.channel_id == target_port.channel_id {

                // Found the port in which the message should be inserted

                mapping.registered_id = Some(message.data_header.new_mapping);

                // Check for sent ports

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

                if !self.workspace_ports.is_empty() {

                    todo!("handle received ports");

                    self.workspace_ports.clear();

                }

                return;

            }

        }

        // If here, then the branch didn't actually own the port? Means the

        // caller made a mistake

        unreachable!("incorrect notify_of_received_message");

    }

    /// Matches the mapping between the branch and the data message. If they

    /// match then the branch can receive the message.

    pub fn branch_can_receive(&self, branch_id: BranchId, message: &DataMessage) -> bool {

        if let Some(peer) = self.peers.iter().find(|v| v.id == message.sync_header.sending_component_id) {

            if message.sync_header.sync_round < peer.expected_sync_round {

                return false;

            }

        }

        let annotation = &self.branch_annotations[branch_id.index as usize];

        for expected in &message.data_header.expected_mapping {

            // If we own the port, then we have an entry in the

            // annotation, check if the current mapping matches

            for current in &annotation.channel_mapping {

                if expected.channel_id == current.channel_id {

                    if expected.registered_id != current.registered_id {

                        // IDs do not match, we cannot receive the

                        // message in this branch

                        return false;

                    }

                }

    pub sync_round: u32,

}

/// The header added to data messages

#[derive(Debug, Clone)]

pub(crate) struct DataHeader {

    pub expected_mapping: Vec<ChannelAnnotation>,

    pub sending_port: PortIdLocal,

    pub target_port: PortIdLocal,

    pub new_mapping: BranchMarker,

}

/// A data message is a message that is intended for the receiver's PDL code,

/// but will also be handled by the consensus algorithm

#[derive(Debug, Clone)]

pub(crate) struct DataMessage {

    pub sync_header: SyncHeader,

    pub data_header: DataHeader,

}

#[derive(Debug)]

pub(crate) enum SyncCompContent {

    LocalFailure, // notifying leader that component has failed (e.g. timeout, whatever)

    LocalSolution(LocalSolution), // sending a local solution to the leader

    PartialSolution(SolutionCombiner), // when new leader is detected, forward all local results

    GlobalSolution(GlobalSolution), // broadcasting to everyone

    GlobalFailure, // broadcasting to everyone

    AckFailure, // acknowledgement of failure to leader

    Notification, // just a notification (so purpose of message is to send the SyncHeader)

    Presence(ConnectorId, Vec<ChannelId>), // notifying leader of component presence (needed to ensure failing a round involves all components in a sync round)

/// A sync message is a message that is intended only for the consensus

/// algorithm. The message goes directly to a component.

#[derive(Debug)]

pub(crate) struct SyncCompMessage {

    pub sync_header: SyncHeader,

    pub target_component_id: ConnectorId,

    pub content: SyncCompContent,

}

#[derive(Debug)]

pub(crate) enum SyncPortContent {

    NotificationWave,

}

#[derive(Debug)]

pub(crate) struct SyncPortMessage {

    pub sync_header: SyncHeader,

    pub source_port: PortIdLocal,

    pub target_port: PortIdLocal,

    pub content: SyncPortContent,

}

/// A control message is a message intended for the scheduler that is executing

pub(crate) enum Message {

    Data(DataMessage),

    SyncComp(SyncCompMessage),

    SyncPort(SyncPortMessage),

    Control(ControlMessage),

}

impl Message {

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

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

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

        // Currently only data messages have a source port

        }

    }

}

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

}

use crate::protocol::ComponentCreationError;

use crate::protocol::eval::ValueGroup;

use crate::runtime2::consensus::RoundConclusion;

use super::{ConnectorKey, ConnectorId, RuntimeInner};

use super::branch::{BranchId, FakeTree, QueueKind, SpeculativeState};

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

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

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

use super::connector::{ConnectorScheduling, ConnectorPDL};

use super::inbox::{

    SyncCompMessage, SyncPortMessage,

    ControlContent, ControlMessage

};

/// Generic connector interface from the scheduler's point of view.

pub(crate) trait Connector {

    /// Should run the connector's behaviour up until the next blocking point.

    /// One should generally request and handle new messages from the component

    /// context. Then perform any logic the component has to do, and in the

    /// process perhaps queue up some state changes using the same context.

    fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling;

}

            let branch = &self.tree[branch_id];

            if branch.awaiting_port != message.data_header.target_port { continue; }

            if !self.consensus.branch_can_receive(branch_id, &message) { continue; }

            // This branch can receive, so fork and given it the message

            let receiving_branch_id = self.tree.fork_branch(branch_id);

            debug_assert!(receiving_branch_id.index as usize == self.branch_extra.len());

            self.branch_extra.push(self.branch_extra[branch_id.index as usize]); // copy instruction index

            self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

            let receiving_branch = &mut self.tree[receiving_branch_id];

            self.consensus.notify_of_received_message(receiving_branch_id, &message, ctx);

            // And prepare the branch for running

            self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);

        }

    }

    pub(crate) fn handle_new_sync_comp_message(&mut self, message: SyncCompMessage, ctx: &mut ComponentCtx) {

        if let Some(conclusion) = self.consensus.handle_new_sync_comp_message(message, ctx) {

            self.collapse_sync_to_conclusion(conclusion, ctx);

        }

    }

            // We still have instructions to perform

            let cur_instruction = &self.sync_desc[instruction_idx];

            self.branch_extra[branch_id.index as usize] += 1;

            match &cur_instruction {

                ApplicationSyncAction::Put(port_id, content) => {

                    let port_id = *port_id;

                    let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);

                    let message = Message::Data(DataMessage {

                        sync_header,

                        data_header,

                    });

                    comp_ctx.submit_message(message);

                    self.tree.push_into_queue(QueueKind::Runnable, branch_id);

                    return ConnectorScheduling::Immediate;

                },

                ApplicationSyncAction::Get(port_id) => {

                    let port_id = *port_id;

                    branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

                    branch.awaiting_port = port_id;

                    self.tree.push_into_queue(QueueKind::AwaitingMessage, branch_id);

                    let mut any_message_received = false;

                    for message in comp_ctx.get_read_data_messages(port_id) {

                        if self.consensus.branch_can_receive(branch_id, &message) {

                            // This branch can receive the message, so we do the

                            // fork-and-receive dance

                            let receiving_branch_id = self.tree.fork_branch(branch_id);

                            let branch = &mut self.tree[receiving_branch_id];

                            debug_assert!(receiving_branch_id.index as usize == self.branch_extra.len());

                            self.branch_extra.push(instruction_idx + 1);

                            self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

                            self.consensus.notify_of_received_message(receiving_branch_id, &message, comp_ctx);

                            self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);

                            any_message_received = true;

                        }

                    }

                    if any_message_received {

                        return ConnectorScheduling::Immediate;

                    }

                        // We're actually done, we can safely destroy the

                        // currently running connector

                        self.runtime.destroy_component(connector_key);

                        continue 'thread_loop;

                    } else {

                        cur_schedule = ConnectorScheduling::NotNow;

                    }

                } else {

                    self.debug_conn(connector_id, "Running ...");

                    let scheduler_ctx = SchedulerCtx{ runtime: &*self.runtime };

                    let new_schedule = scheduled.connector.run(scheduler_ctx, &mut scheduled.ctx);

                    self.debug_conn(connector_id, &format!("Finished running (new scheduling is {:?})", new_schedule));

                    // Handle all of the output from the current run: messages to

                    // send and connectors to instantiate.

                    self.handle_changes_in_context(scheduled);

                    cur_schedule = new_schedule;

                }

            }

            // If here then the connector does not require immediate execution.

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

            // state.

            match cur_schedule {