CSY/reowolf Changeset - 1755ca411ca7 · Centrum Wiskunde & Informatica (CWI)

Changeset - 1755ca411ca7

Parent rev.

Child rev.

[Not reviewed]

0 6 0

MH - 4 years ago 2021-10-21 01:50:31
contact@maxhenger.nl

better exit condition, WIP on bugfixing

6 files changed with 394 insertions and 164 deletions:

src/runtime2/connector.rs

src/runtime2/inbox.rs

src/runtime2/mod.rs

298

src/runtime2/native.rs

src/runtime2/port.rs

src/runtime2/scheduler.rs

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::inbox::{Message, MessageContents, SolutionMessage};
 use crate::runtime2::native::Connector;
 use crate::runtime2::port::{Port, PortKind};
 use crate::runtime2::scheduler::ConnectorCtx;
 use super::global_store::ConnectorId;
 use super::ConnectorId;
 use super::native::Connector;
 use super::scheduler::ConnectorCtx;
 use super::inbox::{
     PrivateInbox, PublicInbox, DataMessage, SyncMessage,
     PrivateInbox, PublicInbox,
     DataMessage, SyncMessage, SolutionMessage, Message, MessageContents,
     SyncBranchConstraint, SyncConnectorSolution
 };
 use super::port::PortIdLocal;
+use super::port::{Port, PortKind, PortIdLocal};
 /// Represents the identifier of a branch (the index within its container). An
 /// ID of `0` generally means "no branch" (e.g. no parent, or a port did not
 /// yet receive anything from any branch).
 #[derive(Clone, Copy, PartialEq, Eq)]
 // TODO: Remove Debug derive
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct BranchId {
     pub index: u32,
+}
 impl BranchId {
     fn new_invalid() -> Self {
@@ @@ -906,13 +907,13 @@ impl ConnectorPDL { @@
         match run_result {
             RunResult::ComponentTerminated => {
                 // Need to wait until all children are terminated
                 // TODO: Think about how to do this?
                 branch.sync_state = SpeculativeState::Finished;
-                return ConnectorScheduling::NotNow;
+                return ConnectorScheduling::Exit;
             },
             RunResult::ComponentAtSyncStart => {
                 // Prepare for sync execution and reschedule immediately
                 self.in_sync = true;
                 let first_sync_branch = Branch::new_sync_branching_from(1, branch);
                 let first_sync_branch_id = first_sync_branch.index;
@@ @@ -1309,12 +1310,13 @@ impl RunDeltaState { @@
 #[derive(Eq, PartialEq)]
 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
+}
 /// Recursively goes through the value group, attempting to find ports.
 /// Duplicates will only be added once.
 pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut Vec<PortIdLocal>) {
     // Helper to check a value for a port and recurse if needed.

src/runtime2/inbox.rs

➞

Show inline comments

 within a certain sync-round.
 **/
 use std::collections::VecDeque;
 use std::sync::Mutex;
 use super::ConnectorId;
 use crate::protocol::eval::ValueGroup;
 use super::connector::BranchId;
 use super::port::PortIdLocal;
 use super::global_store::ConnectorId;
 /// A message that has been delivered (after being imbued with the receiving
 /// port by the scheduler) to a connector.
 #[derive(Clone)]
 // TODO: Remove Debug on messages
 #[derive(Debug, Clone)]
 pub struct DataMessage {
     pub sending_port: PortIdLocal,
     pub sender_prev_branch_id: BranchId,
     pub sender_cur_branch_id: BranchId,
     pub message: ValueGroup,
+}
 #[derive(Clone)]
+#[derive(Debug, Clone)]
 pub enum SyncBranchConstraint {
     SilentPort(PortIdLocal),
     BranchNumber(BranchId),
     PortMapping(PortIdLocal, BranchId),
+}
 #[derive(Clone)]
+#[derive(Debug, Clone)]
 pub struct SyncConnectorSolution {
     pub connector_id: ConnectorId,
     pub terminating_branch_id: BranchId,
     pub execution_branch_ids: Vec<BranchId>, // no particular ordering of IDs enforced
     pub final_port_mapping: Vec<(PortIdLocal, BranchId)>
+}
 #[derive(Clone)]
+#[derive(Debug, Clone)]
 pub struct SyncConnectorConstraints {
     pub connector_id: ConnectorId,
     pub constraints: Vec<SyncBranchConstraint>,
+}
 #[derive(Clone)]
+#[derive(Debug, Clone)]
 pub struct SyncMessage {
     pub local_solutions: Vec<SyncConnectorSolution>,
     pub constraints: Vec<SyncConnectorConstraints>,
     pub to_visit: Vec<ConnectorId>,
+}
@@ @@ -164,45 +165,46 @@ impl SyncMessage { @@
                 return Err(());
             },
+        }
+    }
+}
 #[derive(Clone)]
+#[derive(Debug, Clone)]
 pub struct SolutionMessage {
     pub comparison_number: u64,
     pub connector_origin: ConnectorId,
     pub local_solutions: Vec<(ConnectorId, BranchId)>,
     pub to_visit: Vec<ConnectorId>,
+}
 /// A control message. These might be sent by the scheduler to notify eachother
 /// of asynchronous state changes.
 #[derive(Clone)]
+#[derive(Debug, Clone)]
 pub struct ControlMessage {
     pub id: u32, // generic identifier, used to match request to response
     pub content: ControlMessageVariant,
+}
 #[derive(Clone)]
+#[derive(Debug, Clone)]
 pub enum ControlMessageVariant {
     ChangePortPeer(PortIdLocal, ConnectorId), // specified port has a new peer, sent to owner of said port
     Ack, // acknowledgement of previous control message, matching occurs through control message ID.
+}
 /// Generic message contents.
 #[derive(Clone)]
+#[derive(Debug, Clone)]
 pub enum MessageContents {
     Data(DataMessage),              // data message, handled by connector
     Sync(SyncMessage),              // sync message, handled by both connector/scheduler
     RequestCommit(SolutionMessage), // solution message, requesting participants to commit
     ConfirmCommit(SolutionMessage), // solution message, confirming a solution everyone committed to
     Control(ControlMessage),        // control message, handled by scheduler
     Ping,                           // ping message, intentionally waking up a connector (used for native connectors)
+}
 #[derive(Debug)]
 pub struct Message {
     pub sending_connector: ConnectorId,
     pub receiving_port: PortIdLocal, // may be invalid (in case of messages targeted at the connector)
     pub contents: MessageContents,
+}

src/runtime2/mod.rs

➞

Show inline comments

@@ @@ -2,30 +2,31 @@ @@
 mod runtime;
 mod messages;
 mod connector;
 mod native;
 mod port;
 mod global_store;
 mod scheduler;
 mod inbox;
 #[cfg(test)] mod tests;
 // Imports
 use std::sync::{Arc, Mutex};
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::collections::VecDeque;
 use std::sync::{Arc, Condvar, Mutex, RwLock};
 use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 use std::thread::{self, JoinHandle};
 use crate::collections::RawVec;
 use crate::ProtocolDescription;
 use global_store::{ConnectorVariant, GlobalStore};
 use scheduler::Scheduler;
 use native::{ConnectorApplication, ApplicationInterface};
 use inbox::Message;
 use connector::{ConnectorPDL, ConnectorPublic, ConnectorScheduling, RunDeltaState};
 use scheduler::{Scheduler, ConnectorCtx, Router};
 use native::{Connector, ConnectorApplication, ApplicationInterface};
 /// 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
@@ @@ -94,61 +95,35 @@ pub(crate) struct ScheduledConnector { @@
     pub connector: ConnectorVariant, // access by connector
     pub context: ConnectorCtx, // mutable access by scheduler, immutable by connector
     pub public: ConnectorPublic, // accessible by all schedulers and connectors
     pub router: Router,
+}
 // -----------------------------------------------------------------------------
 // Runtime
 // -----------------------------------------------------------------------------
 /// Externally facing runtime.
 pub struct Runtime {
     inner: Arc<RuntimeInner>,
+}
 pub(crate) struct RuntimeInner {
     // Protocol
     pub(crate) protocol_description: ProtocolDescription,
     // Storage of connectors in a kind of freelist. Note the vector of points to
     // ensure pointer stability: the vector can be changed but the entries
     // themselves remain valid.
     pub connectors_list: RawVec<*mut ScheduledConnector>,
     pub connectors_free: Vec<usize>,
     pub(crate) global_store: GlobalStore,
     schedulers: Mutex<Vec<JoinHandle<()>>>,
     active_interfaces: AtomicU32, // active API interfaces that can add connectors/channels
+}
 impl RuntimeInner {
     #[inline]
     pub(crate) fn increment_active_interfaces(&self) {
         let _old_num = self.active_interfaces.fetch_add(1, Ordering::SeqCst);
         debug_assert_ne!(_old_num, 1); // 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);
         debug_assert!(old_num > 0);
         if old_num == 1 {
             // Became 0
             // TODO: Check num connectors, if 0, then set exit flag
+        }
+    }
+}
 // TODO: Come back to this at some point
 unsafe impl Send for RuntimeInner {}
 unsafe impl Sync for RuntimeInner {}
 impl Runtime {
     pub fn new(num_threads: u32, protocol_description: ProtocolDescription) -> Runtime {
         // Setup global state
         assert!(num_threads > 0, "need a thread to run connectors");
         let runtime_inner = Arc::new(RuntimeInner{
             global_store: GlobalStore::new(),
             protocol_description,
             port_counter: AtomicU32::new(0),
             connectors: RwLock::new(ConnectorStore::with_capacity(32)),
             connector_queue: Mutex::new(VecDeque::with_capacity(32)),
             schedulers: Mutex::new(Vec::new()),
             active_interfaces: AtomicU32::new(1), // we are the active interface
             scheduler_notifier: Condvar::new(),
             active_connectors: AtomicU32::new(0),
             active_interfaces: AtomicU32::new(1), // this `Runtime` instance
             should_exit: AtomicBool::new(false),
         });
         // Launch threads
+        {
             let mut schedulers = Vec::with_capacity(num_threads as usize);
             for thread_index in 0..num_threads {
@@ @@ -172,25 +147,299 @@ impl Runtime { @@
         return Runtime{ inner: runtime_inner };
+    }
     /// Returns a new interface through which channels and connectors can be
     /// created.
     pub fn create_interface(&self) -> ApplicationInterface {
         self.inner.increment_active_interfaces();
         let (connector, mut interface) = ConnectorApplication::new(self.inner.clone());
-        let connector_key = self.inner.global_store.connectors.create_interface(connector);
+        let connector_key = self.inner.create_interface_component(connector);
         interface.set_connector_id(connector_key.downcast());
         // Note that we're not scheduling. That is done by the interface in case
         // it is actually needed.
         return interface;
+    }
+}
 impl Drop for Runtime {
     fn drop(&mut self) {
         self.inner.global_store.should_exit.store(true, Ordering::Release);
         let mut schedulers = self.inner.schedulers.lock().unwrap();
         for scheduler in schedulers.drain(..) {
             scheduler.join().unwrap();
         self.inner.decrement_active_interfaces();
         let mut lock = self.inner.schedulers.lock().unwrap();
         for handle in lock.drain(..) {
             handle.join().unwrap();
+        }
+    }
+}
 // -----------------------------------------------------------------------------
 // RuntimeInner
 // -----------------------------------------------------------------------------
 pub(crate) struct RuntimeInner {
     // Protocol
     pub(crate) protocol_description: ProtocolDescription,
     // Regular counter for port IDs
     port_counter: AtomicU32,
     // Storage of connectors and the work queue
     connectors: RwLock<ConnectorStore>,
     connector_queue: Mutex<VecDeque<ConnectorKey>>,
     schedulers: Mutex<Vec<JoinHandle<()>>>,
     // Conditions to determine whether the runtime can exit
     scheduler_notifier: Condvar,  // coupled to mutex on `connector_queue`.
     // TODO: Figure out if we can simply merge the counters?
     active_connectors: AtomicU32, // active connectors (if sleeping, then still considered active)
     active_interfaces: AtomicU32, // active API interfaces that can add connectors/channels
     should_exit: AtomicBool,
+}
 impl RuntimeInner {
     // --- Managing the components queued for execution
     /// Wait until there is a connector to run. If there is one, then `Some`
     /// will be returned. If there is no more work, then `None` will be
     /// returned.
     pub(crate) fn wait_for_work(&self) -> Option<ConnectorKey> {
         let mut lock = self.connector_queue.lock().unwrap();
         while lock.is_empty() && !self.should_exit.load(Ordering::Acquire) {
             lock = self.scheduler_notifier.wait(lock).unwrap();
+        }
         return lock.pop_front();
+    }
     pub(crate) fn push_work(&self, key: ConnectorKey) {
         let mut lock = self.connector_queue.lock().unwrap();
         lock.push_back(key);
         self.scheduler_notifier.notify_one();
+    }
     // --- Creating/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);
         self.increment_active_components();
         return key;
+    }
     /// Creates a new PDL component. The caller MUST make sure to schedule the
     /// connector.
     // TODO: Nicer code, not forcing the caller to schedule, perhaps?
     pub(crate) fn create_pdl_component(&self, created_by: &mut ScheduledConnector, connector: ConnectorPDL) -> ConnectorKey {
         // Create as not sleeping, as we'll schedule it immediately
         let key = {
             let mut lock = self.connectors.write().unwrap();
             lock.create(ConnectorVariant::UserDefined(connector), true)
         };
         // Transfer the ports
+        {
             let lock = self.connectors.read().unwrap();
             let created = lock.get_private(&key);
             match &created.connector {
                 ConnectorVariant::UserDefined(connector) => {
                     for port_id in connector.ports.owned_ports.iter().copied() {
                         println!("DEBUG: Transferring port {:?} from {} to {}", port_id, created_by.context.id.0, key.index);
                         let mut port = created_by.context.remove_port(port_id);
                         created.context.add_port(port);
+                    }
                 },
                 ConnectorVariant::Native(_) => unreachable!(),
+            }
+        }
         self.increment_active_components();
         return key;
+    }
     pub(crate) fn get_component_private(&self, connector_key: &ConnectorKey) -> &'static mut ScheduledConnector {
         let lock = self.connectors.read().unwrap();
         return lock.get_private(connector_key);
+    }
     pub(crate) fn get_component_public(&self, connector_id: ConnectorId) -> &'static ConnectorPublic {
         let lock = self.connectors.read().unwrap();
         return lock.get_public(connector_id);
+    }
     pub(crate) fn destroy_component(&self, connector_key: ConnectorKey) {
         let mut lock = self.connectors.write().unwrap();
         lock.destroy(connector_key);
         self.decrement_active_components();
+    }
     // --- Managing exit condition
     #[inline]
     pub(crate) fn increment_active_interfaces(&self) {
         let _old_num = self.active_interfaces.fetch_add(1, Ordering::SeqCst);
         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);
         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);
+    }
     fn decrement_active_components(&self) {
         let old_num = self.active_connectors.fetch_sub(1, Ordering::SeqCst);
         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();
+            }
+        }
+    }
     #[inline]
     fn signal_for_shutdown(&self) {
         debug_assert_eq!(self.active_interfaces.load(Ordering::Acquire), 0);
         debug_assert_eq!(self.active_connectors.load(Ordering::Acquire), 0);
         println!("DEBUG: Signaling for shutdown");
         let _lock = self.connector_queue.lock().unwrap();
         let should_signal = self.should_exit
             .compare_exchange(false, true, Ordering::SeqCst, Ordering::Acquire)
             .is_ok();
         if should_signal {
             println!("DEBUG: Notifying all waiting schedulers");
             self.scheduler_notifier.notify_all();
+        }
+    }
+}
 // TODO: Come back to this at some point
 unsafe impl Send for RuntimeInner {}
 unsafe impl Sync for RuntimeInner {}
 // -----------------------------------------------------------------------------
 // ConnectorStore
 // -----------------------------------------------------------------------------
 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.
     connectors: RawVec<*mut ScheduledConnector>,
     free: Vec<usize>,
+}
 impl ConnectorStore {
     fn with_capacity(capacity: usize) -> Self {
         Self {
             connectors: 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 connector = self.connectors.get(id.0 as usize);
             debug_assert!(!connector.is_null());
             return &(**connector).public;
+        }
+    }
     /// Retrieves private part of connector - accessible by one thread at a
     /// time.
     fn get_private(&self, key: &ConnectorKey) -> &'static mut ScheduledConnector {
         unsafe {
             let connector = self.connectors.get_mut(key.index as usize);
             debug_assert!(!connector.is_null());
             return &mut (**connector);
+        }
+    }
     /// Creates a new connector. Caller should ensure ports are set up correctly
     /// and the connector is queued for execution if needed.
     fn create(&mut self, connector: ConnectorVariant, initially_sleeping: bool) -> ConnectorKey {
         let mut connector = ScheduledConnector {
             connector,
             context: ConnectorCtx::new(),
             public: ConnectorPublic::new(initially_sleeping),
             router: Router::new(),
         };
         let index;
         let key;
         if self.free.is_empty() {
             // No free entries, allocate new entry
             index = self.connectors.len();
             key = ConnectorKey{ index: index as u32 };
             connector.context.id = key.downcast();
             let connector = Box::into_raw(Box::new(connector));
             self.connectors.push(connector);
         } else {
             // Free spot available
             index = self.free.pop().unwrap();
             key = ConnectorKey{ index: index as u32 };
             connector.context.id = key.downcast();
             unsafe {
                 let target = self.connectors.get_mut(index);
                 std::ptr::write(*target, connector);
+            }
+        }
         return key;
+    }
     /// Destroys a connector. Caller should make sure it is not scheduled for
     /// execution. Otherwise one experiences "bad stuff" (tm).
     fn destroy(&mut self, key: ConnectorKey) {
         unsafe {
             let target = self.connectors.get_mut(key.index as usize);
             std::ptr::drop_in_place(*target);
             // Note: but not deallocating!
+        }
         self.free.push(key.index as usize);
+    }
+}
 impl Drop for ConnectorStore {
     fn drop(&mut self) {
         // Everything in the freelist already had its destructor called, so only
         // has to be deallocated
         for free_idx in self.free.iter().copied() {
             unsafe {
                 let memory = self.connectors.get_mut(free_idx);
                 let layout = std::alloc::Layout::for_value(&**memory);
                 std::alloc::dealloc(*memory as *mut u8, layout);
                 // mark as null for the remainder
                 *memory = std::ptr::null_mut();
+            }
+        }
         // With the deallocated stuff marked as null, clear the remainder that
         // is not null
         for idx in 0..self.connectors.len() {
             unsafe {
                 let memory = *self.connectors.get_mut(idx);
                 if !memory.is_null() {
                     let _ = Box::from_raw(memory); // take care of deallocation, bit dirty, but meh
+                }
+            }
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/native.rs

➞

Show inline comments

@@ @@ -2,23 +2,18 @@ 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::connector::{Branch, find_ports_in_value_group};
 use crate::runtime2::global_store::ConnectorKey;
 use crate::runtime2::inbox::MessageContents;
 use crate::runtime2::port::{Port, PortKind};
 use crate::runtime2::scheduler::ConnectorCtx;
 use super::RuntimeInner;
 use super::global_store::ConnectorId;
 use super::port::{Channel, PortIdLocal};
 use super::connector::{ConnectorPDL, ConnectorScheduling, RunDeltaState};
 use super::inbox::Message;
 use super::{ConnectorKey, ConnectorId, RuntimeInner, ConnectorCtx};
 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.
 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.
@@ @@ -31,12 +26,13 @@ pub(crate) trait Connector { @@
 type SyncDone = Arc<(Mutex<bool>, Condvar)>;
 type JobQueue = Arc<Mutex<VecDeque<ApplicationJob>>>;
 enum ApplicationJob {
     NewChannel((Port, Port)),
     NewConnector(ConnectorPDL),
     Shutdown,
+}
 /// The connector which an application can directly interface with. Once may set
 /// up the next synchronous round, and retrieve the data afterwards.
 pub struct ConnectorApplication {
     sync_done: SyncDone,
@@ @@ -66,25 +62,29 @@ 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, _protocol_description: &ProtocolDescription, _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);
                     delta_state.new_ports.push(endpoint_a);
                     delta_state.new_ports.push(endpoint_b);
+                }
                 ApplicationJob::NewConnector(connector) => {
                     println!("DEBUG: API creating connector");
                     delta_state.new_connectors.push(connector);
                 },
                 ApplicationJob::Shutdown => {
                     debug_assert!(queue.is_empty());
                     return ConnectorScheduling::Exit;
+                }
+            }
+        }
         return ConnectorScheduling::NotNow;
+    }
@@ @@ -99,25 +99,23 @@ pub struct ApplicationInterface { @@
     connector_id: ConnectorId,
     owned_ports: Vec<PortIdLocal>,
+}
 impl ApplicationInterface {
     fn new(sync_done: SyncDone, job_queue: JobQueue, runtime: Arc<RuntimeInner>) -> Self {
         runtime.active_interfaces += 1;
         return Self{
             sync_done, job_queue, runtime,
             connector_id: ConnectorId::new_invalid(),
             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.global_store.connectors.port_counter.fetch_add(2, Ordering::SeqCst);
         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)
@@ @@ -176,31 +174,13 @@ impl ApplicationInterface { @@
         // Put on job queue
+        {
             let mut queue = self.job_queue.lock().unwrap();
             queue.push_back(ApplicationJob::NewConnector(connector));
+        }
         // Send ping message to wake up connector
         let connector = self.runtime.global_store.connectors.get_shared(self.connector_id);
         connector.inbox.insert_message(Message{
             sending_connector: ConnectorId::new_invalid(),
             receiving_port: PortIdLocal::new_invalid(),
             contents: MessageContents::Ping,
         });
         let should_wake_up = connector.sleeping
             .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)
             .is_ok();
         if should_wake_up {
             println!("DEBUG: Waking up connector");
             let key = unsafe{ ConnectorKey::from_id(self.connector_id) };
             self.runtime.global_store.connector_queue.push_back(key);
         } else {
             println!("DEBUG: NOT waking up connector");
+        }
         self.wake_up_connector_with_ping();
         return Ok(());
+    }
     /// Check if the next sync-round is finished.
     pub fn try_wait(&self) -> bool {
@@ @@ -217,13 +197,40 @@ impl ApplicationInterface { @@
+    }
     /// Called by runtime to set associated connector's ID.
     pub(crate) fn set_connector_id(&mut self, id: ConnectorId) {
         self.connector_id = id;
+    }
     fn wake_up_connector_with_ping(&self) {
         let connector = self.runtime.get_component_public(self.connector_id);
         connector.inbox.insert_message(Message{
             sending_connector: ConnectorId::new_invalid(),
             receiving_port: PortIdLocal::new_invalid(),
             contents: MessageContents::Ping,
         });
         let should_wake_up = connector.sleeping
             .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)
             .is_ok();
         if should_wake_up {
             println!("DEBUG: Waking up connector");
             let key = unsafe{ ConnectorKey::from_id(self.connector_id) };
             self.runtime.push_work(key);
         } else {
             println!("DEBUG: NOT waking up connector");
+        }
+    }
+}
 impl Drop for ApplicationInterface {
     fn drop(&mut self) {
+        {
             let mut lock = self.job_queue.lock().unwrap();
             lock.push_back(ApplicationJob::Shutdown);
+        }
         self.wake_up_connector_with_ping();
         self.runtime.decrement_active_interfaces();
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/port.rs

➞

Show inline comments

-use super::global_store::ConnectorId;
 use super::ConnectorId;
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub struct PortIdLocal {
     pub index: u32,
+}

src/runtime2/scheduler.rs

➞

Show inline comments

 use std::sync::Arc;
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::time::Duration;
 use std::thread;
 use std::sync::atomic::Ordering;
 use crate::runtime2::global_store::ConnectorVariant;
 use crate::runtime2::inbox::MessageContents;
 use crate::runtime2::native::Connector;
 use crate::runtime2::port::{Channel, PortKind};
 use super::RuntimeInner;
 use super::{RuntimeInner, ConnectorId, ConnectorKey};
 use super::port::{Port, PortIdLocal};
-use super::inbox::{Message, ControlMessage, ControlMessageVariant};
+use super::native::Connector;
 use super::connector::{ConnectorScheduling, RunDeltaState};
-use super::global_store::{ConnectorKey, ConnectorId};
+use super::inbox::{Message, MessageContents, ControlMessageVariant, ControlMessage};
 /// Contains fields that are mostly managed by the scheduler, but may be
 /// accessed by the connector
 pub(crate) struct ConnectorCtx {
     pub(crate) id: ConnectorId,
     port_counter: Arc<AtomicU32>,
     pub(crate) ports: Vec<Port>,
+}
 impl ConnectorCtx {
-    pub(crate) fn new(port_counter: Arc<AtomicU32>) -> ConnectorCtx {
     pub(crate) fn new() -> ConnectorCtx {
         Self{
             id: ConnectorId::new_invalid(),
             port_counter,
             ports: Vec::new(),
+        }
+    }
     /// Creates a (putter, getter) port pair belonging to the same channel. The
     /// port will be implicitly owned by the connector.
     pub(crate) fn create_channel(&mut self) -> Channel {
         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);
         self.ports.push(Port{
             self_id: getter_id,
             peer_id: putter_id,
             kind: PortKind::Getter,
             peer_connector: self.id,
         });
         self.ports.push(Port{
             self_id: putter_id,
             peer_id: getter_id,
             kind: PortKind::Putter,
             peer_connector: self.id,
         });
         return Channel{ getter_id, putter_id };
+    }
     pub(crate) fn add_port(&mut self, port: Port) {
         debug_assert!(!self.ports.iter().any(|v| v.self_id == port.self_id));
         self.ports.push(port);
+    }
     pub(crate) fn remove_port(&mut self, id: PortIdLocal) -> Port {
@@ @@ -105,40 +72,34 @@ impl Scheduler { @@
         // connector that we run
         let scheduler_id = self.scheduler_id;
         let mut delta_state = RunDeltaState::new();
         'thread_loop: loop {
             // Retrieve a unit of work
             let mut connector_key = self.runtime.global_store.connector_queue.pop_front();
             while connector_key.is_none() {
                 // TODO: @Performance, needs condition or something, and most
                 //  def' not sleeping
                 println!("DEBUG [{}]: Nothing to do", scheduler_id);
                 thread::sleep(Duration::new(1, 0));
                 if self.runtime.global_store.should_exit.load(Ordering::Acquire) {
                     // Thread exits!
                     println!("DEBUG [{}]: ... So I am quitting", scheduler_id);
             println!("DEBUG [{}]: Waiting for work", scheduler_id);
             let connector_key = self.runtime.wait_for_work();
             if connector_key.is_none() {
                 // We should exit
                 println!("DEBUG [{}]: ... No more work, quitting", scheduler_id);
                 break 'thread_loop;
+            }
                 println!("DEBUG [{}]: ... But I'm still running", scheduler_id);
                 continue 'thread_loop;
+            }
             // We have something to do
             let connector_key = connector_key.unwrap();
             println!("DEBUG [{}]: Running connector {}", scheduler_id, connector_key.index);
             let scheduled = self.runtime.global_store.connectors.get_mut(&connector_key);
             println!("DEBUG [{}]: ... Got work, running {}", scheduler_id, connector_key.index);
             let scheduled = self.runtime.get_component_private(&connector_key);
             // Keep running until we should no longer immediately schedule the
             // connector.
             let mut cur_schedule = ConnectorScheduling::Immediate;
             while cur_schedule == ConnectorScheduling::Immediate {
                 // Check all the message that are in the shared inbox
                 while let Some(message) = scheduled.public.inbox.take_message() {
                     // Check for rerouting
                     println!("DEBUG [{}]: Handling message from {}:{}\n{:#?}", scheduler_id, message.sending_connector.0, message.receiving_port.index, message);
                     if let Some(other_connector_id) = scheduled.router.should_reroute(message.sending_connector, message.receiving_port) {
                         self.send_message_and_wake_up_if_sleeping(other_connector_id, message);
                         continue;
+                    }
                     // Check for messages that requires special action from the
@@ @@ -194,13 +155,13 @@ impl Scheduler { @@
             // 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.global_store.connector_queue.push_back(connector_key);
+                    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`.
                     debug_assert_eq!(scheduled.public.sleeping.load(Ordering::Acquire), false);
@@ @@ -212,15 +173,20 @@ impl Scheduler { @@
                     if !scheduled.public.inbox.is_empty() {
                         let should_reschedule_self = scheduled.public.sleeping
                             .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)
                             .is_ok();
                         if should_reschedule_self {
-                            self.runtime.global_store.connector_queue.push_back(connector_key);
+                            self.runtime.push_work(connector_key);
+                        }
+                    }
                 },
                 ConnectorScheduling::Exit => {
                     // TODO: Better way of doing this, when exiting then
                     //  connected components must know their channels are invalid
                     self.runtime.destroy_component(connector_key);
+                }
+            }
+        }
+    }
     fn handle_delta_state(&mut self, connector_key: &ConnectorKey, context: &mut ConnectorCtx, delta_state: &mut RunDeltaState) {
@@ @@ -280,18 +246,18 @@ impl Scheduler { @@
+            }
+        }
         // Handling any new connectors that were scheduled
         // TODO: Pool outgoing messages to reduce atomic access
         if !delta_state.new_connectors.is_empty() {
-            let cur_connector = self.runtime.global_store.connectors.get_mut(connector_key);
+            let cur_connector = self.runtime.get_component_private(connector_key);
             for new_connector in delta_state.new_connectors.drain(..) {
                 // Add to global registry to obtain key
                 let new_key = self.runtime.global_store.connectors.create_pdl(cur_connector, new_connector);
                 let new_connector = self.runtime.global_store.connectors.get_mut(&new_key);
                 let new_key = self.runtime.create_pdl_component(cur_connector, new_connector);
                 let new_connector = self.runtime.get_component_private(&new_key);
                 // Call above changed ownership of ports, but we still have to
                 // let the other end of the channel know that the port has
                 // changed location.
                 for port in &new_connector.context.ports {
                     let reroute_message = cur_connector.router.prepare_reroute(
@@ @@ -300,28 +266,32 @@ impl Scheduler { @@
                     );
                     self.send_message_and_wake_up_if_sleeping(port.peer_connector, reroute_message);
+                }
                 // Schedule new connector to run
-                self.runtime.global_store.connector_queue.push_back(new_key);
+                self.runtime.push_work(new_key);
+            }
+        }
         debug_assert!(delta_state.outbox.is_empty());
         debug_assert!(delta_state.new_ports.is_empty());
         debug_assert!(delta_state.new_connectors.is_empty());
+    }
     fn send_message_and_wake_up_if_sleeping(&self, connector_id: ConnectorId, message: Message) {
-        let connector = self.runtime.global_store.connectors.get_shared(connector_id);
+        let connector = self.runtime.get_component_public(connector_id);
         connector.inbox.insert_message(message);
         let should_wake_up = connector.sleeping
             .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)
             .is_ok();
         if should_wake_up {
             let key = unsafe { ConnectorKey::from_id(connector_id) };
-            self.runtime.global_store.connector_queue.push_back(key);
+            self.runtime.push_work(key);
+        }
+    }
+}
 /// Represents a rerouting entry due to a moved port
 // TODO: Optimize

0 comments (0 inline, 0 general)