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

Changeset - 4ef99b52a75d

Parent rev.

Child rev.

[Not reviewed]

0 5 0

Christopher Esterhuyse - 5 years ago 2020-07-23 15:37:37
christopher.esterhuyse@gmail.com

tightened visibility. cleaned up and simplified UDP mediator components

5 files changed with 11 insertions and 24 deletions:

Cargo.toml

src/runtime/communication.rs

src/runtime/endpoints.rs

src/runtime/mod.rs

src/runtime/setup.rs

0 comments (0 inline, 0 general)

Cargo.toml

➞

Show inline comments

@@ @@ -38,11 +38,11 @@ lazy_static = "1.4.0" @@
 [lib]
 # compile target: dynamically linked library using C ABI
 crate-type = ["cdylib"]
 [features]
-default = ["ffi", "session_optimization", "ffi_pseudo_socket_api"]
 default = ["ffi", "session_optimization"]
 ffi = [] # see src/ffi/mod.rs
 ffi_pseudo_socket_api = ["ffi", "libc", "os_socketaddr"]# see src/ffi/pseudo_socket_api.rs
 endpoint_logging = [] # see src/macros.rs
 session_optimization = [] # see src/runtime/setup.rs
@@ \ No newline at end of file @@

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -349,14 +349,12 @@ impl Connector { @@
                 // thanks to the native_spec_var, each batch has a distinct predicate
                 unreachable!()
+            }
+        }
         // restore the invariant: !native_batches.is_empty()
         comm.native_batches.push(Default::default());
         comm.endpoint_manager.udp_endpoints_round_start(&mut *cu.inner.logger);
         // Call to another big method; keep running this round until a distributed decision is reached
         let decision = Self::sync_reach_decision(
             cu,
             comm,
             &mut branching_native,
             &mut branching_proto_components,

src/runtime/endpoints.rs

➞

Show inline comments

@@ @@ -231,25 +231,25 @@ impl EndpointManager { @@
+                }
+            }
             // try receive a udp message
             let recv_buffer = self.udp_in_buffer.as_mut_slice();
             while let Some(index) = self.udp_endpoint_store.polled_undrained.pop() {
                 let ee = &mut self.udp_endpoint_store.endpoint_exts[index];
-                if let Some((bytes_written, from)) = ee.sock.recv_from(recv_buffer).ok() {
+                if let Some(bytes_written) = ee.sock.recv(recv_buffer).ok() {
                     // I received a payload!
                     self.udp_endpoint_store.polled_undrained.insert(index);
-                    if !ee.received_from_this_round.is_none() {
+                    if !ee.received_this_round {
                         let payload = Payload::from(&recv_buffer[..bytes_written]);
                         let port_spec_var = port_info.spec_var_for(ee.getter_for_incoming);
                         let predicate = Predicate::singleton(port_spec_var, SpecVal::FIRING);
                         round_ctx.getter_add(
                             ee.getter_for_incoming,
                             SendPayloadMsg { payload, predicate },
                         );
                         some_message_enqueued = true;
-                        ee.received_from_this_round = Some(from);
+                        ee.received_this_round = true;
                     } else {
                         // lose the message!
+                    }
+                }
+            }
             if some_message_enqueued {
@@ @@ -333,22 +333,12 @@ impl EndpointManager { @@
             std::mem::swap(&mut self.delayed_messages, &mut self.undelayed_messages);
             return;
+        }
         // slow path
         self.undelayed_messages.extend(self.delayed_messages.drain(..));
+    }
     pub(super) fn udp_endpoints_round_start(&mut self, logger: &mut dyn Logger) {
         log!(
             logger,
             "Starting round for {} udp endpoints",
             self.udp_endpoint_store.endpoint_exts.len()
         );
         for ee in self.udp_endpoint_store.endpoint_exts.iter_mut() {
             ee.received_from_this_round = None;
+        }
+    }
     pub(super) fn udp_endpoints_round_end(
         &mut self,
         logger: &mut dyn Logger,
         decision: &Decision,
     ) -> Result<(), UnrecoverableSyncError> {
         // retain received_from_this_round for use in pseudo_socket_api::recv_from
@@ @@ -356,32 +346,31 @@ impl EndpointManager { @@
             logger,
             "Ending round for {} udp endpoints",
             self.udp_endpoint_store.endpoint_exts.len()
         );
         use UnrecoverableSyncError as Use;
         if let Decision::Success(solution_predicate) = decision {
             'endpoint_loop: for (index, ee) in
                 self.udp_endpoint_store.endpoint_exts.iter_mut().enumerate()
+            {
                 for (payload_predicate, payload) in ee.outgoing_payloads.drain() {
             for (index, ee) in self.udp_endpoint_store.endpoint_exts.iter_mut().enumerate() {
                 'outgoing_loop: for (payload_predicate, payload) in ee.outgoing_payloads.drain() {
                     if payload_predicate.assigns_subset(solution_predicate) {
                         ee.sock.send(payload.as_slice()).map_err(|e| {
                             println!("{:?}", e);
                             Use::BrokenUdpEndpoint { index }
                         })?;
                         log!(
                             logger,
                             "Sent payload {:?} with pred {:?} through Udp endpoint {}",
                             &payload,
                             &payload_predicate,
                             index
                         );
                         continue 'endpoint_loop; // send at most one payload per endpoint per round
                         // send at most one payload per endpoint per round
                         break 'outgoing_loop;
+                    }
+                }
-                log!(logger, "Sent no message through Udp endpoint {}", index);
+                ee.received_this_round = false;
+            }
+        }
         Ok(())
+    }
+}
 impl Debug for NetEndpoint {

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -168,13 +168,13 @@ struct NetEndpointExt { @@
     net_endpoint: NetEndpoint,
     getter_for_incoming: PortId,
+}
 #[derive(Debug)]
 struct UdpEndpointExt {
     sock: UdpSocket, // already bound and connected
-    received_from_this_round: Option<SocketAddr>,
+    received_this_round: bool,
     outgoing_payloads: HashMap<Predicate, Payload>,
     getter_for_incoming: PortId,
+}
 #[derive(Debug)]
 struct Neighborhood {
     parent: Option<usize>,

src/runtime/setup.rs

➞

Show inline comments

@@ @@ -508,13 +508,13 @@ fn new_endpoint_manager( @@
             let UdpTodo { mut sock, getter_for_incoming } = udp_todo;
             let token = TokenTarget::UdpEndpoint { index }.into();
             poll.registry().reregister(&mut sock, token, Interest::READABLE).unwrap();
             UdpEndpointExt {
                 sock,
                 outgoing_payloads: Default::default(),
-                received_from_this_round: None,
+                received_this_round: false,
                 getter_for_incoming,
+            }
         })
         .collect();
     Ok(EndpointManager {
         poll,

0 comments (0 inline, 0 general)