Changeset - 15b9bb47abdc
Parent rev.
Child rev.
[Not reviewed]
0 4 0
mh - 4 years ago 2021-10-28 11:27:30
contact@maxhenger.nl
WIP on rewriting execution ctx to fix recv bug
4 files changed with 169 insertions and 13 deletions:
src/runtime2/connector.rs
9
9
src/runtime2/native.rs
3
2
src/runtime2/port.rs
1
src/runtime2/scheduler.rs
156
2
0 comments (0 inline, 0 general)
src/runtime2/connector.rs
➞
Show inline comments
 
use std::collections::HashMap;
 
use std::sync::atomic::AtomicBool;
 

			




	
	
	
		
 
use crate::{PortId, ProtocolDescription};

			




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

			




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

			




	
	
	
		
 
use crate::runtime2::scheduler::Scheduler;

			




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

			




	
	
	
		
 

			




	
	
	
		
 
use super::ConnectorId;

			




	
	
	
		
 
use super::native::Connector;

			




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

			




	
	
	
		
 
use super::inbox::{

			




	
	
	
		
 
    PrivateInbox, PublicInbox,

			




	
	
		
 
@@ -407,17 +407,17 @@ impl Connector for ConnectorPDL {

			




	
	
	
		
 
            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 {

			




	
	
	
		
 
    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)) = self.inbox.next_message() {

			




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

			




	
	
		
 
@@ -441,13 +441,13 @@ impl Connector for ConnectorPDL {

			




	
	
	
		
 
                    }

			




	
	
	
		
 

			




	
	
	
		
 
                    branch_idx = next_branch_idx;

			




	
	
	
		
 
                }

			




	
	
	
		
 
            }

			




	
	
	
		
 

			




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

			




	
	
	
		
 
            let scheduling = self.run_in_speculative_mode(sched_ctx, comp_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;

			




	
	
		
 
@@ -490,13 +490,13 @@ impl Connector for ConnectorPDL {

			




	
	
	
		
 

			




	
	
	
		
 
                self.sync_finished_last_handled = next_id;

			




	
	
	
		
 
            }

			




	
	
	
		
 

			




	
	
	
		
 
            return scheduling;

			




	
	
	
		
 
        } else {

			




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

			




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

			




	
	
	
		
 
            return scheduling;

			




	
	
	
		
 
        }

			




	
	
	
		
 
    }

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
impl ConnectorPDL {

			




	
	
		
 
@@ -524,13 +524,13 @@ impl ConnectorPDL {

			




	
	
	
		
 

			




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

			




	
	
	
		
 
    // Handling connector messages

			




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

			




	
	
	
		
 

			




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

			




	
	
	
		
 
        self.inbox.insert_message(target_port, message);

			




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

			




	
	
		
 
@@ -761,13 +761,13 @@ impl ConnectorPDL {

			




	
	
	
		
 

			




	
	
	
		
 
    /// Runs the connector in synchronous mode. Potential changes to the global

			




	
	
	
		
 
    /// system's state are added to the `RunDeltaState` object by the connector,

			




	
	
	
		
 
    /// where it is the caller's responsibility to immediately take care of

			




	
	
	
		
 
    /// those changes. The return value indicates when (and if) the connector

			




	
	
	
		
 
    /// needs to be scheduled again.

			




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

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        if self.sync_active.is_empty() {

			




	
	
	
		
 
            return ConnectorScheduling::NotNow;

			




	
	
	
		
 
        }

			




	
	
	
		
 

			




	
	
		
 
@@ -861,13 +861,13 @@ impl ConnectorPDL {

			




	
	
	
		
 
                    branch.halted_at_port = local_port_id;

			




	
	
	
		
 
                    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 messages = comp_ctx.get_read_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

			




	
	
		
 
@@ -983,13 +983,13 @@ impl ConnectorPDL {

			




	
	
	
		
 
        } else {

			




	
	
	
		
 
            return ConnectorScheduling::Later;

			




	
	
	
		
 
        }

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    /// Runs the connector in non-synchronous mode.

			




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

			




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

			




	
	
	
		
 
        debug_assert!(self.sync_active.is_empty() && self.sync_pending_get.is_empty() && self.sync_finished.is_empty());

			




	
	
	
		
 
        debug_assert!(self.branches.len() == 1);

			




	
	
	
		
 

			




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

			




	
	
	
		
 
        debug_assert!(branch.sync_state == SpeculativeState::RunningNonSync);

			




        

    


    
    

    

        

                

                    src/runtime2/native.rs
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -2,12 +2,13 @@ use std::collections::VecDeque;

			




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

			




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

			




	
	
	
		
 

			




	
	
	
		
 
use crate::protocol::ComponentCreationError;

			




	
	
	
		
 
use crate::protocol::eval::ValueGroup;

			




	
	
	
		
 
use crate::ProtocolDescription;

			




	
	
	
		
 
use crate::runtime2::scheduler::ComponentCtxFancy;

			




	
	
	
		
 

			




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

			




	
	
	
		
 
use super::scheduler::SchedulerCtx;

			




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

			




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

			




	
	
	
		
 
use super::connector::find_ports_in_value_group;

			




	
	
		
 
@@ -18,13 +19,13 @@ pub(crate) trait Connector {

			




	
	
	
		
 
    /// Handle a new message (preprocessed by the scheduler). You probably only

			




	
	
	
		
 
    /// want to handle `Data`, `Sync`, and `Solution` messages. The others are

			




	
	
	
		
 
    /// intended for the scheduler itself.

			




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

			




	
	
	
		
 

			




	
	
	
		
 
    /// Should run the connector's behaviour up until the next blocking point.

			




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

			




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

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
type SyncDone = Arc<(Mutex<bool>, Condvar)>;

			




	
	
	
		
 
type JobQueue = Arc<Mutex<VecDeque<ApplicationJob>>>;

			




	
	
	
		
 

			




	
	
	
		
 
enum ApplicationJob {

			




	
	
		
 
@@ -63,13 +64,13 @@ impl Connector for ConnectorApplication {

			




	
	
	
		
 
            },

			




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

			




	
	
	
		
 
            MC::Ping => {},

			




	
	
	
		
 
        }

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




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

			




	
	
	
		
 
    fn run(&mut self, _sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy, _conn_ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {

			




	
	
	
		
 
        let mut queue = self.job_queue.lock().unwrap();

			




	
	
	
		
 
        while let Some(job) = queue.pop_front() {

			




	
	
	
		
 
            match job {

			




	
	
	
		
 
                ApplicationJob::NewChannel((endpoint_a, endpoint_b)) => {

			




	
	
	
		
 
                    println!("DEBUG: API adopting ports");

			




	
	
	
		
 
                    delta_state.new_ports.reserve(2);

			




        

    


    
    

    

        

                

                    src/runtime2/port.rs
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -34,12 +34,13 @@ pub enum PortState {

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
/// Represents a port inside of the runtime. This is generally the local view of

			




	
	
	
		
 
/// a connector on its port, which may not be consistent with the rest of the

			




	
	
	
		
 
/// global system (e.g. its peer was moved to a new connector, or the peer might

			




	
	
	
		
 
/// have died in the meantime, so it is no longer usable).

			




	
	
	
		
 
#[derive(Clone)]

			




	
	
	
		
 
pub struct Port {

			




	
	
	
		
 
    pub self_id: PortIdLocal,

			




	
	
	
		
 
    pub peer_id: PortIdLocal,

			




	
	
	
		
 
    pub kind: PortKind,

			




	
	
	
		
 
    pub state: PortState,

			




	
	
	
		
 
    pub peer_connector: ConnectorId, // might be temporarily inconsistent while peer port is sent around in non-sync phase

			




        

    


    
    

    

        

                

                    src/runtime2/scheduler.rs
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
use std::sync::Arc;

			




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

			




	
	
	
		
 
use crate::runtime2::ScheduledConnector;

			




	
	
	
		
 
use crate::runtime2::connector::{BranchId, ConnectorPDL};

			




	
	
	
		
 
use crate::runtime2::inbox::{DataMessage, PrivateInbox};

			




	
	
	
		
 

			




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

			




	
	
	
		
 
use super::{ScheduledConnector, 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

			




	
	
		
 
@@ -363,12 +364,165 @@ impl Scheduler {

			




	
	
	
		
 

			




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

			




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

			




	
	
	
		
 
    }

			




	
	
	
		
 
}

			




	
	
	
		
 

			




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

			




	
	
	
		
 
// ComponentCtx

			




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

			




	
	
	
		
 

			




	
	
	
		
 
enum ComponentStateChange {

			




	
	
	
		
 
    CreatedComponent(ConnectorPDL),

			




	
	
	
		
 
    CreatedPort(Port),

			




	
	
	
		
 
    ChangedPort(ComponentPortChange),

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
#[derive(Clone)]

			




	
	
	
		
 
pub(crate) enum ComponentPortChange {

			




	
	
	
		
 
    Acquired(Port),

			




	
	
	
		
 
    Released(Port),

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
struct InboxMessage {

			




	
	
	
		
 
    target_port: PortIdLocal,

			




	
	
	
		
 
    data: DataMessage,

			




	
	
	
		
 
}

			




	
	
	
		
 

			




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

			




	
	
	
		
 
    // Mostly managed by the scheduler

			




	
	
	
		
 
    id: ConnectorId,

			




	
	
	
		
 
    ports: Vec<Port>,

			




	
	
	
		
 
    inbox_messages: Vec<InboxMessage>,

			




	
	
	
		
 
    inbox_len_read: usize,

			




	
	
	
		
 
    // Submitted by the component

			




	
	
	
		
 
    is_in_sync: bool,

			




	
	
	
		
 
    changed_in_sync: bool,

			




	
	
	
		
 
    outbox: Vec<MessageContents>,

			




	
	
	
		
 
    state_changes: Vec<ComponentStateChange>

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
impl ComponentCtxFancy {

			




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

			




	
	
	
		
 
        debug_assert!(!self.is_in_sync);

			




	
	
	
		
 
        self.state_changes.push(ComponentStateChange::CreatedComponent(component));

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    /// 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(ComponentStateChange::CreatedPort(port))

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    /// Notify that component will enter a sync block.

			




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

			




	
	
	
		
 
        debug_assert!(!self.is_in_sync);

			




	
	
	
		
 

			




	
	
	
		
 
        self.is_in_sync = true;

			




	
	
	
		
 
        self.changed_in_sync = true;

			




	
	
	
		
 
        return &self.ports

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




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

			




	
	
	
		
 
        debug_assert!(self.is_in_sync);

			




	
	
	
		
 
        self.outbox.push(contents);

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    /// Notify that component just finished a sync block.

			




	
	
	
		
 
    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(ComponentStateChange::ChangedPort(changed_port.clone()));

			




	
	
	
		
 
        }

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    /// Inserts message into inbox. Generally only called by scheduler.

			




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

			




	
	
	
		
 
        debug_assert!(!self.inbox_messages.iter().any(|v| {

			




	
	
	
		
 
            v.target_port == target_port &&

			




	
	
	
		
 
                v.data.sender_prev_branch_id == data.sender_prev_branch_id &&

			




	
	
	
		
 
                v.data.sender_cur_branch_id == data.sender_cur_branch_id

			




	
	
	
		
 
        }));

			




	
	
	
		
 

			




	
	
	
		
 
        self.inbox_messages.push(InboxMessage{ target_port, data })

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    /// 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_messages(&self, match_port_id: PortIdLocal, match_prev_branch_id: BranchId) -> MessagesIter {

			




	
	
	
		
 
        return MessageIter {

			




	
	
	
		
 
            messages: &self.inbox_messages,

			




	
	
	
		
 
            next_index: 0,

			




	
	
	
		
 
            max_index: self.inbox_len_read,

			




	
	
	
		
 
            match_port_id, match_prev_branch_id

			




	
	
	
		
 
        };

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




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

			




	
	
	
		
 
    /// (new) messages to read.

			




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

			




	
	
	
		
 
        if self.inbox_len_read == self.inbox_messages.len() {

			




	
	
	
		
 
            return None;

			




	
	
	
		
 
        }

			




	
	
	
		
 

			




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

			




	
	
	
		
 
        self.inbox_len_read += 1;

			




	
	
	
		
 
        return Some((&message.target_port, &message.data))

			




	
	
	
		
 
    }

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
pub(crate) struct MessagesIter<'a> {

			




	
	
	
		
 
    messages: &'a [InboxMessage],

			




	
	
	
		
 
    next_index: usize,

			




	
	
	
		
 
    max_index: usize,

			




	
	
	
		
 
    match_port_id: PortIdLocal,

			




	
	
	
		
 
    match_prev_branch_id: BranchId,

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
impl Iterator for MessagesIter {

			




	
	
	
		
 
    type Item = 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 message.target_port == self.match_port_id && message.data.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.data);

			




	
	
	
		
 
    }

			




	
	
	
		
 
}

			




	
	
	
		
 

			




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

			




	
	
	
		
 
// Control messages

			




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

			




	
	
	
		
 

			




	
	
	
		
 
struct ControlEntry {

			




	
	
	
		
 
    id: u32,

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








    
        Server instance: kalli-1039567
    
    
        This site is powered by
            Kallithea,
        which is
        © 2010–2022 by various authors & licensed under GPLv3.
            – Support