CSY/reowolf Changeset - 1b179e5f4579 · Centrum Wiskunde & Informatica (CWI)

Changeset - 1b179e5f4579

Parent rev.

Child rev.

[Not reviewed]

0 4 0

MH - 4 years ago 2021-11-12 13:07:38
contact@maxhenger.nl

Bugfixes in initial API component

4 files changed with 75 insertions and 11 deletions:

src/runtime2/branch.rs

src/runtime2/native.rs

src/runtime2/tests/api_component.rs

src/runtime2/tests/mod.rs

0 comments (0 inline, 0 general)

src/runtime2/branch.rs

➞

Show inline comments

@@ @@ -419,13 +419,13 @@ impl FakeTree { @@
             }],
             queues: [BranchQueue::new(); 3]
+        }
+    }
     fn is_in_sync(&self) -> bool {
-        return !self.branches.is_empty();
+        return self.branches.len() > 1;
+    }
     pub fn queue_is_empty(&self, kind: QueueKind) -> bool {
         return self.queues[kind.as_index()].is_empty();
+    }
@@ @@ -457,13 +457,13 @@ impl FakeTree { @@
+    }
     pub fn start_sync(&mut self) -> BranchId {
         debug_assert!(!self.is_in_sync());
         // Create the first branch
-        let sync_branch = FakeBranch::new_root(0);
+        let sync_branch = FakeBranch::new_root(1);
         let sync_branch_id = sync_branch.id;
         self.branches.push(sync_branch);
         return sync_branch_id;
+    }

src/runtime2/native.rs

➞

Show inline comments

@@ @@ -94,25 +94,25 @@ impl ConnectorApplication { @@
             job_queue: job_queue.clone(),
             is_in_sync: false,
             sync_desc: Vec::new(),
             tree: FakeTree::new(),
             consensus: Consensus::new(),
             last_finished_handled: None,
-            branch_extra: Vec::new(),
+            branch_extra: vec![0],
         };
         let interface = ApplicationInterface::new(sync_done, job_queue, runtime);
         return (connector, interface);
+    }
     fn handle_new_messages(&mut self, comp_ctx: &mut ComponentCtx) {
         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"),
                 Message::Data(message) => self.handle_new_data_message(message, comp_ctx),
                 Message::Sync(message) => self.handle_new_sync_message(message, comp_ctx),
                 Message::Control(_) => unreachable!("control message in native API component"),
+            }
+        }
+    }
     pub(crate) fn handle_new_data_message(&mut self, message: DataMessage, ctx: &mut ComponentCtx) {
         // Go through all branches that are awaiting new messages and see if
@@ @@ -129,12 +129,14 @@ impl ConnectorApplication { @@
             let branch = &self.tree[branch_id];
             if branch.awaiting_port != message.data_header.target_port { continue; }
             if !self.consensus.branch_can_receive(branch_id, &message) { continue; }
             // This branch can receive, so fork and given it the message
             let receiving_branch_id = self.tree.fork_branch(branch_id);
             debug_assert!(receiving_branch_id.index as usize == self.branch_extra.len());
             self.branch_extra.push(self.branch_extra[branch_id.index as usize]); // copy instruction index
             self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
             let receiving_branch = &mut self.tree[receiving_branch_id];
             receiving_branch.insert_message(message.data_header.target_port, message.content.as_message().unwrap().clone());
             self.consensus.notify_of_received_message(receiving_branch_id, &message);
@@ @@ -244,43 +246,55 @@ impl ConnectorApplication { @@
     fn run_in_deterministic_mode(&mut self, _sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         debug_assert!(!self.is_in_sync);
         // In non-sync mode the application component doesn't really do anything
         // except performing jobs submitted from the API. This is the only
         // case where we expect to be woken up.
         // Note that we have to communicate to the scheduler when we've received
         // ports or created components (hence: given away ports) *before* we
         // enter a sync round.
         let mut queue = self.job_queue.lock().unwrap();
         while let Some(job) = queue.pop_front() {
             match job {
                 ApplicationJob::NewChannel((endpoint_a, endpoint_b)) => {
                     comp_ctx.push_port(endpoint_a);
                     comp_ctx.push_port(endpoint_b);
                     return ConnectorScheduling::Immediate;
+                }
                 ApplicationJob::NewConnector(connector, initial_ports) => {
                     comp_ctx.push_component(connector, initial_ports);
                     return ConnectorScheduling::Later;
                 },
                 ApplicationJob::SyncRound(mut description) => {
                     // Entering sync mode
                     comp_ctx.notify_sync_start();
                     self.sync_desc = description;
                     self.is_in_sync = true;
                     debug_assert!(self.last_finished_handled.is_none());
-                    debug_assert!(self.branch_extra.is_empty());
+                    debug_assert!(self.branch_extra.len() == 1);
                     let first_branch_id = self.tree.start_sync();
                     self.tree.push_into_queue(QueueKind::Runnable, first_branch_id);
                     debug_assert!(first_branch_id.index == 1);
                     self.consensus.start_sync(comp_ctx);
                     self.consensus.notify_of_new_branch(BranchId::new_invalid(), first_branch_id);
                     self.branch_extra.push(0); // set first branch to first instruction
                     return ConnectorScheduling::Immediate;
                 },
                 ApplicationJob::Shutdown => {
                     debug_assert!(queue.is_empty());
                     return ConnectorScheduling::Exit;
+                }
+            }
+        }
         // Queue was empty
         return ConnectorScheduling::NotNow;
+    }
     fn collapse_sync_to_solution_branch(&mut self, branch_id: BranchId, comp_ctx: &mut ComponentCtx) {
         debug_assert!(self.branch_extra[branch_id.index as usize] >= self.sync_desc.len()); // finished program
         // Notifying tree, consensus algorithm and context of ending sync
@@ @@ -306,13 +320,13 @@ impl ConnectorApplication { @@
+            }
+        }
         // Notifying interface of ending sync
         self.is_in_sync = false;
         self.sync_desc.clear();
-        self.branch_extra.clear();
+        self.branch_extra.truncate(1);
         self.last_finished_handled = None;
         let (results, notification) = &*self.sync_done;
         let mut results = results.lock().unwrap();
         *results = Some(FinishedSync{ inbox });
         notification.notify_one();
@@ @@ -476,21 +490,22 @@ impl ApplicationInterface { @@
         self.wake_up_connector_with_ping();
         return Ok(())
+    }
     /// Wait until the next sync-round is finished
     pub fn wait(&self) -> Result<Vec<ValueGroup>, ApplicationEndSyncError> {
+    pub fn wait(&mut self) -> Result<Vec<ValueGroup>, ApplicationEndSyncError> {
         if !self.is_in_sync {
             return Err(ApplicationEndSyncError::NotInSync);
+        }
         let (is_done, condition) = &*self.sync_done;
         let mut lock = is_done.lock().unwrap();
         lock = condition.wait_while(lock, |v| v.is_none()).unwrap(); // wait while not done
         self.is_in_sync = false;
         return Ok(lock.take().unwrap().inbox);
+    }
     /// Called by runtime to set associated connector's ID.
     pub(crate) fn set_connector_id(&mut self, id: ConnectorId) {
         self.connector_id = id;

src/runtime2/tests/api_component.rs

➞

Show inline comments

 // Testing the api component.
 //
 // These tests explicitly do not use the "NUM_INSTANCES" constant because we're
 // doing some communication with the native component. Hence only expect one
 use super::*;
 #[test]
 fn test_put_and_get() {
     const CODE: &'static str = "
     primitive handler(in<u32> request, out<u32> response, u32 loops) {
         u32 index = 0;
         while (index < loops) {
             synchronous {
                 auto value = get(request);
                 put(response, value * 2);
+            }
             index += 1;
+        }
+    }
     ";
     let pd = ProtocolDescription::parse(CODE.as_bytes()).unwrap();
     let rt = Runtime::new(NUM_THREADS, pd);
     let mut api = rt.create_interface();
     let req_chan = api.create_channel().unwrap();
     let resp_chan = api.create_channel().unwrap();
     api.create_connector("", "handler", ValueGroup::new_stack(vec![
         Value::Input(PortId::new(req_chan.getter_id.index)),
         Value::Output(PortId::new(resp_chan.putter_id.index)),
         Value::UInt32(NUM_LOOPS),
     ])).unwrap();
     for loop_idx in 0..NUM_LOOPS {
         api.perform_sync_round(vec![
             ApplicationSyncAction::Put(req_chan.putter_id, ValueGroup::new_stack(vec![Value::UInt32(loop_idx)])),
             ApplicationSyncAction::Get(resp_chan.getter_id)
         ]).expect("start sync round");
         let result = api.wait().expect("finish sync round");
         assert!(result.len() == 1);
         if let Value::UInt32(gotten) = result[0].values[0] {
             assert_eq!(gotten, loop_idx * 2);
         } else {
             assert!(false);
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/tests/mod.rs

➞

Show inline comments

@@ @@ -2,12 +2,13 @@ mod network_shapes; @@
 mod api_component;
 use super::*;
 use crate::{PortId, ProtocolDescription};
 use crate::common::Id;
 use crate::protocol::eval::*;
 use crate::runtime2::native::{ApplicationSyncAction};
 //
 // Generic testing constants, use when appropriate to simplify stress-testing
 pub(crate) const NUM_THREADS: u32 = 3;     // number of threads in runtime
 pub(crate) const NUM_INSTANCES: u32 = 5;   // number of test instances constructed
     let runtime = Runtime::new(NUM_THREADS, protocol);
     let mut api = runtime.create_interface();
     for _ in 0..NUM_INSTANCES {
         constructor(&mut api);
+    }
     // Wait until done :)
+}
 pub(crate) struct TestTimer {
     name: &'static str,
     started: std::time::Instant
+}

0 comments (0 inline, 0 general)