mod mpmc_queue;

pub(crate) use mpmc_queue::MpmcQueue;

use crate::{PortId, ProtocolDescription};

use crate::runtime2::scheduler::{ComponentCtxFancy, Scheduler};

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

    PrivateInbox, PublicInbox,

    DataMessage, SyncMessage, SolutionMessage, Message, MessageContents,

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

    ports_delta: Vec<PortOwnershipDelta>,

#[derive(Clone)]

struct PortOwnershipDelta {

    acquired: bool, // if false, then released ownership

    port_id: PortIdLocal,

}

    // State and properties of connector itself

    in_sync: bool,

    pub inbox: PrivateInbox,

    ports_delta: &'a Vec<PortOwnershipDelta>,

        if self.ports_delta.iter().any(|v| v.port_id.index == port.0.u32_suffix) {

        if self.ports_delta.iter().any(|v| v.port_id == port_id) {

    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, comp_ctx: &mut ComponentCtxFancy, conn_ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {

        if self.in_sync {

            // Check for new messages we haven't seen before. If any of the

            // pending branches can accept the message, do so.

            while let Some((target_port_id, message)) = comp_ctx.read_next_message() {

                let mut branch_idx = self.sync_pending_get.first;

                while branch_idx != 0 {

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

                    let next_branch_idx = branch.next_branch_in_queue.unwrap_or(0);

                    let target_port_index = self.ports.get_port_index(*target_port_id).unwrap();

                    let port_mapping = self.ports.get_port(branch_idx, target_port_index);

                    if branch.sync_state == SpeculativeState::HaltedAtBranchPoint &&

                        branch.halted_at_port == *target_port_id &&

                        port_mapping.last_registered_branch_id == message.sender_prev_branch_id {

                        // Branch may accept this mesage, so create a fork that

                        // contains this message in the inbox.

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

                        let new_branch = Branch::new_sync_branching_from(new_branch_idx, branch);

                        self.ports.prepare_sync_branch(branch_idx, new_branch_idx);

                        let mapping = self.ports.get_port_mut(branch_idx, target_port_index);

                        mapping.last_registered_branch_id = message.sender_cur_branch_id;

                        let new_branch_id = BranchId::new(new_branch_idx);

                        self.branches.push(new_branch);

                        Self::push_branch_into_queue(&mut self.branches, &mut self.sync_active, new_branch_id)

                    }

                    branch_idx = next_branch_idx;

                }

            }

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

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

                        self.submit_sync_solution(valid_solution, conn_ctx, delta_state);

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

            in_sync: false,

            inbox: PrivateInbox::new(),

    pub fn is_in_sync_mode(&self) -> bool {

        return self.in_sync;

    }

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

        // self.inbox.insert_message(target_port, message);

    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

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

                    connector_id: ctx.id,

                        let port = ctx.get_port(port_id);

                self.submit_sync_solution(new_solution, ctx, results);

    fn handle_request_commit_message(&mut self, mut message: SolutionMessage, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {

                    if *connector_id != ctx.id {

                self.handle_confirm_commit_message(message.clone(), ctx, delta_state);

                delta_state.outbox.push(MessageContents::ConfirmCommit(message));

                delta_state.outbox.push(MessageContents::RequestCommit(message));

    fn handle_confirm_commit_message(&mut self, message: SolutionMessage, ctx: &ConnectorCtx, _delta_state: &mut RunDeltaState) {

            .find(|(id, _)| *id == ctx.id)

        self.in_sync = false;

        let solution = &mut self.branches[0];

        self.inbox.clear();

        for port_delta in &solution.ports_delta {

            if port_delta.acquired {

                self.ports.add_port(port_delta.port_id);

                self.ports.remove_port(port_delta.port_id);

        solution.commit_to_sync();

    pub fn run_in_speculative_mode(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy, _context: &ConnectorCtx, results: &mut RunDeltaState) -> ConnectorScheduling {

        debug_assert!(self.in_sync);

                    let messages = comp_ctx.get_read_messages(local_port_id, port_mapping.last_registered_branch_id);

                        debug_assert!(results.ports.is_empty());

                        find_ports_in_value_group(&message.message, &mut results.ports);

                        Self::acquire_ports_during_sync(&mut self.ports, &mut new_branch, &results.ports);

                        results.ports.clear();

                    debug_assert!(results.ports.is_empty());

                    find_ports_in_value_group(&message.message, &mut results.ports);

                    Self::release_ports_during_sync(&mut self.ports, branch, &results.ports).unwrap();

                    results.ports.clear();

                    results.outbox.push(MessageContents::Data(message));

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

        debug_assert!(!self.in_sync);

                self.in_sync = true;

                debug_assert!(results.ports.is_empty());

                find_ports_in_value_group(&arguments, &mut results.ports);

                if !results.ports.is_empty() {

                    if let Err(_) = Self::release_ports_during_non_sync(&mut self.ports, branch, &results.ports) {

                let new_connector_ports = results.ports.clone(); // TODO: Do something with this

                let new_connector = ConnectorPDL::new(new_connector_branch, new_connector_ports);

                results.new_connectors.push(new_connector);

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

                results.new_ports.push(putter);

                results.new_ports.push(getter);

                        if delta.port_id == *port_id {

                            debug_assert!(!delta.acquired);

                    branch.ports_delta.push(PortOwnershipDelta{

                        acquired: false,

                        port_id: *port_id,

                    });

                        if delta.port_id == *port_id {

                            debug_assert!(delta.acquired);

                if delta.port_id == *port_id {

                    if delta.acquired {

            branch.ports_delta.push(PortOwnershipDelta{

                acquired: true,

                port_id: *port_id,

            });

    fn generate_initial_solution_for_branch(&self, branch_id: BranchId, ctx: &ConnectorCtx) -> Option<SyncMessage> {

        // Retrieve branchg

            connector_id: ctx.id,

            if !sync_message.check_constraint(ctx.id, SyncBranchConstraint::SilentPort(port_delta.port_id)).unwrap() {

            let port = ctx.get_port(port_delta.port_id);

            let port = ctx.get_port(port_id);

    fn submit_sync_solution(&mut self, partial_solution: SyncMessage, ctx: &ConnectorCtx, results: &mut RunDeltaState) {

                connector_origin: ctx.id,

                if local_solution.connector_id != ctx.id {

            results.outbox.push(MessageContents::RequestCommit(full_solution));

            results.outbox.push(MessageContents::Sync(partial_solution));

/// A data structure passed to a connector whose code is being executed that is

/// used to queue up various state changes that have to be applied after

/// running, e.g. the messages the have to be transferred to other connectors.

// TODO: Come up with a better name

pub(crate) struct RunDeltaState {

    // Variables that allow the thread running the connector to pick up global

    // state changes and try to apply them.

    pub outbox: Vec<MessageContents>,

    pub new_connectors: Vec<ConnectorPDL>,

    pub new_ports: Vec<Port>,

    // Workspaces

    pub ports: Vec<PortIdLocal>,

}

impl RunDeltaState {

    /// Constructs a new `RunDeltaState` object with the default amount of

    /// reserved memory

    pub fn new() -> Self {

        RunDeltaState{

            outbox: Vec::with_capacity(64),

            new_connectors: Vec::new(),

            new_ports: Vec::new(),

            ports: Vec::with_capacity(64),

        }

    }

}

pub(crate) struct PrivateInbox {

    // "Normal" messages, intended for a PDL protocol. These need to stick

    // around during an entire sync-block (to handle `put`s for which the

    // corresponding `get`s have not yet been reached).

    messages: Vec<(PortIdLocal, DataMessage)>,

    len_read: usize,

}

impl PrivateInbox {

    pub fn new() -> Self {

        Self{

            messages: Vec::new(),

            len_read: 0,

        }

    }

    /// Will insert the message into the inbox. Only exception is when the tuple

    /// (prev_branch_id, cur_branch_id, receiving_port_id) already exists, then

    /// nothing is inserted..

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

        for (existing_target_port, existing) in self.messages.iter() {

            if existing.sender_prev_branch_id == message.sender_prev_branch_id &&

                    existing.sender_cur_branch_id == message.sender_cur_branch_id &&

                    *existing_target_port == target_port {

                // Message was already received

                return;

            }

        }

        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.

    pub(crate) fn next_message(&mut self) -> Option<(&PortIdLocal, &DataMessage)> {

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

            return None;

        }

        let (target_port, message) = &self.messages[self.len_read];

        self.len_read += 1;

        return Some((target_port, message));

    }

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

            let (target_port, cur_message) = &self.messages[self.next_index];

            if *target_port == self.match_port_id && cur_message.sender_prev_branch_id == self.match_prev_branch_id {

                // Found a match

                break;

            }

            self.next_index += 1;

        }

        if self.next_index == self.max_index {

            return None;

        }

        let (_, message) = &self.messages[self.next_index];

        self.next_index += 1;

        return Some(message);

    }

}

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

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

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

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

            ConnectorVariant::UserDefined(c) => c.handle_message(message, ctx, delta_state),

            ConnectorVariant::Native(c) => c.handle_message(message, ctx, delta_state),

        }

    }

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

        match self {

            ConnectorVariant::UserDefined(c) => c.run(scheduler_ctx, conn_ctx, delta_state),

            ConnectorVariant::Native(c) => c.run(scheduler_ctx, conn_ctx, delta_state),

    pub context: ConnectorCtx, // mutable access by scheduler, immutable by connector

    pub(crate) fn create_interface_component(&self, component: ConnectorApplication) -> ConnectorKey {