CSY/reowolf Changeset - 305bb82af392 · Centrum Wiskunde & Informatica (CWI)

Changeset - 305bb82af392

Parent rev.

Child rev.

[Not reviewed]

0 3 0

mh - 4 years ago 2021-11-26 10:03:11
contact@maxhenger.nl

Add generation counter to connector store

3 files changed with 78 insertions and 32 deletions:

src/runtime2/consensus.rs

src/runtime2/mod.rs

src/runtime2/scheduler.rs

0 comments (0 inline, 0 general)

src/runtime2/consensus.rs

➞

Show inline comments

@@ @@ -701,193 +701,193 @@ impl Consensus { @@
+                        }
+                    }
                 },
                 SyncCompContent::Presence(component_presence) => {
                     if self.solution_combiner.add_presence_and_check_for_global_failure(component_presence.component_id, &component_presence.channels) {
                         return self.handle_global_failure_as_leader(ctx);
+                    }
                 },
                 SyncCompContent::AckFailure => {
                     debug_assert_eq!(Some(RoundConclusion::Failure), self.conclusion);
                     debug_assert!(self.ack_remaining > 0);
                     self.ack_remaining -= 1;
                     if self.ack_remaining == 0 {
                         return Some(RoundConclusion::Failure);
+                    }
+                }
                 SyncCompContent::Notification | SyncCompContent::GlobalSolution(_) |
                 SyncCompContent::GlobalFailure => {
                     unreachable!("unexpected message content for leader");
                 },
+            }
         } else {
             // Someone else is the leader
             let message = SyncCompMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: self.highest_connector_id,
                 content,
             };
             ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel
+        }
         return None;
+    }
     fn handle_global_solution_as_leader(&mut self, global_solution: GlobalSolution, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         if self.conclusion.is_some() {
             return None;
+        }
         // Handle the global solution
         let mut my_final_branch_id = BranchId::new_invalid();
         for (connector_id, branch_id) in global_solution.component_branches.iter().copied() {
             if connector_id == ctx.id {
                 // This is our solution branch
                 my_final_branch_id = branch_id;
                 continue;
+            }
             let message = SyncCompMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: connector_id,
                 content: SyncCompContent::GlobalSolution(global_solution.clone()),
             };
             ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel
+        }
         debug_assert!(my_final_branch_id.is_valid());
         self.conclusion = Some(RoundConclusion::Success(my_final_branch_id));
         return Some(RoundConclusion::Success(my_final_branch_id));
+    }
     fn handle_global_failure_as_leader(&mut self, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         debug_assert!(self.solution_combiner.failure_reported && self.solution_combiner.check_for_global_failure());
         if self.conclusion.is_some() {
             // Already sent out a failure
             return None;
+        }
         // TODO: Performance
         let mut encountered = VecSet::new();
         for presence in &self.solution_combiner.presence {
             if presence.owner_a != ctx.id {
                 // Did not add it ourselves
                 if encountered.push(presence.owner_a) {
                     // Not yet sent a message
                     let message = SyncCompMessage{
                         sync_header: self.create_sync_header(ctx),
                         target_component_id: presence.owner_a,
                         content: SyncCompContent::GlobalFailure,
                     };
                     ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel
+                }
             } else if let Some(owner_b) = presence.owner_b {
                 if owner_b != ctx.id {
                     if encountered.push(owner_b) {
                         let message = SyncCompMessage{
                             sync_header: self.create_sync_header(ctx),
                             target_component_id: owner_b,
                             content: SyncCompContent::GlobalFailure,
                         };
                         ctx.submit_message(Message::SyncComp(message)).unwrap();
+                    }
+                }
+            }
+        }
-        println!("DEBUERINO: Leader entering error state, we need to wait on {:?}", encountered.iter().map(|v| v.0).collect::<Vec<_>>());
+        println!("DEBUGERINO: Leader entering error state, we need to wait on {:?}", encountered.iter().map(|v| v.index).collect::<Vec<_>>());
         self.conclusion = Some(RoundConclusion::Failure);
         if encountered.is_empty() {
             // We don't have to wait on Acks
             return Some(RoundConclusion::Failure);
         } else {
             self.ack_remaining = encountered.len() as u32;
             return None;
+        }
+    }
     fn initiate_sync_failure(&mut self, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         debug_assert!(self.is_in_sync());
         // Notify leader of our channels and the fact that we just failed
         let channel_mapping = &self.branch_annotations[0].channel_mapping;
         let mut channel_presence = Vec::with_capacity(channel_mapping.len());
         for mapping in channel_mapping {
             let port = ctx.get_port_by_channel_id(mapping.channel_id).unwrap();
             channel_presence.push(LocalChannelPresence{
                 channel_id: mapping.channel_id,
                 is_closed: port.state == PortState::Closed,
             });
+        }
         let maybe_already = self.send_to_leader_or_handle_as_leader(SyncCompContent::Presence(ComponentPresence{
             component_id: ctx.id,
             channels: channel_presence,
         }), ctx);
         if self.handled_wave {
             // Someone (or us) has already initiated a sync failure.
             return maybe_already;
+        }
         let maybe_conclusion = self.send_to_leader_or_handle_as_leader(SyncCompContent::LocalFailure, ctx);
         debug_assert!(if maybe_already.is_some() { maybe_conclusion.is_some() } else { true });
         println!("DEBUG: Maybe conclusion is {:?}", maybe_conclusion);
         // Initiate a discovery wave so peers can do the same
         self.handled_wave = true;
         for mapping in &self.branch_annotations[0].channel_mapping {
             let channel_id = mapping.channel_id;
             let port_info = ctx.get_port_by_channel_id(channel_id).unwrap();
             let message = SyncPortMessage{
                 sync_header: self.create_sync_header(ctx),
                 source_port: port_info.self_id,
                 target_port: port_info.peer_id,
                 content: SyncPortContent::NotificationWave,
             };
             // Note: submitting the message might fail. But we're attempting to
             // handle the error anyway.
             // TODO: Think about this a second time: how do we make sure the
             //  entire network will fail if we reach this condition
             let _unused = ctx.submit_message(Message::SyncPort(message));
+        }
         return maybe_conclusion;
+    }
     #[inline]
     fn create_sync_header(&self, ctx: &ComponentCtx) -> SyncHeader {
         return SyncHeader{
             sending_component_id: ctx.id,
             highest_component_id: self.highest_connector_id,
             sync_round: self.sync_round,
+        }
+    }
     fn forward_local_data_to_new_leader(&mut self, ctx: &mut ComponentCtx) {
         debug_assert_ne!(self.highest_connector_id, ctx.id);
         if let Some(partial_solution) = self.solution_combiner.drain() {
             let message = SyncCompMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: self.highest_connector_id,
                 content: SyncCompContent::PartialSolution(partial_solution),
             };
             ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel
+        }
+    }
+}
 // -----------------------------------------------------------------------------
 // Solution storage and algorithms
 // -----------------------------------------------------------------------------
 // TODO: Remove all debug derives
 #[derive(Debug, Clone)]
 struct MatchedLocalSolution {
     final_branch_id: BranchId,
     channel_mapping: Vec<(ChannelId, BranchMarker)>,
     matches: Vec<ComponentMatches>,
+}
 #[derive(Debug, Clone)]

src/runtime2/mod.rs

➞

Show inline comments

 // Structure of module
 mod branch;
 mod native;
 mod port;
 mod scheduler;
 mod consensus;
 mod inbox;
 #[cfg(test)] mod tests;
 mod connector;
 // Imports
 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 connector::{ConnectorPDL, ConnectorPublic, ConnectorScheduling};
 use scheduler::{Scheduler, ComponentCtx, SchedulerCtx, ControlMessageHandler};
 use native::{Connector, ConnectorApplication, ApplicationInterface};
 use inbox::Message;
 use port::{ChannelId, Port, PortState};
 /// 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
     pub generation: u32,
+}
 impl ConnectorKey {
     /// Downcasts the `ConnectorKey` type, which can be used to obtain mutable
     /// access, to a "regular ID" which can be used to obtain immutable access.
     #[inline]
     pub fn downcast(&self) -> ConnectorId {
         return ConnectorId(self.index);
         return ConnectorId{
             index: self.index,
             generation: self.generation,
         };
+    }
     /// Turns the `ConnectorId` into a `ConnectorKey`, marked as unsafe as it
     /// bypasses the type-enforced `ConnectorKey`/`ConnectorId` system
     #[inline]
     pub unsafe fn from_id(id: ConnectorId) -> ConnectorKey {
         return ConnectorKey{ index: id.0 };
         return ConnectorKey{
             index: id.index,
             generation: id.generation,
         };
+    }
+}
 /// A kind of token that allows shared access to a connector. Multiple threads
 /// may hold this
 #[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
 pub struct ConnectorId(pub u32);
 #[derive(Debug, Copy, Clone)]
 pub struct ConnectorId{
     pub index: u32,
     pub generation: u32,
+}
 impl PartialEq for ConnectorId {
     fn eq(&self, other: &Self) -> bool {
         return self.index.eq(&other.index);
+    }
+}
 impl Eq for ConnectorId{}
 impl PartialOrd for ConnectorId{
     fn partial_cmp(&self, other: &Self) -> Option<crate::common::Ordering> {
         return self.index.partial_cmp(&other.index)
+    }
+}
 impl Ord for ConnectorId{
     fn cmp(&self, other: &Self) -> crate::common::Ordering {
         return self.partial_cmp(other).unwrap();
+    }
+}
 impl ConnectorId {
     // TODO: Like the other `new_invalid`, maybe remove
     #[inline]
     pub fn new_invalid() -> ConnectorId {
         return ConnectorId(u32::MAX);
         return ConnectorId {
             index: u32::MAX,
             generation: 0,
         };
+    }
     #[inline]
     pub(crate) fn is_valid(&self) -> bool {
-        return self.0 != u32::MAX;
+        return self.index != u32::MAX;
+    }
+}
 // TODO: Change this, I hate this. But I also don't want to put `public` and
 //  `router` of `ScheduledConnector` back into `Connector`. The reason I don't
 //  want `Box<dyn Connector>` everywhere is because of the v-table overhead. But
 //  to truly design this properly I need some benchmarks.
 pub(crate) enum ConnectorVariant {
     UserDefined(ConnectorPDL),
     Native(Box<dyn Connector>),
+}
 impl Connector for ConnectorVariant {
     fn run(&mut self, scheduler_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         match self {
             ConnectorVariant::UserDefined(c) => c.run(scheduler_ctx, comp_ctx),
             ConnectorVariant::Native(c) => c.run(scheduler_ctx, comp_ctx),
+        }
+    }
+}
 pub(crate) struct ScheduledConnector {
     pub connector: ConnectorVariant, // access by connector
     pub ctx: ComponentCtx,
     pub public: ConnectorPublic, // accessible by all schedulers and connectors
     pub router: ControlMessageHandler,
     pub shutting_down: bool,
+}
 // -----------------------------------------------------------------------------
 // Runtime
 // -----------------------------------------------------------------------------
 /// Externally facing runtime.
 pub struct Runtime {
     inner: Arc<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{
             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()),
             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 {
                 let cloned_runtime_inner = runtime_inner.clone();
                 let thread = thread::Builder::new()
                     .name(format!("thread-{}", thread_index))
                     .spawn(move || {
                         let mut scheduler = Scheduler::new(cloned_runtime_inner, thread_index);
                         scheduler.run();
                     })
                     .unwrap();
                 schedulers.push(thread);
+            }
             let mut lock = runtime_inner.schedulers.lock().unwrap();
             *lock = schedulers;
+        }
         // Return 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.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.decrement_active_interfaces();
         let mut lock = self.inner.schedulers.lock().unwrap();
@@ @@ -267,212 +300,225 @@ impl RuntimeInner { @@
         self.increment_active_components();
         return key;
+    }
     /// Creates a new PDL component. This function just creates the component.
     /// If you create it initially awake, then you must add it to the work
     /// queue. Other aspects of correctness (i.e. setting initial ports) are
     /// relinquished to the caller!
     pub(crate) fn create_pdl_component(&self, connector: ConnectorPDL, initially_sleeping: bool) -> 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), initially_sleeping)
         };
         self.increment_active_components();
         return key;
+    }
     #[inline]
     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);
+    }
     #[inline]
     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) {
         let _old_num = 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 == 1 { // 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);
         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.
-    connectors: RawVec<*mut ScheduledConnector>,
+    entries: RawVec<*mut StoreEntry>,
     free: Vec<usize>,
+}
 impl ConnectorStore {
     fn with_capacity(capacity: usize) -> Self {
         Self {
-            connectors: RawVec::with_capacity(capacity),
+            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 {
             debug_assert!(!self.free.contains(&(id.0 as usize)));
             let connector = self.connectors.get(id.0 as usize);
             debug_assert!(!connector.is_null());
             return &(**connector).public;
             let entry = self.entries.get(id.index as usize);
             let cur_generation = (**entry).generation.load(Ordering::Acquire);
             debug_assert_eq!(cur_generation, id.generation);
             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 {
             debug_assert!(!self.free.contains(&(key.index as usize)));
             let connector = self.connectors.get_mut(key.index as usize);
             debug_assert!(!connector.is_null());
             return &mut (**connector);
             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.connectors.len();
             key = ConnectorKey{ index: index as u32 };
             index = self.entries.len();
             key = ConnectorKey{
                 index: index as u32, generation: 0
             };
             connector.ctx.id = key.downcast();
             let connector = Box::into_raw(Box::new(connector));
             self.connectors.push(connector);
             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();
             key = ConnectorKey{ index: index as u32 };
             connector.ctx.id = key.downcast();
             unsafe {
                 let target = self.connectors.get_mut(index);
                 std::ptr::write(*target, connector);
                 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;
+    }
     /// 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);
             let target = self.entries.get_mut(key.index as usize);
             (**target).generation.fetch_add(1, Ordering::SeqCst);
             std::ptr::drop_in_place(*target);
             // Note: but not deallocating!
+        }
         println!("DEBUG [ global store  ] Destroyed component at {}", key.index);
         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 memory = self.entries.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() {
+        for idx in 0..self.entries.len() {
             unsafe {
-                let memory = *self.connectors.get_mut(idx);
+                let memory = *self.entries.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/scheduler.rs

➞

Show inline comments

@@ @@ -263,299 +263,299 @@ impl Scheduler { @@
                     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.debug_conn(connector_id, &format!("Sending message [newcon]\n --- {:?}", reroute_message));
                             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.0);
+        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.
         if !connector.public.inbox.is_empty() {
             // Try to wake ourselves up (needed because someone might be trying
             // the exact same atomic compare-and-swap at this point in time)
             let should_wake_up_again = connector.public.sleeping
                 .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)
                 .is_ok();
             if should_wake_up_again {
                 self.runtime.push_work(connector_key)
+            }
+        }
+    }
     // TODO: Remove, this is debugging stuff
     fn debug(&self, message: &str) {
         println!("DEBUG [thrd:{:02} conn:  ]: {}", self.scheduler_id, message);
+    }
     fn debug_conn(&self, conn: ConnectorId, message: &str) {
-        println!("DEBUG [thrd:{:02} conn:{:02}]: {}", self.scheduler_id, conn.0, message);
+        println!("DEBUG [thrd:{:02} conn:{:02}]: {}", self.scheduler_id, conn.index, message);
+    }
+}
 // -----------------------------------------------------------------------------
 // ComponentCtx
 // -----------------------------------------------------------------------------
 enum ComponentStateChange {
     CreatedComponent(ConnectorPDL, Vec<PortIdLocal>),
     CreatedPort(Port),
     ChangedPort(ComponentPortChange),
+}
 #[derive(Clone)]
 pub(crate) struct ComponentPortChange {
     pub is_acquired: bool, // otherwise: released
     pub port: Port,
+}
 /// 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 ComponentCtx {
     // Mostly managed by the scheduler
     pub(crate) id: ConnectorId,
     ports: Vec<Port>,
     inbox: Inbox,
     // Submitted by the component
     is_in_sync: bool,
     changed_in_sync: bool,
     outbox: VecDeque<Message>,
     state_changes: VecDeque<ComponentStateChange>,
     // Workspaces that may be used by components to (generally) prevent
     // allocations. Be a good scout and leave it empty after you've used it.
     // TODO: Move to scheduler ctx, this is the wrong place
     pub workspace_ports: Vec<PortIdLocal>,
     pub workspace_branches: Vec<BranchId>,
+}
 impl ComponentCtx {
     pub(crate) fn new_empty() -> Self {
         return Self{
             id: ConnectorId::new_invalid(),
             ports: Vec::new(),
             inbox: Inbox::new(),
             is_in_sync: false,
             changed_in_sync: false,
             outbox: VecDeque::new(),
             state_changes: VecDeque::new(),
             workspace_ports: Vec::new(),
             workspace_branches: Vec::new(),
         };
+    }
     /// 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, initial_ports: Vec<PortIdLocal>) {
         debug_assert!(!self.is_in_sync);
         self.state_changes.push_back(ComponentStateChange::CreatedComponent(component, initial_ports));
+    }
     /// 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_back(ComponentStateChange::CreatedPort(port))
+    }
     /// Notify the runtime of an error. Note that this will not perform any
     /// special action beyond printing the error. The component is responsible
     /// for waiting until it is appropriate to shut down (i.e. being outside
     /// of a sync region) and returning the `Exit` scheduling code.
     pub(crate) fn push_error(&mut self, error: EvalError) {
-        println!("ERROR: Component ({}) encountered a critical error:\n{}", self.id.0, error);
+        println!("ERROR: Component ({}) encountered a critical error:\n{}", self.id.index, error);
+    }
     #[inline]
     pub(crate) fn get_ports(&self) -> &[Port] {
         return self.ports.as_slice();
+    }
     pub(crate) fn get_port_by_id(&self, id: PortIdLocal) -> Option<&Port> {
         return self.ports.iter().find(|v| v.self_id == id);
+    }
     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 {

0 comments (0 inline, 0 general)