// 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 std::collections::HashMap;

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

use crate::PortId;

use crate::common::ComponentState;

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

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

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

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

use super::inbox::{DataMessage, DataContent, Message, SyncMessage, PublicInbox};

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{

            inbox: PublicInbox::new(),

            sleeping: AtomicBool::new(initialize_as_sleeping),

        }

    }

}

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

}

pub(crate) struct ConnectorPDL {

    tree: ExecTree,

    consensus: Consensus,

    last_finished_handled: Option<BranchId>,

}

struct ConnectorRunContext<'a> {

    branch_id: BranchId,

    consensus: &'a Consensus,

    received: &'a HashMap<PortIdLocal, ValueGroup>,

    scheduler: SchedulerCtx<'a>,

    prepared_channel: Option<(Value, Value)>,

}

impl<'a> RunContext for ConnectorRunContext<'a>{

    fn did_put(&mut self, port: PortId) -> bool {

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

        let annotation = self.consensus.get_annotation(self.branch_id, port_id);

        return annotation.registered_id.is_some();

    }

    fn get(&mut self, port: PortId) -> Option<ValueGroup> {

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

        match self.received.get(&port_id) {

            Some(data) => Some(data.clone()),

            None => None,

        }

    }

    fn fires(&mut self, port: PortId) -> Option<Value> {

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

        let annotation = self.consensus.get_annotation(self.branch_id, port_id);

        return annotation.expected_firing.map(|v| Value::Bool(v));

    }

    fn get_channel(&mut self) -> Option<(Value, Value)> {

        return self.prepared_channel.take();

    }

}

impl Connector for ConnectorPDL {

                                new_connector.ctx.inbox_messages.push(message);

                            } else {

                                message_idx += 1;

                            }

                        }

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

                        let reroute_message = scheduled.router.prepare_reroute(

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

                            port.peer_connector, new_connector.ctx.id

                        );

                        self.debug_conn(connector_id, &format!("Sending message [newcon]\n --- {:?}", reroute_message));

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

                    }

                }

            }

        }

        // Finally, check if we just entered or just left a sync region

        if scheduled.ctx.changed_in_sync {

            if scheduled.ctx.is_in_sync {

                // Just entered sync region

            } else {

                scheduled.ctx.inbox_len_read = 0;

            }

            scheduled.ctx.changed_in_sync = false; // reset flag

        }

    }

    fn try_go_to_sleep(&self, connector_key: ConnectorKey, connector: &mut ScheduledConnector) {

        debug_assert_eq!(connector_key.index, connector.ctx.id.0);

        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)

            }

        }

    }

    #[inline]

    fn get_message_target_port(message: &Message) -> Option<PortIdLocal> {

        match message {

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

            Message::Sync(_) => {},

            Message::Control(control) => {

                match &control.content {

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

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

                    ControlContent::Ping | ControlContent::Ack => {},

                }

            },

        }

        return None

    }

    // TODO: Remove, this is debugging stuff

    fn debug(&self, message: &str) {

        // println!("DEBUG [thrd:{:02} conn:  ]: {}", self.scheduler_id, message);

    }

    fn debug_conn(&self, conn: ConnectorId, message: &str) {

        // println!("DEBUG [thrd:{:02} conn:{:02}]: {}", self.scheduler_id, conn.0, 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

/// scheduler need to be exchanged.

pub(crate) struct ComponentCtx {

    // Mostly managed by the scheduler

    pub(crate) id: ConnectorId,

    ports: Vec<Port>,

    inbox_len_read: usize,

    // Submitted by the component

    is_in_sync: bool,

    changed_in_sync: bool,

    outbox: VecDeque<Message>,

    state_changes: VecDeque<ComponentStateChange>,

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

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

    // TODO: Move to scheduler ctx, this is the wrong place

    pub workspace_ports: Vec<PortIdLocal>,

    pub workspace_branches: Vec<BranchId>,

}

impl ComponentCtx {

    pub(crate) fn new_empty() -> Self {

        return Self{

            id: ConnectorId::new_invalid(),

            ports: Vec::new(),

            inbox_messages: Vec::new(),

            inbox_len_read: 0,

            is_in_sync: false,

            changed_in_sync: false,

            outbox: VecDeque::new(),

            state_changes: VecDeque::new(),

            workspace_ports: Vec::new(),

            workspace_branches: Vec::new(),

        };

    }

    /// Notify the runtime that the component has created a new component. May

    /// only be called outside of a sync block.

    pub(crate) fn push_component(&mut self, component: ConnectorPDL, initial_ports: Vec<PortIdLocal>) {

        debug_assert!(!self.is_in_sync);

        self.state_changes.push_back(ComponentStateChange::CreatedComponent(component, initial_ports));

    }

    /// Notify the runtime that the component has created a new port. May only

    /// be called outside of a sync block (for ports received during a sync

    /// block, pass them when calling `notify_sync_end`).

    pub(crate) fn push_port(&mut self, port: Port) {

        debug_assert!(!self.is_in_sync);

        self.state_changes.push_back(ComponentStateChange::CreatedPort(port))

    }

    #[inline]

    pub(crate) fn get_ports(&self) -> &[Port] {

        return self.ports.as_slice();

    }

//

// These tests explicitly do not use the "NUM_INSTANCES" constant because we're

// doing some communication with the native component. Hence only expect one

use super::*;

#[test]

fn test_put_and_get() {

    const CODE: &'static str = "

    primitive handler(in<u32> request, out<u32> response, u32 loops) {

        u32 index = 0;

        while (index < loops) {

            synchronous {

                auto value = get(request);

                put(response, value * 2);

            }

            index += 1;

        }

    }

    ";

    let pd = ProtocolDescription::parse(CODE.as_bytes()).unwrap();

    let rt = Runtime::new(NUM_THREADS, pd);

    let mut api = rt.create_interface();

    let req_chan = api.create_channel().unwrap();

    let resp_chan = api.create_channel().unwrap();

    api.create_connector("", "handler", ValueGroup::new_stack(vec![

        Value::Input(PortId::new(req_chan.getter_id.index)),

        Value::Output(PortId::new(resp_chan.putter_id.index)),

        Value::UInt32(NUM_LOOPS),

    ])).unwrap();

    for loop_idx in 0..NUM_LOOPS {

        api.perform_sync_round(vec![

            ApplicationSyncAction::Put(req_chan.putter_id, ValueGroup::new_stack(vec![Value::UInt32(loop_idx)])),

            ApplicationSyncAction::Get(resp_chan.getter_id)

        ]).expect("start sync round");

        let result = api.wait().expect("finish sync round");

        assert!(result.len() == 1);

        if let Value::UInt32(gotten) = result[0].values[0] {

            assert_eq!(gotten, loop_idx * 2);

        } else {

            assert!(false);

        }

    }

}