CSY/reowolf Changeset - 7ce4d293022a · Centrum Wiskunde & Informatica (CWI)

Changeset - 7ce4d293022a

Parent rev.

Child rev.

[Not reviewed]

0 6 0

MH - 4 years ago 2021-11-09 18:41:05
contact@maxhenger.nl

remove some debug logging, did basic perf tests

6 files changed with 27 insertions and 22 deletions:

src/runtime2/consensus.rs

src/runtime2/mod.rs

src/runtime2/native.rs

src/runtime2/port.rs

src/runtime2/scheduler.rs

src/runtime2/tests/mod.rs

0 comments (0 inline, 0 general)

src/runtime2/consensus.rs

➞

Show inline comments

@@ @@ -437,50 +437,48 @@ impl Consensus { @@
                 let message = SyncMessage {
                     sync_header: self.create_sync_header(ctx),
                     target_component_id: *encountered_id,
                     content: SyncContent::Notification,
                 };
                 ctx.submit_message(Message::Sync(message));
+            }
             // But also send our locally combined solution
             self.forward_local_solutions(ctx);
         } else if sync_header.highest_component_id < self.highest_connector_id {
             // Sender has lower leader ID, so it should know about our higher
             // one.
             let message = SyncMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: sync_header.sending_component_id,
                 content: SyncContent::Notification
             };
             ctx.submit_message(Message::Sync(message));
         } // else: exactly equal, so do nothing
+    }
     fn send_or_store_local_solution(&mut self, solution: LocalSolution, ctx: &mut ComponentCtx) -> Option<BranchId> {
         println!("DEBUG [....:.. conn:{:02}]: Storing local solution for component {}, branch {}", ctx.id.0, solution.component.0, solution.final_branch_id.index);
         if self.highest_connector_id == ctx.id {
             // We are the leader
             if let Some(global_solution) = self.solution_combiner.add_solution_and_check_for_global_solution(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 = SyncMessage {
                         sync_header: self.create_sync_header(ctx),
                         target_component_id: connector_id,
                         content: SyncContent::GlobalSolution(global_solution.clone()),
                     };
                     ctx.submit_message(Message::Sync(message));
+                }
                 debug_assert!(my_final_branch_id.is_valid());
                 return Some(my_final_branch_id);
             } else {
                 return None;
+            }

src/runtime2/mod.rs

➞

Show inline comments

@@ @@ -285,95 +285,89 @@ impl RuntimeInner { @@
     #[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);
         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) {
         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 == 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);
         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 {

src/runtime2/native.rs

➞

Show inline comments

@@ @@ -48,54 +48,52 @@ impl ConnectorApplication { @@
         };
         let interface = ApplicationInterface::new(sync_done, job_queue, runtime);
         return (connector, interface);
+    }
+}
 impl Connector for ConnectorApplication {
     fn run(&mut self, _sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         // Handle any incoming messages if we're participating in a round
         while let Some(message) = comp_ctx.read_next_message() {
             match message {
                 Message::Data(_) => todo!("data message in API connector"),
                 Message::Sync(_)  => todo!("sync message in API connector"),
                 Message::Control(_) => todo!("impossible control message"),
+            }
+        }
         // Handle requests coming from the API
+        {
             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");
                         comp_ctx.push_port(endpoint_a);
                         comp_ctx.push_port(endpoint_b);
+                    }
                     ApplicationJob::NewConnector(connector, initial_ports) => {
                         println!("DEBUG: API creating connector");
                         comp_ctx.push_component(connector, initial_ports);
                     },
                     ApplicationJob::Shutdown => {
                         debug_assert!(queue.is_empty());
                         return ConnectorScheduling::Exit;
+                    }
+                }
+            }
+        }
         return ConnectorScheduling::NotNow;
+    }
+}
 /// The interface to a `ApplicationConnector`. This allows setting up the
 /// interactions the `ApplicationConnector` performs within a synchronous round.
 pub struct ApplicationInterface {
     sync_done: SyncDone,
     job_queue: JobQueue,
     runtime: Arc<RuntimeInner>,
     connector_id: ConnectorId,
     owned_ports: Vec<PortIdLocal>,
+}
@@ @@ -175,44 +173,41 @@ impl ApplicationInterface { @@
     pub fn wait(&self) {
         let (is_done, condition) = &*self.sync_done;
         let lock = is_done.lock().unwrap();
         condition.wait_while(lock, |v| !*v).unwrap(); // wait while not done
+    }
     /// 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::Control(ControlMessage {
             id: 0,
             sending_component_id: self.connector_id,
             content: ControlContent::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

@@ @@ -37,31 +37,29 @@ pub enum PortKind { @@
     Putter,
     Getter,
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq)]
 pub enum PortState {
     Open,
     Closed,
+}
 /// Represents a port inside of the runtime. This is generally the local view of
 /// a connector on its port, which may not be consistent with the rest of the
 /// global system (e.g. its peer was moved to a new connector, or the peer might
 /// have died in the meantime, so it is no longer usable).
 #[derive(Clone)]
 pub struct Port {
     pub self_id: PortIdLocal,
     pub peer_id: PortIdLocal,
     pub channel_id: ChannelId,
     pub kind: PortKind,
     pub state: PortState,
     pub peer_connector: ConnectorId, // might be temporarily inconsistent while peer port is sent around in non-sync phase
+}
 // TODO: Turn port ID into its own type
 pub struct Channel {
     pub putter_id: PortIdLocal, // can put on it, so from the connector's point of view, this is an output
     pub getter_id: PortIdLocal, // vice versa: can get on it, so an input for the connector
+}
@@ \ No newline at end of file @@

src/runtime2/scheduler.rs

➞

Show inline comments

@@ @@ -326,53 +326,53 @@ impl Scheduler { @@
                 self.runtime.push_work(connector_key)
+            }
+        }
+    }
     #[inline]
     fn get_message_target_port(message: &Message) -> Option<PortIdLocal> {
         match message {
             Message::Data(data) => return Some(data.data_header.target_port),
             Message::Sync(_) => {},
             Message::Control(control) => {
                 match &control.content {
                     ControlContent::PortPeerChanged(port_id, _) => return Some(*port_id),
                     ControlContent::CloseChannel(port_id) => return Some(*port_id),
                     ControlContent::Ping | ControlContent::Ack => {},
+                }
             },
+        }
         return None
+    }
     // TODO: Remove, this is debugging stuff
     fn debug(&self, message: &str) {
         println!("DEBUG [thrd:{:02} conn:  ]: {}", self.scheduler_id, message);
+        // 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.0, 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.

src/runtime2/tests/mod.rs

➞

Show inline comments

 use super::*;
 use crate::{PortId, ProtocolDescription};
 use crate::common::Id;
 use crate::protocol::eval::*;
 const NUM_THREADS: u32 = 10;  // number of threads in runtime
 const NUM_INSTANCES: u32 = 10;  // number of test instances constructed
 const NUM_LOOPS: u32 = 10;      // number of loops within a single test (not used by all tests)
 const NUM_THREADS: u32 = 1;  // number of threads in runtime
 const NUM_INSTANCES: u32 = 4;  // number of test instances constructed
 const NUM_LOOPS: u32 = 4;      // number of loops within a single test (not used by all tests)
 fn create_runtime(pdl: &str) -> Runtime {
     let protocol = ProtocolDescription::parse(pdl.as_bytes()).expect("parse pdl");
     let runtime = Runtime::new(NUM_THREADS, protocol);
     return runtime;
+}
 fn run_test_in_runtime<F: Fn(&mut ApplicationInterface)>(pdl: &str, constructor: F) {
     let protocol = ProtocolDescription::parse(pdl.as_bytes())
         .expect("parse PDL");
     let runtime = Runtime::new(NUM_THREADS, protocol);
     let mut api = runtime.create_interface();
     for _ in 0..NUM_INSTANCES {
         constructor(&mut api);
+    }
     // Wait until done :)
+}
 struct TestTimer {
     name: &'static str,
     started: std::time::Instant
+}
 impl TestTimer {
     fn new(name: &'static str) -> Self {
         Self{ name, started: std::time::Instant::now() }
+    }
+}
 impl Drop for TestTimer {
     fn drop(&mut self) {
         let delta = std::time::Instant::now() - self.started;
         let nanos = (delta.as_secs_f64() * 1_000_000.0) as u64;
         let millis = nanos / 1000;
         let nanos = nanos % 1000;
         println!("[{}] Took {:>4}.{:03} ms", self.name, millis, nanos);
+    }
+}
 #[test]
 fn test_put_and_get() {
     const CODE: &'static str = "
     primitive putter(out<bool> sender, u32 loops) {
         u32 index = 0;
         while (index < loops) {
             synchronous {
                 print(\"putting!\");
                 put(sender, true);
+            }
             index += 1;
+        }
+    }
     primitive getter(in<bool> receiver, u32 loops) {
         u32 index = 0;
         while (index < loops) {
             synchronous {
                 print(\"getting!\");
                 auto result = get(receiver);
                 assert(result);
+            }
             index += 1;
+        }
+    }
     ";
     let thing = TestTimer::new("hello");
     run_test_in_runtime(CODE, |api| {
         let channel = api.create_channel();
         api.create_connector("", "putter", ValueGroup::new_stack(vec![
             Value::Output(PortId(Id{ connector_id: 0, u32_suffix: channel.putter_id.index })),
             Value::UInt32(NUM_LOOPS)
         ])).expect("create putter");
         api.create_connector("", "getter", ValueGroup::new_stack(vec![
             Value::Input(PortId(Id{ connector_id: 0, u32_suffix: channel.getter_id.index })),
             Value::UInt32(NUM_LOOPS)
         ])).expect("create getter");
     });
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)