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

Changeset - 8c5d438b0fa3

Parent rev.

Child rev.

[Not reviewed]

0 4 1

mh - 4 years ago 2021-10-01 17:59:31
contact@maxhenger.nl

rewriting inbox to behave mpsc-like

5 files changed with 103 insertions and 86 deletions:

src/runtime2/connector.rs

src/runtime2/inbox.rs

src/runtime2/messages.rs

src/runtime2/mod.rs

src/runtime2/scheduler.rs

0 comments (0 inline, 0 general)

src/runtime2/connector.rs

➞

Show inline comments

 use std::collections::HashMap;
 use super::messages::{Message, Inbox};
 use crate::protocol::{ComponentState, RunContext, RunResult};
 use crate::{PortId, ProtocolDescription};
 use crate::protocol::eval::{ValueGroup, Value, Prompt};
 use crate::protocol::{ComponentState, RunContext, RunResult};
 use crate::protocol::eval::{Prompt, Value, ValueGroup};
 use crate::runtime2::inbox::{Inbox, Message};
 #[derive(Clone, Copy, PartialEq, Eq)]
 pub(crate) struct PortIdLocal {
     pub id: u32,
+}
 impl PortIdLocal {
     pub fn new(id: u32) -> Self {
         Self{ id }
+    }
     // TODO: Unsure about this, maybe remove, then also remove all struct
     //  instances where I call this
     pub fn new_invalid() -> Self {
         Self{ id: u32::MAX }
+    }
+}
 /// Represents the identifier of a branch (the index within its container). An
 /// ID of `0` generally means "no branch" (e.g. no parent, or a port did not
 /// yet receive anything from any branch).
 #[derive(Clone, Copy, PartialEq, Eq)]
 pub(crate) struct BranchId {
     pub index: u32,

src/runtime2/inbox.rs

➞

Show inline comments

@@ new file 100644 @@
 use crate::common::Ordering;
 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::connector::{BranchId, PortIdLocal};
 /// A message in transit from one connector to another.
 #[derive(Clone)]
 pub struct Message {
     pub sending_port: PortIdLocal,
     pub receiving_port: PortIdLocal,
     pub sender_prev_branch_id: BranchId, // may be invalid, implying no prev branch id
     pub sender_cur_branch_id: BranchId, // always valid
     pub message: ValueGroup,
+}
 /// The inbox of a connector. The owning connector (i.e. the thread that is
 /// executing the connector) should be able to read all messages. Other
 /// connectors (potentially executed by different threads) should be able to
 /// append messages.
 ///
 /// Note that the logic inside of the inbox is strongly connected to deciding
 /// whether or not a connector has nothing to execute, and is waiting on new
 /// messages in order to continue.
 pub struct Inbox {
     messages: Vec<Message>
+}
 impl Inbox {
     pub fn new() -> Self {
         Self{ messages: Vec::new() }
+    }
     /// Will insert the message into the inbox. Only exception is when the tuple
     /// (prev_branch_id, cur_branch_id, receiving_port_id) already exists, then
     /// nothing is inserted..
     pub fn insert_message(&mut self, message: Message) {
         match self.messages.binary_search_by(|a| Self::compare_messages(a, &message)) {
             Ok(_) => {} // message already exists
             Err(idx) => self.messages.insert(idx, message)
+        }
+    }
     /// Retrieves all messages for the provided conditions
     pub fn get_messages(&self, port_id: PortIdLocal, prev_branch_id: BranchId) -> &[Message] {
         // Seek the first message with the appropriate port ID and branch ID
         let num_messages = self.messages.len();
         for first_idx in 0..num_messages {
             let msg = &self.messages[first_idx];
             if msg.receiving_port == port_id && msg.sender_prev_branch_id == prev_branch_id {
                 // Found a match, seek ahead until the condition is no longer true
                 let mut last_idx = first_idx + 1;
                 while last_idx < num_messages {
                     let msg = &self.messages[last_idx];
                     if msg.receiving_port != port_id || msg.sender_prev_branch_id != prev_branch_id {
                         // No longer matching
                         break;
+                    }
                     last_idx += 1;
+                }
                 // Return all the matching messages
                 return &self.messages[first_idx..last_idx];
             } else if msg.receiving_port.id > port_id.id {
                 // Because messages are ordered, this implies we couldn't find
                 // any message
                 break;
+            }
+        }
         return &self.messages[0..0];
+    }
     /// Simply empties the inbox
     pub fn clear(&mut self) {
         self.messages.clear();
+    }
     // Ordering by, consecutively, a) receiving port, b) prev branch id, c) cur
     // branch id.
     fn compare_messages(a: &Message, b: &Message) -> Ordering {
         let mut ord = a.receiving_port.id.cmp(&b.receiving_port.id);
         if ord != Ordering::Equal { return ord; }
         ord = a.sender_prev_branch_id.index.cmp(&b.sender_prev_branch_id.index);
         if ord != Ordering::Equal { return ord; }
         return a.sender_cur_branch_id.index.cmp(&b.sender_cur_branch_id.index);
+    }
+}

src/runtime2/messages.rs

➞

Show inline comments

 use std::collections::HashMap;
 use std::collections::hash_map::Entry;
 use std::cmp::Ordering;
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use super::connector::{PortIdLocal, BranchId};
 use crate::PortId;
 use crate::common::Id;
 use crate::PortId;
 use crate::protocol::*;
 use crate::protocol::eval::*;
 use super::connector::{BranchId, PortIdLocal};
 /// A message residing in a connector's inbox (waiting to be put into some kind
 /// of speculative branch), or a message waiting to be sent.
 #[derive(Clone)]
 pub struct BufferedMessage {
     pub(crate) sending_port: PortId,
     pub(crate) receiving_port: PortId,
     pub(crate) peer_prev_branch_id: Option<u32>,
     pub(crate) peer_cur_branch_id: u32,
     pub(crate) message: ValueGroup,
+}
 #[derive(Clone)]
 pub struct Message {
     pub sending_port: PortIdLocal,
     pub receiving_port: PortIdLocal,
     pub sender_prev_branch_id: BranchId, // may be invalid, implying no prev branch id
     pub sender_cur_branch_id: BranchId, // always valid
     pub message: ValueGroup,
+}
 pub struct Inbox {
     messages: Vec<Message>
+}
 impl Inbox {
     pub fn new() -> Self {
         Self{ messages: Vec::new() }
+    }
     /// Will insert the message into the inbox. Only exception is when the tuple
     /// (prev_branch_id, cur_branch_id, receiving_port_id) already exists, then
     /// nothing is inserted..
     pub fn insert_message(&mut self, message: Message) {
         match self.messages.binary_search_by(|a| Self::compare_messages(a, &message)) {
             Ok(_) => {} // message already exists
             Err(idx) => self.messages.insert(idx, message)
+        }
+    }
     /// Retrieves all messages for the provided conditions
     pub fn get_messages(&self, port_id: PortIdLocal, prev_branch_id: BranchId) -> &[Message] {
         // Seek the first message with the appropriate port ID and branch ID
         let num_messages = self.messages.len();
         for first_idx in 0..num_messages {
             let msg = &self.messages[first_idx];
             if msg.receiving_port == port_id && msg.sender_prev_branch_id == prev_branch_id {
                 // Found a match, seek ahead until the condition is no longer true
                 let mut last_idx = first_idx + 1;
                 while last_idx < num_messages {
                     let msg = &self.messages[last_idx];
                     if msg.receiving_port != port_id || msg.sender_prev_branch_id != prev_branch_id {
                         // No longer matching
                         break;
+                    }
                     last_idx += 1;
+                }
                 // Return all the matching messages
                 return &self.messages[first_idx..last_idx];
             } else if msg.receiving_port.id > port_id.id {
                 // Because messages are ordered, this implies we couldn't find
                 // any message
                 break;
+            }
+        }
         return &self.messages[0..0];
+    }
     /// Simply empties the inbox
     pub fn clear(&mut self) {
         self.messages.clear();
+    }
     // Ordering by, consecutively, a) receiving port, b) prev branch id, c) cur
     // branch id.
     fn compare_messages(a: &Message, b: &Message) -> Ordering {
         let mut ord = a.receiving_port.id.cmp(&b.receiving_port.id);
         if ord != Ordering::Equal { return ord; }
         ord = a.sender_prev_branch_id.index.cmp(&b.sender_prev_branch_id.index);
         if ord != Ordering::Equal { return ord; }
         return a.sender_cur_branch_id.index.cmp(&b.sender_cur_branch_id.index);
+    }
+}
 /// A connector's global inbox. Any received message ends up here. This is
 /// because a message might be received before a branch arrives at the
 /// corresponding `get()` that is supposed to receive that message. Hence we
 /// need to store it for all future branches that might be able to receive it.
 pub struct ConnectorInbox {
     // TODO: @optimize, HashMap + Vec is a bit stupid.
     messages: HashMap<PortAction, Vec<BufferedMessage>>
+}
 /// An action performed on a port. Unsure about this
 #[derive(PartialEq, Eq, Hash)]
 struct PortAction {
     port_id: u32,
     prev_branch_id: Option<u32>,
+}
 // TODO: @remove
 impl ConnectorInbox {
     pub fn new() -> Self {
         Self {
             messages: HashMap::new(),
+        }
+    }

src/runtime2/mod.rs

➞

Show inline comments

 mod runtime;
 mod messages;
 mod connector;
 mod global_store;
 mod scheduler;
 #[cfg(test)] mod tests;
 mod inbox;

src/runtime2/scheduler.rs

➞

Show inline comments

@@ @@ -30,43 +30,47 @@ impl Scheduler { @@
             // Retrieve a unit of work
             let connector_key = self.global.pop_key();
             if connector_key.is_none() {
                 // TODO: @Performance, needs condition variable for waking up
                 thread::sleep(Duration::new(1, 0));
                 continue
+            }
             // We have something to do
             let connector_key = connector_key.unwrap();
             let connector = self.global.get_connector(&connector_key);
             let mut cur_schedule = ConnectorScheduling::Immediate;
             while cur_schedule == ConnectorScheduling::Immediate {
                 let new_schedule;
                 if connector.is_in_sync_mode() {
                     // In synchronous mode, so we can expect messages being sent,
                     // but we never expect the creation of connectors
                     new_schedule = connector.run_in_speculative_mode(self.code.as_ref(), &mut delta_state);
                     debug_assert!(delta_state.new_connectors.is_empty());
                     if !delta_state.outbox.is_empty() {}
                     if !delta_state.outbox.is_empty() {
                         for message in delta_state.outbox.drain(..) {
+                        }
+                    }
                 } else {
                     // In regular running mode (not in a sync block) we cannot send
                     // messages but we can create new connectors
                     new_schedule = connector.run_in_deterministic_mode(self.code.as_ref(), &mut delta_state);
                     debug_assert!(delta_state.outbox.is_empty());
                     if !delta_state.new_connectors.is_empty() {
                         // Push all connectors into the global state and queue them
                         // for execution
+                    }
+                }
                 cur_schedule = new_schedule;
+            }
+        }
+    }
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)