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

Changeset - c03e28261b5d

Parent rev.

Child rev.

[Not reviewed]

0 4 0

mh - 3 years ago 2022-03-30 15:03:03
contact@maxhenger.nl

Initial implementation of random number component

4 files changed with 65 insertions and 12 deletions:

src/runtime2/communication.rs

src/runtime2/component/component.rs

src/runtime2/component/component_ip.rs

src/runtime2/component/component_pdl.rs

0 comments (0 inline, 0 general)

src/runtime2/communication.rs

➞

Show inline comments

 use crate::protocol::eval::*;
 use super::runtime::*;
 use super::component::*;
 // -----------------------------------------------------------------------------
 // Generic types
 // -----------------------------------------------------------------------------
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub struct PortId(pub u32);
 impl PortId {
     /// This value is not significant, it is chosen to make debugging easier: a
     /// very large port number is more likely to shine a light on bugs.
     pub fn new_invalid() -> Self {
         return Self(u32::MAX);
+    }
+}
 pub struct CompPortIds {
     pub comp: CompId,
     pub port: PortId,
+}
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
 pub enum PortKind {
     Putter,
     Getter,
+}
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum PortState {
     Open,
     BlockedDueToPeerChange,
     BlockedDueToFullBuffers,
     Closed,
+}
 impl PortState {
     pub fn is_blocked(&self) -> bool {
         match self {
             PortState::BlockedDueToPeerChange | PortState::BlockedDueToFullBuffers => true,
             PortState::Open | PortState::Closed => false,
+        }
+    }
+}
 pub struct Channel {

src/runtime2/component/component.rs

➞

Show inline comments

@@ @@ -247,49 +247,49 @@ pub(crate) fn default_handle_start_exit( @@
+    }
     return CompScheduling::Immediate; // to check if we can shut down immediately
+}
 /// Handles a component waiting until all peers are notified that it is quitting
 /// (i.e. after calling `default_handle_start_exit`).
 pub(crate) fn default_handle_busy_exit(
     exec_state: &mut CompExecState, control: &ControlLayer,
     sched_ctx: &SchedulerCtx
 ) -> CompScheduling {
     debug_assert_eq!(exec_state.mode, CompMode::BusyExit);
     if control.has_acks_remaining() {
         sched_ctx.log("Component busy exiting, still has `Ack`s remaining");
         return CompScheduling::Sleep;
     } else {
         sched_ctx.log("Component busy exiting, now shutting down");
         exec_state.mode = CompMode::Exit;
         return CompScheduling::Exit;
+    }
+}
 #[inline]
 pub(crate) fn default_handle_exit(_exec_state: &CompExecState) -> CompScheduling {
-    debug_assert_eq!(exec_state.mode, CompMode::Exit);
+    debug_assert_eq!(_exec_state.mode, CompMode::Exit);
     return CompScheduling::Exit;
+}
 // -----------------------------------------------------------------------------
 // Internal messaging/state utilities
 // -----------------------------------------------------------------------------
 /// Handles an `Ack` for the control layer.
 fn default_handle_ack(
     control: &mut ControlLayer, control_id: ControlId,
     sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx
 ) {
     // 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 {
             AckAction::SendMessage(target_comp, message) => {
                 // FIX @NoDirectHandle
                 let mut handle = sched_ctx.runtime.get_component_public(target_comp);
                 handle.send_message(sched_ctx, Message::Control(message), true);
                 let _should_remove = handle.decrement_users();
                 debug_assert!(_should_remove.is_none());
             },
@@ @@ -321,48 +321,49 @@ fn default_send_ack( @@
     sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx
 ) {
     let peer_info = comp_ctx.get_peer(peer_handle);
     peer_info.handle.send_message(sched_ctx, Message::Control(ControlMessage{
         id: causer_of_ack_id,
         sender_comp_id: comp_ctx.id,
         target_port_id: None,
         content: ControlMessageContent::Ack
     }), true);
+}
 /// Handles the unblocking of a putter port. In case there is a pending message
 /// on that port then it will be sent.
 fn default_handle_unblock_put(
     exec_state: &mut CompExecState, consensus: &mut Consensus,
     port_handle: LocalPortHandle, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,
 ) {
     let port_info = comp_ctx.get_port_mut(port_handle);
     let port_id = port_info.self_id;
     debug_assert!(port_info.state.is_blocked());
     port_info.state = PortState::Open;
     if exec_state.is_blocked_on_put(port_id) {
         // Annotate the message that we're going to send
         let port_info = comp_ctx.get_port(port_handle); // for immutable access
         debug_assert_eq!(port_info.kind, PortKind::Putter);
         let to_send = exec_state.mode_value.take();
         let to_send = consensus.annotate_data_message(comp_ctx, port_info, to_send);
         // Retrieve peer to send the message
         let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
         let peer_info = comp_ctx.get_peer(peer_handle);
         peer_info.handle.send_message(sched_ctx, Message::Data(to_send), true);
         exec_state.mode = CompMode::Sync; // because we're blocked on a `put`, we must've started in the sync state.
         exec_state.mode_port = PortId::new_invalid();
+    }
+}
 #[inline]
 pub(crate) fn port_id_from_eval(port_id: EvalPortId) -> PortId {
     return PortId(port_id.id);
+}
 #[inline]
 pub(crate) fn port_id_to_eval(port_id: PortId) -> EvalPortId {
     return EvalPortId{ id: port_id.0 };
+}

src/runtime2/component/component_ip.rs

➞

Show inline comments

 use crate::protocol::eval::{ValueGroup, EvalError};
 use rand::prelude as random;
 use rand::RngCore;
 use crate::protocol::eval::{ValueGroup, Value, EvalError};
 use crate::runtime2::*;
 use super::*;
 use super::component::{self, Component, CompExecState, CompScheduling, CompMode};
 use super::control_layer::*;
 use super::consensus::*;
 /// TODO: Temporary component to figure out what to do with custom components.
 ///     This component sends random numbers between two u32 limits
 pub struct ComponentRandomU32 {
     // Properties for this specific component
     output_port_id: PortId,
     random_minimum: u32,
     random_maximum: u32,
     generator: random::ThreadRng,
     // Generic state-tracking
     exec_state: CompExecState,
     did_perform_send: bool, // when in sync mode
     control: ControlLayer,
     consensus: Consensus,
+}
 impl Component for ComponentRandomU32 {
     fn adopt_message(&mut self, _comp_ctx: &mut CompCtx, _message: DataMessage) {
         // Impossible since this component does not have any input ports in its
         // signature.
         unreachable!();
+    }
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message) {
         match message {
             Message::Data(_message) => unreachable!(),
             Message::Sync(message) => {
                 let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);
                 self.handle_sync_decision(sched_ctx, comp_ctx, decision);
             },
             Message::Control(message) => {
                 component::default_handle_control_message(
                     &mut self.exec_state, &mut self.control, &mut self.consensus,
                     message, sched_ctx, comp_ctx
                 );
+            }
+        }
+    }
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> Result<CompScheduling, EvalError> {
         sched_ctx.log(&format!("Running component ComponentRandomU32 (mode: {:?})", self.exec_state.mode));
         match self.exec_state.mode {
             CompMode::BlockedGet | CompMode::BlockedSelect => {
                 // impossible for this component, no input ports and no select
                 // blocks
                 unreachable!();
+            }
             CompMode::NonSync => {
                 // If in non-sync mode then we check if the arguments make sense
                 // (at some point in the future, this is just a testing
                 // component)
                 // component).
                 if self.random_minimum >= self.random_maximum {
                     // Could throw an evaluation error, but lets just panic
                     panic!("going to crash 'n burn your system now, please provide valid arguments");
+                }
                 sched_ctx.log("Entering sync mode");
                 self.did_perform_send = false;
                 self.consensus.notify_sync_start(comp_ctx);
                 self.exec_state.mode = CompMode::Sync;
                 return Ok(CompScheduling::Immediate);
             },
             CompMode::Sync => {
                 // This component just sends a single message, then waits until
                 // consensus has been reached
                 if !self.did_perform_send {
                     sched_ctx.log("Sending random message");
                     let mut random = self.generator.next_u32() - self.random_minimum;
                     let random_delta = self.random_maximum - self.random_minimum;
                     random %= random_delta;
                     random += self.random_minimum;
                     let value_group = ValueGroup::new_stack(vec![Value::UInt32(random)]);
                     let port_handle = comp_ctx.get_port_handle(self.output_port_id);
                     let port_info = comp_ctx.get_port(port_handle);
                     let scheduling = if port_info.state.is_blocked() {
                         // Need to wait until we can send the message
                         self.exec_state.set_as_blocked_put(self.output_port_id, value_group);
                         CompScheduling::Sleep
                     } else {
                         let message = self.consensus.annotate_data_message(comp_ctx, port_info, value_group);
                         let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
                         let peer_info = comp_ctx.get_peer(peer_handle);
                         peer_info.handle.send_message(sched_ctx, Message::Data(message), true);
                         // Remain in sync mode, but after `did_perform_send` was
                         // set to true.
                         CompScheduling::Immediate
                     };
                     // Blocked or not, we set `did_perform_send` to true. If
                     // blocked then the moment we become unblocked (and are back
                     // at the `Sync` mode) we have sent the message.
                     self.did_perform_send = true;
                     return Ok(scheduling)
                 } else {
                     // Message was sent, finish this sync round
                     sched_ctx.log("Waiting for consensus");
                     self.exec_state.mode = CompMode::SyncEnd;
                     let decision = self.consensus.notify_sync_end(sched_ctx, comp_ctx);
                     self.handle_sync_decision(sched_ctx, comp_ctx, decision);
                     return Ok(CompScheduling::Requeue);
+                }
             },
             CompMode::SyncEnd | CompMode::BlockedPut => return Ok(CompScheduling::Sleep),
             CompMode::StartExit => return Ok(component::default_handle_start_exit(
                 &mut self.exec_state, &mut self.control, sched_ctx, comp_ctx
             )),
             CompMode::BusyExit => return Ok(component::default_handle_busy_exit(
                 &mut self.exec_state, &self.control, sched_ctx
             )),
             CompMode::Exit => return Ok(component::default_handle_exit(&self.exec_state)),
+        }
+    }
+}
 impl ComponentRandomU32 {
     pub(crate) fn new(arguments: ValueGroup) -> Self {
         debug_assert_eq!(arguments.values.len(), 3);
         debug_assert!(arguments.regions.is_empty());
         let port_id = component::port_id_from_eval(arguments.values[0].as_port_id());
         let minimum = arguments.values[1].as_uint32();
         let maximum = arguments.values[2].as_uint32();
         return Self{
             output_port_id: port_id,
             random_minimum: minimum,
             random_maximum: maximum,
             generator: random::thread_rng(),
             exec_state: CompExecState::new(),
             did_perform_send: false,
             control: ControlLayer::default(),
             consensus: Consensus::new(),
+        }
+    }
     fn handle_sync_decision(&mut self, _sched_ctx: &SchedulerCtx, _comp_ctx: &mut CompCtx, decision: SyncRoundDecision) {
         let success = match decision {
             SyncRoundDecision::None => return,
             SyncRoundDecision::Solution => true,
             SyncRoundDecision::Failure => false,
         };
         debug_assert_eq!(self.exec_state.mode, CompMode::SyncEnd);
         if success {
             self.exec_state.mode = CompMode::NonSync;
             self.consensus.notify_sync_decision(decision);
         } else {
             self.exec_state.mode = CompMode::StartExit;
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/component/component_pdl.rs

➞

Show inline comments

@@ @@ -186,49 +186,49 @@ impl SelectState { @@
         'case_loop: for (case_index, case) in self.cases.iter().enumerate() {
             for port_handle in case.involved_ports.iter().copied() {
                 let port_index = comp_ctx.get_port_index(port_handle);
                 if inbox[port_index].is_none() {
                     // Condition not satisfied
                     continue 'case_loop;
+                }
+            }
             // If here then the case guard is satisfied
             self.candidates_workspace.push(case_index);
+        }
         if self.candidates_workspace.is_empty() {
             return SelectDecision::None;
         } else {
             let candidate_index = self.random.get_u64() as usize % self.candidates_workspace.len();
             return SelectDecision::Case(self.candidates_workspace[candidate_index] as u32);
+        }
+    }
+}
 pub(crate) struct CompPDL {
     pub exec_state: CompExecState,
-    pub select_state: SelectState,
     select_state: SelectState,
     pub prompt: Prompt,
     pub control: ControlLayer,
     pub consensus: Consensus,
     pub sync_counter: u32,
     pub exec_ctx: ExecCtx,
     // TODO: Temporary field, simulates future plans of having one storage place
     //  reserved per port.
     // Should be same length as the number of ports. Corresponding indices imply
     // message is intended for that port.
     pub inbox_main: InboxMain,
     pub inbox_backup: Vec<DataMessage>,
+}
 impl Component for CompPDL {
     fn adopt_message(&mut self, comp_ctx: &mut CompCtx, message: DataMessage) {
         let port_handle = comp_ctx.get_port_handle(message.data_header.target_port);
         let port_index = comp_ctx.get_port_index(port_handle);
         if self.inbox_main[port_index].is_none() {
             self.inbox_main[port_index] = Some(message);
         } else {
             self.inbox_backup.push(message);
+        }
+    }
@@ @@ -441,49 +441,49 @@ impl CompPDL { @@
         for message in self.inbox_main.iter() {
             if let Some(message) = message {
                 self.consensus.handle_new_data_message(comp_ctx, message);
+            }
+        }
         debug_assert_eq!(self.exec_state.mode, CompMode::NonSync);
         self.exec_state.mode = CompMode::Sync;
+    }
     /// Handles end of sync. The conclusion to the sync round might arise
     /// immediately (and be handled immediately), or might come later through
     /// messaging. In any case the component should be scheduled again
     /// immediately
     fn handle_sync_end(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {
         sched_ctx.log("Component ending sync mode (now waiting for solution)");
         let decision = self.consensus.notify_sync_end(sched_ctx, comp_ctx);
         self.exec_state.mode = CompMode::SyncEnd;
         self.handle_sync_decision(sched_ctx, comp_ctx, decision);
+    }
     /// Handles decision from the consensus round. This will cause a change in
     /// the internal `Mode`, such that the next call to `run` can take the
     /// appropriate next steps.
     fn handle_sync_decision(&mut self, sched_ctx: &SchedulerCtx, _comp_ctx: &mut CompCtx, decision: SyncRoundDecision) {
         sched_ctx.log(&format!("Handling sync decision: {:?} (in mode {:?})", decision, self.mode));
+        sched_ctx.log(&format!("Handling sync decision: {:?} (in mode {:?})", decision, self.exec_state.mode));
         match decision {
             SyncRoundDecision::None => {
                 // No decision yet
                 return;
             },
             SyncRoundDecision::Solution => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::SyncEnd);
                 self.exec_state.mode = CompMode::NonSync;
                 self.consensus.notify_sync_decision(decision);
             },
             SyncRoundDecision::Failure => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::SyncEnd);
                 self.exec_state.mode = CompMode::StartExit;
             },
+        }
+    }
     fn handle_component_exit(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {
         sched_ctx.log("Component exiting");
         debug_assert_eq!(self.exec_state.mode, CompMode::StartExit);
         self.exec_state.mode = CompMode::BusyExit;
         // Doing this by index, then retrieving the handle is a bit rediculous,
         // but Rust is being Rust with its borrowing rules.

0 comments (0 inline, 0 general)