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

Changeset - 9972a4c3928b

Parent rev.

Child rev.

[Not reviewed]

feat-transmitting-ports

0 3 0

MH - 3 years ago 2022-05-13 19:26:19
contact@maxhenger.nl

Add extra transfer test, fix bug related to delayed update of consensus manager

3 files changed with 85 insertions and 12 deletions:

src/runtime2/component/component.rs

src/runtime2/component/consensus.rs

src/runtime2/tests/transfer_ports.rs

0 comments (0 inline, 0 general)

src/runtime2/component/component.rs

➞

Show inline comments

@@ @@ -409,13 +409,13 @@ pub(crate) fn default_attempt_get( @@
             // Note: we can still run into an unrecoverable error when actually
             // receiving this message
             match default_handle_received_data_message(
                 target_port, target_port_instruction,
                 &mut message, inbox_main, inbox_backup,
                 comp_ctx, sched_ctx, control,
+                comp_ctx, sched_ctx, control, consensus
             ) {
                 Ok(()) => return GetResult::Received(message),
                 Err(location_and_message) => return GetResult::Error(location_and_message)
+            }
         } else {
             // We're not allowed to receive this message. This means that the
@@ @@ -437,13 +437,14 @@ pub(crate) fn default_attempt_get( @@
 /// more messages are waiting, and if the corresponding port was blocked because
 /// of full buffers (hence, will use the control layer to make sure the peer
 /// will become unblocked).
 pub(crate) fn default_handle_received_data_message(
     targeted_port: PortId, _port_instruction: PortInstruction, message: &mut DataMessage,
     inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup,
     comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx, control: &mut ControlLayer
     comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx, control: &mut ControlLayer,
     consensus: &mut Consensus
 ) -> Result<(), (PortInstruction, String)> {
     let port_handle = comp_ctx.get_port_handle(targeted_port);
     let port_index = comp_ctx.get_port_index(port_handle);
     debug_assert!(inbox_main[port_index].is_none()); // because we've just received from it
     // If we received any ports, add them to the port tracking and inbox struct.
@@ @@ -467,12 +468,15 @@ pub(crate) fn default_handle_received_data_message( @@
             received_port.kind, new_port_state
         );
         debug_assert_eq!(_new_inbox_index, comp_ctx.get_port_index(new_port_handle));
         comp_ctx.change_port_peer(sched_ctx, new_port_handle, Some(received_port.peer_comp));
         let new_port = comp_ctx.get_port(new_port_handle);
         // Add the port tho the consensus
         consensus.notify_received_port(_new_inbox_index, new_port_handle, comp_ctx);
         // Replace all references to the port in the received message
         for message_location in received_port.locations.iter().copied() {
             let value = message.content.get_value_mut(message_location);
             match value {
                 Value::Input(_) => {

src/runtime2/component/consensus.rs

➞

Show inline comments

@@ @@ -352,12 +352,26 @@ impl Consensus { @@
             port.mapping = None;
+        }
         self.solution.clear();
+    }
     pub(crate) fn notify_received_port(&mut self, _expected_index: usize, port_handle: LocalPortHandle, comp_ctx: &CompCtx) {
         debug_assert_eq!(_expected_index, self.ports.len());
         let port_info = comp_ctx.get_port(port_handle);
         self.ports.push(PortAnnotation{
             self_comp_id: comp_ctx.id,
             self_port_id: port_info.self_id,
             peer_comp_id: port_info.peer_comp_id,
             peer_port_id: port_info.peer_port_id,
             peer_discovered: false,
             mapping: None,
             kind: port_info.kind,
         });
+    }
     // -------------------------------------------------------------------------
     // Handling inbound and outbound messages
     // -------------------------------------------------------------------------
     /// Prepares a set of values to be sent of a channel.
     pub(crate) fn annotate_data_message(&mut self, comp_ctx: &CompCtx, port_info: &Port, content: ValueGroup) -> DataMessage {

src/runtime2/tests/transfer_ports.rs

➞

Show inline comments

@@ @@ -59,22 +59,12 @@ fn test_transfer_synccreated_port() { @@
         new port_sender(a);
         new port_receiver(b);
+    }
     ", "constructor", no_args());
+}
 #[test]
 fn test_transfer_precreated_port_with_owned_peer_back_and_forth() {
+}
 #[test]
 fn test_transfer_precreated_port_with_foreign_peer_back_and_forth() {
+}
 #[test]
 fn test_transfer_precreated_port_with_owned_peer_and_communication() {
     compile_and_create_component("
     primitive port_sender(out<in<u32>> tx) {
         channel a -> b;
         sync put(tx, b);
         new port_sender(port_tx, value_rx);
         new port_receiver(port_rx);
         new message_transmitter(value_tx);
+    }
     ", "constructor", no_args());
+}
 #[test]
 fn test_transfer_precreated_port_with_owned_peer_back_and_forth() {
     compile_and_create_component("
     primitive port_send_and_receive(out<in<u32>> tx, in<in<u32>> rx) {
         channel a -> b;
         sync {
             put(tx, b);
             b = get(rx);
+        }
+    }
     primitive port_receive_and_send(in<in<u32>> rx, out<in<u32>> tx) {
         channel unused -> transferred; // same problem as in different tests
         sync {
             transferred = get(rx);
             put(tx, transferred);
+        }
+    }
     composite constructor() {
         channel port_tx_forward -> port_rx_forward;
         channel port_tx_backward -> port_rx_backward;
         new port_send_and_receive(port_tx_forward, port_rx_backward);
         new port_receive_and_send(port_rx_forward, port_tx_backward);
     }", "constructor", no_args());
+}
 #[test]
 fn test_transfer_precreated_port_with_foreign_peer_back_and_forth_and_communication() {
     compile_and_create_component("
     primitive port_send_and_receive(out<in<u32>> tx, in<in<u32>> rx, in<u32> to_transfer) {
         sync {
             put(tx, to_transfer);
             to_transfer = get(rx);
+        }
         sync {
             auto value = get(to_transfer);
             while (value != 1337) {}
+        }
+    }
     primitive port_receive_and_send(in<in<u32>> rx, out<in<u32>> tx) {
         channel unused -> transferred;
         sync {
             transferred = get(rx);
             put(tx, transferred);
+        }
+    }
     primitive value_sender(out<u32> tx) {
         sync put(tx, 1337);
+    }
     composite constructor() {
         channel port_tx_forward -> port_rx_forward;
         channel port_tx_backward -> port_rx_backward;
         channel message_tx -> message_rx;
         new port_send_and_receive(port_tx_forward, port_rx_backward, message_rx);
         new port_receive_and_send(port_rx_forward, port_tx_backward);
         new value_sender(message_tx);
+    }
     ", "constructor", no_args());
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)