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

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

use crate::{PortId, ProtocolDescription};

use crate::common::ComponentState;

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

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

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

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

use super::native::Connector;

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

use super::scheduler::{ComponentCtx, SchedulerCtx, MessageTicket};

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

        }

    }

}

#[derive(Debug, PartialEq, Eq, Clone, Copy)]

enum Mode {

    NonSync,    // running non-sync code

    Sync,       // running sync code (in potentially multiple branches)

    SyncError,  // encountered an unrecoverable error in sync mode

    Error,      // encountered an error in non-sync mode (or finished handling the sync mode error).

}

#[derive(Debug)]

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 {

    mode: Mode,

    eval_error: Option<EvalError>,

    tree: ExecTree,

    consensus: Consensus,

    last_finished_handled: Option<BranchId>,

}

// TODO: Remove remaining fields once 'fires()' is removed from language.

struct ConnectorRunContext<'a> {

    branch_id: BranchId,

    consensus: &'a Consensus,

    prepared: PreparedStatement,

}

            EvalContinuation::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_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_prompt);

                comp_ctx.push_component(new_component, comp_ctx.workspace_ports.clone());

                comp_ctx.workspace_ports.clear();

                return ConnectorScheduling::Later;

            },

            EvalContinuation::NewChannel => {

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

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

                branch.prepared = PreparedStatement::CreatedChannel((

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

                    Value::Input(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),

        }

    }

    /// Helper that moves the component's state back into non-sync mode, using

    /// the provided solution branch ID as the branch that should be comitted to

    /// memory. If this function returns false, then the component is supposed

    /// to exit.

    fn enter_non_sync_mode(&mut self, conclusion: RoundConclusion, ctx: &mut ComponentCtx) -> ConnectorScheduling {

        debug_assert!(self.mode == Mode::Sync || self.mode == Mode::SyncError);

        // Depending on local state decide what to do

        let final_branch_id = match conclusion {

            RoundConclusion::Success(branch_id) => Some(branch_id),

            RoundConclusion::Failure => None,

        };

        if let Some(solution_branch_id) = final_branch_id {

            let mut fake_vec = Vec::new();

            self.tree.end_sync(solution_branch_id);

            self.consensus.end_sync(solution_branch_id, &mut fake_vec);

            debug_assert!(fake_vec.is_empty());

            ctx.notify_sync_end(&[]);

            self.last_finished_handled = None;

            self.eval_error = None; // in case we came from the SyncError mode

            self.mode = Mode::NonSync;

            return ConnectorScheduling::Immediate;

        } else {

            // No final branch, because we're supposed to exit!

            self.last_finished_handled = None;

            self.mode = Mode::Error;

            if let Some(eval_error) = self.eval_error.take() {

                ctx.push_error(eval_error);

            }

            return ConnectorScheduling::Exit;

        }

    }

    /// Runs the prompt repeatedly until some kind of execution-blocking

    /// condition appears.

    #[inline]

    fn run_prompt(prompt: &mut Prompt, pd: &ProtocolDescription, ctx: &mut ConnectorRunContext) -> Result<EvalContinuation, EvalError> {

        loop {

            let result = prompt.step(&pd.types, &pd.heap, &pd.modules, ctx);

            if let Ok(EvalContinuation::Stepping) = result {

                continue;

            }

            return result;

        }

    }

}

    CloseChannel(PortIdLocal),

    Ack,

    Ping,

}

/// Combination of data message and control messages.

#[derive(Debug)]

pub(crate) enum Message {

    Data(DataMessage),

    SyncComp(SyncCompMessage),

    SyncPort(SyncPortMessage),

    SyncControl(SyncControlMessage),

    Control(ControlMessage),

}

impl Message {

    /// If the message is sent through a particular channel, then this function

    /// returns the port through which the message was sent.

    pub(crate) fn source_port(&self) -> Option<PortIdLocal> {

        // Currently only data messages have a source port

        match self {

            Message::Data(message) => return Some(message.data_header.sending_port),

            Message::SyncPort(message) => return Some(message.source_port),

            Message::SyncComp(_) => return None,

            Message::SyncControl(_) => return None,

            Message::Control(_) => return None,

        }

    }

    /// If the message is sent through a particular channel, then this function

    /// returns the target port through which the message was sent.

    pub(crate) fn target_port(&self) -> Option<PortIdLocal> {

        match self {

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

            Message::SyncPort(message) => return Some(message.target_port),

            Message::SyncComp(_) => return None,

            Message::SyncControl(_) => return None,

            Message::Control(message) => {

                match &message.content {

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

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

                    ControlContent::Ping => return None,

                    ControlContent::Ack => return None,

                }

            }

        }

    }

    pub(crate) fn as_data(&self) -> &DataMessage {

        match self {

            Message::Data(v) => v,

            _ => unreachable!(),

        }

    }

}

/// The public inbox of a connector. The thread running the connector that owns

/// this inbox may retrieved from it. Non-owning threads may only put new

/// messages inside of it.

// TODO: @Optimize, lazy concurrency. Probably ringbuffer with read/write heads.

//  Should behave as a MPSC queue.

pub struct PublicInbox {

    messages: Mutex<VecDeque<Message>>,

}

impl PublicInbox {

    pub fn new() -> Self {

        Self{

            messages: Mutex::new(VecDeque::new()),

        }

    }

    pub(crate) fn insert_message(&self, message: Message) {

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

        lock.push_back(message);

    }

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

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

        return lock.pop_front();

    }

    pub fn is_empty(&self) -> bool {

        let lock = self.messages.lock().unwrap();

        return lock.is_empty();

    }

}

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

        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 {

            self.scheduler_notifier.notify_all();

        }

    }

}

// TODO: Come back to this at some point

unsafe impl Send for RuntimeInner {}

unsafe impl Sync for RuntimeInner {}

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

// ConnectorStore

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

struct StoreEntry {

    connector: ScheduledConnector,

    generation: std::sync::atomic::AtomicU32,

}

struct ConnectorStore {

    // Freelist storage of connectors. Storage should be pointer-stable as

    // someone might be mutating the vector while we're executing one of the

    // connectors.

    entries: RawVec<*mut StoreEntry>,

    free: Vec<usize>,

}

impl ConnectorStore {

    fn with_capacity(capacity: usize) -> Self {

        Self {

            entries: RawVec::with_capacity(capacity),

            free: Vec::with_capacity(capacity),

        }

    }

    /// Retrieves public part of connector - accessible by many threads at once.

    fn get_public(&self, id: ConnectorId) -> &'static ConnectorPublic {

        unsafe {

            let entry = self.entries.get(id.index as usize);

            debug_assert!(!entry.is_null());

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

            debug_assert!(!entry.is_null());

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

            ctx: ComponentCtx::new_empty(),

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

            key = ConnectorKey{

                index: index as u32, generation: 0

            };

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

            let connector = Box::into_raw(Box::new(StoreEntry{

                connector,

                generation: AtomicU32::new(0),

            }));

            self.entries.push(connector);

        } else {

            // Free spot available

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

            unsafe {

                let target = &mut **self.entries.get_mut(index);

                std::ptr::write(&mut target.connector as *mut _, connector);

                let generation = target.generation.load(Ordering::Acquire);

                key = ConnectorKey{ index: index as u32, generation };

                target.connector.ctx.id = key.downcast();

            }

        }

        println!("DEBUG [ global store  ] Created component at {}", key.index);

        return key;

    }

                    self.debug_conn(connector_id, &format!("Shutting down, {} Acks remaining", scheduled.router.num_pending_acks()));

                    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 {

                    self.debug_conn(connector_id, "Running ...");

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

                    let new_schedule = scheduled.connector.run(scheduler_ctx, &mut scheduled.ctx);

                    self.debug_conn(connector_id, &format!("Finished running (new scheduling is {:?})", new_schedule));

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

                    // send and connectors to instantiate.

                    self.handle_changes_in_context(scheduled);

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

                    self.try_go_to_sleep(connector_key, scheduled);

                },

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

                        if port.state != PortState::Closed {

                            let message = scheduled.router.prepare_closing_channel(

                                port.self_id, port.peer_id,

                                connector_id

                            );

                            self.runtime.send_message(port.peer_connector, Message::Control(message));

                        }

                    }

                    // Any messages still in the public inbox should be handled

                    scheduled.ctx.inbox.clear_read_messages();

                    while let Some(ticket) = scheduled.ctx.get_next_message_ticket_even_if_not_in_sync() {

                        let message = scheduled.ctx.take_message_using_ticket(ticket);

                        self.handle_message_while_shutting_down(message, scheduled);

                    }

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

                        // All ports (if any) already closed

                        self.runtime.destroy_component(connector_key);

                        continue 'thread_loop;

                    }

                    self.try_go_to_sleep(connector_key, scheduled);

                },

            }

        }

    }

    /// Receiving messages from the public inbox and handling them or storing

    /// them in the component's private inbox

    fn handle_inbox_messages(&mut self, scheduled: &mut ScheduledConnector) {

        let connector_id = scheduled.ctx.id;

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

            // Check if the message has to be rerouted because we have moved the

            // target port to another component.

            self.debug_conn(connector_id, &format!("Handling message\n --- {:#?}", message));

            if let Some(target_port) = message.target_port() {

                if let Some(other_component_id) = scheduled.router.should_reroute(target_port) {

                    self.debug_conn(connector_id, " ... Rerouting the message");

                    self.runtime.send_message(other_component_id, message);

                    continue;

                }

            }

            // If here, then we should handle the message

            self.debug_conn(connector_id, " ... Handling the message");

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

                match message.content {

                    ControlContent::PortPeerChanged(port_id, new_target_connector_id) => {

                        // Need to change port target

                        let port = scheduled.ctx.get_port_mut_by_id(port_id).unwrap();

                        port.peer_connector = new_target_connector_id;

                        // Note: for simplicity we program the scheduler to always finish

                        // running a connector with an empty outbox. If this ever changes

                        // then accepting the "port peer changed" message implies we need

                        // to change the recipient of the message in the outbox.

                        debug_assert!(scheduled.ctx.outbox.is_empty());

                        // And respond with an Ack

                        let ack_message = Message::Control(ControlMessage {

                            id: message.id,

                            sending_component_id: connector_id,

                            content: ControlContent::Ack,

                        });

                        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.runtime.send_message(message.sending_component_id, ack_message);

                    },

                    ControlContent::Ack => {

                        if let Some((target_component, new_control_message)) = scheduled.router.handle_ack(connector_id, message.id) {

                            self.runtime.send_message(target_component, new_control_message);

                        };

                    },

                    ControlContent::Ping => {},

                }

            } else {

                // Not a control message

                if scheduled.shutting_down {

                    // Since we're shutting down, we just want to respond with a

                    // message saying the message did not arrive.

                    debug_assert!(scheduled.ctx.inbox.get_next_message_ticket().is_none()); // public inbox should be completely cleared

                    self.handle_message_while_shutting_down(message, scheduled);

                } else {

                    scheduled.ctx.inbox.insert_new(message);

                }

            }

        }

    }

    fn handle_message_while_shutting_down(&mut self, message: Message, scheduled: &mut ScheduledConnector) {

        let target_port_and_round_number = match message {

            Message::Data(msg) => Some((msg.data_header.target_port, msg.sync_header.sync_round)),

            Message::SyncComp(_) => None,

            Message::SyncPort(msg) => Some((msg.target_port, msg.sync_header.sync_round)),

            Message::SyncControl(_) => None,

            Message::Control(_) => None,

        };

        if let Some((target_port, sync_round)) = target_port_and_round_number {

            // This message is aimed at a port, but we're shutting down, so

            // notify the peer that its was not received properly.

            // (also: since we're shutting down, we're not in sync mode and

            // the context contains the definitive set of owned ports)

            let port = scheduled.ctx.get_port_by_id(target_port).unwrap();

        }

    }

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

            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::SyncComp(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::SyncPort(content) => {

                    let port_desc = scheduled.ctx.get_port_by_id(content.source_port).unwrap();

                    debug_assert_eq!(port_desc.peer_id, content.target_port);

                    if port_desc.state == PortState::Closed {

                        todo!("handle sending over a closed port")

                    }

                    port_desc.peer_connector

                },

                Message::SyncControl(_) => unreachable!("component sending 'SyncControl' messages directly"),

                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

                        scheduled.ctx.inbox.transfer_messages_for_port(port_id, &mut new_connector.ctx.inbox);

                        // 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, but only

                        // if the port wasn't already closed (otherwise the peer

                        // is gone).

                        if port.state == PortState::Open {

                            let reroute_message = scheduled.router.prepare_reroute(

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

                                port.peer_connector, new_connector.ctx.id,

                                &mut new_connector.router

                            );

                            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 prepare inbox for the next sync

                // round

                scheduled.ctx.inbox.clear_read_messages();

            }

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

        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.

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

        return self.ports.iter().find(|v| v.channel_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) -> Result<(), ()> {

        debug_assert!(self.is_in_sync);

        if let Some(port_id) = contents.source_port() {

            let port_info = self.get_port_by_id(port_id);

            let is_valid = match port_info {

                Some(port_info) => {

                    port_info.state == PortState::Open

                },

                None => false,

            };

            if !is_valid {

                // We don't own the port

                println!(" ****** DEBUG ****** : Sending through closed port!!! {}", port_id.index);

                return Err(());

            }

        }

        self.outbox.push_back(contents);

        return Ok(());

    }

    /// 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 self.inbox.get_read_data_messages(match_port_id);

    }

    pub(crate) fn get_next_message_ticket(&mut self) -> Option<MessageTicket> {

        if !self.is_in_sync { return None; }

        return self.inbox.get_next_message_ticket();

    }

    #[inline]

    pub(crate) fn get_next_message_ticket_even_if_not_in_sync(&mut self) -> Option<MessageTicket> {

        return self.inbox.get_next_message_ticket();

    }

    #[inline]

    pub(crate) fn read_message_using_ticket(&self, ticket: MessageTicket) -> &Message {

        return self.inbox.read_message_using_ticket(ticket);

    }

    #[inline]

    pub(crate) fn take_message_using_ticket(&mut self, ticket: MessageTicket) -> Message {

        return self.inbox.take_message_using_ticket(ticket)

    }

    /// Puts back a message back into the inbox. The reason being that the

    /// message is actually part of the next sync round. This will

    pub(crate) fn put_back_message(&mut self, message: Message) {

        self.inbox.put_back_message(message);

    }

}

    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;

        }