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

Changeset - 63f3efc9c886

Parent rev.

Child rev.

[Not reviewed]

0 8 1

Christopher Esterhuyse - 5 years ago 2020-06-26 09:26:02
christopher.esterhuyse@gmail.com

more logging, tests and comments. re-implemented checking for consistency for natives (eagerly) and proto components (after run_to_blocker). deduplicated each connector's failure requests toward the sink

9 files changed with 275 insertions and 91 deletions:

examples/6_amy_log.txt

examples/6_atomic/amy.c

examples/reowolf_rs.dll

bin+mod

reowolf.h

src/common.rs

src/runtime/communication.rs

160

src/runtime/ffi.rs

src/runtime/mod.rs

src/runtime/tests.rs

0 comments (0 inline, 0 general)

examples/6_amy_log.txt

➞

Show inline comments

@@ new file 100644 @@
 CID(3382080479): Created with connector_id 3382080479
 CID(3382080479): Added port pair (out->in) ptID(3382080479'0) -> ptID(3382080479'1)
 CID(3382080479): ~~~ CONNECT called timeout None
 CID(3382080479): Successfully connected 0 endpoints
 CID(3382080479): beginning neighborhood construction
 CID(3382080479): Edge case of no neighbors! No parent an no children!
 CID(3382080479): Successfully created neighborhood Neighborhood { parent: None, children: {} }
 CID(3382080479): Beginning session optimization
 CID(3382080479): Gathered all children's maps. ConnectorId set is... []
 CID(3382080479): Inserting my own info. Unoptimized subtree map is {3382080479: SessionInfo { serde_proto_description: SerdeProtocolDescription((A big honkin' protocol description)), port_info: PortInfo { polarities: {ptID(3382080479'1): Getter, ptID(3382080479'0): Putter}, peers: {ptID(3382080479'0): ptID(3382080479'1), ptID(3382080479'1): ptID(3382080479'0)}, routes: {ptID(3382080479'0): LocalComponent(Native), ptID(3382080479'1): LocalComponent(Native)} }, proto_components: {} }}
 CID(3382080479): I am the leader! I will optimize this session
 CID(3382080479): Session map optimize START
 CID(3382080479): Session map optimize END
 CID(3382080479): Optimized info map is {3382080479: SessionInfo { serde_proto_description: SerdeProtocolDescription((A big honkin' protocol description)), port_info: PortInfo { polarities: {ptID(3382080479'1): Getter, ptID(3382080479'0): Putter}, peers: {ptID(3382080479'0): ptID(3382080479'1), ptID(3382080479'1): ptID(3382080479'0)}, routes: {ptID(3382080479'0): LocalComponent(Native), ptID(3382080479'1): LocalComponent(Native)} }, proto_components: {} }}. Sending to children Iter([])
 CID(3382080479): All session info dumped!: {
     3382080479: SessionInfo {
         serde_proto_description: SerdeProtocolDescription(
             (A big honkin' protocol description),
         ),
         port_info: PortInfo {
             polarities: {
                 ptID(3382080479'1): Getter,
                 ptID(3382080479'0): Putter,
             },
             peers: {
                 ptID(3382080479'0): ptID(3382080479'1),
                 ptID(3382080479'1): ptID(3382080479'0),
             },
             routes: {
                 ptID(3382080479'0): LocalComponent(
                     Native,
                 ),
                 ptID(3382080479'1): LocalComponent(
                     Native,
                 ),
             },
         },
         proto_components: {},
     },
+}
 CID(3382080479): Session optimizations applied
 CID(3382080479): connect() finished. setup phase complete
 CID(3382080479): ~~~ SYNC called with timeout Some(5s); starting round 0
 CID(3382080479): Nonsync running 0 proto components...
 CID(3382080479): All 0 proto components are now done with Nonsync phase
 CID(3382080479): Solution storage initialized
 CID(3382080479): Translating 1 native batches into branches...
 CID(3382080479): Native branch index=0 contains internal inconsistency wrt. fvID(3382080479'1). Skipping
 CID(3382080479): Native starts with no branches! Failure!
 CID(3382080479): No parent. Deciding on failure
 CID(3382080479): Committing to decision Failure!
 CID(3382080479): Announcing decision CommMsg(CommMsg { round_index: 0, contents: Announce { decision: Failure } }) through child endpoints {}
 CID(3382080479): Connector dropping. Goodbye!

examples/6_atomic/amy.c

➞

Show inline comments

@@ @@ -7,7 +7,9 @@ int main(int argc, char** argv) { @@
 	char * pdl_ptr = buffer_pdl("eg_protocols.pdl");
 	size_t pdl_len = strlen(pdl_ptr);
 	Arc_ProtocolDescription * pd = protocol_description_parse(pdl_ptr, pdl_len);
 	Connector * c = connector_new(pd);
 	char logpath[] = "./6_amy_log.txt";
 	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
 	printf("Err %s\n", reowolf_error_peek(NULL));
 	PortId putter, getter;
 	connector_add_port_pair(c, &putter, &getter);
@@ @@ -21,6 +23,7 @@ int main(int argc, char** argv) { @@
 	int err = connector_sync(c, 5000);
 	printf("Error code %d with string `%s`\n", err, reowolf_error_peek(NULL));
 	/*
 	printf("Let's try again, doing both\n");
 	connector_put_bytes(c, putter, "hello", 5);
 	connector_get(c, getter);
@@ @@ -29,6 +32,7 @@ int main(int argc, char** argv) { @@
 	size_t msg_len;
 	const char * msg_ptr = connector_gotten_bytes(c, getter, &msg_len);
 	printf("Got msg `%.*s`\n", msg_len, msg_ptr);
 	*/
 	protocol_description_destroy(pd);
 	connector_destroy(c);

examples/reowolf_rs.dll

➞

Show inline comments

binary diff not shown

reowolf.h

➞

Show inline comments

@@ @@ -89,17 +89,15 @@ ErrorCode connector_get(Connector *connector, PortId port); @@
 const uint8_t *connector_gotten_bytes(Connector *connector, PortId port, uintptr_t *len);
 /**
  * Allocates a new connector on the heap and returning a pointer,
  * given an initialized protocol description.
  */
 Connector *connector_new(const Arc_ProtocolDescription *pd);
 /**
  * Initializes `out` with a new connector using the given protocol description as its configuration.
  * The connector uses the given (internal) connector ID.
  */
 Connector *connector_new_with_id(const Arc_ProtocolDescription *pd, ConnectorId cid);
 Connector *connector_new(const Arc_ProtocolDescription *pd);
 Connector *connector_new_logging(const Arc_ProtocolDescription *pd,
                                  const uint8_t *path_ptr,
                                  uintptr_t path_len);
 intptr_t connector_next_batch(Connector *connector);

src/common.rs

➞

Show inline comments

@@ @@ -186,17 +186,17 @@ impl From<Vec<u8>> for Payload { @@
+}
 impl Debug for PortId {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
-        write!(f, "PID<{},{}>", self.0.connector_id, self.0.u32_suffix)
+        write!(f, "ptID({}'{})", self.0.connector_id, self.0.u32_suffix)
+    }
+}
 impl Debug for FiringVar {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
-        write!(f, "VID<{},{}>", (self.0).0.connector_id, (self.0).0.u32_suffix)
+        write!(f, "fvID({}'{})", (self.0).0.connector_id, (self.0).0.u32_suffix)
+    }
+}
 impl Debug for ProtoComponentId {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
-        write!(f, "ProtoComponentId({},{})", self.0.connector_id, self.0.u32_suffix)
+        write!(f, "pcID({}'{})", self.0.connector_id, self.0.u32_suffix)
+    }
+}
 impl std::ops::Not for Polarity {

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -45,6 +45,16 @@ trait PayloadMsgSender { @@
         msg: SendPayloadMsg,
     ) -> Result<(), SyncError>;
+}
 trait ReplaceBoolTrue {
     fn replace_with_true(&mut self) -> bool;
+}
 impl ReplaceBoolTrue for bool {
     fn replace_with_true(&mut self) -> bool {
         let was = *self;
         *self = true;
         !was
+    }
+}
 ////////////////
 impl Connector {
@@ @@ -137,7 +147,7 @@ impl Connector { @@
         timeout: Option<Duration>,
     ) -> Result<Option<RoundOk>, SyncError> {
         use SyncError as Se;
-        let mut deadline = timeout.map(|to| Instant::now() + to);
         let deadline = timeout.map(|to| Instant::now() + to);
         log!(
             cu.logger,
             "~~~ SYNC called with timeout {:?}; starting round {}",
@@ @@ -214,22 +224,29 @@ impl Connector { @@
             comm.native_batches.len()
         );
         let mut branching_native = BranchingNative { branches: Default::default() };
         for (index, NativeBatch { to_get, to_put }) in comm.native_batches.drain(..).enumerate() {
         'native_branches: for (index, NativeBatch { to_get, to_put }) in
             comm.native_batches.drain(..).enumerate()
+        {
             let predicate = {
                 let mut predicate = Predicate::default();
                 // assign trues
                 // assign trues for ports that fire
                 let firing_ports: HashSet<PortId> =
                     to_get.iter().chain(to_put.keys()).copied().collect();
                 for &port in to_get.iter().chain(to_put.keys()) {
                     let var = cu.port_info.firing_var_for(port);
                     predicate.assigned.insert(var, true);
+                }
                 // assign falses
                 for &port in cu.native_ports.iter() {
                 // assign falses for silent ports
                 for &port in cu.native_ports.difference(&firing_ports) {
                     let var = cu.port_info.firing_var_for(port);
                     predicate.assigned.entry(var).or_insert(false);
                     if let Some(true) = predicate.assigned.insert(var, false) {
                         log!(cu.logger, "Native branch index={} contains internal inconsistency wrt. {:?}. Skipping", index, var);
                         continue 'native_branches;
+                    }
+                }
                 predicate
             };
-            log!(cu.logger, "Native branch {} has pred {:?}", index, &predicate);
+            log!(cu.logger, "Native branch index={:?} has consistent {:?}", index, &predicate);
             // put all messages
             for (putter, payload) in to_put {
@@ @@ -256,6 +273,85 @@ impl Connector { @@
                 return Err(Se::IndistinguishableBatches([index, existing.index]));
+            }
+        }
         let decision = Self::sync_reach_decision(
             cu,
             comm,
             &mut branching_native,
             &mut branching_proto_components,
             solution_storage,
             payloads_to_get,
             deadline,
         )?;
         log!(cu.logger, "Committing to decision {:?}!", &decision);
         // propagate the decision to children
         let msg = Msg::CommMsg(CommMsg {
             round_index: comm.round_index,
             contents: CommMsgContents::Announce { decision: decision.clone() },
         });
         log!(
             cu.logger,
             "Announcing decision {:?} through child endpoints {:?}",
             &msg,
             &comm.neighborhood.children
         );
         for &child in comm.neighborhood.children.iter() {
             comm.endpoint_manager.send_to_comms(child, &msg)?;
+        }
         let ret = match decision {
             Decision::Failure => {
                 // dropping {branching_proto_components, branching_native}
                 Err(Se::RoundFailure)
+            }
             Decision::Success(predicate) => {
                 // commit changes to component states
                 cu.proto_components.clear();
                 cu.proto_components.extend(
                     // consume branching proto components
                     branching_proto_components
                         .into_iter()
                         .map(|(id, bpc)| (id, bpc.collapse_with(&predicate))),
                 );
                 log!(
                     cu.logger,
                     "End round with (updated) component states {:?}",
                     cu.proto_components.keys()
                 );
                 // consume native
                 Ok(Some(branching_native.collapse_with(&predicate)))
+            }
         };
         log!(cu.logger, "Sync round ending! Cleaning up");
         // dropping {solution_storage, payloads_to_get}
         ret
+    }
     fn sync_reach_decision(
         cu: &mut ConnectorUnphased,
         comm: &mut ConnectorCommunication,
         branching_native: &mut BranchingNative,
         branching_proto_components: &mut HashMap<ProtoComponentId, BranchingProtoComponent>,
         mut solution_storage: SolutionStorage,
         mut payloads_to_get: Vec<(PortId, SendPayloadMsg)>,
         mut deadline: Option<Instant>,
     ) -> Result<Decision, SyncError> {
         let mut already_requested_failure = false;
         if branching_native.branches.is_empty() {
             log!(cu.logger, "Native starts with no branches! Failure!");
             match comm.neighborhood.parent {
                 Some(parent) => {
                     if already_requested_failure.replace_with_true() {
                         Self::request_failure(cu, comm, parent)?
                     } else {
                         log!(cu.logger, "Already requested failure");
+                    }
+                }
                 None => {
                     log!(cu.logger, "No parent. Deciding on failure");
                     return Ok(Decision::Failure);
+                }
+            }
+        }
         log!(cu.logger, "Done translating native batches into branches");
         comm.native_batches.push(Default::default());
@@ @@ -281,12 +377,25 @@ impl Connector { @@
             )?;
             // swap the blocked branches back
             std::mem::swap(&mut blocked, branches);
             if branches.is_empty() {
                 log!(cu.logger, "{:?} has become inconsistent!", proto_component_id);
                 if let Some(parent) = comm.neighborhood.parent {
                     if already_requested_failure.replace_with_true() {
                         Self::request_failure(cu, comm, parent)?
                     } else {
                         log!(cu.logger, "Already requested failure");
+                    }
                 } else {
                     log!(cu.logger, "As the leader, deciding on timeout");
                     return Ok(Decision::Failure);
+                }
+            }
+        }
         log!(cu.logger, "All proto components are blocked");
         log!(cu.logger, "Entering decision loop...");
         comm.endpoint_manager.undelay_all();
-        let decision = 'undecided: loop {
         'undecided: loop {
             // drain payloads_to_get, sending them through endpoints / feeding them to components
             while let Some((getter, send_payload_msg)) = payloads_to_get.pop() {
                 assert!(cu.port_info.polarities.get(&getter).copied() == Some(Getter));
@@ @@ -327,6 +436,23 @@ impl Connector { @@
                                 getter,
                                 &send_payload_msg,
                             )?;
                             if branching_component.branches.is_empty() {
                                 log!(
                                     cu.logger,
                                     "{:?} has become inconsistent!",
                                     proto_component_id
                                 );
                                 if let Some(parent) = comm.neighborhood.parent {
                                     if already_requested_failure.replace_with_true() {
                                         Self::request_failure(cu, comm, parent)?
                                     } else {
                                         log!(cu.logger, "Already requested failure");
+                                    }
                                 } else {
                                     log!(cu.logger, "As the leader, deciding on timeout");
                                     return Ok(Decision::Failure);
+                                }
+                            }
                         } else {
                             log!(
                                 cu.logger,
@@ @@ -356,7 +482,7 @@ impl Connector { @@
+                    }
                     None => {
                         log!(cu.logger, "No parent. Deciding on solution {:?}", &solution);
-                        break 'undecided Decision::Success(solution);
+                        return Ok(Decision::Success(solution));
+                    }
+                }
+            }
@@ @@ -370,21 +496,14 @@ impl Connector { @@
                         None => {
                             log!(cu.logger, "Reached user-defined deadling without decision...");
                             if let Some(parent) = comm.neighborhood.parent {
                                 log!(
                                     cu.logger,
                                     "Sending failure request to parent index {}",
                                     parent
                                 );
                                 let msg = Msg::CommMsg(CommMsg {
                                     round_index: comm.round_index,
                                     contents: CommMsgContents::Suggest {
                                         suggestion: Decision::Failure,
                                     },
                                 });
                                 comm.endpoint_manager.send_to_comms(parent, &msg)?;
                                 if already_requested_failure.replace_with_true() {
                                     Self::request_failure(cu, comm, parent)?
                                 } else {
                                     log!(cu.logger, "Already requested failure");
+                                }
                             } else {
                                 log!(cu.logger, "As the leader, deciding on timeout");
-                                break 'undecided Decision::Failure;
+                                return Ok(Decision::Failure);
+                            }
                             deadline = None;
+                        }
@@ @@ -452,21 +571,16 @@ impl Connector { @@
                                 Decision::Failure => {
                                     match comm.neighborhood.parent {
                                         None => {
                                             log!(
                                                 cu.logger,
                                                 "As sink, I decide on my child's failure"
                                             );
                                             // I am the sink. Decide on failed
                                             break 'undecided Decision::Failure;
                                             log!(cu.logger, "I decide on my child's failure");
                                             break 'undecided Ok(Decision::Failure);
+                                        }
                                         Some(parent) => {
                                             log!(cu.logger, "Forwarding failure through my parent endpoint {:?}", parent);
                                             // I've got a parent. Forward the failure suggestion.
                                             let msg = Msg::CommMsg(CommMsg {
                                                 round_index: comm.round_index,
                                                 contents: CommMsgContents::Suggest { suggestion },
                                             });
                                             comm.endpoint_manager.send_to_comms(parent, &msg)?;
                                             if already_requested_failure.replace_with_true() {
                                                 Self::request_failure(cu, comm, parent)?
                                             } else {
                                                 log!(cu.logger, "Already requested failure");
+                                            }
+                                        }
+                                    }
+                                }
@@ @@ -483,7 +597,7 @@ impl Connector { @@
                     CommMsgContents::Announce { decision } => {
                         if Some(endpoint_index) == comm.neighborhood.parent {
                             // adopt this decision
-                            break 'undecided decision;
+                            return Ok(decision);
                         } else {
                             log!(
                                 cu.logger,
@@ @@ -496,36 +610,20 @@ impl Connector { @@
+                }
+            }
             log!(cu.logger, "Endpoint msg recv done");
         };
         log!(cu.logger, "Committing to decision {:?}!", &decision);
         // propagate the decision to children
+        }
+    }
     fn request_failure(
         cu: &mut ConnectorUnphased,
         comm: &mut ConnectorCommunication,
         parent: usize,
     ) -> Result<(), SyncError> {
         log!(cu.logger, "Forwarding to my parent {:?}", parent);
         let suggestion = Decision::Failure;
         let msg = Msg::CommMsg(CommMsg {
             round_index: comm.round_index,
-            contents: CommMsgContents::Announce { decision: decision.clone() },
+            contents: CommMsgContents::Suggest { suggestion },
         });
         log!(
             cu.logger,
             "Announcing decision {:?} through child endpoints {:?}",
             &msg,
             &comm.neighborhood.children
         );
         for &child in comm.neighborhood.children.iter() {
             comm.endpoint_manager.send_to_comms(child, &msg)?;
+        }
         match decision {
             Decision::Failure => Err(Se::RoundFailure),
             Decision::Success(predicate) => {
                 // commit changes to component states
                 cu.proto_components.clear();
                 cu.proto_components.extend(
                     branching_proto_components
                         .into_iter()
                         .map(|(id, bpc)| (id, bpc.collapse_with(&predicate))),
                 );
                 Ok(Some(branching_native.collapse_with(&predicate)))
+            }
+        }
         comm.endpoint_manager.send_to_comms(parent, &msg)
+    }
+}
 impl BranchingNative {
@@ @@ -624,7 +722,6 @@ impl BranchingNative { @@
         panic!("Native had no branches matching pred {:?}", solution_predicate);
+    }
+}
 // |putter, m| {
 //     let getter = *cu.port_info.peers.get(&putter).unwrap();
 //     payloads_to_get.push((getter, m));

src/runtime/ffi.rs

➞

Show inline comments

@@ @@ -116,11 +116,33 @@ pub unsafe extern "C" fn protocol_description_clone( @@
 ///////////////////// CONNECTOR //////////////////////////
 /// Allocates a new connector on the heap and returning a pointer,
 /// given an initialized protocol description.
 #[no_mangle]
 pub unsafe extern "C" fn connector_new(pd: &Arc<ProtocolDescription>) -> *mut Connector {
     connector_new_with_id(pd, random_connector_id())
 pub unsafe extern "C" fn connector_new_logging(
     pd: &Arc<ProtocolDescription>,
     path_ptr: *const u8,
     path_len: usize,
 ) -> *mut Connector {
     StoredError::tl_clear();
     let path_bytes = &*slice_from_parts(path_ptr, path_len);
     let path_str = match std::str::from_utf8(path_bytes) {
         Ok(path_str) => path_str,
         Err(err) => {
             StoredError::tl_debug_store(&err);
             return std::ptr::null_mut();
+        }
     };
     match std::fs::File::create(path_str) {
         Ok(file) => {
             let connector_id = Connector::random_id();
             let file_logger = Box::new(FileLogger::new(connector_id, file));
             let c = Connector::new(file_logger, pd.clone(), connector_id, 8);
             Box::into_raw(Box::new(c))
+        }
         Err(err) => {
             StoredError::tl_debug_store(&err);
             std::ptr::null_mut()
+        }
+    }
+}
 #[no_mangle]
 /// Initializes `out` with a new connector using the given protocol description as its configuration.
 /// The connector uses the given (internal) connector ID.
 #[no_mangle]
 pub unsafe extern "C" fn connector_new_with_id(
     pd: &Arc<ProtocolDescription>,
     cid: ConnectorId,
 ) -> *mut Connector {
     let c = Connector::new(Box::new(DummyLogger), pd.clone(), cid, 8);
 pub unsafe extern "C" fn connector_new(pd: &Arc<ProtocolDescription>) -> *mut Connector {
     let c = Connector::new(Box::new(DummyLogger), pd.clone(), Connector::random_id(), 8);
     Box::into_raw(Box::new(c))
+}

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -16,7 +16,6 @@ pub struct RoundOk { @@
     batch_index: usize,
     gotten: HashMap<PortId, Payload>,
+}
 #[derive(Debug)]
 pub struct VecSet<T: std::cmp::Ord> {
     // invariant: ordered, deduplicated
     vec: Vec<T>,
@@ @@ -204,12 +203,6 @@ pub struct SyncProtoContext<'a> { @@
     inbox: &'a HashMap<PortId, Payload>,
+}
 ////////////////
 pub fn random_connector_id() -> ConnectorId {
     type Bytes8 = [u8; std::mem::size_of::<ConnectorId>()];
     let mut bytes = Bytes8::default();
     getrandom::getrandom(&mut bytes).unwrap();
     unsafe { std::mem::transmute::<Bytes8, ConnectorId>(bytes) }
+}
 pub fn would_block(err: &std::io::Error) -> bool {
     err.kind() == std::io::ErrorKind::WouldBlock
+}
@@ @@ -268,6 +261,12 @@ impl Drop for Connector { @@
+    }
+}
 impl Connector {
     fn random_id() -> ConnectorId {
         type Bytes8 = [u8; std::mem::size_of::<ConnectorId>()];
         let mut bytes = Bytes8::default();
         getrandom::getrandom(&mut bytes).unwrap();
         unsafe { std::mem::transmute::<Bytes8, ConnectorId>(bytes) }
+    }
     pub fn swap_logger(&mut self, mut new_logger: Box<dyn Logger>) -> Box<dyn Logger> {
         std::mem::swap(&mut self.unphased.logger, &mut new_logger);
         new_logger
@@ @@ -436,6 +435,11 @@ impl Predicate { @@
         self.assigned.get(&var).copied()
+    }
+}
 impl<T: Debug + std::cmp::Ord> Debug for VecSet<T> {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.debug_set().entries(self.vec.iter()).finish()
+    }
+}
 impl Debug for Predicate {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         struct MySet<'a>(&'a Predicate, bool);
@@ @@ -457,7 +461,6 @@ impl Debug for Predicate { @@
             .finish()
+    }
+}
 impl serde::Serialize for SerdeProtocolDescription {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where

src/runtime/tests.rs

➞

Show inline comments

@@ @@ -276,6 +276,16 @@ fn connector_pair_nondet() { @@
     .unwrap();
+}
 #[test]
 fn native_immediately_inconsistent() {
     let test_log_path = Path::new("./logs/native_immediately_inconsistent");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, g] = c.new_port_pair();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.get(g).unwrap();
     c.sync(Some(Duration::from_secs(30))).unwrap_err();
+}
 #[test]
 fn cannot_use_moved_ports() {
     /*

0 comments (0 inline, 0 general)