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

Changeset - c502fc0c252a

Parent rev.

Child rev.

[Not reviewed]

0 3 0

MH - 4 years ago 2021-10-26 20:16:25
contact@maxhenger.nl

Preparing to fix 'delayed port transfer' bug

3 files changed with 75 insertions and 57 deletions:

src/runtime2/mod.rs

src/runtime2/scheduler.rs

src/runtime2/tests/mod.rs

0 comments (0 inline, 0 general)

src/runtime2/mod.rs

➞

Show inline comments

@@ @@ -280,24 +280,26 @@ impl RuntimeInner { @@
         let key = {
             let mut lock = self.connectors.write().unwrap();
             lock.create(ConnectorVariant::UserDefined(connector), false)
         };
         // Transfer the ports
+        {
             let lock = self.connectors.read().unwrap();
             let created = lock.get_private(&key);
             match &created.connector {
                 ConnectorVariant::UserDefined(connector) => {
                     println!("DEBUG: The connector {} owns the ports: {:?}", key.index, connector.ports.owned_ports.iter().map(|v| v.index).collect::<Vec<_>>());
                     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 port = created_by.context.remove_port(port_id);
                         created.context.add_port(port);
+                    }
                 },
                 ConnectorVariant::Native(_) => unreachable!(),
+            }
+        }
         self.increment_active_components();
         return key;

src/runtime2/scheduler.rs

➞

Show inline comments

@@ @@ -114,53 +114,51 @@ impl Scheduler { @@
                             ControlMessageVariant::ChangePortPeer(port_id, new_target_connector_id) => {
                                 // Need to change port target
                                 let port = scheduled.context.get_port_mut(port_id);
                                 port.peer_connector = new_target_connector_id;
                                 // Note: for simplicity we program the scheduler to always finish
                                 // running a connector with an empty outbox. If this ever changes
                                 // then accepting the "port peer changed" message implies we need
                                 // to change the recipient of the message in the outbox.
                                 debug_assert!(delta_state.outbox.is_empty());
                                 // And respond with an Ack
                                 self.runtime.send_message(
                                     message.sending_connector,
                                     Message{
                                         sending_connector: connector_id,
                                         receiving_port: PortIdLocal::new_invalid(),
                                         contents: MessageContents::Control(ControlMessage{
                                             id: content.id,
                                             content: ControlMessageVariant::Ack,
                                         }),
+                                    }
                                 );
                                 let ack_message = Message{
                                     sending_connector: connector_id,
                                     receiving_port: PortIdLocal::new_invalid(),
                                     contents: MessageContents::Control(ControlMessage{
                                         id: content.id,
                                         content: ControlMessageVariant::Ack,
                                     }),
                                 };
                                 self.debug_conn(connector_id, &format!("Sending message [pp ack]\n --- {:?}", ack_message));
                                 self.runtime.send_message(message.sending_connector, ack_message);
                             },
                             ControlMessageVariant::CloseChannel(port_id) => {
                                 // Mark the port as being closed
                                 let port = scheduled.context.get_port_mut(port_id);
                                 port.state = PortState::Closed;
                                 // Send an Ack
                                 self.runtime.send_message(
                                     message.sending_connector,
                                     Message{
                                         sending_connector: connector_id,
                                         receiving_port: PortIdLocal::new_invalid(),
                                         contents: MessageContents::Control(ControlMessage{
                                             id: content.id,
                                             content: ControlMessageVariant::Ack,
                                         }),
+                                    }
                                 );
                                 let ack_message = Message{
                                     sending_connector: connector_id,
                                     receiving_port: PortIdLocal::new_invalid(),
                                     contents: MessageContents::Control(ControlMessage{
                                         id: content.id,
                                         content: ControlMessageVariant::Ack,
                                     }),
                                 };
                                 self.debug_conn(connector_id, &format!("Sending message [cc ack] \n --- {:?}", ack_message));
                                 self.runtime.send_message(message.sending_connector, ack_message);
                             },
                             ControlMessageVariant::Ack => {
                                 scheduled.router.handle_ack(content.id);
+                            }
+                        }
                     } else {
                         // Let connector handle message
                         scheduled.connector.handle_message(message, &scheduled.context, &mut delta_state);
+                    }
+                }
@@ @@ -210,49 +208,50 @@ impl Scheduler { @@
                     self.try_go_to_sleep(connector_key, scheduled);
                 },
                 ConnectorScheduling::Exit => {
                     // Prepare for exit. Set the shutdown flag and broadcast
                     // messages to notify peers of closing channels
                     scheduled.shutting_down = true;
                     for port in &scheduled.context.ports {
                         if port.state != PortState::Closed {
                             let message = scheduled.router.prepare_closing_channel(
                                 port.self_id, port.peer_id,
                                 connector_id
                             );
                             self.debug_conn(connector_id, &format!("Sending message [ exit ] \n --- {:?}", message));
                             self.runtime.send_message(port.peer_connector, message);
+                        }
+                    }
                     if scheduled.router.num_pending_acks() == 0 {
                         self.runtime.destroy_component(connector_key);
                         continue 'thread_loop;
+                    }
                     self.try_go_to_sleep(connector_key, scheduled);
+                }
+            }
+        }
+    }
     fn handle_delta_state(&mut self,
         cur_connector: &mut ScheduledConnector, connector_id: ConnectorId,
         delta_state: &mut RunDeltaState
     ) {
         // Handling any messages that were sent
         if !delta_state.outbox.is_empty() {
             for mut message in delta_state.outbox.drain(..) {
                 // Based on the message contents, decide where the message
                 // should be sent to. This might end up modifying the message.
                 self.debug_conn(connector_id, &format!("Sending message\n --- {:?}", message));
+                self.debug_conn(connector_id, &format!("Sending message [outbox] \n --- {:?}", message));
                 let (peer_connector, self_port, peer_port) = match &mut message {
                     MessageContents::Data(contents) => {
                         let port = cur_connector.context.get_port(contents.sending_port);
                         (port.peer_connector, contents.sending_port, port.peer_id)
                     },
                     MessageContents::Sync(contents) => {
                         let connector = contents.to_visit.pop().unwrap();
                         (connector, PortIdLocal::new_invalid(), PortIdLocal::new_invalid())
                     },
                     MessageContents::RequestCommit(contents)=> {
                         let connector = contents.to_visit.pop().unwrap();
                         (connector, PortIdLocal::new_invalid(), PortIdLocal::new_invalid())
@@ @@ -312,24 +311,25 @@ impl Scheduler { @@
                 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(
                         port.self_id, port.peer_id, cur_connector.context.id,
                         port.peer_connector, new_connector.context.id
                     );
                     self.debug_conn(connector_id, &format!("Sending message [newcon]\n --- {:?}", reroute_message));
                     self.runtime.send_message(port.peer_connector, reroute_message);
+                }
                 // Schedule new connector to run
                 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());
+    }
@@ @@ -456,26 +456,25 @@ impl ControlMessageHandler { @@
             contents: MessageContents::Control(ControlMessage{
                 id,
                 content: ControlMessageVariant::ChangePortPeer(peer_port_id, new_owner_connector_id),
             })
         };
+    }
     /// Returns true if the supplied message should be rerouted. If so then this
     /// function returns the connector that should retrieve this message.
     pub fn should_reroute(&self, sending_connector: ConnectorId, target_port: PortIdLocal) -> Option<ConnectorId> {
         for entry in &self.active {
             if let ControlVariant::ChangedPort(entry) = &entry.variant {
                 if entry.source_connector == sending_connector &&
                     entry.target_port == target_port {
                 if entry.target_port == target_port {
                     // Need to reroute this message
                     return Some(entry.target_connector);
+                }
+            }
+        }
         return None;
+    }
     /// Handles an Ack as an answer to a previously sent control message
     pub fn handle_ack(&mut self, id: u32) {
         let index = self.active.iter()

src/runtime2/tests/mod.rs

➞

Show inline comments

 use std::sync::Arc;
 use super::*;
 use crate::{PortId, ProtocolDescription};
 use crate::common::Id;
 use crate::protocol::eval::*;
 fn runtime_for(num_threads: u32, pdl: &str) -> Runtime {
 const NUM_THREADS: u32 = 4;     // number of threads in runtime
 const NUM_INSTANCES: u32 = 10;   // number of test instances constructed
 const NUM_LOOPS: u32 = 1;       // 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);
+    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 :)
+}
 #[test]
 fn test_put_and_get() {
     let rt = runtime_for(4, "
 primitive putter(out<bool> sender, u32 loops) {
     u32 index = 0;
     while (index < loops) {
         synchronous {
             print(\"putting!\");
             put(sender, true);
     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;
+        }
         index += 1;
+    }
+}
 primitive getter(in<bool> receiver, u32 loops) {
     u32 index = 0;
     while (index < loops) {
         synchronous {
             print(\"getting!\");
             auto result = get(receiver);
             assert(result);
     primitive getter(in<bool> receiver, u32 loops) {
         u32 index = 0;
         while (index < loops) {
             synchronous {
                 print(\"getting!\");
                 auto result = get(receiver);
                 assert(result);
+            }
             index += 1;
+        }
         index += 1;
+    }
+}
     ");
     ";
     let mut api = rt.create_interface();
     let channel = api.create_channel();
     let num_loops = 1;
     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("", "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");
         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)