Changeset - ce39b1540ff5
Parent rev.
Child rev.
[Not reviewed]
0 5 0
mh - 4 years ago 2021-10-22 13:49:49
contact@maxhenger.nl
WIP on port fixing
4 files changed with 95 insertions and 38 deletions:
src/runtime2/connector.rs
51
15
src/runtime2/mod.rs
33
1
src/runtime2/native.rs
7
22
src/runtime2/scheduler.rs
4
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 super::ConnectorId;

			




	
	
	
		
 
use super::native::Connector;

			




	
	
	
		
 
use super::scheduler::ConnectorCtx;

			




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

			




	
	
	
		
 
use super::inbox::{

			




	
	
	
		
 
    PrivateInbox, PublicInbox,

			




	
	
	
		
 
    DataMessage, SyncMessage, SolutionMessage, Message, MessageContents,

			




	
	
	
		
 
    SyncBranchConstraint, SyncConnectorSolution

			




	
	
	
		
 
};

			




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

			




	
	
		
 
@@ -329,28 +330,59 @@ pub(crate) struct ConnectorPDL {

			




	
	
	
		
 
    // Port/message management

			




	
	
	
		
 
    pub committed_to: Option<(ConnectorId, u64)>,

			




	
	
	
		
 
    pub inbox: PrivateInbox,

			




	
	
	
		
 
    pub ports: ConnectorPorts,

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
struct TempCtx {}

			




	
	
	
		
 
impl RunContext for TempCtx {

			




	
	
	
		
 
struct ConnectorRunContext<'a> {

			




	
	
	
		
 
    inbox: &'a PrivateInbox,

			




	
	
	
		
 
    ports: &'a ConnectorPorts,

			




	
	
	
		
 
    branch: &'a Branch,

			




	
	
	
		
 
    scheduler: SchedulerCtx<'a>,

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
impl<'a> RunContext for ConnectorRunContext<'a> {

			




	
	
	
		
 
    fn did_put(&mut self, port: PortId) -> bool {

			




	
	
	
		
 
        todo!()

			




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

			




	
	
	
		
 
            // Either acquired or released, must be silent

			




	
	
	
		
 
            return false;

			




	
	
	
		
 
        }

			




	
	
	
		
 

			




	
	
	
		
 
        let port_index = self.ports.get_port_index(PortIdLocal::new(port.0.u32_suffix)).unwrap();

			




	
	
	
		
 
        let mapping = self.ports.get_port(self.branch.index.index, port_index);

			




	
	
	
		
 
        return mapping.is_assigned;

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    fn get(&mut self, port: PortId) -> Option<ValueGroup> {

			




	
	
	
		
 
        todo!()

			




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

			




	
	
	
		
 
        match self.branch.received.get(&port_id) {

			




	
	
	
		
 
            Some(message) => Some(message.message.clone()),

			




	
	
	
		
 
            None => None,

			




	
	
	
		
 
        }

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    fn fires(&mut self, port: PortId) -> Option<Value> {

			




	
	
	
		
 
        todo!()

			




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

			




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

			




	
	
	
		
 
            return None

			




	
	
	
		
 
        }

			




	
	
	
		
 

			




	
	
	
		
 
        let port_index = self.ports.get_port_index(port_id).unwrap();

			




	
	
	
		
 
        let mapping = self.ports.get_port(self.branch.index.index, port_index);

			




	
	
	
		
 

			




	
	
	
		
 
        if mapping.is_assigned {

			




	
	
	
		
 
            return Some(Value::Bool(mapping.num_times_fired != 0));

			




	
	
	
		
 
        } else {

			




	
	
	
		
 
            return None;

			




	
	
	
		
 
        }

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




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

			




	
	
	
		
 
        todo!()

			




	
	
	
		
 
        let (getter, putter) = self.scheduler.runtime.create_channel();

			




	
	
	
		
 
        debug_assert_eq!(getter.kind, PortKind::Getter);

			




	
	
	
		
 

			




	
	
	
		
 
    }

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
impl Connector for ConnectorPDL {

			




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

			




	
	
	
		
 
        use MessageContents as MC;

			




	
	
		
 
@@ -361,15 +393,15 @@ 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, pd: &ProtocolDescription, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {

			




	
	
	
		
 
    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(pd, ctx, delta_state);

			




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

			




	
	
		
 
@@ -385,15 +417,15 @@ impl Connector for ConnectorPDL {

			




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

			




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

			




	
	
	
		
 
                    if let Some(valid_solution) = solution_message {

			




	
	
	
		
 
                        self.submit_sync_solution(valid_solution, ctx, delta_state);

			




	
	
	
		
 
                        self.submit_sync_solution(valid_solution, conn_ctx, delta_state);

			




	
	
	
		
 
                    } else {

			




	
	
	
		
 
                        // Branch is actually invalid, but we only just figured

			




	
	
	
		
 
                        // it out. We need to mark it as invalid to prevent

			




	
	
	
		
 
                        // future use

			




	
	
	
		
 
                        Self::remove_branch_from_queue(&mut self.branches, &mut self.sync_finished, branch_id);

			




	
	
	
		
 
                        if branch_id.index == self.sync_finished_last_handled {

			




	
	
		
 
@@ -412,13 +444,13 @@ impl Connector for ConnectorPDL {

			




	
	
	
		
 

			




	
	
	
		
 
                self.sync_finished_last_handled = next_id;

			




	
	
	
		
 
            }

			




	
	
	
		
 

			




	
	
	
		
 
            return scheduling;

			




	
	
	
		
 
        } else {

			




	
	
	
		
 
            let scheduling = self.run_in_deterministic_mode(pd, ctx, delta_state);

			




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

			




	
	
	
		
 
            return scheduling;

			




	
	
	
		
 
        }

			




	
	
	
		
 
    }

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
impl ConnectorPDL {

			




	
	
		
 
@@ -691,13 +723,13 @@ impl ConnectorPDL {

			




	
	
	
		
 
        debug_assert!(self.in_sync);

			




	
	
	
		
 
        debug_assert!(!self.sync_active.is_empty());

			




	
	
	
		
 

			




	
	
	
		
 
        let branch = Self::pop_branch_from_queue(&mut self.branches, &mut self.sync_active);

			




	
	
	
		
 

			




	
	
	
		
 
        // Run the branch to the next blocking point

			




	
	
	
		
 
        let mut run_context = TempCtx{};

			




	
	
	
		
 
        let mut run_context = ConnectorRunContext {};

			




	
	
	
		
 
        let run_result = branch.code_state.run(&mut run_context, pd);

			




	
	
	
		
 

			




	
	
	
		
 
        // Match statement contains `return` statements only if the particular

			




	
	
	
		
 
        // run result behind handled requires an immediate re-run of the

			




	
	
	
		
 
        // connector.

			




	
	
	
		
 
        match run_result {

			




	
	
		
 
@@ -891,21 +923,25 @@ impl ConnectorPDL {

			




	
	
	
		
 
        } else {

			




	
	
	
		
 
            return ConnectorScheduling::Later;

			




	
	
	
		
 
        }

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




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

			




	
	
	
		
 
    pub fn run_in_deterministic_mode(&mut self, pd: &ProtocolDescription, _context: &ConnectorCtx, results: &mut RunDeltaState) -> ConnectorScheduling {

			




	
	
	
		
 
    pub fn run_in_deterministic_mode(&mut self, sched_ctx: &SchedulerCtx, _context: &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);

			




	
	
	
		
 

			




	
	
	
		
 
        let mut run_context = TempCtx{};

			




	
	
	
		
 
        let mut run_context = ConnectorRunContext{

			




	
	
	
		
 
            inbox: &self.inbox,

			




	
	
	
		
 
            ports: &self.ports,

			




	
	
	
		
 
            branch: &Branch {}

			




	
	
	
		
 
        };

			




	
	
	
		
 
        let run_result = branch.code_state.run(&mut run_context, pd);

			




	
	
	
		
 

			




	
	
	
		
 
        match run_result {

			




	
	
	
		
 
            RunResult::ComponentTerminated => {

			




	
	
	
		
 
                // Need to wait until all children are terminated

			




	
	
	
		
 
                // TODO: Think about how to do this?

			




        

    


    
    

    

        

                

                    src/runtime2/mod.rs
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -21,12 +21,13 @@ use crate::collections::RawVec;

			




	
	
	
		
 
use crate::ProtocolDescription;

			




	
	
	
		
 

			




	
	
	
		
 
use inbox::Message;

			




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

			




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

			




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

			




	
	
	
		
 
use crate::runtime2::port::Port;

			




	
	
	
		
 

			




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

			




	
	
		
 
@@ -210,12 +211,39 @@ impl RuntimeInner {

			




	
	
	
		
 
    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 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) -> (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: self.connector_id,

			




	
	
	
		
 
        };

			




	
	
	
		
 
        let putter_port = Port{

			




	
	
	
		
 
            self_id: putter_id,

			




	
	
	
		
 
            peer_id: getter_id,

			




	
	
	
		
 
            kind: PortKind::Putter,

			




	
	
	
		
 
            peer_connector: self.connector_id,

			




	
	
	
		
 
        };

			




	
	
	
		
 

			




	
	
	
		
 
        return (getter_port, putter_port);

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    // --- Creating/retrieving/destroying components

			




	
	
	
		
 

			




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

			




	
	
	
		
 
        // Initialize as sleeping, as it will be scheduled by the programmer.

			




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

			




	
	
	
		
 
        let key = lock.create(ConnectorVariant::Native(Box::new(component)), true);

			




	
	
		
 
@@ -273,33 +301,37 @@ impl RuntimeInner {

			




	
	
	
		
 

			




	
	
	
		
 
    // --- Managing exit condition

			




	
	
	
		
 

			




	
	
	
		
 
    #[inline]

			




	
	
	
		
 
    pub(crate) fn increment_active_interfaces(&self) {

			




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

			




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

			




	
	
	
		
 
        debug_assert_ne!(_old_num, 0); // once it hits 0, it stays zero

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




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

			




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

			




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

			




	
	
	
		
 
        if old_num == 0 { // such that we have no more active connectors (for now!)

			




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

			




	
	
	
		
 
            if num_interfaces == 0 {

			




	
	
	
		
 
                self.signal_for_shutdown();

			




	
	
	
		
 
            }

			




        

    


    
    

    

        

                

                    src/runtime2/native.rs
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -4,12 +4,13 @@ use std::sync::atomic::Ordering;

			




	
	
	
		
 

			




	
	
	
		
 
use crate::protocol::ComponentCreationError;

			




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

			




	
	
	
		
 
use crate::ProtocolDescription;

			




	
	
	
		
 

			




	
	
	
		
 
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;

			




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

			




	
	
	
		
 

			




	
	
	
		
 
/// Generic connector interface from the scheduler's point of view.

			




	
	
		
 
@@ -17,13 +18,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, protocol_description: &ProtocolDescription, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling;

			




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

			




	
	
	
		
 
}

			




	
	
	
		
 

			




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

			




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

			




	
	
	
		
 

			




	
	
	
		
 
enum ApplicationJob {

			




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

			




	
	
	
		
 
            },

			




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

			




	
	
	
		
 
            MC::Ping => {},

			




	
	
	
		
 
        }

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    fn run(&mut self, _protocol_description: &ProtocolDescription, _ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {

			




	
	
	
		
 
    fn run(&mut self, _sched_ctx: &SchedulerCtx, _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);

			




	
	
		
 
@@ -108,32 +109,16 @@ impl ApplicationInterface {

			




	
	
	
		
 
            owned_ports: Vec::new(),

			




	
	
	
		
 
        }

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    /// Creates a new channel.

			




	
	
	
		
 
    pub fn create_channel(&mut self) -> Channel {

			




	
	
	
		
 
        // TODO: Duplicated logic in scheduler

			




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

			




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

			




	
	
	
		
 
        let getter_id = PortIdLocal::new(getter_id);

			




	
	
	
		
 

			




	
	
	
		
 
        // Create ports and add a job such that they are transferred to the

			




	
	
	
		
 
        // API component. (note that we do not send a ping, this is only

			




	
	
	
		
 
        // necessary once we create a connector)

			




	
	
	
		
 
        let getter_port = Port{

			




	
	
	
		
 
            self_id: getter_id,

			




	
	
	
		
 
            peer_id: putter_id,

			




	
	
	
		
 
            kind: PortKind::Getter,

			




	
	
	
		
 
            peer_connector: self.connector_id,

			




	
	
	
		
 
        };

			




	
	
	
		
 
        let putter_port = Port{

			




	
	
	
		
 
            self_id: putter_id,

			




	
	
	
		
 
            peer_id: getter_id,

			




	
	
	
		
 
            kind: PortKind::Putter,

			




	
	
	
		
 
            peer_connector: self.connector_id,

			




	
	
	
		
 
        };

			




	
	
	
		
 
        let (getter_port, putter_port) = self.runtime.create_channel();

			




	
	
	
		
 
        debug_assert_eq!(getter_port.kind, PortKind::Getter);

			




	
	
	
		
 
        let getter_id = getter_port.self_id;

			




	
	
	
		
 
        let putter_id = putter_port.self_id;

			




	
	
	
		
 

			




	
	
	
		
 
        {

			




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

			




	
	
	
		
 
            lock.push_back(ApplicationJob::NewChannel((getter_port, putter_port)));

			




	
	
	
		
 
        }

			




	
	
	
		
 

			




        

    


    
    

    

        

                

                    src/runtime2/scheduler.rs
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -50,12 +50,16 @@ impl ConnectorCtx {

			




	
	
	
		
 
        }

			




	
	
	
		
 

			




	
	
	
		
 
        panic!("port {:?}, not owned by connector", id);

			




	
	
	
		
 
    }

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
pub(crate) struct SchedulerCtx<'a> {

			




	
	
	
		
 
    pub(crate) runtime: &'a RuntimeInner

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
pub(crate) struct Scheduler {

			




	
	
	
		
 
    runtime: Arc<RuntimeInner>,

			




	
	
	
		
 
    scheduler_id: u32,

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
// Thinking aloud: actual ports should be accessible by connector, but managed

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








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