use crate::protocol::eval::{Value, ValueGroup};

use crate::runtime2::port::{Port, PortIdLocal};

    pub inbox: HashMap<PortIdLocal, ValueGroup>, // TODO: Remove, currently only valid in single-get/put mode

    pub prepared_channel: Option<(Value, Value)>, // TODO: Maybe remove?

            prepared_channel: None,

            prepared_channel: None,

    /// Returns the non-sync branch (TODO: better name?)

    pub fn base_branch_mut(&mut self) -> &mut Branch {

        debug_assert!(!self.is_in_sync());

        return &mut self.branches[0];

    }

    /// as the root of the speculative tree. The id of this root sync branch is

    /// returned.

    pub fn start_sync(&mut self) -> BranchId {

        return sync_branch_id;

/// connector.rs

///

/// Represents a component. A component (and the scheduler that is running it)

/// has many properties that are not easy to subdivide into aspects that are

/// conceptually handled by particular data structures. That is to say: the code

/// that we run governs: running PDL code, keeping track of ports, instantiating

/// new components and transports (i.e. interacting with the runtime), running

/// a consensus algorithm, etc. But on the other hand, our data is rather

/// simple: we have a speculative execution tree, a set of ports that we own,

/// and a bit of code that we should run.

///

/// So currently the code is organized as following:

/// - The scheduler that is running the component is the authoritative source on

///     ports during *non-sync* mode. The consensus algorithm is the

///     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 crate::common::ComponentState;

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

use crate::runtime2::consensus::find_ports_in_value_group;

use crate::runtime2::port::PortKind;

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

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

use super::inbox2::{DataMessageFancy, MessageFancy, SyncMessageFancy};

use super::inbox::PublicInbox;

use super::native::Connector;

use super::port::PortIdLocal;

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

#[derive(Eq, PartialEq)]

pub(crate) enum ConnectorScheduling {

    Immediate,      // Run again, immediately

    Later,          // Schedule for running, at some later point in time

    NotNow,         // Do not reschedule for running

    Exit,          // Connector has exited

}

struct ConnectorRunContext {}

        debug_assert!(ctx.workspace_branches.is_empty());

        self.consensus.handle_received_data_header(&self.tree, &message.data_header, &mut ctx.workspace_branches);

        for branch_id in ctx.workspace_branches.drain(..) {

        debug_assert!(self.tree.is_in_sync() && self.consensus.is_in_sync());

                            self.consensus.notify_of_received_message(recv_branch_id, &message.data_header, &message.content);

            RunResult::BranchPut(port_id, content) => {

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

                    comp_ctx.submit_message(MessageFancy::Data(DataMessageFancy{

                        sync_header, data_header, content

                    }));

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

                    return ConnectorScheduling::Immediate;

    pub fn run_in_deterministic_mode(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling {

        debug_assert!(!self.tree.is_in_sync() && !self.consensus.is_in_sync());

        let branch = self.tree.base_branch_mut();

        debug_assert!(branch.sync_state == SpeculativeState::RunningNonSync);

        let mut run_context = ConnectorRunContext{};

        let run_result = branch.code_state.run(&mut run_context, &sched_ctx.runtime.protocol_description);

        match run_result {

            RunResult::ComponentTerminated => {

                branch.sync_state = SpeculativeState::Finished;

                return ConnectorScheduling::Exit;

            },

            RunResult::ComponentAtSyncStart => {

                let current_ports = comp_ctx.notify_sync_start();

                let sync_branch_id = self.tree.start_sync();

                self.consensus.start_sync(current_ports);

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

                return ConnectorScheduling::Immediate;

            },

            RunResult::NewComponent(definition_id, monomorph_idx, arguments) => {

                // Note: we're relinquishing ownership of ports. But because

                // we are in non-sync mode the scheduler will handle and check

                // port ownership transfer.

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

                find_ports_in_value_group(&arguments, &mut comp_ctx.workspace_ports);

                let new_state = ComponentState {

                    prompt: Prompt::new(

                        &sched_ctx.runtime.protocol_description.types,

                        &sched_ctx.runtime.protocol_description.heap,

                        definition_id, monomorph_idx, arguments

                    ),

                };

                let new_component = ConnectorPDL::new(new_state, comp_ctx.workspace_ports.clone());

                comp_ctx.push_component(new_component);

                return ConnectorScheduling::Later;

            },

            RunResult::NewChannel => {

                let (getter, putter) = sched_ctx.runtime.create_channel(comp_ctx.id);

                debug_assert!(getter.kind == PortKind::Getter && putter.kind == PortKind::Putter);

                branch.prepared_channel = Some((

                    Value::Input(PortId::new(putter.self_id.index)),

                    Value::Output(PortId::new(getter.self_id.index)),

                ));

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

        }

    }

use crate::runtime2::port::{Port, PortIdLocal};

// TODO: Have a "branch+port position hint" in case multiple operations are

//  performed on the same port to prevent repeated lookups

    workspace_ports: Vec<PortIdLocal>,

    /// Returns whether the consensus algorithm is running in sync mode

    pub fn is_in_sync(&self) -> bool {

        return !self.branch_annotations.is_empty();

    }

    pub fn start_sync(&mut self, ports: &[Port]) {

                    port_id: v.self_id,

    pub fn handle_message_to_send(&mut self, branch_id: BranchId, source_port_id: PortIdLocal, content: &ValueGroup, ctx: &mut ComponentCtxFancy) -> (SyncHeader, DataHeader) {

        debug_assert!(self.is_in_sync());

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

        // Check for ports that are begin sent

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

        find_ports_in_value_group(content, &mut self.workspace_ports);

        if !self.workspace_ports.is_empty() {

            todo!("handle sending ports");

            self.workspace_ports.clear();

        }

        let sync_header = SyncHeader{

            sending_component_id: ctx.id,

            highest_component_id: self.highest_connector_id,

        };

        // TODO: Handle multiple firings. Right now we just assign the current

        //  branch to the `None` value because we know we can only send once.

        debug_assert!(branch.port_mapping.iter().find(|v| v.port_id == source_port_id).unwrap().registered_id.is_none());

        let port_info = ctx.get_port_by_id(source_port_id).unwrap();

        let data_header = DataHeader{

            expected_mapping: branch.port_mapping.clone(),

            target_port: port_info.peer_id,

            new_mapping: branch_id

        };

        for mapping in &mut branch.port_mapping {

            if mapping.port_id == source_port_id {

                mapping.expected_firing = Some(true);

                mapping.registered_id = Some(branch_id);

            }

        }

        return (sync_header, data_header);

    pub fn notify_of_received_message(&mut self, branch_id: BranchId, data_header: &DataHeader, content: &ValueGroup) {

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

                // Check for sent ports

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

                find_ports_in_value_group(content, &mut self.workspace_ports);

                if !self.workspace_ports.is_empty() {

                    todo!("handle received ports");

                    self.workspace_ports.clear();

                }

/// Recursively goes through the value group, attempting to find ports.

/// Duplicates will only be added once.

pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut Vec<PortIdLocal>) {

    // Helper to check a value for a port and recurse if needed.

    use crate::protocol::eval::Value;

    fn find_port_in_value(group: &ValueGroup, value: &Value, ports: &mut Vec<PortIdLocal>) {

        match value {

            Value::Input(port_id) | Value::Output(port_id) => {

                // This is an actual port

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

                for prev_port in ports.iter() {

                    if *prev_port == cur_port {

                        // Already added

                        return;

                    }

                }

                ports.push(cur_port);

            },

            Value::Array(heap_pos) |

            Value::Message(heap_pos) |

            Value::String(heap_pos) |

            Value::Struct(heap_pos) |

            Value::Union(_, heap_pos) => {

                // Reference to some dynamic thing which might contain ports,

                // so recurse

                let heap_region = &group.regions[*heap_pos as usize];

                for embedded_value in heap_region {

                    find_port_in_value(group, embedded_value, ports);

                }

            },

            _ => {}, // values we don't care about

        }

    }

    // Clear the ports, then scan all the available values

    ports.clear();

    for value in &value_group.values {

        find_port_in_value(value_group, value, ports);

    }

}

    outbox: VecDeque<MessageFancy>,

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

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

    pub workspace_ports: Vec<PortIdLocal>,

    pub workspace_branches: Vec<BranchId>,

    pub(crate) fn submit_message(&mut self, contents: MessageFancy) {