id: message.id,

                                sending_component_id: connector_id,

                                content: ControlContent::Ack,

                            });

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

                            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.debug_conn(connector_id, &format!("Sending message [cc ack] \n --- {:?}", ack_message));

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

                        },

                        ControlContent::Ack => {

                            scheduled.router.handle_ack(message.id);

                        },

                        ControlContent::Ping => {},

                    }

                },

                _ => {

                    // All other cases have to be handled by the component

                    scheduled.ctx.inbox_messages.push(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() {

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

            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::Sync(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::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

                        let mut message_idx = 0;

                        while message_idx < scheduled.ctx.inbox_messages.len() {

                            let message = &scheduled.ctx.inbox_messages[message_idx];

                            if Self::get_message_target_port(message) == Some(port_id) {

                                // Need to transfer this message

                                let message = scheduled.ctx.inbox_messages.remove(message_idx);

                                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 {

                // Just left sync region. So clear inbox up until the last

                // message that was read.

                scheduled.ctx.inbox_messages.drain(0..scheduled.ctx.inbox_len_read);

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

    }

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

    }

}

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

// 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_messages: Vec<Message>,

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

    }

    pub(crate) fn get_port_by_id(&self, id: PortIdLocal) -> Option<&Port> {

        return self.ports.iter().find(|v| v.self_id == id);

    }

    fn get_port_mut_by_id(&mut self, id: PortIdLocal) -> Option<&mut Port> {

        return self.ports.iter_mut().find(|v| v.self_id == id);

    }

    /// Notify that component will enter a sync block. Note that after calling

    /// this function you must allow the scheduler to pick up the changes in the

    /// context by exiting your code-executing loop, and to continue executing

    /// code the next time the scheduler picks up the component.

    pub(crate) fn notify_sync_start(&mut self) {

        debug_assert!(!self.is_in_sync);

        self.is_in_sync = true;

        self.changed_in_sync = true;

    }

    #[inline]

    pub(crate) fn is_in_sync(&self) -> bool {

        return self.is_in_sync;

    }

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

        debug_assert!(self.is_in_sync);

        self.outbox.push_back(contents);

    }

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

            messages: &self.inbox_messages,

            next_index: 0,

            max_index: self.inbox_len_read,

            match_port_id

        };

    }

    /// Retrieves the next unread message from the inbox `None` if there are no

    /// (new) messages to read.

    // TODO: Fix the clone of the data message, entirely unnecessary

    pub(crate) fn read_next_message(&mut self) -> Option<Message> {

        if !self.is_in_sync { return None; }

        if self.inbox_len_read == self.inbox_messages.len() { return None; }

        // We want to keep data messages in the inbox, because we need to check

        // them in the future. We don't want to keep sync messages around, we

        // should only handle them once. Control messages should never be in

        // here.

        let message = &self.inbox_messages[self.inbox_len_read];

        match message {

            Message::Data(content) => {

                self.inbox_len_read += 1;

                return Some(Message::Data(content.clone()));

            },

            Message::Sync(_) => {

                let message = self.inbox_messages.remove(self.inbox_len_read);

                return Some(message);

            },

            Message::Control(_) => unreachable!("control message ended up in component inbox"),

        }

    }

}

pub(crate) struct MessagesIter<'a> {

    messages: &'a [Message],

    next_index: usize,

    max_index: usize,

    match_port_id: PortIdLocal,

}

impl<'a> Iterator for MessagesIter<'a> {

    type Item = &'a DataMessage;

    fn next(&mut self) -> Option<Self::Item> {

        // Loop until match is found or at end of messages

        while self.next_index < self.max_index {

            let message = &self.messages[self.next_index];

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

                if message.data_header.target_port == self.match_port_id {

                    // Found a match

                    self.next_index += 1;

                    return Some(message);

                }

            } else {

                // Unreachable because:

                //  1. We only iterate over messages that were previously retrieved by `read_next_message`.

                //  2. Inbox does not contain control/ping messages.

                //  3. If `read_next_message` encounters anything else than a data message, it is removed from the inbox.

                unreachable!();

            }

            self.next_index += 1;

        }

mod network_shapes;

mod api_component;

mod speculation_basic;

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

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

}

impl TestTimer {

    pub(crate) fn new(name: &'static str) -> Self {

        Self{ name, started: std::time::Instant::now() }

    }

}

impl Drop for TestTimer {

    fn drop(&mut self) {

        let delta = std::time::Instant::now() - self.started;

        let nanos = (delta.as_secs_f64() * 1_000_000.0) as u64;

        let millis = nanos / 1000;

        let nanos = nanos % 1000;

        println!("[{}] Took {:>4}.{:03} ms", self.name, millis, nanos);

    }

}

// Testing particular graph shapes

use super::*;

#[test]

fn test_star_shaped_request() {

    const CODE: &'static str = "

    primitive edge(in<u32> input, out<u32> output, u32 loops) {

        u32 index = 0;

        while (index < loops) {

            sync {

                auto req = get(input);

                put(output, req * 2);

            }

            index += 1;

        }

    }

    primitive center(out<u32>[] requests, in<u32>[] responses, u32 loops) {

        u32 loop_index = 0;

        auto num_edges = length(requests);

        while (loop_index < loops) {

            // print(\"starting loop\");

            sync {

                u32 edge_index = 0;

                u32 sum = 0;

                while (edge_index < num_edges) {

                    put(requests[edge_index], edge_index);

                    auto response = get(responses[edge_index]);

                    sum += response;

                    edge_index += 1;

                }

                assert(sum == num_edges * (num_edges - 1));

            }

            // print(\"ending loop\");

            loop_index += 1;

        }

    }

    composite constructor(u32 num_edges, u32 num_loops) {

        auto requests = {};

        auto responses = {};

        u32 edge_index = 0;

        while (edge_index < num_edges) {

            channel req_put -> req_get;

            channel resp_put -> resp_get;

            new edge(req_get, resp_put, num_loops);

            requests @= { req_put };

            responses @= { resp_get };

            edge_index += 1;

        }

        new center(requests, responses, num_loops);

    }

    ";

    let thing = TestTimer::new("star_shaped_request");

    run_test_in_runtime(CODE, |api| {

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

            Value::UInt32(5),

            Value::UInt32(NUM_LOOPS),

        ])).expect("create connector");

    });

}

#[test]

fn test_conga_line_request() {

    const CODE: &'static str = "

    primitive start(out<u32> req, in<u32> resp, u32 num_nodes, u32 num_loops) {

        u32 loop_index = 0;

        u32 initial_value = 1337;

        while (loop_index < num_loops) {

            sync {

                put(req, initial_value);

                auto result = get(resp);

                assert(result == initial_value + num_nodes * 2);

            }

            loop_index += 1;

        }

    }

    primitive middle(

        in<u32> req_in, out<u32> req_forward,

        in<u32> resp_in, out<u32> resp_forward,

        u32 num_loops

    ) {

        u32 loop_index = 0;

        while (loop_index < num_loops) {

            sync {

                auto req = get(req_in);

                put(req_forward, req + 1);

                auto resp = get(resp_in);

                put(resp_forward, resp + 1);

            }

            loop_index += 1;

        }

    }

    primitive end(in<u32> req_in, out<u32> resp_out, u32 num_loops) {

        u32 loop_index = 0;

        while (loop_index < num_loops) {

            sync {

                auto req = get(req_in);

                put(resp_out, req);

            }

            loop_index += 1;

        }

    }

    composite constructor(u32 num_nodes, u32 num_loops) {

        channel initial_req -> req_in;

        channel resp_out -> final_resp;

        new start(initial_req, final_resp, num_nodes, num_loops);

        in<u32> last_req_in = req_in;

        out<u32> last_resp_out = resp_out;

        u32 node = 0;

        while (node < num_nodes) {

            channel new_req_fw -> new_req_in;

            channel new_resp_out -> new_resp_in;

            new middle(last_req_in, new_req_fw, new_resp_in, last_resp_out, num_loops);

            last_req_in = new_req_in;

            last_resp_out = new_resp_out;

            node += 1;

        }

        new end(last_req_in, last_resp_out, num_loops);

    }

    ";

    let thing = TestTimer::new("conga_line_request");

    run_test_in_runtime(CODE, |api| {

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

            Value::UInt32(5),

            Value::UInt32(NUM_LOOPS)

        ])).expect("create connector");

    });

}

// Testing speculation - Basic forms

use super::*;

#[test]

fn test_maybe_do_nothing() {

    // Three variants in which the behaviour in which nothing is performed is

    // somehow not allowed. Note that we "check" by seeing if the test finishes.

    // Only the branches in which ports fire increment the loop index

    const CODE: &'static str = "

    primitive only_puts(out<bool> output, u32 num_loops) {

        u32 index = 0;

        while (index < num_loops) {

            sync { put(output, true); }

            index += 1;

        }

    }

    primitive might_put(out<bool> output, u32 num_loops) {

        u32 index = 0;

        while (index < num_loops) {

            sync {

                fork { put(output, true); index += 1; }

                or   {}

            }

        }

    }

    primitive only_gets(in<bool> input, u32 num_loops) {

        u32 index = 0;

        while (index < num_loops) {

            sync { auto res = get(input); assert(res); }

            index += 1;

        }

    }

    primitive might_get(in<bool> input, u32 num_loops) {

        u32 index = 0;

        while (index < num_loops) {

            sync fork { auto res = get(input); assert(res); index += 1; } or {}

        }

    }

    ";

    // Construct all variants which should work and wait until the runtime exits

    run_test_in_runtime(CODE, |api| {

        // only putting -> maybe getting

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

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

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

            Value::UInt32(NUM_LOOPS),

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

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

            Value::UInt32(NUM_LOOPS),

        // maybe putting -> only getting

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

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

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

            Value::UInt32(NUM_LOOPS),

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

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

            Value::UInt32(NUM_LOOPS),

    })

}