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

Changeset - 93081320c9fa

Parent rev.

Child rev.

[Not reviewed]

0 6 0

mh - 3 years ago 2022-03-23 11:25:13
contact@maxhenger.nl

Introduce trait for components

6 files changed with 78 insertions and 54 deletions:

src/runtime2/component/component.rs

src/runtime2/component/component_pdl.rs

src/runtime2/component/mod.rs

src/runtime2/mod.rs

src/runtime2/runtime.rs

src/runtime2/scheduler.rs

0 comments (0 inline, 0 general)

src/runtime2/component/component.rs

➞

Show inline comments

 use crate::protocol::eval::EvalError;
 use crate::runtime2::*;
 use super::CompCtx;
 pub enum CompScheduling {
     Immediate,
     Requeue,
     Sleep,
     Exit,
+}
 /// Generic representation of a component (as viewed by a scheduler).
 pub(crate) trait Component {
     fn handle_message(sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message);
     fn run(sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling;
     /// Called if the component is created by another component and the messages
     /// are being transferred between the two.
     fn adopt_message(&mut self, comp_ctx: &mut CompCtx, message: DataMessage);
     /// Called if the component receives a new message. The component is
     /// responsible for deciding where that messages goes.
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message);
     /// Called if the component's routine should be executed. The return value
     /// can be used to indicate when the routine should be run again.
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> Result<CompScheduling, EvalError>;
+}
@@ \ No newline at end of file @@

src/runtime2/component/component_pdl.rs

➞

Show inline comments

@@ @@ -7,13 +7,13 @@ use crate::protocol::eval::{ @@
     EvalContinuation, EvalResult, EvalError
 };
 use crate::runtime2::scheduler::SchedulerCtx;
 use crate::runtime2::communication::*;
-use super::component::CompScheduling;
+use super::component::*;
 use super::component_context::*;
 use super::control_layer::*;
 use super::consensus::Consensus;
 pub enum ExecStmt {
     CreatedChannel((Value, Value)),
@@ @@ -240,38 +240,24 @@ pub(crate) struct CompPDL { @@
     // 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 CompPDL {
     pub(crate) fn new(initial_state: Prompt, num_ports: usize) -> Self {
         let mut inbox_main = Vec::new();
         inbox_main.reserve(num_ports);
         for _ in 0..num_ports {
             inbox_main.push(None);
+        }
         return Self{
             mode: Mode::NonSync,
             mode_port: PortId::new_invalid(),
             mode_value: ValueGroup::default(),
             select: SelectState::new(),
             prompt: initial_state,
             control: ControlLayer::default(),
             consensus: Consensus::new(),
             sync_counter: 0,
             exec_ctx: ExecCtx{
                 stmt: ExecStmt::None,
             },
             inbox_main,
             inbox_backup: Vec::new(),
 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);
+        }
+    }
-    pub(crate) fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, mut message: Message) {
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, mut message: Message) {
         sched_ctx.log(&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);
             target.send_message(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());
@@ @@ -288,17 +274,13 @@ impl CompPDL { @@
             Message::Sync(message) => {
                 self.handle_incoming_sync_message(sched_ctx, comp_ctx, message);
+            }
+        }
+    }
     // -------------------------------------------------------------------------
     // Running component and handling changes in global component state
     // -------------------------------------------------------------------------
     pub(crate) fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> Result<CompScheduling, EvalError> {
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> Result<CompScheduling, EvalError> {
         use EvalContinuation as EC;
         sched_ctx.log(&format!("Running component (mode: {:?})", self.mode));
         // Depending on the mode don't do anything at all, take some special
         // actions, or fall through and run the PDL code.
@@ @@ -439,12 +421,42 @@ impl CompPDL { @@
                 self.inbox_main.push(None);
                 self.inbox_main.push(None);
                 return Ok(CompScheduling::Immediate);
+            }
+        }
+    }
+}
 impl CompPDL {
     pub(crate) fn new(initial_state: Prompt, num_ports: usize) -> Self {
         let mut inbox_main = Vec::new();
         inbox_main.reserve(num_ports);
         for _ in 0..num_ports {
             inbox_main.push(None);
+        }
         return Self{
             mode: Mode::NonSync,
             mode_port: PortId::new_invalid(),
             mode_value: ValueGroup::default(),
             select: SelectState::new(),
             prompt: initial_state,
             control: ControlLayer::default(),
             consensus: Consensus::new(),
             sync_counter: 0,
             exec_ctx: ExecCtx{
                 stmt: ExecStmt::None,
             },
             inbox_main,
             inbox_backup: Vec::new(),
+        }
+    }
     // -------------------------------------------------------------------------
     // Running component and handling changes in global component state
     // -------------------------------------------------------------------------
     fn execute_prompt(&mut self, sched_ctx: &SchedulerCtx) -> EvalResult {
         let mut step_result = EvalContinuation::Stepping;
         while let EvalContinuation::Stepping = step_result {
             step_result = self.prompt.step(
                 &sched_ctx.runtime.protocol.types, &sched_ctx.runtime.protocol.heap,
@@ @@ -876,60 +888,59 @@ impl CompPDL { @@
             definition_id, type_id, arguments,
         );
         let component = CompPDL::new(prompt, port_id_pairs.len());
         let (created_key, component) = sched_ctx.runtime.finish_create_pdl_component(
             reservation, component, created_ctx, false,
         );
         let created_ctx = &component.ctx;
         // Now modify the creator's ports: remove every transferred port and
         // potentially remove the peer component. Here is also where we will
         // transfer messages in the main inbox.
         // potentially remove the peer component.
         for pair in port_id_pairs.iter() {
             // Remove peer if appropriate
             let creator_port_info = creator_ctx.get_port(pair.creator_handle);
             let creator_port_index = creator_ctx.get_port_index(pair.creator_handle);
             let creator_peer_comp_id = creator_port_info.peer_comp_id;
             creator_ctx.remove_peer(sched_ctx, pair.creator_handle, creator_peer_comp_id, false);
             creator_ctx.remove_port(pair.creator_handle);
             // Transfer any messages
             let created_port_index = created_ctx.get_port_index(pair.created_handle);
             let created_port_info = created_ctx.get_port(pair.created_handle);
             debug_assert!(component.code.inbox_main[created_port_index].is_none());
             if let Some(mut message) = self.inbox_main.remove(creator_port_index) {
                 message.data_header.target_port = pair.created_id;
-                component.code.inbox_main[created_port_index] = Some(message);
+                component.component.adopt_message(&mut component.ctx, message)
+            }
             let mut message_index = 0;
             while message_index < self.inbox_backup.len() {
                 let message = &self.inbox_backup[message_index];
                 if message.data_header.target_port == pair.creator_id {
                     // transfer message
                     let mut message = self.inbox_backup.remove(message_index);
                     message.data_header.target_port = pair.created_id;
-                    component.code.inbox_backup.push(message);
+                    component.component.adopt_message(&mut component.ctx, message);
                 } else {
                     message_index += 1;
+                }
+            }
             // Handle potential channel between creator and created component
             let created_port_info = component.ctx.get_port(pair.created_handle);
             if created_port_info.peer_comp_id == creator_ctx.id {
                 let peer_port_handle = creator_ctx.get_port_handle(created_port_info.peer_port_id);
                 let peer_port_info = creator_ctx.get_port_mut(peer_port_handle);
-                peer_port_info.peer_comp_id = created_ctx.id;
+                peer_port_info.peer_comp_id = component.ctx.id;
                 peer_port_info.peer_port_id = created_port_info.self_id;
-                creator_ctx.add_peer(peer_port_handle, sched_ctx, created_ctx.id, None);
+                creator_ctx.add_peer(peer_port_handle, sched_ctx, component.ctx.id, None);
+            }
+        }
         // By now all ports have been transferred. We'll now do any of the setup
         // for rerouting/messaging
         // By now all ports and messages have been transferred. If there are any
         // peers that need to be notified about this new component, then we
         // initiate the protocol that will notify everyone here.
         if created_component_has_remote_peers {
             let created_ctx = &component.ctx;
             let schedule_entry_id = self.control.add_schedule_entry(created_ctx.id);
             for pair in port_id_pairs.iter() {
                 let port_info = created_ctx.get_port(pair.created_handle);
                 if port_info.peer_comp_id != creator_ctx.id && port_info.peer_comp_id != created_ctx.id {
                     let message = self.control.add_reroute_entry(
                         creator_ctx.id, port_info.peer_port_id, port_info.peer_comp_id,

src/runtime2/component/mod.rs

➞

Show inline comments

 mod component_pdl;
 mod component_context;
 mod control_layer;
 mod consensus;
 mod component;
 pub(crate) use component_pdl::{CompPDL, CompScheduling};
 pub(crate) use component::{Component, CompScheduling};
 pub(crate) use component_pdl::{CompPDL};
 pub(crate) use component_context::CompCtx;
 pub(crate) use control_layer::{ControlId};
 use super::scheduler::*;
 use super::runtime::*;

src/runtime2/mod.rs

➞

Show inline comments

@@ @@ -4,7 +4,7 @@ mod component; @@
 mod communication;
 mod scheduler;
 #[cfg(test)] mod tests;
 pub use runtime::Runtime;
 pub(crate) use scheduler::SchedulerCtx;
 pub(crate) use communication::Message;
@@ \ No newline at end of file @@
 pub(crate) use communication::{Message, ControlMessage, SyncMessage, DataMessage};
@@ \ No newline at end of file @@

src/runtime2/runtime.rs

➞

Show inline comments

@@ @@ -2,13 +2,13 @@ use std::sync::{Arc, Mutex, Condvar}; @@
 use std::sync::atomic::{AtomicU32, AtomicBool, Ordering};
 use std::collections::VecDeque;
 use crate::protocol::*;
 use super::communication::Message;
 use super::component::{wake_up_if_sleeping, CompPDL, CompCtx};
+use super::component::{Component, wake_up_if_sleeping, CompPDL, CompCtx};
 use super::store::{ComponentStore, ComponentReservation, QueueDynMpsc, QueueDynProducer};
 use super::scheduler::*;
 // -----------------------------------------------------------------------------
 // Component
 // -----------------------------------------------------------------------------
@@ @@ -50,17 +50,18 @@ pub(crate) struct CompReserved { @@
 impl CompReserved {
     pub(crate) fn id(&self) -> CompId {
         return CompId(self.reservation.index)
+    }
+}
 /// Private fields of a component, may only be modified by a single thread at
 /// a time.
 /// Representation of a runtime component. Contains the bookkeeping variables
 /// for the schedulers, the publicly accessible fields, and the private fields
 /// that should only be accessed by the thread running the component's routine.
 pub(crate) struct RuntimeComp {
     pub public: CompPublic,
-    pub code: CompPDL,
+    pub component: Box<dyn Component>,
     pub ctx: CompCtx,
     pub inbox: QueueDynMpsc<Message>,
     pub exiting: bool,
+}
 /// Should contain everything that is accessible in a thread-safe manner
@@ @@ -243,28 +244,28 @@ impl RuntimeInner { @@
     pub(crate) fn start_create_pdl_component(&self) -> CompReserved {
         self.increment_active_components();
         let reservation = self.components.reserve();
         return CompReserved{ reservation };
+    }
     pub(crate) fn finish_create_pdl_component(
+    pub(crate) fn finish_create_pdl_component<C: 'static + Component>(
         &self, reserved: CompReserved,
-        component: CompPDL, mut context: CompCtx, initially_sleeping: bool,
+        component: C, mut context: CompCtx, initially_sleeping: bool,
     ) -> (CompKey, &mut RuntimeComp) {
         let inbox_queue = QueueDynMpsc::new(16);
         let inbox_producer = inbox_queue.producer();
         let _id = reserved.id();
         context.id = reserved.id();
         let component = RuntimeComp {
             public: CompPublic{
                 sleeping: AtomicBool::new(initially_sleeping),
                 num_handles: AtomicU32::new(1), // the component itself acts like a handle
                 inbox: inbox_producer,
             },
-            code: component,
+            component: Box::new(component),
             ctx: context,
             inbox: inbox_queue,
             exiting: false,
         };
         let index = self.components.submit(reserved.reservation, component);

src/runtime2/scheduler.rs

➞

Show inline comments

@@ @@ -58,15 +58,15 @@ impl Scheduler { @@
             // Run the component until it no longer indicates that it needs to
             // be re-executed immediately.
             let mut new_scheduling = CompScheduling::Immediate;
             while let CompScheduling::Immediate = new_scheduling {
                 while let Some(message) = component.inbox.pop() {
-                    component.code.handle_message(&mut scheduler_ctx, &mut component.ctx, message);
+                    component.component.handle_message(&mut scheduler_ctx, &mut component.ctx, message);
+                }
-                new_scheduling = component.code.run(&mut scheduler_ctx, &mut component.ctx).expect("TODO: Handle error");
+                new_scheduling = component.component.run(&mut scheduler_ctx, &mut component.ctx).expect("TODO: Handle error");
+            }
             // Handle the new scheduling
             match new_scheduling {
                 CompScheduling::Immediate => unreachable!(),
                 CompScheduling::Requeue => { self.runtime.enqueue_work(comp_key); },

0 comments (0 inline, 0 general)