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

Changeset - 800041f16514

Parent rev.

Child rev.

[Not reviewed]

0 6 1

Christopher Esterhuyse - 5 years ago 2020-02-12 18:08:48
christopheresterhuyse@gmail.com

playing around with new predicate=>state,payload storages. toward faster querying but also toward deviation detection by way of intelligible error handling

7 files changed with 330 insertions and 37 deletions:

src/protocol/eval.rs

src/runtime/actors.rs

src/runtime/communication.rs

src/runtime/mod.rs

src/runtime/polyp.rs

279

src/runtime/predicate.rs

src/test/connector.rs

0 comments (0 inline, 0 general)

src/protocol/eval.rs

➞

Show inline comments

@@ @@ -208,205 +208,199 @@ impl Value { @@
+            }
             Value::ByteArray(ByteArrayValue(vec)) => {
                 Some(Value::Int(IntValue((vec.len()).try_into().unwrap())))
+            }
             Value::ShortArray(ShortArrayValue(vec)) => {
                 Some(Value::Int(IntValue((vec.len()).try_into().unwrap())))
+            }
             Value::IntArray(IntArrayValue(vec)) => {
                 Some(Value::Int(IntValue((vec.len()).try_into().unwrap())))
+            }
             Value::LongArray(LongArrayValue(vec)) => {
                 Some(Value::Int(IntValue((vec.len()).try_into().unwrap())))
+            }
             _ => unreachable!(),
+        }
+    }
     fn plus(&self, other: &Value) -> Value {
         match (self, other) {
             (Value::Byte(ByteValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Byte(ByteValue(*s + *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Short(ShortValue(*s as i16 + *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s as i32 + *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 + *o))
+            }
             (Value::Short(ShortValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Short(ShortValue(*s + *o as i16))
+            }
             (Value::Short(ShortValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Short(ShortValue(*s + *o))
+            }
             (Value::Short(ShortValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s as i32 + *o))
+            }
             (Value::Short(ShortValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 + *o))
+            }
             (Value::Int(IntValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Int(IntValue(*s + *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Int(IntValue(*s + *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s + *o))
+            }
             (Value::Int(IntValue(s)), Value::Int(IntValue(o))) => Value::Int(IntValue(*s + *o)),
             (Value::Int(IntValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 + *o))
+            }
             (Value::Long(LongValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Long(LongValue(*s + *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Long(LongValue(*s + *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Int(IntValue(o))) => {
                 Value::Long(LongValue(*s + *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s + *o))
+            }
             _ => unimplemented!(),
+        }
+    }
     fn minus(&self, other: &Value) -> Value {
         match (self, other) {
             (Value::Byte(ByteValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Byte(ByteValue(*s - *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Short(ShortValue(*s as i16 - *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s as i32 - *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 - *o))
+            }
             (Value::Short(ShortValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Short(ShortValue(*s - *o as i16))
+            }
             (Value::Short(ShortValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Short(ShortValue(*s - *o))
+            }
             (Value::Short(ShortValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s as i32 - *o))
+            }
             (Value::Short(ShortValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 - *o))
+            }
             (Value::Int(IntValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Int(IntValue(*s - *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Int(IntValue(*s - *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s - *o))
+            }
             (Value::Int(IntValue(s)), Value::Int(IntValue(o))) => Value::Int(IntValue(*s - *o)),
             (Value::Int(IntValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 - *o))
+            }
             (Value::Long(LongValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Long(LongValue(*s - *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Long(LongValue(*s - *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Int(IntValue(o))) => {
                 Value::Long(LongValue(*s - *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s - *o))
+            }
             _ => unimplemented!(),
+        }
+    }
     fn modulus(&self, other: &Value) -> Value {
         match (self, other) {
             (Value::Byte(ByteValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Byte(ByteValue(*s % *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Short(ShortValue(*s as i16 % *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s as i32 % *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 % *o))
+            }
             (Value::Short(ShortValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Short(ShortValue(*s % *o as i16))
+            }
             (Value::Short(ShortValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Short(ShortValue(*s % *o))
+            }
             (Value::Short(ShortValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s as i32 % *o))
+            }
             (Value::Short(ShortValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 % *o))
+            }
             (Value::Int(IntValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Int(IntValue(*s % *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Int(IntValue(*s % *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s % *o))
+            }
             (Value::Int(IntValue(s)), Value::Int(IntValue(o))) => Value::Int(IntValue(*s % *o)),
             (Value::Int(IntValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 % *o))
+            }
             (Value::Long(LongValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Long(LongValue(*s % *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Long(LongValue(*s % *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Int(IntValue(o))) => {
                 Value::Long(LongValue(*s % *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s % *o))
+            }
             _ => unimplemented!(),
+        }
+    }
     fn eq(&self, other: &Value) -> Value {
         match (self, other) {
             (Value::Byte(ByteValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Boolean(BooleanValue(*s == *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Boolean(BooleanValue(*s as i16 == *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Int(IntValue(o))) => {
                 Value::Boolean(BooleanValue(*s as i32 == *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Long(LongValue(o))) => {
                 Value::Boolean(BooleanValue(*s as i64 == *o))
+            }
             (Value::Short(ShortValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Boolean(BooleanValue(*s == *o as i16))
+            }
             (Value::Short(ShortValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Boolean(BooleanValue(*s == *o))
+            }
             (Value::Short(ShortValue(s)), Value::Int(IntValue(o))) => {
                 Value::Boolean(BooleanValue(*s as i32 == *o))
+            }
             (Value::Short(ShortValue(s)), Value::Long(LongValue(o))) => {
                 Value::Boolean(BooleanValue(*s as i64 == *o))
+            }
             (Value::Int(IntValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Boolean(BooleanValue(*s == *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Short(ShortValue(o))) => {
@@ @@ -1402,320 +1396,323 @@ impl Store { @@
     fn initialize(&mut self, h: &Heap, var: VariableId, value: Value) {
         // Ensure value is compatible with type of variable
         let the_type = h[var].the_type(h);
         assert!(value.is_type_compatible(the_type));
         // Overwrite mapping
         self.map.insert(var, value.clone());
+    }
     fn update(
         &mut self,
         h: &Heap,
         ctx: &mut EvalContext,
         lexpr: ExpressionId,
         value: Value,
     ) -> EvalResult {
         match &h[lexpr] {
             Expression::Variable(var) => {
                 let var = var.declaration.unwrap();
                 // Ensure value is compatible with type of variable
                 let the_type = h[var].the_type(h);
                 assert!(value.is_type_compatible(the_type));
                 // Overwrite mapping
                 self.map.insert(var, value.clone());
                 Ok(value)
+            }
             Expression::Indexing(indexing) => {
                 // Evaluate index expression, which must be some integral type
                 let index = self.eval(h, ctx, indexing.index)?;
                 // Mutable reference to the subject
                 let subject;
                 match &h[indexing.subject] {
                     Expression::Variable(var) => {
                         let var = var.declaration.unwrap();
                         subject = self.map.get_mut(&var).unwrap();
+                    }
                     _ => unreachable!(),
+                }
                 match subject.set(&index, &value) {
                     Some(value) => Ok(value),
                     None => Err(EvalContinuation::Inconsistent),
+                }
+            }
             _ => unimplemented!("{:?}", h[lexpr]),
+        }
+    }
     fn get(&mut self, h: &Heap, ctx: &mut EvalContext, rexpr: ExpressionId) -> EvalResult {
         match &h[rexpr] {
             Expression::Variable(var) => {
                 let var = var.declaration.unwrap();
                 let value = self.map.get(&var).expect(&format!("Uninitialized variable {:?}", h[h[var].identifier()]));
                 let value = self
                     .map
                     .get(&var)
                     .expect(&format!("Uninitialized variable {:?}", h[h[var].identifier()]));
                 Ok(value.clone())
+            }
             Expression::Indexing(indexing) => {
                 // Evaluate index expression, which must be some integral type
                 let index = self.eval(h, ctx, indexing.index)?;
                 // Reference to subject
                 let subject;
                 match &h[indexing.subject] {
                     Expression::Variable(var) => {
                         let var = var.declaration.unwrap();
                         subject = self.map.get(&var).unwrap();
+                    }
                     _ => unreachable!(),
+                }
                 match subject.get(&index) {
                     Some(value) => Ok(value),
                     None => Err(EvalContinuation::Inconsistent),
+                }
+            }
             Expression::Select(selecting) => {
                 // Reference to subject
                 let subject;
                 match &h[selecting.subject] {
                     Expression::Variable(var) => {
                         let var = var.declaration.unwrap();
                         subject = self.map.get(&var).unwrap();
+                    }
                     _ => unreachable!(),
+                }
                 match subject.length() {
                     Some(value) => Ok(value),
                     None => Err(EvalContinuation::Inconsistent),
+                }
+            }
             _ => unimplemented!("{:?}", h[rexpr]),
+        }
+    }
     fn eval(&mut self, h: &Heap, ctx: &mut EvalContext, expr: ExpressionId) -> EvalResult {
         match &h[expr] {
             Expression::Assignment(expr) => {
                 let value = self.eval(h, ctx, expr.right)?;
                 match expr.operation {
                     AssignmentOperator::Set => {
                         self.update(h, ctx, expr.left, value.clone())?;
+                    }
                     AssignmentOperator::Added => {
                         let old = self.get(h, ctx, expr.left)?;
                         self.update(h, ctx, expr.left, old.plus(&value))?;
+                    }
                     AssignmentOperator::Subtracted => {
                         let old = self.get(h, ctx, expr.left)?;
                         self.update(h, ctx, expr.left, old.minus(&value))?;
+                    }
                     _ => unimplemented!("{:?}", expr),
+                }
                 Ok(value)
+            }
             Expression::Conditional(expr) => {
                 let test = self.eval(h, ctx, expr.test)?;
                 if test.as_boolean().0 {
                     self.eval(h, ctx, expr.true_expression)
                 } else {
                     self.eval(h, ctx, expr.false_expression)
+                }
+            }
             Expression::Binary(expr) => {
                 let left = self.eval(h, ctx, expr.left)?;
                 let right;
                 match expr.operation {
                     BinaryOperator::LogicalAnd => {
                         if left.as_boolean().0 == false {
                             return Ok(left)
+                            return Ok(left);
+                        }
                         right = self.eval(h, ctx, expr.right)?;
                         right.as_boolean(); // panics if not a boolean
                         return Ok(right);
+                    }
                     BinaryOperator::LogicalOr => {
                         if left.as_boolean().0 == true {
                             return Ok(left)
+                            return Ok(left);
+                        }
                         right = self.eval(h, ctx, expr.right)?;
                         right.as_boolean(); // panics if not a boolean
                         return Ok(right);
+                    }
                     _ => {}
+                }
                 right = self.eval(h, ctx, expr.right)?;
                 match expr.operation {
                     BinaryOperator::Equality => Ok(left.eq(&right)),
                     BinaryOperator::Inequality => Ok(left.neq(&right)),
                     BinaryOperator::LessThan => Ok(left.lt(&right)),
                     BinaryOperator::LessThanEqual => Ok(left.lte(&right)),
                     BinaryOperator::GreaterThan => Ok(left.gt(&right)),
                     BinaryOperator::GreaterThanEqual => Ok(left.gte(&right)),
                     BinaryOperator::Remainder => Ok(left.modulus(&right)),
                     _ => unimplemented!("{:?}", expr.operation),
+                }
+            }
             Expression::Unary(expr) => {
                 let mut value = self.eval(h, ctx, expr.expression)?;
                 match expr.operation {
                     UnaryOperation::PostIncrement => {
                         self.update(h, ctx, expr.expression, value.plus(&ONE))?;
+                    }
                     UnaryOperation::PreIncrement => {
                         value = value.plus(&ONE);
                         self.update(h, ctx, expr.expression, value.clone())?;
+                    }
                     UnaryOperation::PostDecrement => {
                         self.update(h, ctx, expr.expression, value.minus(&ONE))?;
+                    }
                     UnaryOperation::PreDecrement => {
                         value = value.minus(&ONE);
                         self.update(h, ctx, expr.expression, value.clone())?;
+                    }
                     _ => unimplemented!(),
+                }
                 Ok(value)
+            }
             Expression::Indexing(expr) => self.get(h, ctx, expr.this.upcast()),
             Expression::Slicing(expr) => unimplemented!(),
             Expression::Select(expr) => self.get(h, ctx, expr.this.upcast()),
             Expression::Array(expr) => {
                 let mut elements = Vec::new();
                 for &elem in expr.elements.iter() {
                     elements.push(self.eval(h, ctx, elem)?);
+                }
                 todo!()
-            },
+            }
             Expression::Constant(expr) => Ok(Value::from_constant(&expr.value)),
             Expression::Call(expr) => match expr.method {
                 Method::Create => {
                     assert_eq!(1, expr.arguments.len());
                     let length = self.eval(h, ctx, expr.arguments[0])?;
                     Ok(Value::create_message(length))
+                }
                 Method::Fires => {
                     assert_eq!(1, expr.arguments.len());
                     let value = self.eval(h, ctx, expr.arguments[0])?;
                     match ctx.fires(value.clone()) {
                         None => Err(EvalContinuation::BlockFires(value)),
                         Some(result) => Ok(result),
+                    }
+                }
                 Method::Get => {
                     assert_eq!(1, expr.arguments.len());
                     let value = self.eval(h, ctx, expr.arguments[0])?;
                     match ctx.get(value.clone()) {
                         None => Err(EvalContinuation::BlockGet(value)),
                         Some(result) => Ok(result),
+                    }
+                }
                 Method::Symbolic(symbol) => unimplemented!(),
             },
             Expression::Variable(expr) => self.get(h, ctx, expr.this.upcast()),
+        }
+    }
+}
 type EvalResult = Result<Value, EvalContinuation>;
 pub enum EvalContinuation {
     Stepping,
     Inconsistent,
     Terminal,
     SyncBlockStart,
     SyncBlockEnd,
     NewComponent(DeclarationId, Vec<Value>),
     BlockFires(Value),
     BlockGet(Value),
     Put(Value, Value),
+}
 #[derive(Debug, Clone)]
 pub struct Prompt {
     definition: DefinitionId,
     store: Store,
     position: Option<StatementId>,
+}
 impl Prompt {
     pub fn new(h: &Heap, def: DefinitionId, args: &Vec<Value>) -> Self {
         let mut prompt =
             Prompt { definition: def, store: Store::new(), position: Some((&h[def]).body()) };
         prompt.set_arguments(h, args);
         prompt
+    }
     fn set_arguments(&mut self, h: &Heap, args: &Vec<Value>) {
         let def = &h[self.definition];
         let params = def.parameters();
         assert_eq!(params.len(), args.len());
         for (param, value) in params.iter().zip(args.iter()) {
             let hparam = &h[*param];
             let type_annot = &h[hparam.type_annotation];
             assert!(value.is_type_compatible(&type_annot.the_type));
             self.store.initialize(h, param.upcast(), value.clone());
+        }
+    }
     pub fn step(&mut self, h: &Heap, ctx: &mut EvalContext) -> EvalResult {
         if self.position.is_none() {
             return Err(EvalContinuation::Terminal);
+        }
         let stmt = &h[self.position.unwrap()];
         match stmt {
             Statement::Block(stmt) => {
                 // Continue to first statement
                 self.position = Some(stmt.first());
                 Err(EvalContinuation::Stepping)
+            }
             Statement::Local(stmt) => {
                 match stmt {
                     LocalStatement::Memory(stmt) => {
                         // Evaluate initial expression
                         let value = self.store.eval(h, ctx, stmt.initial)?;
                         // Update store
                         self.store.initialize(h, stmt.variable.upcast(), value);
+                    }
                     LocalStatement::Channel(stmt) => {
                         let [from, to] = ctx.new_channel();
                         // Store the values in the declared variables
                         self.store.initialize(h, stmt.from.upcast(), from);
                         self.store.initialize(h, stmt.to.upcast(), to);
-                    },
+                    }
+                }
                 // Continue to next statement
                 self.position = stmt.next();
                 Err(EvalContinuation::Stepping)
+            }
             Statement::Skip(stmt) => {
                 // Continue to next statement
                 self.position = stmt.next;
                 Err(EvalContinuation::Stepping)
+            }
             Statement::Labeled(stmt) => {
                 // Continue to next statement
                 self.position = Some(stmt.body);
                 Err(EvalContinuation::Stepping)
+            }
             Statement::If(stmt) => {
                 // Evaluate test
                 let value = self.store.eval(h, ctx, stmt.test)?;
                 // Continue with either branch
                 if value.as_boolean().0 {
                     self.position = Some(stmt.true_body);
                 } else {
                     self.position = Some(stmt.false_body);
+                }
                 Err(EvalContinuation::Stepping)
+            }
             Statement::EndIf(stmt) => {
                 // Continue to next statement
                 self.position = stmt.next;
                 Err(EvalContinuation::Stepping)
+            }
             Statement::While(stmt) => {
                 // Evaluate test
                 let value = self.store.eval(h, ctx, stmt.test)?;
                 // Either continue with body, or go to next
                 if value.as_boolean().0 {
                     self.position = Some(stmt.body);
                 } else {
                     self.position = stmt.next.map(|x| x.upcast());
+                }
                 Err(EvalContinuation::Stepping)
+            }
             Statement::EndWhile(stmt) => {
                 // Continue to next statement
                 self.position = stmt.next;
                 Err(EvalContinuation::Stepping)
+            }
             Statement::Synchronous(stmt) => {

src/runtime/actors.rs

➞

Show inline comments

@@ @@ -116,195 +116,212 @@ impl PolyP { @@
                         "~ ... ran {:?} reached SyncBlockEnd with pred {:?} ...",
                         m_ctx.my_subtree_id,
                         &predicate,
                     );
                     // come up with the predicate for this local solution
                     for ekey in self.ekeys.iter() {
                         let channel_id = endpoint_exts.get(*ekey).unwrap().info.channel_id;
                         let fired =
                             branch.inbox.contains_key(ekey) || branch.outbox.contains_key(ekey);
                         match predicate.query(channel_id) {
                             Some(true) => {
                                 if !fired {
                                     // This branch should have fired but didn't!
                                     log!(
                                         logger,
                                         "~ ... ... should have fired {:?} and didn't! pruning!",
                                         channel_id,
                                     );
                                     continue 'to_run_loop;
+                                }
+                            }
                             Some(false) => assert!(!fired),
                             None => {
                                 predicate.replace_assignment(channel_id, false);
                                 assert!(!fired)
+                            }
+                        }
+                    }
                     log!(logger, "~ ... ... and finished just fine!",);
                     m_ctx.solution_storage.submit_and_digest_subtree_solution(
                         &mut m_ctx.inner.logger,
                         m_ctx.my_subtree_id,
                         predicate.clone(),
                     );
                     self.complete.insert(predicate, branch);
+                }
                 Sb::PutMsg(ekey, payload) => {
                     assert!(self.ekeys.contains(&ekey));
                     let EndpointExt { info, endpoint } =
                         m_ctx.inner.endpoint_exts.get_mut(ekey).unwrap();
                     if predicate.replace_assignment(info.channel_id, true) != Some(false) {
                         branch.outbox.insert(ekey, payload.clone());
                         let msg = CommMsgContents::SendPayload {
                             payload_predicate: predicate.clone(),
                             payload,
+                        }
                         .into_msg(m_ctx.inner.round_index);
                         log!(
                             &mut m_ctx.inner.logger,
                             "~ ... ... PolyP sending msg {:?} to {:?} ({:?}) now!",
                             &msg,
                             ekey,
                             (info.channel_id.controller_id, info.channel_id.channel_index),
                         );
                         endpoint.send(msg)?;
                         to_run.push((predicate, branch));
+                    }
                     // ELSE DROP
+                }
+            }
+        }
         // all in self.incomplete most recently returned Blocker::CouldntReadMsg
         Ok(if self.incomplete.is_empty() {
             if self.complete.is_empty() {
                 Srr::NoBranches
             } else {
                 Srr::AllBranchesComplete
+            }
         } else {
             Srr::BlockingForRecv
         })
+    }
     pub(crate) fn poly_recv_run(
         &mut self,
         m_ctx: PolyPContext,
         protocol_description: &ProtocolD,
         ekey: Key,
         payload_predicate: Predicate,
         payload: Payload,
     ) -> Result<SyncRunResult, EndpointErr> {
         // try exact match
         let to_run = if self.complete.contains_key(&payload_predicate) {
             // exact match with stopped machine
             log!(
                 &mut m_ctx.inner.logger,
                 "... poly_recv_run matched stopped machine exactly! nothing to do here",
             );
             vec![]
         } else if let Some(mut branch) = self.incomplete.remove(&payload_predicate) {
             // exact match with running machine
             log!(
                 &mut m_ctx.inner.logger,
                 "... poly_recv_run matched running machine exactly! pred is {:?}",
                 &payload_predicate
             );
             branch.inbox.insert(ekey, payload);
             vec![(payload_predicate, branch)]
         } else {
             log!(
                 &mut m_ctx.inner.logger,
                 "... poly_recv_run didn't have any exact matches... Let's try feed it to all branches",
             );
             let mut incomplete2 = HashMap::<_, _>::default();
             let to_run = self
                 .incomplete
                 .drain()
                 .filter_map(|(old_predicate, mut branch)| {
                     use CommonSatResult as Csr;
                     match old_predicate.common_satisfier(&payload_predicate) {
                         Csr::FormerNotLatter | Csr::Equivalent => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run This branch is compatible unaltered! branch pred: {:?}",
                                 &old_predicate
                             );
                             // old_predicate COVERS the assumptions of payload_predicate
                             let was = branch.inbox.insert(ekey, payload.clone());
                             assert!(was.is_none()); // INBOX MUST BE EMPTY!
                             if let Some(prev_payload) = branch.inbox.get(&ekey) {
                                 // Incorrect to receive two distinct messages in same branch!
                                 assert_eq!(prev_payload, &payload);
+                            }
                             branch.inbox.insert(ekey, payload.clone());
                             Some((old_predicate, branch))
+                        }
                         Csr::New(new) => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run payloadpred {:?} and branchpred {:?} satisfied by new pred {:?}. FORKING",
                                 &payload_predicate,
                                 &old_predicate,
                                 &new,
                             );
                             // payload_predicate has new assumptions. FORK!
                             let mut payload_branch = branch.clone();
                             let was = payload_branch.inbox.insert(ekey, payload.clone());
                             assert!(was.is_none()); // INBOX MUST BE EMPTY!
                             if let Some(prev_payload) = payload_branch.inbox.get(&ekey) {
                                 // Incorrect to receive two distinct messages in same branch!
                                 assert_eq!(prev_payload, &payload);
+                            }
                             payload_branch.inbox.insert(ekey, payload.clone());
                             // put the original back untouched
                             incomplete2.insert(old_predicate, branch);
                             Some((new, payload_branch))
+                        }
                         Csr::LatterNotFormer => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run payloadpred {:?} subsumes branch pred {:?}. FORKING",
                                 &old_predicate,
                                 &payload_predicate,
                             );
                             // payload_predicate has new assumptions. FORK!
                             let mut payload_branch = branch.clone();
                             let was = payload_branch.inbox.insert(ekey, payload.clone());
                             assert!(was.is_none()); // INBOX MUST BE EMPTY!
                             if let Some(prev_payload) = payload_branch.inbox.get(&ekey) {
                                 // Incorrect to receive two distinct messages in same branch!
                                 assert_eq!(prev_payload, &payload);
+                            }
                             payload_branch.inbox.insert(ekey, payload.clone());
                             // put the original back untouched
                             incomplete2.insert(old_predicate, branch);
                             Some((payload_predicate.clone(), payload_branch))
+                        }
                         Csr::Nonexistant => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run SKIPPING because branchpred={:?}. payloadpred={:?}",
                                 &old_predicate,
                                 &payload_predicate,
                             );
                             // predicates contradict
                             incomplete2.insert(old_predicate, branch);
                             None
+                        }
+                    }
                 })
                 .collect();
             std::mem::swap(&mut self.incomplete, &mut incomplete2);
             to_run
         };
         log!(
             &mut m_ctx.inner.logger,
             "... DONE FEEDING BRANCHES. {} branches to run!",
             to_run.len(),
         );
         self.poly_run_these_branches(m_ctx, protocol_description, to_run)
+    }
     pub(crate) fn become_mono(
         mut self,
         decision: &Predicate,
         table_row: &mut HashMap<Key, Payload>,
     ) -> MonoP {
         if let Some((_, branch)) = self.complete.drain().find(|(p, _)| decision.satisfies(p)) {
             let BranchP { inbox, state, outbox } = branch;
             for (key, payload) in inbox.into_iter().chain(outbox.into_iter()) {
                 table_row.insert(key, payload);
+            }
             self.incomplete.clear();
             MonoP { state, ekeys: self.ekeys }
         } else {
             panic!("No such solution!")
+        }
+    }
+}

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -237,138 +237,134 @@ impl Controller { @@
             log!(
                 &mut self.inner.logger,
                 "PolyN kicked off, and has branches with predicates... {:?}",
                 poly_n.branches.keys().collect::<Vec<_>>()
             );
             Some(poly_n)
         } else {
             log!(&mut self.inner.logger, "NO NATIVE COMPONENT");
             None
         };
         // 6. Kick off the synchronous round of each protocol actor
         //    If just one actor becomes inconsistent now, there can be no solution!
         //    TODO distinguish between completed and not completed poly_p's?
         log!(&mut self.inner.logger, "Kicking off {} PolyP's.", self.ephemeral.poly_ps.len());
         for (index, poly_p) in self.ephemeral.poly_ps.iter_mut().enumerate() {
             let my_subtree_id = SubtreeId::PolyP { index };
             let m_ctx = PolyPContext {
                 my_subtree_id,
                 inner: &mut self.inner,
                 solution_storage: &mut self.ephemeral.solution_storage,
             };
             use SyncRunResult as Srr;
             let blocker = poly_p.poly_run(m_ctx, &self.protocol_description)?;
             log!(&mut self.inner.logger, "... PolyP's poly_run got blocker {:?}", &blocker);
             match blocker {
                 Srr::NoBranches => return Err(SyncErr::Inconsistent),
                 Srr::AllBranchesComplete | Srr::BlockingForRecv => (),
+            }
+        }
         log!(&mut self.inner.logger, "All Poly machines have been kicked off!");
         // 7. `solution_storage` may have new solutions for this controller
         //    handle their discovery. LEADER => announce, otherwise => send to parent
+        {
             let peeked = self.ephemeral.solution_storage.peek_new_locals().collect::<Vec<_>>();
             log!(
                 &mut self.inner.logger,
                 "Got {} controller-local solutions before a single RECV: {:?}",
                 peeked.len(),
                 peeked
             );
+        }
         if self.handle_locals_maybe_decide()? {
             return Ok(());
+        }
         // 4. Receive incoming messages until the DECISION is made
         log!(&mut self.inner.logger, "No decision yet. Time to recv messages");
+        log!(&mut self.inner.logger, "`No decision yet`. Time to recv messages");
         self.undelay_all();
         'recv_loop: loop {
             log!(&mut self.inner.logger, "`POLLING`...");
             let received = self.recv(deadline)?.ok_or_else(|| {
                 log!(
                     &mut self.inner.logger,
                     ":( timing out. Solutions storage in state... {:#?}",
                     &self.ephemeral.solution_storage
                 );
                 SyncErr::Timeout
             })?;
             log!(&mut self.inner.logger, "::: message {:?}...", &received);
             let current_content = match received.msg {
                 Msg::SetupMsg(_) => {
                     log!(&mut self.inner.logger, "recvd message {:?} and its SETUP :(", &received);
                     // This occurs in the event the connector was malformed during connect()
                     return Err(SyncErr::UnexpectedSetupMsg);
+                }
                 Msg::CommMsg(CommMsg { round_index, .. })
                     if round_index < self.inner.round_index =>
+                {
                     // Old message! Can safely discard
-                    log!(&mut self.inner.logger, "recvd message {:?} and its OLD! :(", &received);
+                    log!(&mut self.inner.logger, "...and its OLD! :(");
                     drop(received);
                     continue 'recv_loop;
+                }
                 Msg::CommMsg(CommMsg { round_index, .. })
                     if round_index > self.inner.round_index =>
+                {
                     // Message from a next round. Keep for later!
                     log!(
                         &mut self.inner.logger,
                         "ecvd message {:?} and its for later. DELAY! :(",
                         &received
                     );
                     log!(&mut self.inner.logger, "... DELAY! :(");
                     self.delay(received);
                     continue 'recv_loop;
+                }
                 Msg::CommMsg(CommMsg { contents, round_index }) => {
                     log!(
                         &mut self.inner.logger,
                         "recvd a round-appropriate CommMsg {:?} with key {:?}",
                         &contents,
                         "... its a round-appropriate CommMsg with key {:?}",
                         received.recipient
                     );
                     assert_eq!(round_index, self.inner.round_index);
                     contents
+                }
             };
             match current_content {
                 CommMsgContents::Elaborate { partial_oracle } => {
                     // Child controller submitted a subtree solution.
                     if !self.inner.family.children_ekeys.contains(&received.recipient) {
                         return Err(SyncErr::ElaborateFromNonChild);
+                    }
                     let subtree_id = SubtreeId::ChildController { ekey: received.recipient };
                     log!(
                         &mut self.inner.logger,
                         "Received elaboration from child for subtree {:?}: {:?}",
                         subtree_id,
                         &partial_oracle
                     );
                     self.ephemeral.solution_storage.submit_and_digest_subtree_solution(
                         &mut self.inner.logger,
                         subtree_id,
                         partial_oracle,
                     );
                     if self.handle_locals_maybe_decide()? {
                         return Ok(());
+                    }
+                }
                 CommMsgContents::Announce { oracle } => {
                     if self.inner.family.parent_ekey != Some(received.recipient) {
                         return Err(SyncErr::AnnounceFromNonParent);
+                    }
                     log!(
                         &mut self.inner.logger,
                         "Received ANNOUNCEMENT from from parent {:?}: {:?}",
                         received.recipient,
                         &oracle
                     );
                     return self.end_round_with_decision(oracle);
+                }
                 CommMsgContents::SendPayload { payload_predicate, payload } => {
                     assert_eq!(
                         Getter,
                         self.inner.endpoint_exts.get(received.recipient).unwrap().info.polarity
                     );
                     // message for some actor. Feed it to the appropriate actor
                     // and then give them another chance to run.

src/runtime/mod.rs

➞

Show inline comments

 #[cfg(feature = "ffi")]
 pub mod ffi;
 mod actors;
 pub(crate) mod communication;
 pub(crate) mod connector;
 pub(crate) mod endpoint;
 pub mod errors;
 mod predicate; // TODO later
 // mod predicate; // TODO later
 mod polyp;
 mod serde;
 pub(crate) mod setup;
 pub(crate) type ProtocolD = crate::protocol::ProtocolDescriptionImpl;
 pub(crate) type ProtocolS = crate::protocol::ComponentStateImpl;
 use crate::common::*;
 use actors::*;
 use endpoint::*;
 use errors::*;
 #[derive(Debug, PartialEq)]
 pub(crate) enum CommonSatResult {
     FormerNotLatter,
     LatterNotFormer,
     Equivalent,
     New(Predicate),
     Nonexistant,
+}
 #[derive(Clone, Eq, PartialEq, Hash)]
 pub(crate) struct Predicate {
     pub assigned: BTreeMap<ChannelId, bool>,
+}
 #[derive(Debug, Default)]
 struct SyncBatch {
     puts: HashMap<Key, Payload>,
     gets: HashSet<Key>,
+}
 #[derive(Debug)]
 pub enum Connector {
     Unconfigured(Unconfigured),
     Configured(Configured),
     Connected(Connected), // TODO consider boxing. currently takes up a lot of stack real estate
+}
 #[derive(Debug)]
 pub struct Unconfigured {
     pub controller_id: ControllerId,
+}
 #[derive(Debug)]
 pub struct Configured {
     controller_id: ControllerId,
     polarities: Vec<Polarity>,
     bindings: HashMap<usize, PortBinding>,
     protocol_description: Arc<ProtocolD>,
     main_component: Vec<u8>,
@@ @@ -189,97 +190,97 @@ trait Messengerlike { @@
         self.get_state_mut().delayed.push(received);
+    }
     fn undelay_all(&mut self) {
         let MessengerState { delayed, undelayed, .. } = self.get_state_mut();
         undelayed.extend(delayed.drain(..))
+    }
     fn send(&mut self, to: Key, msg: Msg) -> Result<(), EndpointErr> {
         self.get_endpoint_mut(to).send(msg)
+    }
     // attempt to receive a message from one of the endpoints before the deadline
     fn recv(&mut self, deadline: Instant) -> Result<Option<ReceivedMsg>, MessengerRecvErr> {
         // try get something buffered
         if let Some(x) = self.get_state_mut().undelayed.pop() {
             return Ok(Some(x));
+        }
         loop {
             // polled_undrained may not be empty
             while let Some(eekey) = self.get_state_mut().polled_undrained.pop() {
                 if let Some(msg) = self.get_endpoint_mut(eekey).recv()? {
                     // this endpoint MAY still have messages! check again in future
                     self.get_state_mut().polled_undrained.insert(eekey);
                     return Ok(Some(ReceivedMsg { recipient: eekey, msg }));
+                }
+            }
             let state = self.get_state_mut();
             match state.poll_events(deadline) {
                 Ok(()) => {
                     for e in state.events.iter() {
                         state.polled_undrained.insert(Key::from_token(e.token()));
+                    }
+                }
                 Err(PollDeadlineErr::PollingFailed) => return Err(MessengerRecvErr::PollingFailed),
                 Err(PollDeadlineErr::Timeout) => return Ok(None),
+            }
+        }
+    }
+}
 /////////////////////////////////
 impl Debug for SolutionStorage {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         f.pad("Solutions: [")?;
         for (subtree_id, &index) in self.subtree_id_to_index.iter() {
             let sols = &self.subtree_solutions[index];
-            f.write_fmt(format_args!("{:?} => {:?}, ", subtree_id, sols))?;
+            f.write_fmt(format_args!("{:?}: {:?}, ", subtree_id, sols))?;
+        }
         f.pad("]")
+    }
+}
 impl From<EvalErr> for SyncErr {
     fn from(e: EvalErr) -> SyncErr {
         SyncErr::EvalErr(e)
+    }
+}
 impl From<MessengerRecvErr> for SyncErr {
     fn from(e: MessengerRecvErr) -> SyncErr {
         SyncErr::MessengerRecvErr(e)
+    }
+}
 impl From<MessengerRecvErr> for ConnectErr {
     fn from(e: MessengerRecvErr) -> ConnectErr {
         ConnectErr::MessengerRecvErr(e)
+    }
+}
 impl From<EndpointErr> for MessengerRecvErr {
     fn from(e: EndpointErr) -> MessengerRecvErr {
         MessengerRecvErr::EndpointErr(e)
+    }
+}
 impl<T> Default for Arena<T> {
     fn default() -> Self {
         Self { storage: vec![] }
+    }
+}
 impl<T> Arena<T> {
     pub fn alloc(&mut self, t: T) -> Key {
         self.storage.push(t);
         Key::from_raw(self.storage.len() as u64 - 1)
+    }
     pub fn get(&self, key: Key) -> Option<&T> {
         self.storage.get(key.to_raw() as usize)
+    }
     pub fn get_mut(&mut self, key: Key) -> Option<&mut T> {
         self.storage.get_mut(key.to_raw() as usize)
+    }
     pub fn type_convert<X>(self, f: impl FnMut((Key, T)) -> X) -> Arena<X> {
         Arena { storage: self.keyspace().zip(self.storage.into_iter()).map(f).collect() }
+    }
     pub fn iter(&self) -> impl Iterator<Item = (Key, &T)> {
         self.keyspace().zip(self.storage.iter())
+    }
     pub fn len(&self) -> usize {
         self.storage.len()

src/runtime/polyp.rs

➞

Show inline comments

@@ new file 100644 @@
 use crate::common::*;
 use crate::runtime::{Predicate, ProtocolS};
 use core::ops::{Index, IndexMut};
 use std::collections::HashMap;
 use std::num::NonZeroU32;
 // struct SpeculationBranch {
 //     inbox: HashMap<Key, Payload>,
 //     outbox: HashMap<Key, Payload>,
 //     inner: SpeculationBranchInner,
 //     known: HashMap<ChannelId, bool>,
 // }
 // enum SpeculationBranchInner {
 //     Leaf(ProtocolS),
 //     // invariant: channel_id branching is redundantly represented by true_false branches' known assignments
 //     // => true_false[0].known[channel_id] == Some(true)
 //     // => true_false[1].known[channel_id] == Some(false)
 //     Fork { channel_id: ChannelId, true_false: Box<[SpeculationBranch; 2]> },
 // }
 // impl SpeculationBranch {
 //     fn new_tree(init: ProtocolS) -> Self {
 //         SpeculationBranch {
 //             inbox: Default::default(),
 //             outbox: Default::default(),
 //             known: Default::default(),
 //             inner: SpeculationBranchInner::Leaf(init),
 //         }
 //     }
 //     fn feed_msg(
 //         &mut self,
 //         ekey: Key,
 //         payload: &Payload,
 //         predicate: &Predicate,
 //         pred: Option<Pred>,
 //     ) {
 //         use SpeculationBranchInner as Sbi;
 //         let next_pred = Some(Pred { known: &self.known, prev: pred.as_ref() });
 //         match &mut self.inner {
 //             Sbi::Leaf(_state) => {
 //                 if self.inbox.insert(ekey, payload.clone()).is_none() {
 //                     // run this machine
 //                 }
 //             }
 //             Sbi::Fork { channel_id, true_false } => match predicate.query(*channel_id) {
 //                 Some(true) => true_false[0].feed_msg(ekey, payload, predicate, next_pred), // feed true
 //                 Some(false) => true_false[1].feed_msg(ekey, payload, predicate, next_pred), // feed false
 //                 None => {
 //                     // feed to both true and false branches
 //                     for x in true_false.iter_mut() {
 //                         x.feed_msg(ekey, payload, predicate, next_pred);
 //                     }
 //                 }
 //             },
 //         }
 //     }
 // }
 // #[derive(Copy, Clone)]
 // struct Pred<'a> {
 //     known: &'a HashMap<ChannelId, bool>,
 //     prev: Option<&'a Pred<'a>>,
 // }
 struct Branch {
     state: Option<StateKey>,
     speculation: Option<Speculation>,
+}
 struct Speculation {
     on: ChannelId,
     t: Option<BranchKey>,
     f: Option<BranchKey>,
+}
 struct Tree {
     branches: Vec<Branch>, // invariant: non-empty. root at index 0
     states: Vec<ProtocolS>,
+}
 impl Tree {
     /// determine where in the tree the given message should be inserted (based on the predicate).
     /// run all machines
     fn feed_and_run(&mut self, predicate: Predicate, payload: &Payload) {
         let q = Queryable::new(&predicate);
         let mut qs = QueryableSubset::new(&q);
         self.branches[0].feed_and_run(payload, &q, &mut qs);
+    }
+}
 struct Queryable(Vec<(ChannelId, bool)>);
 impl Queryable {
     fn new(predicate: &Predicate) -> Self {
         let mut vec: Vec<_> = predicate.assigned.iter().map(|(&k, &v)| (k, v)).collect();
         vec.sort_by(|(a, _), (b, _)| a.cmp(b));
         Self(vec)
+    }
     fn query(&self, channel_id: ChannelId) -> Option<(usize, bool)> {
         self.0
             .binary_search_by(|(cid, _)| cid.cmp(&channel_id))
             .ok()
             .map(|index| (index, self.0[index].1))
+    }
+}
 struct QueryableSubset {
     buf: Vec<usize>,
     prefix_end: usize,
+}
 impl QueryableSubset {
     fn new(q: &Queryable) -> Self {
         let prefix_end = q.0.len();
         Self { buf: (0..prefix_end).collect(), prefix_end }
+    }
     fn remove(&mut self, at: usize) {
         self.prefix_end -= 1;
         self.buf.swap(self.prefix_end, at);
+    }
     fn undo_remove(&mut self, at: usize) {
         self.buf.swap(self.prefix_end, at);
         self.prefix_end += 1;
+    }
     fn iter_q<'a: 'b, 'b>(
         &'a self,
         q: &'b Queryable,
     ) -> impl Iterator<Item = &'b (ChannelId, bool)> {
         self.buf[..self.prefix_end].iter().map(move |&index| &q.0[index])
+    }
+}
 impl Branch {
     // invariant: q.0 is sorted
     //
     // invariant: qs.buf[0..qs.prefix_end] is a slice that encodes the set of INDICES in q.0
     // which the path to this branch has NOT queried.
     //
     // ie. for a given predicate {X=>true, Z=>true, Y=>false}
     // => q is [(X,true), (Y,false), (Z,true)]
     // => qs is initially [0,1,2]
     // and if this branch queries 1, the subtree will receive qs as [0,2]
     fn feed_and_run(&mut self, payload: &Payload, q: &Queryable, qs: &mut QueryableSubset) {
         match &mut self.speculation {
             Some(Speculation { on, t, f }) => {
                 if let Some((index, assignment)) = q.query(*on) {
                     // if assignment
                 } else {
+                }
                 todo!()
+            }
             None => {
                 //
                 todo!()
+            }
+        }
+    }
+}
 impl Index<BranchKey> for Tree {
     type Output = Branch;
     fn index(&self, k: BranchKey) -> &Self::Output {
         &self.branches[(k.index_plus_one.get() - 1) as usize]
+    }
+}
 impl IndexMut<BranchKey> for Tree {
     fn index_mut(&mut self, k: BranchKey) -> &mut Self::Output {
         &mut self.branches[(k.index_plus_one.get() - 1) as usize]
+    }
+}
 impl Index<StateKey> for Tree {
     type Output = ProtocolS;
     fn index(&self, k: StateKey) -> &Self::Output {
         &self.states[(k.index_plus_one.get() - 1) as usize]
+    }
+}
 impl IndexMut<StateKey> for Tree {
     fn index_mut(&mut self, k: StateKey) -> &mut Self::Output {
         &mut self.states[(k.index_plus_one.get() - 1) as usize]
+    }
+}
 struct BranchKey {
     index_plus_one: NonZeroU32,
+}
 struct StateKey {
     index_plus_one: NonZeroU32,
+}
 struct Bitset {
     bits: Vec<u32>,
+}
 struct Polyp {
     inbox: Vec<Payload>,
     inbox_masks: BitMasks,
     states: Vec<ProtocolS>,
     states_masks: BitMasks,
+}
 // invariant: last element is not zero.
 // => all values out of bounds are implicitly absent
 #[derive(Debug, Default)]
 struct BitSet(Vec<u32>);
 #[derive(Debug, Default)]
 struct BitMasks(HashMap<(ChannelId, bool), BitSet>);
 struct BitSetAndIter<'a> {
     // this value is immutable
     // invariant: !sets.is_empty()
     sets: &'a [&'a [u32]],
     next_u32_index: usize, // invariant: in 0..32 while iterating
     next_bit_index: usize,
     cached: Option<u32>, // None <=> iterator is done
+}
 impl<'a> BitSetAndIter<'a> {
     fn new(sets: &'a [&'a [u32]]) -> Self {
         const EMPTY_SINGLETON: &[&[u32]] = &[&[]];
         let sets = if sets.is_empty() { EMPTY_SINGLETON } else { sets };
         Self { sets, next_u32_index: 0, next_bit_index: 0, cached: Self::nth_u32(sets, 0) }
+    }
     fn nth_u32(sets: &'a [&'a [u32]], index: usize) -> Option<u32> {
         sets.iter().fold(Some(!0), |a, b| {
             let b = b.get(index)?;
             Some(a? & b)
         })
+    }
     fn next_chunk(&mut self) {
         self.next_bit_index = 0;
         self.next_u32_index += 1;
         self.cached = Self::nth_u32(self.sets, self.next_u32_index);
+    }
+}
 impl Iterator for BitSetAndIter<'_> {
     type Item = usize;
     fn next(&mut self) -> Option<Self::Item> {
         loop {
             // get cached chunk. If none exists, iterator is done.
             let mut chunk = self.cached?;
             if chunk == 0 {
                 self.next_chunk();
                 continue;
+            }
             // this chunk encodes 1+ Items to yield
             // shift the contents of chunk until the least significant bit is 1
             #[inline(always)]
             fn shifty(chunk: &mut u32, shift_by: usize, next_bit_index: &mut usize) {
                 if *chunk & ((1 << shift_by) - 1) == 0 {
                     *next_bit_index += shift_by;
                     *chunk >>= shift_by;
+                }
+            }
             shifty(&mut chunk, 16, &mut self.next_bit_index);
             shifty(&mut chunk, 08, &mut self.next_bit_index);
             shifty(&mut chunk, 04, &mut self.next_bit_index);
             shifty(&mut chunk, 02, &mut self.next_bit_index);
             shifty(&mut chunk, 01, &mut self.next_bit_index);
             // assert(chunk & 1 == 1)
             let index = self.next_u32_index * 32 + self.next_bit_index;
             self.next_bit_index += 1;
             self.cached = Some(chunk >> 1);
             if chunk > 0 {
                 // assert(self.next_bit_index <= 32)
                 // because index was calculated with self.next_bit_index - 1
                 return Some(index);
+            }
+        }
+    }
+}
 #[test]
 fn test_bit_iter() {
     static SETS: &[&[u32]] = &[
         //
         &[0b100011000000100101101],
         &[0b100001000000000110100],
         &[0b100001000100010100110],
     ];
     let indices = BitSetAndIter::new(SETS).collect::<Vec<_>>();
     println!("indices {:?}", indices);
+}

src/runtime/predicate.rs

➞

Show inline comments

 use crate::common::ChannelId;
 use crate::common::ChannelIndex;
 use crate::common::ControllerId;
 use crate::common::*;
 use crate::runtime::ProtocolS;
 use core::ops::Index;
 use core::ops::IndexMut;
 use std::collections::BTreeMap;
 // we assume a dense ChannelIndex domain!
 enum CommonSatisfier<T> {
     FormerNotLatter,
     LatterNotFormer,
     Equivalent,
     New(T),
     Nonexistant,
+}
 type ChunkType = u16;
 const MASK_BITS: ChunkType = 0x_AA_AA; // 101010...
 #[test]
 fn mask_ok() {
     assert_eq!(!0, MASK_BITS | (MASK_BITS >> 1));
     assert_eq!(0, MASK_BITS & (MASK_BITS >> 1));
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 struct TernChunk(ChunkType); // invariant: no pair is 01
 impl TernChunk {
     fn overwrite(&mut self, index: usize, value: bool) -> Option<bool> {
         assert!(index < Self::vars_per_chunk());
         let mask_bit_mask = 1 << (index * 2 + 1);
         let bool_bit_mask = 1 << (index * 2);
         let ret = if self.0 & mask_bit_mask != 0 {
             let was_value = self.0 & bool_bit_mask != 0;
             if was_value != value {
                 // flip the value bit
                 self.0 ^= bool_bit_mask;
+            }
             Some(was_value)
         } else {
             if value {
                 // set the value bit
                 self.0 |= bool_bit_mask;
+            }
             None
         };
         // set the mask bit
         self.0 |= mask_bit_mask;
         ret
+    }
@@ @@ -190,48 +191,50 @@ impl TernSet { @@
 struct Predicate(BTreeMap<ControllerId, TernSet>);
 impl Predicate {
     pub fn overwrite(&mut self, channel_id: ChannelId, value: bool) -> Option<bool> {
         let ChannelId { controller_id, channel_index } = channel_id;
         use std::collections::btree_map::Entry;
         match self.0.entry(controller_id) {
             Entry::Occupied(mut x) => x.get_mut().overwrite(channel_index, value),
             Entry::Vacant(x) => {
                 x.insert(TernSet::new_singleton(channel_index, value));
                 None
+            }
+        }
+    }
     pub fn query(&self, channel_id: ChannelId) -> Option<bool> {
         let ChannelId { controller_id, channel_index } = channel_id;
         self.0.get(&controller_id).and_then(move |tern_set| tern_set.query(channel_index))
+    }
     pub fn satisfies(&self, other: &Self) -> bool {
         let mut s_it = self.0.iter();
         let mut s = if let Some(s) = s_it.next() {
+            s
         } else {
             return other.0.is_empty();
         };
         for (oid, ob) in other.0.iter() {
             while s.0 < oid {
                 s = if let Some(s) = s_it.next() {
+                    s
                 } else {
                     return false;
                 };
+            }
             if s.0 > oid || !s.1.satisfies(ob) {
                 return false;
+            }
+        }
         true
+    }
     pub fn common_satisfier(&self, othe: &Self) -> CommonSatisfier<Self> {
         // use CommonSatisfier as Cs;
         // let [slen, olen] = [self.0.len(), othe.0.len()];
         // let [mut s_sat_o, mut o_sat_s] = [slen >= olen, slen <= olen];
         // let [mut s_it, mut o_it] = [self.0.iter(), othe.0.iter()];
         // let [mut s, mut o] = [s_it.next(), o_it.next()];
         todo!()
+    }
+}
 ////////////////////////////

src/test/connector.rs

➞

Show inline comments

@@ @@ -548,99 +548,99 @@ fn exchange() { @@
         \<--/      \<--P|A<--/      \<--/
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr(), next_addr()];
     const N: usize = 1;
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"exchange").unwrap();
             x.bind_port(0, Native).unwrap(); // native in
             x.bind_port(1, Native).unwrap(); // native out
             x.bind_port(2, Passive(addrs[0])).unwrap(); // peer out
             x.bind_port(3, Passive(addrs[1])).unwrap(); // peer in
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 assert_eq!(Ok(()), x.put(0, b"A->B".to_vec()));
                 assert_eq!(Ok(()), x.get(1));
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(b"B->A" as &[u8]), x.read_gotten(1));
+            }
         },
         &|x| {
             // Bob
             x.configure(PDL, b"exchange").unwrap();
             x.bind_port(0, Native).unwrap(); // native in
             x.bind_port(1, Native).unwrap(); // native out
             x.bind_port(2, Active(addrs[1])).unwrap(); // peer out
             x.bind_port(3, Active(addrs[0])).unwrap(); // peer in
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 assert_eq!(Ok(()), x.put(0, b"B->A".to_vec()));
                 assert_eq!(Ok(()), x.get(1));
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(b"A->B" as &[u8]), x.read_gotten(1));
+            }
         },
     ]));
+}
 #[test]
 fn routing_filter() {
     // Make a protocol whose behavior is a function of the contents of
     // a message. Here, the putter determines what is sent, and the proto
     // determines how it is routed
     /*
     Sender -->filter-->P|A-->sync--> Receiver
     */
-    let timeout = Duration::from_millis(1_500);
+    let timeout = Duration::from_millis(3_000);
     let addrs = [next_addr()];
-    const N: usize = 10;
+    const N: usize = 1;
     assert!(run_connector_set(&[
         //
         &|x| {
             // Sender
             x.configure(PDL, b"filter").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Passive(addrs[0])).unwrap();
             x.bind_port(2, Native).unwrap(); // err channel
             x.connect(timeout).unwrap();
             for i in (0..3).cycle().take(N) {
                 // messages cycle [], [4], [4,4], ...
                 let msg: Payload = std::iter::repeat(4).take(i).collect();
                 // batch 0: passes through filter!
                 x.put(0, msg.clone()).unwrap();
                 x.next_batch().unwrap();
                 // batch 1: gets returned!
                 x.put(0, msg.clone()).unwrap();
                 x.get(1).unwrap();
                 match x.sync(timeout).unwrap() {
 => assert_ne!(msg.len(), 0), // ok
 => assert_eq!(msg.len(), 0), // err
                     _ => unreachable!(),
+                }
+            }
         },
         &|x| {
             // Receiver
             x.configure(PDL, b"sync").unwrap();
             x.bind_port(0, Active(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 // empty batch
                 x.next_batch().unwrap();
                 // got a message
                 x.get(0).unwrap();
                 match x.sync(timeout).unwrap() {
 => assert_eq!(Err(ReadGottenErr::DidNotGet), x.read_gotten(0)),
 => assert_ne!(Ok(&[] as &[u8]), x.read_gotten(0)),
                     _ => unreachable!(),
+                }
+            }
         },
     ]));

0 comments (0 inline, 0 general)