return self.index != 0;

    }

}

#[derive(Debug, PartialEq, Eq)]

pub(crate) enum SpeculativeState {

    // Non-synchronous variants

    RunningNonSync,         // regular execution of code

    Error,                  // encountered a runtime error

    Finished,               // finished executing connector's code

    // Synchronous variants

    RunningInSync,          // running within a sync block

    HaltedAtBranchPoint,    // at a branching point (at a `get` call)

    ReachedSyncEnd,         // reached end of sync block, branch represents a local solution

    Inconsistent,           // branch can never represent a local solution, so halted

}

pub(crate) struct Branch {

    index: BranchId,

    parent_index: BranchId,

    // Code execution state

    code_state: ComponentState,

    prepared_channel: Option<(Value, Value)>,

    sync_state: SpeculativeState,

    next_branch_in_queue: Option<u32>,

    // Message/port state

    received: HashMap<PortIdLocal, DataMessage>, // TODO: @temporary, remove together with fires()

    ports_delta: Vec<PortOwnershipDelta>,

}

impl Branch {

    /// Constructs a non-sync branch. It is assumed that the code is at the

    /// first instruction

    pub(crate) fn new_initial_branch(component_state: ComponentState) -> Self {

        Branch{

            index: BranchId::new_invalid(),

            parent_index: BranchId::new_invalid(),

            code_state: component_state,

            prepared_channel: None,

            sync_state: SpeculativeState::RunningNonSync,

            next_branch_in_queue: None,

            received: HashMap::new(),

            ports_delta: Vec::new(),

        }

    }

    /// Constructs a sync branch. The provided branch is assumed to be the

    /// parent of the new branch within the execution tree.

    fn new_sync_branching_from(new_index: u32, parent_branch: &Branch) -> Self {

        debug_assert!(

            (parent_branch.sync_state == SpeculativeState::RunningNonSync && !parent_branch.parent_index.is_valid()) ||

            (parent_branch.sync_state == SpeculativeState::HaltedAtBranchPoint)

        );

        debug_assert!(parent_branch.prepared_channel.is_none());

        Branch{

            index: BranchId::new(new_index),

            parent_index: parent_branch.index,

            code_state: parent_branch.code_state.clone(),

            prepared_channel: None,

            sync_state: SpeculativeState::RunningInSync,

            next_branch_in_queue: None,

            received: parent_branch.received.clone(),

            ports_delta: parent_branch.ports_delta.clone(),

        }

    }

    fn commit_to_sync(&mut self) {

        self.parent_index = BranchId::new_invalid();

        self.sync_state = SpeculativeState::RunningNonSync;

        self.next_branch_in_queue = None;

        self.received.clear();

        self.ports_delta.clear();

    }

}

#[derive(Clone)]

struct PortAssignment {

    is_assigned: bool,

    last_registered_branch_id: BranchId, // invalid branch ID implies not assigned yet

    num_times_fired: u32,

}

impl PortAssignment {

    fn new_unassigned() -> Self {

        Self{

            is_assigned: false,

            last_registered_branch_id: BranchId::new_invalid(),

            num_times_fired: 0,

        }

    }

            return None;

        }

    }

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

        return self.prepared_channel.take();

    }

}

impl Connector for ConnectorPDL {

    fn handle_message(&mut self, message: Message, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {

        use MessageContents as MC;

        match message.contents {

            MC::Data(content) => self.handle_data_message(message.receiving_port, content),

            MC::Sync(content) => self.handle_sync_message(content, ctx, delta_state),

            MC::RequestCommit(content) => self.handle_request_commit_message(content, ctx, delta_state),

            MC::ConfirmCommit(content) => self.handle_confirm_commit_message(content, ctx, delta_state),

            MC::Control(_) | MC::Ping => {},

        }

    }

    fn run(&mut self, sched_ctx: SchedulerCtx, conn_ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {

        if self.in_sync {

            let scheduling = self.run_in_speculative_mode(sched_ctx, conn_ctx, delta_state);

            // When in speculative mode we might have generated new sync

            // solutions, we need to turn them into proposed solutions here.

            if self.sync_finished_last_handled != self.sync_finished.last {

                // Retrieve first element in queue

                let mut next_id;

                if self.sync_finished_last_handled == 0 {

                    next_id = self.sync_finished.first;

                } else {

                    let last_handled = &self.branches[self.sync_finished_last_handled as usize];

                    debug_assert!(last_handled.next_branch_in_queue.is_some()); // because "last handled" != "last in queue"

                    next_id = last_handled.next_branch_in_queue.unwrap();

                }

                loop {

                    let branch_id = BranchId::new(next_id);

                    let branch = &self.branches[next_id as usize];

                    let branch_next = branch.next_branch_in_queue;

                    // Turn local solution into a message and send it along

                    // TODO: Like `ports` access, also revise the construction of this `key`, should not be needed

                    let solution_message = self.generate_initial_solution_for_branch(branch_id, conn_ctx);

                    if let Some(valid_solution) = solution_message {

    /// hence is in a non-sync state.

    pub fn new(initial_branch: Branch, owned_ports: Vec<PortIdLocal>) -> Self {

        Self{

            in_sync: false,

            branches: vec![initial_branch],

            sync_active: BranchQueue::new(),

            sync_pending_get: BranchQueue::new(),

            sync_finished: BranchQueue::new(),

            sync_finished_last_handled: 0, // none at all

            cur_round: 0,

            committed_to: None,

            inbox: PrivateInbox::new(),

            ports: ConnectorPorts::new(owned_ports),

        }

    }

    pub fn is_in_sync_mode(&self) -> bool {

        return self.in_sync;

    }

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

    // Handling connector messages

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

    pub fn handle_data_message(&mut self, target_port: PortIdLocal, message: DataMessage) {

        self.inbox.insert_message(target_port, message);

    }

    /// Accepts a synchronous message and combines it with the locally stored

    /// solution(s). Then queue new `Sync`/`Solution` messages when appropriate.

    pub fn handle_sync_message(&mut self, message: SyncMessage, ctx: &ConnectorCtx, results: &mut RunDeltaState) {

        debug_assert!(!message.to_visit.contains(&ctx.id)); // own ID already removed

        debug_assert!(message.constraints.iter().any(|v| v.connector_id == ctx.id)); // we have constraints

        // TODO: Optimize, use some kind of temp workspace vector

        let mut execution_path_branch_ids = Vec::new();

        if self.sync_finished_last_handled != 0 {

            // We have some solutions to match against

            let constraints_index = message.constraints

                .iter()

                .position(|v| v.connector_id == ctx.id)

                .unwrap();

            let constraints = &message.constraints[constraints_index].constraints;

            debug_assert!(!constraints.is_empty());

            // Note that we only iterate over the solutions we've already

            // handled ourselves, not necessarily

                // Branch performed a `get` on a port that has not yet received

                // a value in its inbox.

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

                let local_port_index = self.ports.get_port_index(local_port_id);

                if local_port_index.is_none() {

                    todo!("deal with the case where the port is acquired");

                }

                let local_port_index = local_port_index.unwrap();

                let port_mapping = self.ports.get_port_mut(branch.index.index, local_port_index);

                // Check for port mapping assignment and, if present, if it is

                // consistent

                let is_valid_get = if port_mapping.is_assigned {

                    assert!(port_mapping.num_times_fired <= 1); // temporary, until we get rid of `fires`

                    port_mapping.num_times_fired == 1

                } else {

                    // Not yet assigned

                    port_mapping.mark_speculative(1);

                    true

                };

                if is_valid_get {

                    // Mark as a branching point for future messages

                    branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

                    let branch_id = branch.index;

                    Self::push_branch_into_queue(&mut self.branches, &mut self.sync_pending_get, branch_id);

                    // But if some messages can be immediately applied, do so

                    // now.

                    let messages = self.inbox.get_messages(local_port_id, port_mapping.last_registered_branch_id);

                    let mut did_have_messages = false;

                    for message in messages {

                        did_have_messages = true;

                        // For each message prepare a new branch to execute

                        let parent_branch = &self.branches[branch_id.index as usize];

                        let new_branch_index = self.branches.len() as u32;

                        let mut new_branch = Branch::new_sync_branching_from(new_branch_index, parent_branch);

                        self.ports.prepare_sync_branch(branch_id.index, new_branch_index);

                        let port_mapping = self.ports.get_port_mut(new_branch_index, local_port_index);

                        port_mapping.last_registered_branch_id = message.sender_cur_branch_id;

                        debug_assert!(port_mapping.is_assigned && port_mapping.num_times_fired == 1);

                        new_branch.received.insert(local_port_id, message.clone());

                        // If the message contains any ports then they will now

        self.messages.push((target_port, message));

    }

    /// Retrieves all previously read messages that satisfy the provided

    /// speculative conditions. Note that the inbox remains read-locked until

    /// the returned iterator is dropped. Should only be called by the

    /// inbox-reader (i.e. the thread executing a connector's PDL code).

    ///

    /// This function should only be used to check if already-received messages

    /// could be received by a newly encountered `get` call in a connector's

    /// PDL code.

    pub(crate) fn get_messages(&self, port_id: PortIdLocal, prev_branch_id: BranchId) -> InboxMessageIter {

        return InboxMessageIter {

            messages: &self.messages,

            next_index: 0,

            max_index: self.len_read,

            match_port_id: port_id,

            match_prev_branch_id: prev_branch_id,

        };

    }

    /// Retrieves the next unread message. Should only be called by the

    /// inbox-reader.

        if self.len_read == self.messages.len() {

            return None;

        }

        self.len_read += 1;

    }

    /// Simply empties the inbox

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

        self.messages.clear();

        self.len_read = 0;

    }

}

/// Iterator over previously received messages in the inbox.

pub(crate) struct InboxMessageIter<'i> {

    messages: &'i Vec<(PortIdLocal, DataMessage)>,

    next_index: usize,

    max_index: usize,

    match_port_id: PortIdLocal,

    match_prev_branch_id: BranchId,

}

impl<'i> Iterator for InboxMessageIter<'i> {

    type Item = &'i 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 {

mod runtime;

mod messages;

mod connector;

mod native;

mod port;

mod scheduler;

mod inbox;

#[cfg(test)] mod tests;

// Imports

use std::collections::VecDeque;

use std::sync::{Arc, Condvar, Mutex, RwLock};

use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};

use std::thread::{self, JoinHandle};

use crate::collections::RawVec;

use crate::ProtocolDescription;

use inbox::Message;

use connector::{ConnectorPDL, ConnectorPublic, ConnectorScheduling, RunDeltaState};

use scheduler::{Scheduler, ConnectorCtx, ControlMessageHandler};

use native::{Connector, ConnectorApplication, ApplicationInterface};

use crate::runtime2::scheduler::SchedulerCtx;

/// A kind of token that, once obtained, allows mutable access to a connector.

/// We're trying to use move semantics as much as possible: the owner of this

/// key is the only one that may execute the connector's code.

pub(crate) struct ConnectorKey {

    pub index: u32, // of connector

}

impl ConnectorKey {

    /// Downcasts the `ConnectorKey` type, which can be used to obtain mutable

    /// access, to a "regular ID" which can be used to obtain immutable access.

    #[inline]

    pub fn downcast(&self) -> ConnectorId {

        return ConnectorId(self.index);

    }

    /// Turns the `ConnectorId` into a `ConnectorKey`, marked as unsafe as it

    /// bypasses the type-enforced `ConnectorKey`/`ConnectorId` system

    #[inline]

    pub unsafe fn from_id(id: ConnectorId) -> ConnectorKey {

        return ConnectorKey{ index: id.0 };

    }

}

        return lock.pop_front();

    }

    pub(crate) fn push_work(&self, key: ConnectorKey) {

        let mut lock = self.connector_queue.lock().unwrap();

        lock.push_back(key);

        self.scheduler_notifier.notify_one();

    }

    // --- Creating/using ports

    /// Creates a new port pair. Note that these are stored globally like the

    /// connectors are. Ports stored by components belong to those components.

    pub(crate) fn create_channel(&self, creating_connector: ConnectorId) -> (Port, Port) {

        use port::{PortIdLocal, PortKind};

        let getter_id = self.port_counter.fetch_add(2, Ordering::SeqCst);

        let putter_id = PortIdLocal::new(getter_id + 1);

        let getter_id = PortIdLocal::new(getter_id);

        let getter_port = Port{

            self_id: getter_id,

            peer_id: putter_id,

            kind: PortKind::Getter,

            peer_connector: creating_connector,

        };

        let putter_port = Port{

            self_id: putter_id,

            peer_id: getter_id,

            kind: PortKind::Putter,

            peer_connector: creating_connector,

        };

        return (getter_port, putter_port);

    }

    /// Sends a message to a particular connector. If the connector happened to

    /// be sleeping then it will be scheduled for execution.

    pub(crate) fn send_message(&self, target_id: ConnectorId, message: Message) {

        let target = self.get_component_public(target_id);

        target.inbox.insert_message(message);

        let should_wake_up = target.sleeping

            .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)

            .is_ok();

        if should_wake_up {

            let key = unsafe{ ConnectorKey::from_id(target_id) };

            self.push_work(key);

        }

    }

    // --- Creating/retrieving/destroying components

        self.increment_active_components();

        return key;

    }

    /// Creates a new PDL component. The caller MUST make sure to schedule the

    /// connector.

    // TODO: Nicer code, not forcing the caller to schedule, perhaps?

    pub(crate) fn create_pdl_component(&self, created_by: &mut ScheduledConnector, connector: ConnectorPDL) -> ConnectorKey {

        // Create as not sleeping, as we'll schedule it immediately

        let key = {

            let mut lock = self.connectors.write().unwrap();

            lock.create(ConnectorVariant::UserDefined(connector), false)

        };

        // Transfer the ports

        {

            let lock = self.connectors.read().unwrap();

            let created = lock.get_private(&key);

            match &created.connector {

                ConnectorVariant::UserDefined(connector) => {

                    for port_id in connector.ports.owned_ports.iter().copied() {

                        println!("DEBUG: Transferring port {:?} from {} to {}", port_id, created_by.context.id.0, key.index);

                        created.context.add_port(port);

                    }

                },

                ConnectorVariant::Native(_) => unreachable!(),

            }

        }

        self.increment_active_components();

        return key;

    }

    #[inline]

    pub(crate) fn get_component_private(&self, connector_key: &ConnectorKey) -> &'static mut ScheduledConnector {

        let lock = self.connectors.read().unwrap();

        return lock.get_private(connector_key);

    }

    #[inline]

    pub(crate) fn get_component_public(&self, connector_id: ConnectorId) -> &'static ConnectorPublic {

        let lock = self.connectors.read().unwrap();

        return lock.get_public(connector_id);

    }

    pub(crate) fn destroy_component(&self, connector_key: ConnectorKey) {

    }

    pub(crate) fn decrement_active_interfaces(&self) {

        let old_num = self.active_interfaces.fetch_sub(1, Ordering::SeqCst);

        println!("DEBUG: Decremented active interfaces to {}", old_num - 1);

        debug_assert!(old_num > 0);

        if old_num == 1 { // such that active interfaces is now 0

            let num_connectors = self.active_connectors.load(Ordering::Acquire);

            if num_connectors == 0 {

                self.signal_for_shutdown();

            }

        }

    }

    #[inline]

    fn increment_active_components(&self) {

        let _old_num = self.active_connectors.fetch_add(1, Ordering::SeqCst);

        println!("DEBUG: Incremented components to {}", _old_num + 1);

    }

    fn decrement_active_components(&self) {

        let old_num = self.active_connectors.fetch_sub(1, Ordering::SeqCst);

        println!("DEBUG: Decremented components to {}", old_num - 1);

        debug_assert!(old_num > 0);

            let num_interfaces = self.active_interfaces.load(Ordering::Acquire);

            if num_interfaces == 0 {

                self.signal_for_shutdown();

            }

        }

    }

    #[inline]

    fn signal_for_shutdown(&self) {

        debug_assert_eq!(self.active_interfaces.load(Ordering::Acquire), 0);

        debug_assert_eq!(self.active_connectors.load(Ordering::Acquire), 0);

        println!("DEBUG: Signaling for shutdown");

        let _lock = self.connector_queue.lock().unwrap();

        let should_signal = self.should_exit

            .compare_exchange(false, true, Ordering::SeqCst, Ordering::Acquire)

            .is_ok();

        if should_signal {

            println!("DEBUG: Notifying all waiting schedulers");

            self.scheduler_notifier.notify_all();

        }

    }

}

            debug_assert!(!connector.is_null());

            return &(**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 connector = self.connectors.get_mut(key.index as usize);

            debug_assert!(!connector.is_null());

            return &mut (**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,

            context: ConnectorCtx::new(),

            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.connectors.len();

            key = ConnectorKey{ index: index as u32 };

            connector.context.id = key.downcast();

            let connector = Box::into_raw(Box::new(connector));

            self.connectors.push(connector);

        } else {

            // Free spot available

            index = self.free.pop().unwrap();

            key = ConnectorKey{ index: index as u32 };

            connector.context.id = key.downcast();

            unsafe {

                let target = self.connectors.get_mut(index);

                std::ptr::write(*target, connector);

            }

        }

use std::sync::Arc;

use std::sync::atomic::Ordering;

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

use super::port::{Port, PortState, PortIdLocal};

use super::native::Connector;

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

use super::inbox::{Message, MessageContents, ControlMessageVariant, ControlMessage};

/// Contains fields that are mostly managed by the scheduler, but may be

/// accessed by the connector

pub(crate) struct ConnectorCtx {

    pub(crate) id: ConnectorId,

    pub(crate) ports: Vec<Port>,

}

impl ConnectorCtx {

    pub(crate) fn new() -> ConnectorCtx {

        Self{

            id: ConnectorId::new_invalid(),

            ports: Vec::new(),

        }

    }

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

        debug_assert!(!self.ports.iter().any(|v| v.self_id == port.self_id));

// Because it contains pointers we're going to do a copy by value on this one

#[derive(Clone, Copy)]

pub(crate) struct SchedulerCtx<'a> {

    pub(crate) runtime: &'a RuntimeInner

}

pub(crate) struct Scheduler {

    runtime: Arc<RuntimeInner>,

    scheduler_id: u32,

}

impl Scheduler {

    pub fn new(runtime: Arc<RuntimeInner>, scheduler_id: u32) -> Self {

        return Self{ runtime, scheduler_id };

    }

    pub fn run(&mut self) {

        // Setup global storage and workspaces that are reused for every

        // connector that we run

        let scheduler_id = self.scheduler_id;

        let mut delta_state = RunDeltaState::new();

        'thread_loop: loop {

            // Retrieve a unit of work

            let connector_key = self.runtime.wait_for_work();

            if connector_key.is_none() {

                // We should exit

                break 'thread_loop;

            }

            // We have something to do

            let connector_key = connector_key.unwrap();

            let connector_id = connector_key.downcast();

            let scheduled = self.runtime.get_component_private(&connector_key);

            // Keep running until we should no longer immediately schedule the

            // connector.

            let mut cur_schedule = ConnectorScheduling::Immediate;

            while cur_schedule == ConnectorScheduling::Immediate {

                // Check all the message that are in the shared inbox

                while let Some(message) = scheduled.public.inbox.take_message() {

                    // Check for rerouting

                    if let Some(other_connector_id) = scheduled.router.should_reroute(message.sending_connector, message.receiving_port) {

                        continue;

                    }

                    // Check for messages that requires special action from the

                    // scheduler.

                    if let MessageContents::Control(content) = message.contents {

                        match content.content {

                            ControlMessageVariant::ChangePortPeer(port_id, new_target_connector_id) => {

                                // Need to change port target

                                let port = scheduled.context.get_port_mut(port_id);

                                port.peer_connector = new_target_connector_id;

                                debug_assert!(delta_state.outbox.is_empty());

                                // And respond with an Ack

                                self.runtime.send_message(

                                    message.sending_connector,

                                    Message{

                                        sending_connector: connector_id,

                                        receiving_port: PortIdLocal::new_invalid(),

                                        contents: MessageContents::Control(ControlMessage{

                                            id: content.id,

                                            content: ControlMessageVariant::Ack,

                                        }),

                                    }

                                );

                            },

                            ControlMessageVariant::CloseChannel(port_id) => {

                                // Mark the port as being closed

                                let port = scheduled.context.get_port_mut(port_id);

                                port.state = PortState::Closed;

                                // Send an Ack

                                self.runtime.send_message(

                                    message.sending_connector,

                                    Message{

                                        sending_connector: connector_id,

                                        receiving_port: PortIdLocal::new_invalid(),

                                        contents: MessageContents::Control(ControlMessage{

                                            id: content.id,

                                            content: ControlMessageVariant::Ack,

                                        }),

                                    }

                                );

                            },

                            ControlMessageVariant::Ack => {

                                scheduled.router.handle_ack(content.id);

                            }

                        }

                    } else {

                        // Let connector handle message

                        scheduled.connector.handle_message(message, &scheduled.context, &mut delta_state);

                    }

                }

                // Run the main behaviour of the connector, depending on its

                // current state.

                if scheduled.shutting_down {

                    // Nothing to do. But we're stil waiting for all our pending

                    // control messages to be answered.

                    if scheduled.router.num_pending_acks() == 0 {

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

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

                    let new_schedule = scheduled.connector.run(

                        scheduler_ctx, &scheduled.context, &mut delta_state

                    );

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

                    // send and connectors to instantiate.

                    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 {

                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`.

                },

                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.context.ports {

                    }

                }

            }

        }

    }

        // Handling any messages that were sent

        if !delta_state.outbox.is_empty() {

            for mut message in delta_state.outbox.drain(..) {

                // Based on the message contents, decide where the message

                // should be sent to. This might end up modifying the message.

                    MessageContents::Data(contents) => {