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

Changeset - b4a8c628800d

Parent rev.

Child rev.

[Not reviewed]

0 5 0

MH - 4 years ago 2021-11-19 19:59:49
contact@maxhenger.nl

Refactored data messages

5 files changed with 30 insertions and 60 deletions:

src/runtime2/connector.rs

src/runtime2/consensus.rs

src/runtime2/inbox.rs

src/runtime2/native.rs

src/runtime2/scheduler.rs

0 comments (0 inline, 0 general)

src/runtime2/connector.rs

➞

Show inline comments

@@ @@ -30,19 +30,16 @@ use std::collections::HashMap; @@
 use std::sync::atomic::AtomicBool;
 use crate::{PortId, ProtocolDescription};
 use crate::common::ComponentState;
 use crate::protocol::eval::{EvalContinuation, EvalError, Prompt, Value, ValueGroup};
 use crate::protocol::{RunContext, RunResult};
 use crate::runtime2::branch::PreparedStatement;
 use crate::runtime2::consensus::RoundConclusion;
 use crate::runtime2::inbox::SyncPortMessage;
 use super::branch::{BranchId, ExecTree, QueueKind, SpeculativeState};
 use super::consensus::{Consensus, Consistency, find_ports_in_value_group};
 use super::inbox::{DataMessage, DataContent, Message, SyncCompMessage, PublicInbox};
 use super::branch::{BranchId, ExecTree, QueueKind, SpeculativeState, PreparedStatement};
 use super::consensus::{Consensus, Consistency, RoundConclusion, find_ports_in_value_group};
 use super::inbox::{DataMessage, Message, SyncCompMessage, SyncPortMessage, PublicInbox};
 use super::native::Connector;
 use super::port::{PortKind, PortIdLocal};
 use super::scheduler::{ComponentCtx, SchedulerCtx};
 pub(crate) struct ConnectorPublic {
     pub inbox: PublicInbox,
@@ @@ -225,13 +222,13 @@ impl ConnectorPDL { @@
             let receiving_branch_id = self.tree.fork_branch(branch_id);
             self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
             let receiving_branch = &mut self.tree[receiving_branch_id];
             debug_assert!(receiving_branch.awaiting_port == message.data_header.target_port);
             receiving_branch.awaiting_port = PortIdLocal::new_invalid();
-            receiving_branch.prepared = PreparedStatement::PerformedGet(message.content.as_message().unwrap().clone());
             receiving_branch.prepared = PreparedStatement::PerformedGet(message.content.clone());
             self.consensus.notify_of_received_message(receiving_branch_id, &message, ctx);
             // And prepare the branch for running
             self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
+        }
+    }
@@ @@ -323,13 +320,13 @@ impl ConnectorPDL { @@
                     if self.consensus.branch_can_receive(branch_id, &message) {
                         // This branch can receive the message, so we do the
                         // fork-and-receive dance
                         let receiving_branch_id = self.tree.fork_branch(branch_id);
                         let branch = &mut self.tree[receiving_branch_id];
                         branch.awaiting_port = PortIdLocal::new_invalid();
-                        branch.prepared = PreparedStatement::PerformedGet(message.content.as_message().unwrap().clone());
                         branch.prepared = PreparedStatement::PerformedGet(message.content.clone());
                         self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
                         self.consensus.notify_of_received_message(receiving_branch_id, &message, comp_ctx);
                         self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
                         any_message_received = true;
@@ @@ -367,13 +364,13 @@ impl ConnectorPDL { @@
             EvalContinuation::Put(port_id, content) => {
                 // Branch is attempting to send data
                 let port_id = PortIdLocal::new(port_id.0.u32_suffix);
                 let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);
                 if let Err(_) = comp_ctx.submit_message(Message::Data(DataMessage {
                     sync_header, data_header,
-                    content: DataContent::Message(content),
                     content,
                 })) {
                     // We don't own the port
                     let pd = &sched_ctx.runtime.protocol_description;
                     let eval_error = branch.code_state.new_error_at_expr(
                         &pd.modules, &pd.heap,
                         String::from("attempted to 'put' on port that is no longer owned")
@@ @@ -497,13 +494,12 @@ impl ConnectorPDL { @@
             self.eval_error = None; // in case we came from the SyncError mode
             self.mode = Mode::NonSync;
             return ConnectorScheduling::Immediate;
         } else {
             // No final branch, because we're supposed to exit!
             panic!("TEMPTEMP: NOOOOOOOOO 1");
             self.last_finished_handled = None;
             self.mode = Mode::Error;
             return ConnectorScheduling::Exit;
+        }
+    }

src/runtime2/consensus.rs

➞

Show inline comments

 use crate::collections::VecSet;
 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::inbox::BranchMarker;
 use crate::runtime2::scheduler::ComponentPortChange;
 use super::ConnectorId;
 use super::branch::BranchId;
 use super::port::{ChannelId, PortIdLocal};
 use super::inbox::{
     Message, ChannelAnnotation,
     DataMessage, DataContent, DataHeader,
     Message, ChannelAnnotation, BranchMarker, DataMessage, DataHeader,
     SyncCompMessage, SyncCompContent, SyncPortMessage, SyncPortContent, SyncHeader,
 };
 use super::scheduler::ComponentCtx;
+use super::scheduler::{ComponentCtx, ComponentPortChange};
 struct BranchAnnotation {
     channel_mapping: Vec<ChannelAnnotation>,
     cur_marker: BranchMarker,
+}
@@ @@ -262,25 +259,21 @@ impl Consensus { @@
             let port_desc = ctx.get_port_by_channel_id(port.channel_id).unwrap();
             let self_port_id = port_desc.self_id;
             let peer_port_id = port_desc.peer_id;
             let channel_id = port_desc.channel_id;
             if !self.encountered_ports.contains(&self_port_id) {
-                ctx.submit_message(Message::Data(DataMessage {
+                ctx.submit_message(Message::SyncPort(SyncPortMessage {
                     sync_header: SyncHeader{
                         sending_component_id: ctx.id,
                         highest_component_id: self.highest_connector_id,
                         sync_round: self.sync_round
                     },
                     data_header: DataHeader{
                         expected_mapping: source_mapping.clone(),
                         sending_port: self_port_id,
                         target_port: peer_port_id,
                         new_mapping: BranchMarker::new_invalid(),
                     },
                     content: DataContent::SilentPortNotification,
                     source_port: self_port_id,
                     target_port: peer_port_id,
                     content: SyncPortContent::SilentPortNotification,
                 }));
                 self.encountered_ports.push(self_port_id);
+            }
             target_mapping.push((
                 channel_id,
@@ @@ -436,12 +429,17 @@ impl Consensus { @@
             return;
+        }
         debug_assert!(self.is_in_sync());
         debug_assert!(ctx.get_port_by_id(message.target_port).is_some());
         match message.content {
             SyncPortContent::SilentPortNotification => {
                 // The point here is to let us become part of the sync round and
                 // take note of the leader in case all of our ports are silent.
                 self.encountered_ports.push(message.target_port);
+            }
             SyncPortContent::NotificationWave => {
                 // Wave to discover everyone in the network, handling sync
                 // header takes care of leader discovery, here we need to make
                 // sure we propagate the wave
                 if self.handled_wave {
                     return;
@@ @@ -480,13 +478,13 @@ impl Consensus { @@
             if mapping.channel_id == target_port.channel_id {
                 // Found the port in which the message should be inserted
                 mapping.registered_id = Some(message.data_header.new_mapping);
                 // Check for sent ports
                 debug_assert!(self.workspace_ports.is_empty());
-                find_ports_in_value_group(message.content.as_message().unwrap(), &mut self.workspace_ports);
+                find_ports_in_value_group(&message.content, &mut self.workspace_ports);
                 if !self.workspace_ports.is_empty() {
                     todo!("handle received ports");
                     self.workspace_ports.clear();
+                }
                 return;
@@ @@ -504,17 +502,12 @@ impl Consensus { @@
         if let Some(peer) = self.peers.iter().find(|v| v.id == message.sync_header.sending_component_id) {
             if message.sync_header.sync_round < peer.expected_sync_round {
                 return false;
+            }
+        }
         if let DataContent::SilentPortNotification = message.content {
             // No port can receive a "silent" notification.
             return false;
+        }
         let annotation = &self.branch_annotations[branch_id.index as usize];
         for expected in &message.data_header.expected_mapping {
             // If we own the port, then we have an entry in the
             // annotation, check if the current mapping matches
             for current in &annotation.channel_mapping {
                 if expected.channel_id == current.channel_id {

src/runtime2/inbox.rs

➞

Show inline comments

@@ @@ -55,40 +55,19 @@ pub(crate) struct DataHeader { @@
     pub expected_mapping: Vec<ChannelAnnotation>,
     pub sending_port: PortIdLocal,
     pub target_port: PortIdLocal,
     pub new_mapping: BranchMarker,
+}
 // TODO: Very much on the fence about this. On one hand I thought making it a
 //  data message was neat because "silent port notification" should be rerouted
 //  like any other data message to determine the component ID of the receiver
 //  and to make it part of the leader election algorithm for the sync leader.
 //  However: it complicates logic quite a bit. Really it might be easier to
 //  create `Message::SyncAtComponent` and `Message::SyncAtPort` messages...
 #[derive(Debug, Clone)]
 pub(crate) enum DataContent {
     SilentPortNotification,
     Message(ValueGroup),
+}
 impl DataContent {
     pub(crate) fn as_message(&self) -> Option<&ValueGroup> {
         match self {
             DataContent::SilentPortNotification => None,
             DataContent::Message(message) => Some(message),
+        }
+    }
+}
 /// A data message is a message that is intended for the receiver's PDL code,
 /// but will also be handled by the consensus algorithm
 #[derive(Debug, Clone)]
 pub(crate) struct DataMessage {
     pub sync_header: SyncHeader,
     pub data_header: DataHeader,
-    pub content: DataContent,
+    pub content: ValueGroup,
+}
 #[derive(Debug)]
 pub(crate) enum SyncCompContent {
     LocalFailure, // notifying leader that component has failed (e.g. timeout, whatever)
     LocalSolution(LocalSolution), // sending a local solution to the leader
@@ @@ -108,12 +87,13 @@ pub(crate) struct SyncCompMessage { @@
     pub target_component_id: ConnectorId,
     pub content: SyncCompContent,
+}
 #[derive(Debug)]
 pub(crate) enum SyncPortContent {
     SilentPortNotification,
     NotificationWave,
+}
 #[derive(Debug)]
 pub(crate) struct SyncPortMessage {
     pub sync_header: SyncHeader,
@@ @@ -150,16 +130,17 @@ pub(crate) enum Message { @@
 impl Message {
     /// If the message is sent through a particular channel, then this function
     /// returns the port through which the message was sent.
     pub(crate) fn source_port(&self) -> Option<PortIdLocal> {
         // Currently only data messages have a source port
         if let Message::Data(message) = self {
             return Some(message.data_header.sending_port);
         } else {
             return None;
         match self {
             Message::Data(message) => return Some(message.data_header.sending_port),
             Message::SyncPort(message) => return Some(message.source_port),
             Message::SyncComp(_) => return None,
             Message::Control(_) => return None,
+        }
+    }
+}
 /// The public inbox of a connector. The thread running the connector that owns
 /// this inbox may retrieved from it. Non-owning threads may only put new

src/runtime2/native.rs

➞

Show inline comments

@@ @@ -11,13 +11,13 @@ use super::{ConnectorKey, ConnectorId, RuntimeInner}; @@
 use super::branch::{BranchId, FakeTree, QueueKind, SpeculativeState};
 use super::scheduler::{SchedulerCtx, ComponentCtx};
 use super::port::{Port, PortIdLocal, Channel, PortKind};
 use super::consensus::{Consensus, Consistency, find_ports_in_value_group};
 use super::connector::{ConnectorScheduling, ConnectorPDL};
 use super::inbox::{
-    Message, DataContent, DataMessage,
     Message, DataMessage,
     SyncCompMessage, SyncPortMessage,
     ControlContent, ControlMessage
 };
 /// Generic connector interface from the scheduler's point of view.
 pub(crate) trait Connector {
@@ @@ -141,13 +141,13 @@ impl ConnectorApplication { @@
             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());
             receiving_branch.insert_message(message.data_header.target_port, message.content.clone());
             self.consensus.notify_of_received_message(receiving_branch_id, &message, ctx);
             // And prepare the branch for running
             self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
+        }
+    }
@@ @@ -196,13 +196,13 @@ impl ConnectorApplication { @@
                     let port_id = *port_id;
                     let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);
                     let message = Message::Data(DataMessage {
                         sync_header,
                         data_header,
-                        content: DataContent::Message(content.clone()),
                         content: content.clone(),
                     });
                     comp_ctx.submit_message(message);
                     self.tree.push_into_queue(QueueKind::Runnable, branch_id);
                     return ConnectorScheduling::Immediate;
                 },
                 ApplicationSyncAction::Get(port_id) => {
@@ @@ -219,13 +219,13 @@ impl ConnectorApplication { @@
                             // fork-and-receive dance
                             let receiving_branch_id = self.tree.fork_branch(branch_id);
                             let branch = &mut self.tree[receiving_branch_id];
                             debug_assert!(receiving_branch_id.index as usize == self.branch_extra.len());
                             self.branch_extra.push(instruction_idx + 1);
-                            branch.insert_message(port_id, message.content.as_message().unwrap().clone());
                             branch.insert_message(port_id, message.content.clone());
                             self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
                             self.consensus.notify_of_received_message(receiving_branch_id, &message, comp_ctx);
                             self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
                             any_message_received = true;

src/runtime2/scheduler.rs

➞

Show inline comments

@@ @@ -69,13 +69,13 @@ impl Scheduler { @@
+                    }
                 } else {
                     self.debug_conn(connector_id, "Running ...");
                     let scheduler_ctx = SchedulerCtx{ runtime: &*self.runtime };
                     let new_schedule = scheduled.connector.run(scheduler_ctx, &mut scheduled.ctx);
                     self.debug_conn(connector_id, &format!("Finished running (new scheduling is {:?})", new_schedule));
                     // Handle all of the output from the current run: messages to
                     // send and connectors to instantiate.
                     self.handle_changes_in_context(scheduled);
                     cur_schedule = new_schedule;
+                }

0 comments (0 inline, 0 general)