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

Changeset - 18c70be7107f

Parent rev.

Child rev.

[Not reviewed]

0 8 0

mh - 4 years ago 2021-11-25 12:38:15
contact@maxhenger.nl

WIP on fixing delayed message bug

8 files changed with 397 insertions and 130 deletions:

src/collections/raw_vec.rs

src/runtime2/connector.rs

src/runtime2/consensus.rs

src/runtime2/inbox.rs

src/runtime2/native.rs

src/runtime2/scheduler.rs

240

src/runtime2/tests/mod.rs

src/runtime2/tests/network_shapes.rs

0 comments (0 inline, 0 general)

src/collections/raw_vec.rs

➞

Show inline comments

@@ @@ -11,10 +11,12 @@ enum AllocError { @@
 /// ensuring that no illegal mutable access occurs.
 /// A lot of the logic is simply stolen from the std lib. The destructor will
 /// free the backing memory, but will not run any destructors.
 /// Try to use functions to modify the length. But feel free if you know what
 /// you're doing
 pub struct RawVec<T: Sized> {
     base: *mut T,
     cap: usize,
     len: usize,
+    pub len: usize,
+}
 impl<T: Sized> RawVec<T> {
@@ @@ -38,16 +40,19 @@ impl<T: Sized> RawVec<T> { @@
         return result;
+    }
     #[inline]
     pub unsafe fn get(&self, idx: usize) -> *const T {
         debug_assert!(idx < self.len);
         return self.base.add(idx);
+    }
     #[inline]
     pub unsafe fn get_mut(&self, idx: usize) -> *mut T {
         debug_assert!(idx < self.len);
         return self.base.add(idx);
+    }
     /// Pushes a new element to the end of the list.
     pub fn push(&mut self, item: T) {
         self.ensure_space(1).unwrap();
         unsafe {
@@ @@ -57,10 +62,30 @@ impl<T: Sized> RawVec<T> { @@
+        }
+    }
     /// Moves the elements in the range [from_idx, from_idx + num_to_move) to
     /// the range [to_idx, to_idx + num_to_move). Caller must make sure that all
     /// non-overlapping elements of the second range had their destructor called
     /// in case those elements were used.
     pub fn move_range(&mut self, from_idx: usize, to_idx: usize, num_to_move: usize) {
         debug_assert!(from_idx + num_to_move <= self.len);
         debug_assert!(to_idx + num_to_move <= self.len); // maybe not in future, for now this is fine
         unsafe {
             let source = self.base.add(from_idx);
             let target = self.base.add(to_idx);
             std::ptr::copy(source, target, num_to_move);
+        }
+    }
     pub fn len(&self) -> usize {
         return self.len;
+    }
     pub fn as_slice(&self) -> &[T] {
         return unsafe{
             std::slice::from_raw_parts(self.base, self.len)
         };
+    }
     fn ensure_space(&mut self, additional: usize) -> Result<(), AllocError>{
         debug_assert!(Self::T_SIZE != 0);
         debug_assert!(self.cap >= self.len);

src/runtime2/connector.rs

➞

Show inline comments

 use super::inbox::{DataMessage, Message, SyncCompMessage, SyncPortMessage, SyncControlMessage, PublicInbox};
 use super::native::Connector;
 use super::port::{PortKind, PortIdLocal};
 use super::scheduler::{ComponentCtx, SchedulerCtx};
+use super::scheduler::{ComponentCtx, SchedulerCtx, MessageTicket};
 pub(crate) struct ConnectorPublic {
     pub inbox: PublicInbox,
@@ @@ -185,35 +185,45 @@ impl ConnectorPDL { @@
     // --- Handling messages
     pub fn handle_new_messages(&mut self, ctx: &mut ComponentCtx) -> Option<ConnectorScheduling> {
         while let Some(message) = ctx.read_next_message() {
             match message {
                 Message::Data(message) => self.handle_new_data_message(message, ctx),
         while let Some(ticket) = ctx.get_next_message_ticket() {
             let message = ctx.read_message_using_ticket(ticket);
             let immediate_result = if let Message::Data(_) = message {
                 self.handle_new_data_message(ticket, ctx);
                 None
             } else {
                 match ctx.take_message_using_ticket(ticket) {
                     Message::Data(_) => unreachable!(),
                     Message::SyncComp(message) => {
                     if let Some(result) = self.handle_new_sync_comp_message(message, ctx) {
                         return Some(result);
+                    }
                         self.handle_new_sync_comp_message(message, ctx)
                     },
                     Message::SyncPort(message) => {
                         self.handle_new_sync_port_message(message, ctx);
                         None
                     },
                 Message::SyncPort(message) => self.handle_new_sync_port_message(message, ctx),
                     Message::SyncControl(message) => {
                     if let Some(result) = self.handle_new_sync_control_message(message, ctx) {
                         return Some(result);
+                    }
                         self.handle_new_sync_control_message(message, ctx)
                     },
                     Message::Control(_) => unreachable!("control message in component"),
+                }
             };
             if let Some(result) = immediate_result {
                 return Some(result);
+            }
+        }
         return None;
+    }
-    pub fn handle_new_data_message(&mut self, message: DataMessage, ctx: &mut ComponentCtx) {
+    pub fn handle_new_data_message(&mut self, ticket: MessageTicket, ctx: &mut ComponentCtx) {
         // Go through all branches that are awaiting new messages and see if
         // there is one that can receive this message.
         if !self.consensus.handle_new_data_message(&message, ctx) {
             // Old message, so drop it
         if self.consensus.handle_new_data_message(ticket, ctx) {
             // Message should not be handled now
             return;
+        }
         let message = ctx.read_message_using_ticket(ticket).as_data();
         let mut iter_id = self.tree.get_queue_first(QueueKind::AwaitingMessage);
         while let Some(branch_id) = iter_id {
             iter_id = self.tree.get_queue_next(branch_id);

src/runtime2/consensus.rs

➞

Show inline comments

@@ @@ -12,7 +12,7 @@ use super::inbox::{ @@
     SyncPortMessage, SyncPortContent,
     SyncControlMessage, SyncControlContent
 };
 use super::scheduler::{ComponentCtx, ComponentPortChange};
+use super::scheduler::{ComponentCtx, ComponentPortChange, MessageTicket};
 struct BranchAnnotation {
     channel_mapping: Vec<ChannelAnnotation>,
@@ @@ -88,6 +88,13 @@ pub(crate) enum Consistency { @@
     Inconsistent,
+}
 #[derive(PartialEq, Eq)]
 pub(crate) enum MessageOrigin {
     Past,
     Present,
     Future
+}
 impl Consensus {
     pub fn new() -> Self {
         return Self {
@@ @@ -375,20 +382,34 @@ impl Consensus { @@
     /// Handles a new data message by handling the sync header. The caller is
     /// responsible for checking for branches that might be able to receive
     /// the message.
     pub fn handle_new_data_message(&mut self, message: &DataMessage, ctx: &mut ComponentCtx) -> bool {
         let handled = self.handle_received_sync_header(&message.sync_header, ctx);
         if handled {
             self.encountered_ports.push(message.data_header.target_port);
     pub fn handle_new_data_message(&mut self, ticket: MessageTicket, ctx: &mut ComponentCtx) -> bool {
         let message = ctx.read_message_using_ticket(ticket).as_data();
         let target_port = message.data_header.target_port;
         match self.handle_received_sync_header(message.sync_header, ctx) {
             MessageOrigin::Past => return false,
             MessageOrigin::Present => {
                 self.encountered_ports.push(target_port);
                 return true;
             },
             MessageOrigin::Future => {
                 let message = ctx.take_message_using_ticket(ticket);
                 ctx.put_back_message(message);
                 return false;
+            }
+        }
         return handled;
+    }
     /// Handles a new sync message by handling the sync header and the contents
     /// of the message. Returns `Some` with the branch ID of the global solution
     /// if the sync solution has been found.
     pub fn handle_new_sync_comp_message(&mut self, message: SyncCompMessage, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         if !self.handle_received_sync_header(&message.sync_header, ctx) {
             return None;
         match self.handle_received_sync_header(message.sync_header, ctx) {
             MessageOrigin::Past => return None,
             MessageOrigin::Present => {},
             MessageOrigin::Future => {
                 ctx.put_back_message(Message::SyncComp(message));
                 return None
+            }
+        }
         // And handle the contents
@@ @@ -427,9 +448,14 @@ impl Consensus { @@
+    }
     pub fn handle_new_sync_port_message(&mut self, message: SyncPortMessage, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         if !self.handle_received_sync_header(&message.sync_header, ctx) {
         match self.handle_received_sync_header(message.sync_header, ctx) {
             MessageOrigin::Past => return None,
             MessageOrigin::Present => {},
             MessageOrigin::Future => {
                 ctx.put_back_message(Message::SyncPort(message));
                 return None;
+            }
+        }
         debug_assert!(self.is_in_sync());
         debug_assert!(ctx.get_port_by_id(message.target_port).is_some());
@@ @@ -482,11 +508,11 @@ impl Consensus { @@
                     });
+                }
-                let maybe_conlusion = self.send_to_leader_or_handle_as_leader(SyncCompContent::Presence(ComponentPresence{
+                let maybe_conclusion = self.send_to_leader_or_handle_as_leader(SyncCompContent::Presence(ComponentPresence{
                     component_id: ctx.id,
                     channels,
                 }), ctx);
-                return maybe_conlusion;
+                return maybe_conclusion;
+            }
+        }
+    }
@@ @@ -497,6 +523,10 @@ impl Consensus { @@
             return None
+        }
         // Because the message is always sent in response to a message
         // originating here, the sync round number can never be larger than the
         // currently stored one.
         debug_assert_eq!(message.in_response_to_sync_round, self.sync_round);
         match message.content {
             SyncControlContent::ChannelIsClosed(_) => {
                 return self.initiate_sync_failure(ctx);
@@ @@ -560,11 +590,12 @@ impl Consensus { @@
     // --- Internal helpers
-    fn handle_received_sync_header(&mut self, sync_header: &SyncHeader, ctx: &mut ComponentCtx) -> bool {
+    fn handle_received_sync_header(&mut self, sync_header: SyncHeader, ctx: &mut ComponentCtx) -> MessageOrigin {
         debug_assert!(sync_header.sending_component_id != ctx.id); // not sending to ourselves
         if !self.handle_peer(sync_header) {
             // We can drop this package
             return false;
         let origin = self.handle_peer(&sync_header);
         if origin != MessageOrigin::Present {
             // We do not have to handle it now
             return origin;
+        }
         if sync_header.highest_component_id > self.highest_connector_id {
@@ @@ -598,23 +629,35 @@ impl Consensus { @@
             ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel
         } // else: exactly equal, so do nothing
-        return true;
+        return MessageOrigin::Present;
+    }
     /// Handles a (potentially new) peer. Returns `false` if the provided sync
     /// number is different then the expected one.
-    fn handle_peer(&mut self, sync_header: &SyncHeader) -> bool {
+    fn handle_peer(&mut self, sync_header: &SyncHeader) -> MessageOrigin {
         let position = self.peers.iter().position(|v| v.id == sync_header.sending_component_id);
         match position {
             Some(index) => {
                 let entry = &mut self.peers[index];
                 entry.encountered_this_round = true;
                 if entry.encountered_this_round {
                     // Already encountered this round
                     if sync_header.sync_round < entry.expected_sync_round {
                         return MessageOrigin::Past;
                     } else if sync_header.sync_round == entry.expected_sync_round {
                         return MessageOrigin::Present;
                     } else {
                         return MessageOrigin::Future;
+                    }
                 } else {
                     // TODO: Proper handling of potential overflow
                     entry.encountered_this_round = true;
                     if sync_header.sync_round >= entry.expected_sync_round {
                         entry.expected_sync_round = sync_header.sync_round;
-                    return true;
+                        return MessageOrigin::Present;
                     } else {
                     return false;
                         return MessageOrigin::Past;
+                    }
+                }
             },
             None => {
@@ @@ -623,7 +666,7 @@ impl Consensus { @@
                     encountered_this_round: true,
                     expected_sync_round: sync_header.sync_round,
                 });
-                return true;
+                return MessageOrigin::Present;
+            }
+        }
+    }

src/runtime2/inbox.rs

➞

Show inline comments

@@ @@ -42,7 +42,7 @@ impl BranchMarker { @@
+}
 /// The header added by the synchronization algorithm to all.
 #[derive(Debug, Clone)]
+#[derive(Debug, Clone, Copy)]
 pub(crate) struct SyncHeader {
     pub sending_component_id: ConnectorId,
     pub highest_component_id: ConnectorId,
@@ @@ -164,6 +164,32 @@ impl Message { @@
             Message::Control(_) => return None,
+        }
+    }
     /// If the message is sent through a particular channel, then this function
     /// returns the target port through which the message was sent.
     pub(crate) fn target_port(&self) -> Option<PortIdLocal> {
         match self {
             Message::Data(message) => return Some(message.data_header.target_port),
             Message::SyncPort(message) => return Some(message.target_port),
             Message::SyncComp(_) => return None,
             Message::SyncControl(_) => return None,
             Message::Control(message) => {
                 match &message.content {
                     ControlContent::PortPeerChanged(port_id, _) => return Some(*port_id),
                     ControlContent::CloseChannel(port_id) => return Some(*port_id),
                     ControlContent::Ping => return None,
                     ControlContent::Ack => return None,
+                }
+            }
+        }
+    }
     pub(crate) fn as_data(&self) -> &DataMessage {
         match self {
             Message::Data(v) => v,
             _ => unreachable!(),
+        }
+    }
+}
 /// The public inbox of a connector. The thread running the connector that owns

src/runtime2/native.rs

➞

Show inline comments

@@ @@ -9,7 +9,7 @@ use crate::runtime2::consensus::RoundConclusion; @@
 use super::{ConnectorKey, ConnectorId, RuntimeInner};
 use super::branch::{BranchId, FakeTree, QueueKind, SpeculativeState};
 use super::scheduler::{SchedulerCtx, ComponentCtx};
+use super::scheduler::{SchedulerCtx, ComponentCtx, MessageTicket};
 use super::port::{Port, PortIdLocal, Channel, PortKind};
 use super::consensus::{Consensus, Consistency, find_ports_in_value_group};
 use super::connector::{ConnectorScheduling, ConnectorPDL};
@@ @@ -111,9 +111,13 @@ impl ConnectorApplication { @@
+    }
     fn handle_new_messages(&mut self, comp_ctx: &mut ComponentCtx) {
         while let Some(message) = comp_ctx.read_next_message() {
             match message {
                 Message::Data(message) => self.handle_new_data_message(message, comp_ctx),
         while let Some(ticket) = comp_ctx.get_next_message_ticket() {
             let message = comp_ctx.read_message_using_ticket(ticket);
             if let Message::Data(_) = message {
                 self.handle_new_data_message(ticket, comp_ctx)
             } else {
                 match comp_ctx.take_message_using_ticket(ticket) {
                     Message::Data(message) => unreachable!(),
                     Message::SyncComp(message) => self.handle_new_sync_comp_message(message, comp_ctx),
                     Message::SyncPort(message) => self.handle_new_sync_port_message(message, comp_ctx),
                     Message::SyncControl(message) => todo!("implement"),
@@ @@ -121,17 +125,19 @@ impl ConnectorApplication { @@
+                }
+            }
+        }
+    }
-    pub(crate) fn handle_new_data_message(&mut self, message: DataMessage, ctx: &mut ComponentCtx) {
+    pub(crate) fn handle_new_data_message(&mut self, ticket: MessageTicket, ctx: &mut ComponentCtx) {
         // Go through all branches that are awaiting new messages and see if
         // there is one that can receive this message.
-        if !self.consensus.handle_new_data_message(&message, ctx) {
+        if !self.consensus.handle_new_data_message(ticket, ctx) {
             // Old message, so drop it
             return;
+        }
         let mut iter_id = self.tree.get_queue_first(QueueKind::AwaitingMessage);
         while let Some(branch_id) = iter_id {
             let message = ctx.read_message_using_ticket(ticket).as_data();
             iter_id = self.tree.get_queue_next(branch_id);
             let branch = &self.tree[branch_id];

src/runtime2/scheduler.rs

➞

Show inline comments

 use std::collections::VecDeque;
 use std::mem::MaybeUninit;
 use std::sync::Arc;
 use std::sync::atomic::Ordering;
 use crate::collections::RawVec;
 use crate::protocol::eval::EvalError;
 use crate::runtime2::port::ChannelId;
@@ @@ -137,7 +139,7 @@ impl Scheduler { @@
             // Check if the message has to be rerouted because we have moved the
             // target port to another component.
             self.debug_conn(connector_id, &format!("Handling message\n --- {:#?}", message));
-            if let Some(target_port) = Self::get_message_target_port(&message) {
+            if let Some(target_port) = message.target_port() {
                 if let Some(other_component_id) = scheduled.router.should_reroute(target_port) {
                     self.debug_conn(connector_id, " ... Rerouting the message");
                     self.runtime.send_message(other_component_id, message);
@@ @@ -186,14 +188,17 @@ impl Scheduler { @@
                             self.runtime.send_message(message.sending_component_id, ack_message);
                         },
                         ControlContent::Ack => {
                             scheduled.router.handle_ack(message.id);
                             if let Some((target_component, new_control_message)) = scheduled.router.handle_ack(connector_id, message.id) {
                                 self.debug_conn(connector_id, &format!("Sending message [ack ack] \n --- {:?}", new_control_message));
                                 self.runtime.send_message(target_component, new_control_message);
                             };
                         },
                         ControlContent::Ping => {},
+                    }
                 },
                 _ => {
                     // All other cases have to be handled by the component
-                    scheduled.ctx.inbox_messages.push(message);
+                    scheduled.ctx.inbox.insert_new(message);
+                }
+            }
+        }
@@ @@ -208,8 +213,10 @@ impl Scheduler { @@
         // Note: we're not handling the public inbox, we're dealing with the
         // private one!
         debug_assert!(scheduled.shutting_down);
         while let Some(message) = scheduled.ctx.read_next_message_even_if_not_in_sync() {
             let target_port_and_round_number = match &message {
         while let Some(ticket) = scheduled.ctx.get_next_message_ticket_even_if_not_in_sync() {
             let message = scheduled.ctx.read_message_using_ticket(ticket);
             let target_port_and_round_number = match message {
                 Message::Data(msg) => Some((msg.data_header.target_port, msg.sync_header.sync_round)),
                 Message::SyncComp(_) => None,
                 Message::SyncPort(msg) => Some((msg.target_port, msg.sync_header.sync_round)),
@@ @@ -293,21 +300,7 @@ impl Scheduler { @@
                     for port_id in initial_ports {
                         // Transfer messages associated with the transferred port
                         let mut message_idx = 0;
                         while message_idx < scheduled.ctx.inbox_messages.len() {
                             let message = &scheduled.ctx.inbox_messages[message_idx];
                             if Self::get_message_target_port(message) == Some(port_id) {
                                 // Need to transfer this message
                                 // TODO: Revise messages, this is becoming messy and error-prone
                                 let message = scheduled.ctx.inbox_messages.remove(message_idx);
                                 if message_idx < scheduled.ctx.inbox_len_read {
                                     scheduled.ctx.inbox_len_read -= 1;
+                                }
                                 new_connector.ctx.inbox_messages.push(message);
                             } else {
                                 message_idx += 1;
+                            }
+                        }
                         scheduled.ctx.inbox.transfer_messages_for_port(port_id, &mut new_connector.ctx.inbox);
                         // Transfer the port itself
                         let port_index = scheduled.ctx.ports.iter()
@@ @@ -322,7 +315,8 @@ impl Scheduler { @@
                         if port.state == PortState::Open {
                             let reroute_message = scheduled.router.prepare_reroute(
                                 port.self_id, port.peer_id, scheduled.ctx.id,
                                 port.peer_connector, new_connector.ctx.id
                                 port.peer_connector, new_connector.ctx.id,
                                 &mut new_connector.router
                             );
                             self.debug_conn(connector_id, &format!("Sending message [newcon]\n --- {:?}", reroute_message));
@@ @@ -355,10 +349,9 @@ impl Scheduler { @@
             if scheduled.ctx.is_in_sync {
                 // Just entered sync region
             } else {
                 // Just left sync region. So clear inbox up until the last
                 // message that was read.
                 scheduled.ctx.inbox_messages.drain(0..scheduled.ctx.inbox_len_read);
                 scheduled.ctx.inbox_len_read = 0;
                 // Just left sync region. So prepare inbox for the next sync
                 // round
                 scheduled.ctx.inbox.prepare_for_next_round();
+            }
             scheduled.ctx.changed_in_sync = false; // reset flag
@@ @@ -388,25 +381,6 @@ impl Scheduler { @@
+        }
+    }
     #[inline]
     fn get_message_target_port(message: &Message) -> Option<PortIdLocal> {
         match message {
             Message::Data(data) => return Some(data.data_header.target_port),
             Message::SyncComp(_) => {},
             Message::SyncPort(content) => return Some(content.target_port),
             Message::SyncControl(_) => return None,
             Message::Control(control) => {
                 match &control.content {
                     ControlContent::PortPeerChanged(port_id, _) => return Some(*port_id),
                     ControlContent::CloseChannel(port_id) => return Some(*port_id),
                     ControlContent::Ping | ControlContent::Ack => {},
+                }
             },
+        }
         return None
+    }
     // TODO: Remove, this is debugging stuff
     fn debug(&self, message: &str) {
         println!("DEBUG [thrd:{:02} conn:  ]: {}", self.scheduler_id, message);
@@ @@ -442,8 +416,7 @@ pub(crate) struct ComponentCtx { @@
     // Mostly managed by the scheduler
     pub(crate) id: ConnectorId,
     ports: Vec<Port>,
     inbox_messages: Vec<Message>,
     inbox_len_read: usize,
     inbox: Inbox,
     // Submitted by the component
     is_in_sync: bool,
     changed_in_sync: bool,
@@ @@ -462,8 +435,7 @@ impl ComponentCtx { @@
         return Self{
             id: ConnectorId::new_invalid(),
             ports: Vec::new(),
             inbox_messages: Vec::new(),
             inbox_len_read: 0,
             inbox: Inbox::new(),
             is_in_sync: false,
             changed_in_sync: false,
             outbox: VecDeque::new(),
@@ @@ -570,43 +542,33 @@ impl ComponentCtx { @@
     /// those messages that have been previously received with
     /// `read_next_message`.
     pub(crate) fn get_read_data_messages(&self, match_port_id: PortIdLocal) -> MessagesIter {
         return MessagesIter {
             messages: &self.inbox_messages,
             next_index: 0,
             max_index: self.inbox_len_read,
             match_port_id
         };
         return self.inbox.get_read_data_messages(match_port_id);
+    }
     /// Retrieves the next unread message from the inbox `None` if there are no
     /// (new) messages to read.
     // TODO: Fix the clone of the data message, entirely unnecessary
     pub(crate) fn read_next_message(&mut self) -> Option<Message> {
     pub(crate) fn get_next_message_ticket(&mut self) -> Option<MessageTicket> {
         if !self.is_in_sync { return None; }
-        return self.read_next_message_even_if_not_in_sync();
+        return self.inbox.get_next_message_ticket();
+    }
     pub(crate) fn read_next_message_even_if_not_in_sync(&mut self) -> Option<Message> {
         if self.inbox_len_read == self.inbox_messages.len() { return None; }
     #[inline]
     pub(crate) fn get_next_message_ticket_even_if_not_in_sync(&mut self) -> Option<MessageTicket> {
         return self.inbox.get_next_message_ticket();
+    }
         // We want to keep data messages in the inbox, because we need to check
         // them in the future. We don't want to keep sync messages around, we
         // should only handle them once. Control messages should never be in
         // here.
         let message = &self.inbox_messages[self.inbox_len_read];
         match &message {
             Message::Data(content) => {
                 // Keep message in inbox for later reading
                 self.inbox_len_read += 1;
                 return Some(Message::Data(content.clone()));
             },
             Message::SyncComp(_) | Message::SyncPort(_) | Message::SyncControl(_) => {
                 // Remove message from inbox
                 let message = self.inbox_messages.remove(self.inbox_len_read);
                 return Some(message);
             },
             Message::Control(_) => unreachable!("control message ended up in component inbox"),
     #[inline]
     pub(crate) fn read_message_using_ticket(&self, ticket: MessageTicket) -> &Message {
         return self.inbox.read_message_using_ticket(ticket);
+    }
     #[inline]
     pub(crate) fn take_message_using_ticket(&mut self, ticket: MessageTicket) -> Message {
         return self.inbox.take_message_using_ticket(ticket)
+    }
     /// Puts back a message back into the inbox. The reason being that the
     /// message is actually part of the next sync round. This will
     pub(crate) fn put_back_message(&mut self, message: Message) {
         self.inbox.put_back_message(message);
+    }
+}
@@ @@ -646,6 +608,173 @@ impl<'a> Iterator for MessagesIter<'a> { @@
+    }
+}
 // -----------------------------------------------------------------------------
 // Private Inbox
 // -----------------------------------------------------------------------------
 /// A structure that contains inbox messages. Some messages are left inside and
 /// continuously re-read. Others are taken out, but may potentially be put back
 /// for later reading. Later reading in this case implies that they are put back
 /// for reading in the next sync round.
 struct Inbox {
     messages: RawVec<Message>,
     next_delay_idx: u32,
     start_read_idx: u32,
     next_read_idx: u32,
     generation: u32,
+}
 #[derive(Clone, Copy)]
 pub(crate) struct MessageTicket {
     index: u32,
     generation: u32,
+}
 impl Inbox {
     fn new() -> Self {
         return Inbox {
             messages: RawVec::new(),
             next_delay_idx: 0,
             start_read_idx: 0,
             next_read_idx: 0,
             generation: 0,
+        }
+    }
     fn insert_new(&mut self, message: Message) {
         assert!(self.messages.len() < u32::MAX as usize); // TODO: @Size
         self.messages.push(message);
+    }
     fn get_next_message_ticket(&mut self) -> Option<MessageTicket> {
         let cur_read_idx = self.next_read_idx as usize;
         if cur_read_idx >= self.messages.len() {
             return None;
+        }
         self.generation += 1;
         self.next_read_idx += 1;
         return Some(MessageTicket{
             index: cur_read_idx as u32,
             generation: self.generation
         });
+    }
     fn read_message_using_ticket(&self, ticket: MessageTicket) -> &Message {
         debug_assert_eq!(self.generation, ticket.generation);
         return unsafe{ &*self.messages.get(ticket.index as usize) }
+    }
     fn take_message_using_ticket(&mut self, ticket: MessageTicket) -> Message {
         debug_assert_eq!(self.generation, ticket.generation);
         unsafe {
             let take_idx = ticket.index as usize;
             let val = std::ptr::read(self.messages.get(take_idx));
             // Move messages to the right, clearing up space in the
             // front.
             let num_move_right = take_idx - self.start_read_idx as usize;
             self.messages.move_range(
                 self.start_read_idx as usize,
                 self.start_read_idx as usize + 1,
                 num_move_right
             );
             self.start_read_idx += 1;
             return val;
+        }
+    }
     fn put_back_message(&mut self, message: Message) {
         // We have space in front of the array because we've taken out a message
         // before.
         debug_assert!(self.next_delay_idx < self.start_read_idx);
         unsafe {
             // Write to front of the array
             std::ptr::write(self.messages.get_mut(self.next_delay_idx as usize), message);
             self.next_delay_idx += 1;
+        }
+    }
     fn get_read_data_messages(&self, match_port_id: PortIdLocal) -> MessagesIter {
         return MessagesIter{
             messages: self.messages.as_slice(),
             next_index: self.start_read_idx as usize,
             max_index: self.next_read_idx as usize,
             match_port_id
         };
+    }
     fn prepare_for_next_round(&mut self) {
         // Deallocate everything that was read
         self.destroy_range(self.start_read_idx, self.next_read_idx);
         self.generation += 1;
         // Join up all remaining values with the delayed ones in the front
         let num_to_move = self.messages.len() - self.next_read_idx as usize;
         self.messages.move_range(
             self.next_read_idx as usize,
             self.next_delay_idx as usize,
             num_to_move
         );
         // Set all indices (and the RawVec len) to make sense in this new state
         let new_len = self.next_delay_idx as usize + num_to_move;
         self.next_delay_idx = 0;
         self.start_read_idx = 0;
         self.next_read_idx = 0;
         self.messages.len = new_len;
+    }
     fn transfer_messages_for_port(&mut self, port: PortIdLocal, new_inbox: &mut Inbox) {
         // Convoluted assert to make sure we're in non-sync mode, as that is
         // when this is called, and that makes our lives easier
         let mut idx = 0;
         while idx < self.messages.len() {
             let message = unsafe{ &*self.messages.get(idx) };
             if let Some(target_port) = message.target_port() {
                 if target_port == port {
                     // Transfer port
                     unsafe {
                         let message = std::ptr::read(message as *const _);
                         let remaining = self.messages.len() - idx;
                         if remaining > 1 {
                             self.messages.move_range(idx + 1, idx, remaining - 1);
+                        }
                         self.messages.len -= 1;
                         new_inbox.insert_new(message);
+                    }
                 } else {
                     // Do not transfer port
                     idx += 1;
+                }
+            }
+        }
+    }
     #[inline]
     fn destroy_range(&mut self, start_idx: u32, end_idx: u32) {
         for idx in (start_idx as usize)..(end_idx as usize) {
             unsafe {
                 let msg = self.messages.get_mut(idx);
                 std::ptr::drop_in_place(msg);
+            }
+        }
+    }
+}
 impl Drop for Inbox {
     fn drop(&mut self) {
         // Whether in sync or not in sync. We have two ranges of allocated
         // messages:
         // - delayed messages: from 0 to `next_delay_idx` (which is 0 if in non-sync)
         // - readable messages: from `start_read_idx` to `messages.len`
         self.destroy_range(0, self.next_delay_idx);
         self.destroy_range(self.start_read_idx, self.messages.len as u32);
+    }
+}
 // -----------------------------------------------------------------------------
 // Control messages
 // -----------------------------------------------------------------------------
@@ @@ -658,12 +787,14 @@ struct ControlEntry { @@
 enum ControlVariant {
     ChangedPort(ControlChangedPort),
     ClosedChannel(ControlClosedChannel),
     ReroutePending,
+}
 struct ControlChangedPort {
     target_port: PortIdLocal,       // if send to this port, then reroute
     source_connector: ConnectorId,  // connector we expect messages from
     target_connector: ConnectorId,  // connector we need to reroute to
     id_of_ack_after_confirmation: u32, // control message ID we need to send to the target upon receiving an ack
+}
 struct ControlClosedChannel {
@@ @@ -714,19 +845,26 @@ impl ControlMessageHandler { @@
         &mut self,
         port_id: PortIdLocal, peer_port_id: PortIdLocal,
         self_connector_id: ConnectorId, peer_connector_id: ConnectorId,
         new_owner_connector_id: ConnectorId
+        new_owner_connector_id: ConnectorId, new_owner_ctrl_handler: &mut ControlMessageHandler,
     ) -> ControlMessage {
         let id = self.take_id();
         let new_owner_id = new_owner_ctrl_handler.take_id();
         self.active.push(ControlEntry{
             id,
             variant: ControlVariant::ChangedPort(ControlChangedPort{
                 target_port: port_id,
                 source_connector: peer_connector_id,
                 target_connector: new_owner_connector_id,
                 id_of_ack_after_confirmation: new_owner_id,
             }),
         });
         new_owner_ctrl_handler.active.push(ControlEntry{
             id: new_owner_id,
             variant: ControlVariant::ReroutePending,
         });
         return ControlMessage {
             id,
             sending_component_id: self_connector_id,
@@ @@ -749,14 +887,33 @@ impl ControlMessageHandler { @@
         return None;
+    }
     /// Handles an Ack as an answer to a previously sent control message
     pub fn handle_ack(&mut self, id: u32) {
     /// Handles an Ack as an answer to a previously sent control message.
     /// Handling an Ack might spawn a new message that needs to be sent.
     pub fn handle_ack(&mut self, handler_component_id: ConnectorId, id: u32) -> Option<(ConnectorId, Message)> {
         let index = self.active.iter()
             .position(|v| v.id == id);
         match index {
             Some(index) => { self.active.remove(index); },
             None => { todo!("handling of nefarious ACKs"); },
             Some(index) => {
                 let removed = self.active.remove(index);
                 match removed.variant {
                     ControlVariant::ChangedPort(message) => {
                         return Some((
                             message.target_connector,
                             Message::Control(ControlMessage{
                                 id: message.id_of_ack_after_confirmation,
                                 sending_component_id: handler_component_id,
                                 content: ControlContent::Ack
                             })
                         ));
                     },
                     _ => return None,
+                }
             },
             None => {
                 todo!("handling of nefarious ACKs");
                 return None;
             },
+        }
+    }

src/runtime2/tests/mod.rs

➞

Show inline comments

@@ @@ -15,9 +15,9 @@ use crate::runtime2::native::{ApplicationSyncAction}; @@
 // pub(crate) const NUM_INSTANCES: u32 = 7;   // number of test instances constructed
 // pub(crate) const NUM_LOOPS: u32 = 8;       // number of loops within a single test (not used by all tests)
 pub(crate) const NUM_THREADS: u32 = 6;
 pub(crate) const NUM_INSTANCES: u32 = 2;
 pub(crate) const NUM_LOOPS: u32 = 5;
 pub(crate) const NUM_THREADS: u32 = 4;
 pub(crate) const NUM_INSTANCES: u32 = 1;
 pub(crate) const NUM_LOOPS: u32 = 3;
 fn create_runtime(pdl: &str) -> Runtime {

src/runtime2/tests/network_shapes.rs

➞

Show inline comments

@@ @@ -138,7 +138,7 @@ fn test_conga_line_request() { @@
     let thing = TestTimer::new("conga_line_request");
     run_test_in_runtime(CODE, |api| {
         api.create_connector("", "constructor", ValueGroup::new_stack(vec![
-            Value::UInt32(5),
+            Value::UInt32(1),
             Value::UInt32(NUM_LOOPS)
         ])).expect("create connector");
     });

0 comments (0 inline, 0 general)