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

Changeset - 935c576f54d0

Parent rev.

Child rev.

[Not reviewed]

0 4 0

MH - 3 years ago 2022-05-11 00:41:52
contact@maxhenger.nl

Fixing rerouting+port-transfer bug

4 files changed with 9 insertions and 4 deletions:

src/runtime2/component/component.rs

src/runtime2/component/component_pdl.rs

src/runtime2/component/control_layer.rs

src/runtime2/tests/transfer_ports.rs

0 comments (0 inline, 0 general)

src/runtime2/component/component.rs

➞

Show inline comments

@@ @@ -867,6 +867,7 @@ fn prepare_send_message_with_ports( @@
         // Block the peer of the port
         let message = control.create_port_transfer_message(unblock_put_control_id, comp_ctx.id, peer_port_id);
         println!("DEBUG: Port transfer message\nControl ID: {:?}\nMessage: {:?}", unblock_put_control_id, message);
         let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);
         let peer_info = comp_ctx.get_peer(peer_handle);
@@ @@ -938,6 +939,7 @@ fn default_handle_ack( @@
 ) {
     // Since an `Ack` may cause another one, handle them in a loop
     let mut to_ack = control_id;
     loop {
         let (action, new_to_ack) = control.handle_ack(to_ack, sched_ctx, comp_ctx);
         match action {
@@ @@ -962,6 +964,7 @@ fn default_handle_ack( @@
             },
             AckAction::UnblockPutWithPorts => {
                 // Send the message (containing ports) to the recipient
                 println!("DEBUG: Unblocking put with ports");
                 perform_send_message_with_ports(exec_state, sched_ctx, comp_ctx, consensus, inbox_main, inbox_backup);
                 exec_state.mode = CompMode::Sync;
                 exec_state.mode_port = PortId::new_invalid();

src/runtime2/component/component_pdl.rs

➞

Show inline comments

@@ @@ -241,15 +241,17 @@ impl Component for CompPDL { @@
+    }
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, mut message: Message) {
-        // sched_ctx.log(&format!("handling message: {:?}", message));
+        sched_ctx.debug(&format!("handling message: {:?}", message));
         if let Some(new_target) = self.control.should_reroute(&mut message) {
             let mut target = sched_ctx.runtime.get_component_public(new_target); // TODO: @NoDirectHandle
             sched_ctx.debug(&format!("rerouting to: {:?}", new_target));
             target.send_message_logged(sched_ctx, message, false); // not waking up: we schedule once we've received all PortPeerChanged Acks
             let _should_remove = target.decrement_users();
             debug_assert!(_should_remove.is_none());
             return;
+        }
         sched_ctx.debug("handling message myself");
         match message {
             Message::Data(message) => {
                 self.handle_incoming_data_message(sched_ctx, comp_ctx, message);

src/runtime2/component/control_layer.rs

➞

Show inline comments

@@ @@ -5,7 +5,7 @@ use crate::runtime2::component::*; @@
 use super::component_context::*;
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
 pub(crate) struct ControlId(u32);
+pub(crate) struct ControlId(pub(crate) u32);
 impl ControlId {
     /// Like other invalid IDs, this one doesn't care any significance, but is
@@ @@ -162,7 +162,7 @@ impl ControlLayer { @@
         let entry_id = self.take_id();
         self.entries.push(ControlEntry{
             id: entry_id,
-            ack_countdown: 1, // incremented by calls to `add_reroute_entry`
+            ack_countdown: 0, // incremented by calls to `add_reroute_entry`
             content: ControlContent::UnblockPutWithPorts,
         });

src/runtime2/tests/transfer_ports.rs

➞

Show inline comments

 use super::*;
 #[test]
-fn test_transfer_precreated_port_without_using() {
+fn test_transfer_precreated_port_with_owned_peer() {
     compile_and_create_component("
     primitive port_sender(out<in<u32>> tx) {
         channel a -> b;

0 comments (0 inline, 0 general)