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

Changeset - cdb4810532c2

Parent rev.

Child rev.

[Not reviewed]

0 3 0

mh - 3 years ago 2022-05-11 15:19:14
contact@maxhenger.nl

Fix rerouting bug for transmitted ports

3 files changed with 76 insertions and 5 deletions:

src/runtime2/component/component.rs

src/runtime2/component/component_pdl.rs

src/runtime2/tests/transfer_ports.rs

0 comments (0 inline, 0 general)

src/runtime2/component/component.rs

➞

Show inline comments

@@ @@ -574,13 +574,13 @@ pub(crate) fn default_handle_control_message( @@
                 // component attempted a `get`, but there were no messages, so
                 // now it is in the `BlockedGet` state.
                 let port_was_used = last_instruction != PortInstruction::None;
                 if exec_state.mode.is_in_sync_block() {
                     let closed_during_sync_round = content.closed_in_sync_round && port_was_used;
                     let closed_before_sync_round = !content.closed_in_sync_round && !port_has_had_message;
+                    let closed_before_sync_round = !content.closed_in_sync_round && !port_has_had_message && port_was_used;
                     if closed_during_sync_round || closed_before_sync_round {
                         return Err((
                             last_instruction,
                             format!("Peer component (id:{}) shut down, so communication cannot (have) succeed(ed)", peer_comp_id.0)
                         ));
@@ @@ -687,12 +687,19 @@ pub(crate) fn default_handle_sync_end( @@
 #[must_use]
 pub(crate) fn default_handle_start_exit(
     exec_state: &mut CompExecState, control: &mut ControlLayer,
     sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus
 ) -> CompScheduling {
     debug_assert_eq!(exec_state.mode, CompMode::StartExit);
     for port_index in 0..comp_ctx.num_ports() {
         let port_info = comp_ctx.get_port_by_index_mut(port_index);
         if port_info.state.is_blocked() {
             return CompScheduling::Sleep;
+        }
+    }
     sched_ctx.info(&format!("Component starting exit (reason: {:?})", exec_state.exit_reason));
     exec_state.mode = CompMode::BusyExit;
     let exit_inside_sync = exec_state.exit_reason.is_in_sync();
     // If exiting while inside sync mode, report to the leader of the current
     // round that we've failed.
@@ @@ -701,13 +708,14 @@ pub(crate) fn default_handle_start_exit( @@
         default_handle_sync_decision(sched_ctx, exec_state, comp_ctx, decision, consensus);
+    }
     // Iterating over ports by index to work around borrowing rules
     for port_index in 0..comp_ctx.num_ports() {
         let port = comp_ctx.get_port_by_index_mut(port_index);
         if port.state.is_closed() || port.close_at_sync_end {
         println!("DEBUG: Considering port:\n{:?}", port);
         if port.state.is_closed() || port.state.is_set(PortStateFlag::Transmitted) || port.close_at_sync_end {
             // Already closed, or in the process of being closed
             continue;
+        }
         // Mark as closed
         let port_id = port.self_id;
@@ @@ -901,13 +909,13 @@ fn prepare_send_message_with_ports( @@
+}
 /// Performs the transmission of a data message that contains ports. These were
 /// all stored in the component's execution state by the
 /// `prepare_send_message_with_ports` function. Port must be ready to send!
 fn perform_send_message_with_ports(
     exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, consensus: &mut Consensus,
+    exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus,
     inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup
 ) {
     debug_assert_eq!(exec_state.mode, CompMode::BlockedPutPortsReady);
     // Find all ports again
     let mut transmit_ports = Vec::new();
@@ @@ -938,13 +946,15 @@ fn perform_send_message_with_ports( @@
             locations: port_locations,
             messages: transmit_messages,
             peer_comp: transmit_port_info.peer_comp_id,
             peer_port: transmit_port_info.peer_port_id,
             kind: transmit_port_info.kind,
             state: transmit_port_state
         })
         });
         comp_ctx.change_port_peer(sched_ctx, transmit_port_handle, None);
+    }
     // And finally, send the message to the peer
     let peer_info = comp_ctx.get_peer(peer_handle);
     peer_info.handle.send_message_logged(sched_ctx, Message::Data(annotated_message), true);
+}

src/runtime2/component/component_pdl.rs

➞

Show inline comments

@@ @@ -428,13 +428,12 @@ impl Component for CompPDL { @@
                     sched_ctx, comp_ctx,
                     definition_id, type_id, arguments
                 );
                 return CompScheduling::Requeue;
             },
             EC::NewChannel => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::NonSync);
                 debug_assert!(self.exec_ctx.stmt.is_none());
                 let channel = comp_ctx.create_channel();
                 self.exec_ctx.stmt = ExecStmt::CreatedChannel((
                     Value::Output(port_id_to_eval(channel.putter_id)),
                     Value::Input(port_id_to_eval(channel.getter_id))
                 ));

src/runtime2/tests/transfer_ports.rs

➞

Show inline comments

@@ @@ -16,6 +16,68 @@ fn test_transfer_precreated_port_with_owned_peer() { @@
         channel a -> b;
         new port_sender(a);
         new port_receiver(b);
+    }
     ", "constructor", no_args());
+}
 #[test]
 fn test_transfer_precreated_port_with_foreign_peer() {
     compile_and_create_component("
     primitive port_sender(out<in<u32>> tx, in<u32> to_send) {
         sync put(tx, to_send);
+    }
     primitive port_receiver(in<in<u32>> rx) {
         sync auto a = get(rx);
+    }
     composite constructor() {
         channel tx -> rx;
         channel forgotten -> to_send;
         new port_sender(tx, to_send);
         new port_receiver(rx);
+    }
     ", "constructor", no_args());
+}
 #[test]
 fn test_transfer_synccreated_port() {
     compile_and_create_component("
     primitive port_sender(out<in<u32>> tx) {
         sync {
             channel a -> b;
             put(tx, b);
+        }
+    }
     primitive port_receiver(in<in<u32>> rx) {
         sync auto a = get(rx);
+    }
     composite constructor() {
         channel a -> b;
         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() {
+}
 #[test]
 fn test_transfer_precreated_port_with_foreign_peer_and_communication() {
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)