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

Changeset - 26d47db4f922

Parent rev.

Child rev.

[Not reviewed]

0 4 0

mh - 4 years ago 2021-10-19 12:29:52
contact@maxhenger.nl

WIP on second rewrite of port management

4 files changed with 15 insertions and 11 deletions:

src/runtime2/global_store.rs

src/runtime2/native.rs

src/runtime2/port.rs

src/runtime2/scheduler.rs

0 comments (0 inline, 0 general)

src/runtime2/global_store.rs

➞

Show inline comments

@@ @@ -135,14 +135,16 @@ impl ConnectorStore { @@
             debug_assert!(!connector.is_null());
             return *connector as &mut _;
+        }
+    }
     /// Create a new connector, returning the key that can be used to retrieve
     /// and/or queue it.
     pub(crate) fn create(&self, created_by: &mut ScheduledConnector, connector: ConnectorVariant) -> ConnectorKey {
     /// and/or queue it. The caller must make sure that the constructed
     /// connector's code is initialized with the same ports as the ports in the
     /// `initial_ports` array.
     pub(crate) fn create(&self, created_by: &mut ScheduledConnector, connector: ConnectorVariant, initial_ports: Vec<Port>) -> ConnectorKey {
         // Creation of the connector in the global store, requires a lock
+        {
             let lock = self.inner.write().unwrap();
             let connector = ScheduledConnector {
                 connector,
                 context: ConnectorCtx::new(self.port_counter.clone()),
@@ @@ -178,13 +180,12 @@ impl ConnectorStore { @@
         // Transferring ownership of ports (and crashing if there is a
         // programmer's mistake in port management)
         match &new_connector.connector {
             ConnectorVariant::UserDefined(connector) => {
                 for port_id in &connector.ports.owned_ports {
                     let mut port = created_by.context.remove_port(*port_id);
                     port.owning_connector = new_connector.context.id;
                     new_connector.context.add_port(port);
+                }
             },
             ConnectorVariant::Native(_) => {}, // no initial ports (yet!)
+        }

src/runtime2/native.rs

➞

Show inline comments

@@ @@ -77,13 +77,13 @@ impl Connector for ConnectorApplication { @@
 /// interactions the `ApplicationConnector` performs within a synchronous round.
 pub struct ApplicationInterface {
     sync_done: SyncDone,
     job_queue: JobQueue,
     runtime: Arc<RuntimeInner>,
     connector_id: ConnectorId,
-    owned_ports: Vec<PortIdLocal>,
+    owned_ports: Vec<Port>,
+}
 impl ApplicationInterface {
     pub(crate) fn new(sync_done: SyncDone, job_queue: JobQueue, runtime: Arc<RuntimeInner1>) -> Self {
         return Self{
             sync_done, job_queue, runtime,
@@ @@ -96,21 +96,27 @@ impl ApplicationInterface { @@
     pub fn create_channel(&mut self) -> Channel {
         // TODO: Duplicated logic in scheduler
         let getter_id = self.runtime.global_store.connectors.port_counter.fetch_add(2, Ordering::SeqCst);
         let putter_id = PortIdLocal::new(getter_id + 1);
         let getter_id = PortIdLocal::new(getter_id);
-        self.ports.push(Port{
+        self.owned_ports.push(Port{
             self_id: getter_id,
             peer_id: putter_id,
             kind: PortKind::Getter,
             owning_connector: self.connector_id,
             peer_connector: self.connector_id,
         });
         return channel;
         self.owned_ports.push(Port{
             self_id: putter_id,
             peer_id: getter_id,
             kind: PortKind::Putter,
             peer_connector: self.connector_id,
         });
         return Channel{ putter_id, getter_id };
+    }
     /// Creates a new connector. Note that it is not scheduled immediately, but
     /// depends on the `ApplicationConnector` to run, followed by the created
     /// connector being scheduled.
     // TODO: Optimize by yanking out scheduler logic for common use.

src/runtime2/port.rs

➞

Show inline comments

@@ @@ -30,13 +30,12 @@ pub enum PortKind { @@
 /// created by the application interfacing with the runtime, instead of being
 /// created by a connector.
 pub struct Port {
     pub self_id: PortIdLocal,
     pub peer_id: PortIdLocal,
     pub kind: PortKind,
     pub owning_connector: ConnectorId,
     pub peer_connector: ConnectorId, // might be temporarily inconsistent while peer port is sent around in non-sync phase.
+}
 // TODO: Turn port ID into its own type

src/runtime2/scheduler.rs

➞

Show inline comments

@@ @@ -26,13 +26,13 @@ pub(crate) struct ConnectorCtx { @@
 impl ConnectorCtx {
     pub(crate) fn new(port_counter: Arc<AtomicU32>) -> ConnectorCtx {
         Self{
             id: ConnectorId::new_invalid(),
             port_counter,
-            ports: Vec::new(),
+            ports: initial_ports,
+        }
+    }
     /// Creates a (putter, getter) port pair belonging to the same channel. The
     /// port will be implicitly owned by the connector.
     pub(crate) fn create_channel(&mut self) -> Channel {
@@ @@ -41,21 +41,19 @@ impl ConnectorCtx { @@
         let getter_id = PortIdLocal::new(getter_id);
         self.ports.push(Port{
             self_id: getter_id,
             peer_id: putter_id,
             kind: PortKind::Getter,
             owning_connector: self.id,
             peer_connector: self.id,
         });
         self.ports.push(Port{
             self_id: putter_id,
             peer_id: getter_id,
             kind: PortKind::Putter,
             owning_connector: self.id,
             peer_connector: self.id,
         });
         return Channel{ getter_id, putter_id };
+    }

0 comments (0 inline, 0 general)