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

Changeset - dedf89df1602

Parent rev.

Child rev.

[Not reviewed]

0 9 0

Christopher Esterhuyse - 5 years ago 2020-05-28 11:14:52
christopher.esterhuyse@gmail.com

retiring `Endpoint Key` term in favor of port

9 files changed with 253 insertions and 258 deletions:

src/common.rs

src/protocol/eval.rs

src/protocol/inputsource.rs

src/protocol/mod.rs

src/runtime/actors.rs

src/runtime/communication.rs

src/runtime/connector.rs

src/runtime/mod.rs

src/runtime/setup.rs

0 comments (0 inline, 0 general)

src/common.rs

➞

Show inline comments

@@ @@ -21,13 +21,12 @@ pub use std::{ @@
     time::Instant,
 };
 pub use Polarity::*;
 ///////////////////// DEFS /////////////////////
 // pub type Payload = Vec<u8>;
 pub type ControllerId = u32;
 pub type ChannelIndex = u32;
 #[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd)]
 pub struct Payload(Arc<Vec<u8>>);
@@ @@ -44,13 +43,12 @@ pub enum Polarity { @@
     Getter, // input port (from the perspective of the component)
+}
 #[derive(Eq, PartialEq, Ord, PartialOrd, Hash, Copy, Clone)]
 #[repr(C)]
 pub struct Port(pub usize); // ports are COPY
 pub type Key = Port;
 #[derive(Eq, PartialEq, Copy, Clone, Debug)]
 pub enum MainComponentErr {
     NoSuchComponent,
     NonPortTypeParameters,
     CannotMovePort(Port),
@@ @@ -61,13 +59,13 @@ pub enum MainComponentErr { @@
+}
 pub trait ProtocolDescription: Sized {
     type S: ComponentState<D = Self>;
     fn parse(pdl: &[u8]) -> Result<Self, String>;
     fn component_polarities(&self, identifier: &[u8]) -> Result<Vec<Polarity>, MainComponentErr>;
-    fn new_main_component(&self, identifier: &[u8], ports: &[Key]) -> Self::S;
+    fn new_main_component(&self, identifier: &[u8], ports: &[Port]) -> Self::S;
+}
 pub trait ComponentState: Sized + Clone {
     type D: ProtocolDescription;
     fn pre_sync_run<C: MonoContext<D = Self::D, S = Self>>(
         &mut self,
@@ @@ -90,30 +88,30 @@ pub enum MonoBlocker { @@
+}
 #[derive(Debug, Clone)]
 pub enum PolyBlocker {
     Inconsistent,
     SyncBlockEnd,
     CouldntReadMsg(Key),
     CouldntCheckFiring(Key),
     PutMsg(Key, Payload),
     CouldntReadMsg(Port),
     CouldntCheckFiring(Port),
     PutMsg(Port, Payload),
+}
 pub trait MonoContext {
     type D: ProtocolDescription;
     type S: ComponentState<D = Self::D>;
     fn new_component(&mut self, moved_keys: HashSet<Key>, init_state: Self::S);
     fn new_channel(&mut self) -> [Key; 2];
     fn new_component(&mut self, moved_ports: HashSet<Port>, init_state: Self::S);
     fn new_channel(&mut self) -> [Port; 2];
     fn new_random(&mut self) -> u64;
+}
 pub trait PolyContext {
     type D: ProtocolDescription;
     fn is_firing(&mut self, ekey: Key) -> Option<bool>;
     fn read_msg(&mut self, ekey: Key) -> Option<&Payload>;
     fn is_firing(&mut self, port: Port) -> Option<bool>;
     fn read_msg(&mut self, port: Port) -> Option<&Payload>;
+}
 ///////////////////// IMPL /////////////////////
 impl Payload {
     pub fn new(len: usize) -> Payload {
         let mut v = Vec::with_capacity(len);
@@ @@ -149,13 +147,13 @@ impl From<Vec<u8>> for Payload { @@
+}
 impl Debug for Port {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "Port({})", self.0)
+    }
+}
-impl Key {
+impl Port {
     pub fn from_raw(raw: usize) -> Self {
         Self(raw)
+    }
     pub fn to_raw(self) -> usize {
         self.0
+    }

src/protocol/eval.rs

➞

Show inline comments

@@ @@ -883,13 +883,13 @@ impl Display for Value { @@
+        }
         disp.fmt(f)
+    }
+}
 #[derive(Debug, Clone)]
-pub struct InputValue(pub Key);
+pub struct InputValue(pub Port);
 impl Display for InputValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "#in")
+    }
+}
@@ @@ -908,13 +908,13 @@ impl ValueImpl for InputValue { @@
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone)]
-pub struct OutputValue(pub Key);
+pub struct OutputValue(pub Port);
 impl Display for OutputValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "#out")
+    }
+}

src/protocol/inputsource.rs

➞

Show inline comments

@@ @@ -18,16 +18,16 @@ static STD_LIB_PDL: &'static [u8] = b" @@
 primitive forward(in i, out o) {
     while(true) synchronous() put(o, get(i));
+}
 primitive sync(in i, out o) {
     while(true) synchronous() if(fires(i)) put(o, get(i));
+}
-primitive alternator_2(in i, out a, out b) {
+primitive alternator_2(in i, out l, out r) {
     while(true) {
         synchronous() put(a, get(i));
         synchronous() put(b, get(i));
         synchronous() put(l, get(i));
         synchronous() put(r, get(i));
+    }
+}
 primitive replicator_2(in i, out l, out r) {
     while(true) synchronous() if(fires(i)) {
         msg m = get(i);
         put(l, m);
@@ @@ -40,13 +40,13 @@ primitive merger_2(in l, in r, out o) { @@
         else         put(o, get(r));
+    }
 }";
 impl InputSource {
     // Constructors
-    pub fn new<A: io::Read, S: ToString>(filename: S, reader: &mut A) -> io::Result<InputSource> {
+    pub fn new<R: io::Read, S: ToString>(filename: S, reader: &mut R) -> io::Result<InputSource> {
         let mut vec = STD_LIB_PDL.to_vec();
         reader.read_to_end(&mut vec)?;
         Ok(InputSource {
             filename: filename.to_string(),
             input: vec,
             line: 1,

src/protocol/mod.rs

➞

Show inline comments

@@ @@ -77,13 +77,13 @@ impl ProtocolDescription for ProtocolDescriptionImpl { @@
             } else {
                 unreachable!()
+            }
+        }
         Ok(result)
+    }
-    fn new_main_component(&self, identifier: &[u8], ports: &[Key]) -> ComponentStateImpl {
+    fn new_main_component(&self, identifier: &[u8], ports: &[Port]) -> ComponentStateImpl {
         let mut args = Vec::new();
         for (&x, y) in ports.iter().zip(self.component_polarities(identifier).unwrap()) {
             match y {
                 Polarity::Getter => args.push(Value::Input(InputValue(x))),
                 Polarity::Putter => args.push(Value::Output(OutputValue(x))),
+            }
@@ @@ -157,37 +157,37 @@ impl ComponentState for ComponentStateImpl { @@
                     // No nested synchronous blocks
                     EvalContinuation::SyncBlockStart => unreachable!(),
                     EvalContinuation::SyncBlockEnd => return PolyBlocker::SyncBlockEnd,
                     // Not possible to create component in sync block
                     EvalContinuation::NewComponent(_, _) => unreachable!(),
                     EvalContinuation::BlockFires(port) => match port {
                         Value::Output(OutputValue(key)) => {
                             return PolyBlocker::CouldntCheckFiring(key);
                         Value::Output(OutputValue(port)) => {
                             return PolyBlocker::CouldntCheckFiring(port);
+                        }
                         Value::Input(InputValue(key)) => {
                             return PolyBlocker::CouldntCheckFiring(key);
                         Value::Input(InputValue(port)) => {
                             return PolyBlocker::CouldntCheckFiring(port);
+                        }
                         _ => unreachable!(),
                     },
                     EvalContinuation::BlockGet(port) => match port {
                         Value::Output(OutputValue(key)) => {
                             return PolyBlocker::CouldntReadMsg(key);
                         Value::Output(OutputValue(port)) => {
                             return PolyBlocker::CouldntReadMsg(port);
+                        }
                         Value::Input(InputValue(key)) => {
                             return PolyBlocker::CouldntReadMsg(key);
                         Value::Input(InputValue(port)) => {
                             return PolyBlocker::CouldntReadMsg(port);
+                        }
                         _ => unreachable!(),
                     },
                     EvalContinuation::Put(port, message) => {
-                        let key;
+                        let value;
                         match port {
                             Value::Output(OutputValue(the_key)) => {
                                 key = the_key;
                             Value::Output(OutputValue(port_value)) => {
                                 value = port_value;
+                            }
                             Value::Input(InputValue(the_key)) => {
                                 key = the_key;
                             Value::Input(InputValue(port_value)) => {
                                 value = port_value;
+                            }
                             _ => unreachable!(),
+                        }
                         let payload;
                         match message {
                             Value::Message(MessageValue(None)) => {
@@ @@ -197,13 +197,13 @@ impl ComponentState for ComponentStateImpl { @@
                             Value::Message(MessageValue(Some(buffer))) => {
                                 // Create a copy of the payload
                                 payload = buffer.clone();
+                            }
                             _ => unreachable!(),
+                        }
-                        return PolyBlocker::PutMsg(key, payload);
+                        return PolyBlocker::PutMsg(value, payload);
+                    }
                 },
+            }
+        }
+    }
+}
@@ @@ -211,36 +211,36 @@ impl ComponentState for ComponentStateImpl { @@
 pub enum EvalContext<'a> {
     Mono(&'a mut dyn MonoContext<D = ProtocolDescriptionImpl, S = ComponentStateImpl>),
     Poly(&'a mut dyn PolyContext<D = ProtocolDescriptionImpl>),
     None,
+}
 impl EvalContext<'_> {
     fn random(&mut self) -> LongValue {
         match self {
             EvalContext::None => unreachable!(),
             EvalContext::Mono(context) => todo!(),
             EvalContext::Poly(_) => unreachable!(),
+        }
+    }
     // fn random(&mut self) -> LongValue {
     //     match self {
     //         EvalContext::None => unreachable!(),
     //         EvalContext::Mono(_context) => todo!(),
     //         EvalContext::Poly(_) => unreachable!(),
     //     }
     // }
     fn new_component(&mut self, args: &[Value], init_state: ComponentStateImpl) -> () {
         match self {
             EvalContext::None => unreachable!(),
             EvalContext::Mono(context) => {
-                let mut moved_keys = HashSet::new();
+                let mut moved_ports = HashSet::new();
                 for arg in args.iter() {
                     match arg {
                         Value::Output(OutputValue(key)) => {
                             moved_keys.insert(*key);
                         Value::Output(OutputValue(port)) => {
                             moved_ports.insert(*port);
+                        }
                         Value::Input(InputValue(key)) => {
                             moved_keys.insert(*key);
                         Value::Input(InputValue(port)) => {
                             moved_ports.insert(*port);
+                        }
                         _ => {}
+                    }
+                }
-                context.new_component(moved_keys, init_state)
+                context.new_component(moved_ports, init_state)
+            }
             EvalContext::Poly(_) => unreachable!(),
+        }
+    }
     fn new_channel(&mut self) -> [Value; 2] {
         match self {
@@ @@ -256,26 +256,28 @@ impl EvalContext<'_> { @@
+    }
     fn fires(&mut self, port: Value) -> Option<Value> {
         match self {
             EvalContext::None => unreachable!(),
             EvalContext::Mono(_) => unreachable!(),
             EvalContext::Poly(context) => match port {
                 Value::Output(OutputValue(key)) => context.is_firing(key).map(Value::from),
                 Value::Input(InputValue(key)) => context.is_firing(key).map(Value::from),
                 Value::Output(OutputValue(port)) => context.is_firing(port).map(Value::from),
                 Value::Input(InputValue(port)) => context.is_firing(port).map(Value::from),
                 _ => unreachable!(),
             },
+        }
+    }
     fn get(&mut self, port: Value) -> Option<Value> {
         match self {
             EvalContext::None => unreachable!(),
             EvalContext::Mono(_) => unreachable!(),
             EvalContext::Poly(context) => match port {
                 Value::Output(OutputValue(key)) => {
                     context.read_msg(key).map(Value::receive_message)
                 Value::Output(OutputValue(port)) => {
                     context.read_msg(port).map(Value::receive_message)
+                }
                 Value::Input(InputValue(port)) => {
                     context.read_msg(port).map(Value::receive_message)
+                }
                 Value::Input(InputValue(key)) => context.read_msg(key).map(Value::receive_message),
                 _ => unreachable!(),
             },
+        }
+    }
+}

src/runtime/actors.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::{endpoint::*, *};
 #[derive(Debug, Clone)]
 pub(crate) struct MonoN {
     pub ekeys: HashSet<Key>,
     pub result: Option<(usize, HashMap<Key, Payload>)>,
     pub ports: HashSet<Port>,
     pub result: Option<(usize, HashMap<Port, Payload>)>,
+}
 #[derive(Debug)]
 pub(crate) struct PolyN {
-    pub ekeys: HashSet<Key>,
+    pub ports: HashSet<Port>,
     pub branches: HashMap<Predicate, BranchN>,
+}
 #[derive(Debug, Clone)]
 pub(crate) struct BranchN {
     pub to_get: HashSet<Key>,
     pub gotten: HashMap<Key, Payload>,
     pub to_get: HashSet<Port>,
     pub gotten: HashMap<Port, Payload>,
     pub sync_batch_index: usize,
+}
 #[derive(Debug, Clone)]
 pub struct MonoP {
     pub state: ProtocolS,
-    pub ekeys: HashSet<Key>,
+    pub ports: HashSet<Port>,
+}
 #[derive(Debug)]
 pub(crate) struct PolyP {
     pub incomplete: HashMap<Predicate, BranchP>,
     pub complete: HashMap<Predicate, BranchP>,
-    pub ekeys: HashSet<Key>,
+    pub ports: HashSet<Port>,
+}
 #[derive(Debug, Clone)]
 pub(crate) struct BranchP {
     pub blocking_on: Option<Key>,
     pub outbox: HashMap<Key, Payload>,
     pub inbox: HashMap<Key, Payload>,
     pub blocking_on: Option<Port>,
     pub outbox: HashMap<Port, Payload>,
     pub inbox: HashMap<Port, Payload>,
     pub state: ProtocolS,
+}
 //////////////////////////////////////////////////////////////////
 impl PolyP {
@@ @@ -57,13 +57,13 @@ impl PolyP { @@
     ) -> Result<SyncRunResult, EndpointErr> {
         use SyncRunResult as Srr;
         log!(&mut m_ctx.inner.logger, "~ Running branches for PolyP {:?}!", m_ctx.my_subtree_id,);
         'to_run_loop: while let Some((mut predicate, mut branch)) = to_run.pop() {
             let mut r_ctx = BranchPContext {
                 m_ctx: m_ctx.reborrow(),
-                ekeys: &self.ekeys,
+                ports: &self.ports,
                 predicate: &predicate,
                 inbox: &branch.inbox,
             };
             use PolyBlocker as Sb;
             let blocker = branch.state.sync_run(&mut r_ctx, protocol_description);
             log!(
@@ @@ -72,38 +72,38 @@ impl PolyP { @@
                 r_ctx.m_ctx.my_subtree_id,
                 &predicate,
                 &blocker
             );
             match blocker {
                 Sb::Inconsistent => {} // DROP
                 Sb::CouldntReadMsg(ekey) => {
                     assert!(self.ekeys.contains(&ekey));
                 Sb::CouldntReadMsg(port) => {
                     assert!(self.ports.contains(&port));
                     let channel_id =
-                        r_ctx.m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;
+                        r_ctx.m_ctx.inner.endpoint_exts.get(port).unwrap().info.channel_id;
                     log!(
                         &mut r_ctx.m_ctx.inner.logger,
                         "~ ... {:?} couldnt read msg for port {:?}. has inbox {:?}",
                         r_ctx.m_ctx.my_subtree_id,
                         channel_id,
                         &branch.inbox,
                     );
                     if predicate.replace_assignment(channel_id, true) != Some(false) {
                         // don't rerun now. Rerun at next `sync_run`
                         log!(&mut m_ctx.inner.logger, "~ ... Delay {:?}", m_ctx.my_subtree_id,);
-                        branch.blocking_on = Some(ekey);
+                        branch.blocking_on = Some(port);
                         self.incomplete.insert(predicate, branch);
                     } else {
                         log!(&mut m_ctx.inner.logger, "~ ... Drop {:?}", m_ctx.my_subtree_id,);
+                    }
                     // ELSE DROP
+                }
                 Sb::CouldntCheckFiring(ekey) => {
                     assert!(self.ekeys.contains(&ekey));
                 Sb::CouldntCheckFiring(port) => {
                     assert!(self.ports.contains(&port));
                     let channel_id =
-                        r_ctx.m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;
+                        r_ctx.m_ctx.inner.endpoint_exts.get(port).unwrap().info.channel_id;
                     // split the branch!
                     let branch_f = branch.clone();
                     let mut predicate_f = predicate.clone();
                     if predicate_f.replace_assignment(channel_id, false).is_some() {
                         panic!("OI HANS QUERY FIRST!");
+                    }
@@ @@ -118,16 +118,16 @@ 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;
                     for port in self.ports.iter() {
                         let channel_id = endpoint_exts.get(*port).unwrap().info.channel_id;
                         let fired =
-                            branch.inbox.contains_key(ekey) || branch.outbox.contains_key(ekey);
+                            branch.inbox.contains_key(port) || branch.outbox.contains_key(port);
                         match predicate.query(channel_id) {
                             Some(true) => {
                                 if !fired {
                                     // This branch should have fired but didn't!
                                     log!(
                                         logger,
@@ @@ -139,26 +139,26 @@ impl PolyP { @@
+                            }
                             Some(false) => {
                                 if fired {
                                     println!(
                                         "pred {:#?} in {:#?} out {:#?}",
                                         &predicate,
                                         branch.inbox.get(ekey),
                                         branch.outbox.get(ekey)
                                         branch.inbox.get(port),
                                         branch.outbox.get(port)
                                     );
                                     panic!("channel_id {:?} fired (based on outbox/inbox) but the predicate had Some(false)!" ,channel_id)
+                                }
+                            }
                             None => {
                                 predicate.replace_assignment(channel_id, false);
                                 if fired {
                                     println!(
                                         "pred {:#?} in {:#?} out {:#?}",
                                         &predicate,
                                         branch.inbox.get(ekey),
                                         branch.outbox.get(ekey)
                                         branch.inbox.get(port),
                                         branch.outbox.get(port)
                                     );
                                     panic!("channel_id {:?} fired (based on outbox/inbox) but the predicate had None!" ,channel_id)
+                                }
+                            }
+                        }
+                    }
@@ @@ -167,28 +167,28 @@ impl PolyP { @@
                         &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));
                 Sb::PutMsg(port, payload) => {
                     assert!(self.ports.contains(&port));
                     let EndpointExt { info, endpoint } =
-                        m_ctx.inner.endpoint_exts.get_mut(ekey).unwrap();
+                        m_ctx.inner.endpoint_exts.get_mut(port).unwrap();
                     if predicate.replace_assignment(info.channel_id, true) != Some(false) {
-                        branch.outbox.insert(ekey, payload.clone());
+                        branch.outbox.insert(port, 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,
+                            port,
                             (info.channel_id.controller_id, info.channel_id.channel_index),
                         );
                         endpoint.send(msg)?;
                         to_run.push((predicate, branch));
+                    }
                     // ELSE DROP
@@ @@ -208,13 +208,13 @@ impl PolyP { @@
+    }
     pub(crate) fn poly_recv_run(
         &mut self,
         m_ctx: PolyPContext,
         protocol_description: &ProtocolD,
-        ekey: Key,
+        port: Port,
         payload_predicate: Predicate,
         payload: Payload,
     ) -> Result<SyncRunResult, EndpointErr> {
         // try exact match
         let to_run = if self.complete.contains_key(&payload_predicate) {
@@ @@ -230,14 +230,14 @@ impl PolyP { @@
             log!(
                 &mut m_ctx.inner.logger,
                 "... poly_recv_run matched running machine exactly! pred is {:?}",
                 &payload_predicate
             );
             branch.inbox.insert(ekey, payload);
             if branch.blocking_on == Some(ekey) {
             branch.inbox.insert(port, payload);
             if branch.blocking_on == Some(port) {
                 branch.blocking_on = None;
                 vec![(payload_predicate, branch)]
             } else {
                 vec![]
+            }
         } else {
@@ @@ -257,18 +257,18 @@ impl PolyP { @@
                                 &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
-                            if let Some(prev_payload) = branch.inbox.get(&ekey) {
+                            if let Some(prev_payload) = branch.inbox.get(&port) {
                                 // Incorrect to receive two distinct messages in same branch!
                                 assert_eq!(prev_payload, &payload);
+                            }
                             branch.inbox.insert(ekey, payload.clone());
                             if branch.blocking_on == Some(ekey) {
                             branch.inbox.insert(port, payload.clone());
                             if branch.blocking_on == Some(port) {
                                 // run.
                                 branch.blocking_on = None;
                                 Some((old_predicate, branch))
                             } else {
                                 // don't bother running. its awaiting something else
                                 incomplete2.insert(old_predicate, branch);
@@ @@ -282,21 +282,21 @@ impl PolyP { @@
                                 &payload_predicate,
                                 &old_predicate,
                                 &new,
                             );
                             // payload_predicate has new assumptions. FORK!
                             let mut payload_branch = branch.clone();
-                            if let Some(prev_payload) = payload_branch.inbox.get(&ekey) {
+                            if let Some(prev_payload) = payload_branch.inbox.get(&port) {
                                 // Incorrect to receive two distinct messages in same branch!
                                 assert_eq!(prev_payload, &payload);
+                            }
-                            payload_branch.inbox.insert(ekey, payload.clone());
+                            payload_branch.inbox.insert(port, payload.clone());
                             // put the original back untouched
                             incomplete2.insert(old_predicate, branch);
-                            if payload_branch.blocking_on == Some(ekey) {
+                            if payload_branch.blocking_on == Some(port) {
                                 // run the fork
                                 payload_branch.blocking_on = None;
                                 Some((new, payload_branch))
                             } else {
                                 // don't bother running. its awaiting something else
                                 incomplete2.insert(new, payload_branch);
@@ @@ -309,21 +309,21 @@ impl PolyP { @@
                                 "... poly_recv_run payloadpred {:?} subsumes branch pred {:?}. FORKING",
                                 &old_predicate,
                                 &payload_predicate,
                             );
                             // payload_predicate has new assumptions. FORK!
                             let mut payload_branch = branch.clone();
-                            if let Some(prev_payload) = payload_branch.inbox.get(&ekey) {
+                            if let Some(prev_payload) = payload_branch.inbox.get(&port) {
                                 // Incorrect to receive two distinct messages in same branch!
                                 assert_eq!(prev_payload, &payload);
+                            }
-                            payload_branch.inbox.insert(ekey, payload.clone());
+                            payload_branch.inbox.insert(port, payload.clone());
                             // put the original back untouched
                             incomplete2.insert(old_predicate.clone(), branch);
-                            if payload_branch.blocking_on == Some(ekey) {
+                            if payload_branch.blocking_on == Some(port) {
                                 // run the fork
                                 payload_branch.blocking_on = None;
                                 Some((payload_predicate.clone(), payload_branch))
                             } else {
                                 // don't bother running. its awaiting something else
                                 incomplete2.insert(payload_predicate.clone(), payload_branch);
@@ @@ -356,20 +356,20 @@ impl PolyP { @@
+    }
     pub(crate) fn choose_mono(&self, decision: &Predicate) -> Option<MonoP> {
         self.complete
             .iter()
             .find(|(p, _)| decision.satisfies(p))
-            .map(|(_, branch)| MonoP { state: branch.state.clone(), ekeys: self.ekeys.clone() })
+            .map(|(_, branch)| MonoP { state: branch.state.clone(), ports: self.ports.clone() })
+    }
+}
 impl PolyN {
     pub fn sync_recv(
         &mut self,
-        ekey: Key,
+        port: Port,
         logger: &mut String,
         payload: Payload,
         payload_predicate: Predicate,
         solution_storage: &mut SolutionStorage,
     ) {
         let mut branches2: HashMap<_, _> = Default::default();
@@ @@ -396,31 +396,31 @@ impl PolyN { @@
                 &case
             );
             match case {
                 Csr::Nonexistant => { /* skip branch */ }
                 Csr::LatterNotFormer | Csr::Equivalent => {
                     // Feed the message to this branch in-place. no need to modify pred.
                     if branch.to_get.remove(&ekey) {
                         branch.gotten.insert(ekey, payload.clone());
                     if branch.to_get.remove(&port) {
                         branch.gotten.insert(port, payload.clone());
                         report_if_solution(&branch, &old_predicate, logger);
+                    }
+                }
                 Csr::FormerNotLatter => {
                     // create a new branch with the payload_predicate.
                     let mut forked = branch.clone();
                     if forked.to_get.remove(&ekey) {
                         forked.gotten.insert(ekey, payload.clone());
                     if forked.to_get.remove(&port) {
                         forked.gotten.insert(port, payload.clone());
                         report_if_solution(&forked, &payload_predicate, logger);
                         branches2.insert(payload_predicate.clone(), forked);
+                    }
+                }
                 Csr::New(new) => {
                     // create a new branch with the newly-created predicate
                     let mut forked = branch.clone();
                     if forked.to_get.remove(&ekey) {
                         forked.gotten.insert(ekey, payload.clone());
                     if forked.to_get.remove(&port) {
                         forked.gotten.insert(port, payload.clone());
                         report_if_solution(&forked, &new, logger);
                         branches2.insert(new.clone(), forked);
+                    }
+                }
+            }
             // unlike PolyP machines, Native branches do NOT become inconsistent
@@ @@ -433,10 +433,10 @@ impl PolyN { @@
     pub fn choose_mono(&self, decision: &Predicate) -> Option<MonoN> {
         self.branches
             .iter()
             .find(|(p, branch)| branch.to_get.is_empty() && decision.satisfies(p))
             .map(|(_, branch)| {
                 let BranchN { gotten, sync_batch_index, .. } = branch.clone();
-                MonoN { ekeys: self.ekeys.clone(), result: Some((sync_batch_index, gotten)) }
+                MonoN { ports: self.ports.clone(), result: Some((sync_batch_index, gotten)) }
             })
+    }
+}

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -25,47 +25,47 @@ impl Controller { @@
                 );
                 Ok(())
+            }
             Decision::Failure => Err(SyncErr::Timeout),
         };
         let announcement = CommMsgContents::Announce { decision }.into_msg(self.inner.round_index);
-        for &child_ekey in self.inner.family.children_ekeys.iter() {
+        for &child_port in self.inner.family.children_ports.iter() {
             log!(
                 &mut self.inner.logger,
-                "Forwarding {:?} to child with ekey {:?}",
+                "Forwarding {:?} to child with port {:?}",
                 &announcement,
-                child_ekey
+                child_port
             );
             self.inner
                 .endpoint_exts
-                .get_mut(child_ekey)
+                .get_mut(child_port)
                 .expect("eefef")
                 .endpoint
                 .send(announcement.clone())?;
+        }
         self.inner.round_index += 1;
         self.ephemeral.clear();
         ret
+    }
     // Drain self.ephemeral.solution_storage and handle the new locals. Return decision if one is found
     fn handle_locals_maybe_decide(&mut self) -> Result<bool, SyncErr> {
-        if let Some(parent_ekey) = self.inner.family.parent_ekey {
+        if let Some(parent_port) = self.inner.family.parent_port {
             // I have a parent -> I'm not the leader
             let parent_endpoint =
-                &mut self.inner.endpoint_exts.get_mut(parent_ekey).expect("huu").endpoint;
+                &mut self.inner.endpoint_exts.get_mut(parent_port).expect("huu").endpoint;
             for partial_oracle in self.ephemeral.solution_storage.iter_new_local_make_old() {
                 let msg =
                     CommMsgContents::Elaborate { partial_oracle }.into_msg(self.inner.round_index);
-                log!(&mut self.inner.logger, "Sending {:?} to parent {:?}", &msg, parent_ekey);
+                log!(&mut self.inner.logger, "Sending {:?} to parent {:?}", &msg, parent_port);
                 parent_endpoint.send(msg)?;
+            }
             Ok(false)
         } else {
             // I have no parent -> I'm the leader
-            assert!(self.inner.family.parent_ekey.is_none());
+            assert!(self.inner.family.parent_port.is_none());
             let maybe_predicate = self.ephemeral.solution_storage.iter_new_local_make_old().next();
             Ok(if let Some(predicate) = maybe_predicate {
                 let decision = Decision::Success(predicate);
                 log!(&mut self.inner.logger, "DECIDE ON {:?} AS LEADER!", &decision);
                 self.end_round_with_decision(decision)?;
                 true
@@ @@ -76,25 +76,25 @@ impl Controller { @@
+    }
     fn kick_off_native(
         &mut self,
         sync_batches: impl Iterator<Item = SyncBatch>,
     ) -> Result<PolyN, EndpointErr> {
-        let MonoN { ekeys, .. } = self.inner.mono_n.clone();
+        let MonoN { ports, .. } = self.inner.mono_n.clone();
         let Self { inner: ControllerInner { endpoint_exts, round_index, .. }, .. } = self;
         let mut branches = HashMap::<_, _>::default();
         for (sync_batch_index, SyncBatch { puts, gets }) in sync_batches.enumerate() {
             let ekey_to_channel_id = |ekey| endpoint_exts.get(ekey).unwrap().info.channel_id;
             let all_ekeys = ekeys.iter().copied();
             let all_channel_ids = all_ekeys.map(ekey_to_channel_id);
             let port_to_channel_id = |port| endpoint_exts.get(port).unwrap().info.channel_id;
             let all_ports = ports.iter().copied();
             let all_channel_ids = all_ports.map(port_to_channel_id);
             let mut predicate = Predicate::new_trivial();
             // assign TRUE for puts and gets
             let true_ekeys = puts.keys().chain(gets.iter()).copied();
             let true_channel_ids = true_ekeys.clone().map(ekey_to_channel_id);
             let true_ports = puts.keys().chain(gets.iter()).copied();
             let true_channel_ids = true_ports.clone().map(port_to_channel_id);
             predicate.batch_assign_nones(true_channel_ids, true);
             // assign FALSE for all in interface not assigned true
             predicate.batch_assign_nones(all_channel_ids.clone(), false);
             if branches.contains_key(&predicate) {
@@ @@ -103,24 +103,24 @@ impl Controller { @@
                     "Duplicate predicate {:#?}!\nHaving multiple batches with the same
                     predicate requires the support of oracle boolean variables",
                     &predicate,
+                )
+            }
             let branch = BranchN { to_get: gets, gotten: Default::default(), sync_batch_index };
-            for (ekey, payload) in puts {
+            for (port, payload) in puts {
                 log!(
                     &mut self.inner.logger,
                     "... ... Initial native put msg {:?} pred {:?} batch {:?}",
                     &payload,
                     &predicate,
                     sync_batch_index,
                 );
                 let msg =
                     CommMsgContents::SendPayload { payload_predicate: predicate.clone(), payload }
                         .into_msg(*round_index);
-                endpoint_exts.get_mut(ekey).unwrap().endpoint.send(msg)?;
+                endpoint_exts.get_mut(port).unwrap().endpoint.send(msg)?;
+            }
             log!(
                 &mut self.inner.logger,
                 "... Initial native branch batch index={} with pred {:?}",
                 sync_batch_index,
                 &predicate
@@ @@ -131,13 +131,13 @@ impl Controller { @@
                     SubtreeId::PolyN,
                     predicate.clone(),
                 );
+            }
             branches.insert(predicate, branch);
+        }
-        Ok(PolyN { ekeys, branches })
+        Ok(PolyN { ports, branches })
+    }
     pub fn sync_round(
         &mut self,
         deadline: Option<Instant>,
         sync_batches: Option<impl Iterator<Item = SyncBatch>>,
     ) -> Result<(), SyncErr> {
@@ @@ -175,13 +175,13 @@ impl Controller { @@
         //    Some actors are dropped. some new actors are created.
         //    Ultimately, we have 0 Mono actors and a list of unnamed sync_actors
         self.ephemeral.mono_ps.extend(self.inner.mono_ps.iter().cloned());
         log!(&mut self.inner.logger, "Got {} MonoP's to run!", self.ephemeral.mono_ps.len());
         while let Some(mut mono_p) = self.ephemeral.mono_ps.pop() {
             let mut m_ctx = MonoPContext {
-                ekeys: &mut mono_p.ekeys,
+                ports: &mut mono_p.ports,
                 mono_ps: &mut self.ephemeral.mono_ps,
                 inner: &mut self.inner,
             };
             // cross boundary into crate::protocol
             let blocker = mono_p.state.pre_sync_run(&mut m_ctx, &self.protocol_description);
             log!(&mut self.inner.logger, "... MonoP's pre_sync_run got blocker {:?}", &blocker);
@@ @@ -194,44 +194,44 @@ impl Controller { @@
         log!(
             &mut self.inner.logger,
             "Finished running all MonoPs! Have {} PolyPs waiting",
             self.ephemeral.poly_ps.len()
         );
-        // 3. define the mapping from ekey -> actor
+        // 3. define the mapping from port -> actor
         //    this is needed during the event loop to determine which actor
         //    should receive the incoming message.
         //    TODO: store and update this mapping rather than rebuilding it each round.
-        let ekey_to_holder: HashMap<Key, PolyId> = {
+        let port_to_holder: HashMap<Port, PolyId> = {
             use PolyId::*;
-            let n = self.inner.mono_n.ekeys.iter().map(move |&e| (e, N));
+            let n = self.inner.mono_n.ports.iter().map(move |&e| (e, N));
             let p = self
                 .ephemeral
                 .poly_ps
                 .iter()
                 .enumerate()
-                .flat_map(|(index, m)| m.ekeys.iter().map(move |&e| (e, P { index })));
+                .flat_map(|(index, m)| m.ports.iter().map(move |&e| (e, P { index })));
             n.chain(p).collect()
         };
         log!(
             &mut self.inner.logger,
             "SET OF PolyPs and MonoPs final! ekey lookup map is {:?}",
             &ekey_to_holder
             "SET OF PolyPs and MonoPs final! port lookup map is {:?}",
             &port_to_holder
         );
         // 4. Create the solution storage. it tracks the solutions of "subtrees"
         //    of the controller in the overlay tree.
         self.ephemeral.solution_storage.reset({
             let n = std::iter::once(SubtreeId::PolyN);
             let m = (0..self.ephemeral.poly_ps.len()).map(|index| SubtreeId::PolyP { index });
             let c = self
                 .inner
                 .family
-                .children_ekeys
+                .children_ports
                 .iter()
-                .map(|&ekey| SubtreeId::ChildController { ekey });
+                .map(|&port| SubtreeId::ChildController { port });
             let subtree_id_iter = n.chain(m).chain(c);
             log!(
                 &mut self.inner.logger,
                 "Solution Storage has subtree Ids: {:?}",
                 &subtree_id_iter.clone().collect::<Vec<_>>()
             );
@@ @@ -306,32 +306,32 @@ impl Controller { @@
                     // we have not yet timed out. performed a time-limited poll
                     Some(received) => received,
                     None => {
                         // timed out! send a FAILURE message to the sink,
                         // and henceforth don't time out on polling.
                         deadline = None;
-                        match self.inner.family.parent_ekey {
+                        match self.inner.family.parent_port {
                             None => {
                                 // I am the sink! announce failure and return.
                                 return self.end_round_with_decision(Decision::Failure);
+                            }
-                            Some(parent_ekey) => {
+                            Some(parent_port) => {
                                 // I am not the sink! send a failure message.
                                 let announcement = Msg::CommMsg(CommMsg {
                                     round_index: self.inner.round_index,
                                     contents: CommMsgContents::Failure,
                                 });
                                 log!(
                                     &mut self.inner.logger,
-                                    "Forwarding {:?} to parent with ekey {:?}",
+                                    "Forwarding {:?} to parent with port {:?}",
                                     &announcement,
-                                    parent_ekey
+                                    parent_port
                                 );
                                 self.inner
                                     .endpoint_exts
-                                    .get_mut(parent_ekey)
+                                    .get_mut(parent_port)
                                     .expect("ss")
                                     .endpoint
                                     .send(announcement.clone())?;
                                 continue; // poll some more
+                            }
+                        }
@@ @@ -361,47 +361,47 @@ impl Controller { @@
                     self.delay(received);
                     continue 'recv_loop;
+                }
                 Msg::CommMsg(CommMsg { contents, round_index }) => {
                     log!(
                         &mut self.inner.logger,
-                        "... its a round-appropriate CommMsg with key {:?}",
+                        "... its a round-appropriate CommMsg with port {:?}",
                         received.recipient
                     );
                     assert_eq!(round_index, self.inner.round_index);
                     contents
+                }
             };
             match current_content {
                 CommMsgContents::Failure => match self.inner.family.parent_ekey {
                     Some(parent_ekey) => {
                 CommMsgContents::Failure => match self.inner.family.parent_port {
                     Some(parent_port) => {
                         let announcement = Msg::CommMsg(CommMsg {
                             round_index: self.inner.round_index,
                             contents: CommMsgContents::Failure,
                         });
                         log!(
                             &mut self.inner.logger,
-                            "Forwarding {:?} to parent with ekey {:?}",
+                            "Forwarding {:?} to parent with port {:?}",
                             &announcement,
-                            parent_ekey
+                            parent_port
                         );
                         self.inner
                             .endpoint_exts
-                            .get_mut(parent_ekey)
+                            .get_mut(parent_port)
                             .expect("ss")
                             .endpoint
                             .send(announcement.clone())?;
+                    }
                     None => return self.end_round_with_decision(Decision::Failure),
                 },
                 CommMsgContents::Elaborate { partial_oracle } => {
                     // Child controller submitted a subtree solution.
-                    if !self.inner.family.children_ekeys.contains(&received.recipient) {
+                    if !self.inner.family.children_ports.contains(&received.recipient) {
                         return Err(SyncErr::ElaborateFromNonChild);
+                    }
-                    let subtree_id = SubtreeId::ChildController { ekey: received.recipient };
+                    let subtree_id = SubtreeId::ChildController { port: received.recipient };
                     log!(
                         &mut self.inner.logger,
                         "Received elaboration from child for subtree {:?}: {:?}",
                         subtree_id,
                         &partial_oracle
                     );
@@ @@ -412,13 +412,13 @@ impl Controller { @@
                     );
                     if self.handle_locals_maybe_decide()? {
                         return Ok(());
+                    }
+                }
                 CommMsgContents::Announce { decision } => {
-                    if self.inner.family.parent_ekey != Some(received.recipient) {
+                    if self.inner.family.parent_port != Some(received.recipient) {
                         return Err(SyncErr::AnnounceFromNonParent);
+                    }
                     log!(
                         &mut self.inner.logger,
                         "Received ANNOUNCEMENT from from parent {:?}: {:?}",
                         received.recipient,
@@ @@ -432,13 +432,13 @@ impl Controller { @@
                         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.
-                    let subtree_id = ekey_to_holder.get(&received.recipient);
+                    let subtree_id = port_to_holder.get(&received.recipient);
                     log!(
                         &mut self.inner.logger,
                         "Received SendPayload for subtree {:?} with pred {:?} and payload {:?}",
                         subtree_id,
                         &payload_predicate,
                         &payload
@@ @@ -527,19 +527,19 @@ impl Controller { @@
 impl ControllerEphemeral {
     fn is_clear(&self) -> bool {
         self.solution_storage.is_clear()
             && self.poly_n.is_none()
             && self.poly_ps.is_empty()
             && self.mono_ps.is_empty()
-            && self.ekey_to_holder.is_empty()
+            && self.port_to_holder.is_empty()
+    }
     fn clear(&mut self) {
         self.solution_storage.clear();
         self.poly_n.take();
         self.poly_ps.clear();
-        self.ekey_to_holder.clear();
+        self.port_to_holder.clear();
+    }
+}
 impl Into<PolyP> for MonoP {
     fn into(self) -> PolyP {
         PolyP {
             complete: Default::default(),
@@ @@ -549,13 +549,13 @@ impl Into<PolyP> for MonoP { @@
                     state: self.state,
                     inbox: Default::default(),
                     outbox: Default::default(),
                     blocking_on: None,
+                }
             },
-            ekeys: self.ekeys,
+            ports: self.ports,
+        }
+    }
+}
 impl From<EndpointErr> for SyncErr {
     fn from(e: EndpointErr) -> SyncErr {
@@ @@ -563,47 +563,47 @@ impl From<EndpointErr> for SyncErr { @@
+    }
+}
 impl MonoContext for MonoPContext<'_> {
     type D = ProtocolD;
     type S = ProtocolS;
-    fn new_component(&mut self, moved_ekeys: HashSet<Key>, init_state: Self::S) {
+    fn new_component(&mut self, moved_ports: HashSet<Port>, init_state: Self::S) {
         log!(
             &mut self.inner.logger,
             "!! MonoContext callback to new_component with ekeys {:?}!",
             &moved_ekeys,
             "!! MonoContext callback to new_component with ports {:?}!",
             &moved_ports,
         );
         if moved_ekeys.is_subset(self.ekeys) {
             self.ekeys.retain(|x| !moved_ekeys.contains(x));
             self.mono_ps.push(MonoP { state: init_state, ekeys: moved_ekeys });
         if moved_ports.is_subset(self.ports) {
             self.ports.retain(|x| !moved_ports.contains(x));
             self.mono_ps.push(MonoP { state: init_state, ports: moved_ports });
         } else {
-            panic!("MachineP attempting to move alien ekey!");
+            panic!("MachineP attempting to move alien port!");
+        }
+    }
-    fn new_channel(&mut self) -> [Key; 2] {
+    fn new_channel(&mut self) -> [Port; 2] {
         let [a, b] = Endpoint::new_memory_pair();
         let channel_id = self.inner.channel_id_stream.next();
         let mut clos = |endpoint, polarity| {
             let endpoint_ext =
                 EndpointExt { info: EndpointInfo { polarity, channel_id }, endpoint };
             let ekey = self.inner.endpoint_exts.alloc(endpoint_ext);
             let endpoint = &self.inner.endpoint_exts.get(ekey).unwrap().endpoint;
             let token = Key::to_token(ekey);
             let port = self.inner.endpoint_exts.alloc(endpoint_ext);
             let endpoint = &self.inner.endpoint_exts.get(port).unwrap().endpoint;
             let token = Port::to_token(port);
             self.inner
                 .messenger_state
                 .poll
                 .register(endpoint, token, Ready::readable(), PollOpt::edge())
                 .expect("AAGAGGGGG");
             self.ekeys.insert(ekey);
             ekey
             self.ports.insert(port);
             port
         };
         let [kp, kg] = [clos(a, Putter), clos(b, Getter)];
         log!(
             &mut self.inner.logger,
-            "!! MonoContext callback to new_channel. returning ekeys {:?}!",
+            "!! MonoContext callback to new_channel. returning ports {:?}!",
             [kp, kg],
         );
         [kp, kg]
+    }
     fn new_random(&mut self) -> u64 {
         type Bytes8 = [u8; std::mem::size_of::<u64>()];
@@ @@ -710,27 +710,27 @@ impl SolutionStorage { @@
+        }
+    }
+}
 impl PolyContext for BranchPContext<'_, '_> {
     type D = ProtocolD;
     fn is_firing(&mut self, ekey: Key) -> Option<bool> {
         assert!(self.ekeys.contains(&ekey));
         let channel_id = self.m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;
     fn is_firing(&mut self, port: Port) -> Option<bool> {
         assert!(self.ports.contains(&port));
         let channel_id = self.m_ctx.inner.endpoint_exts.get(port).unwrap().info.channel_id;
         let val = self.predicate.query(channel_id);
         log!(
             &mut self.m_ctx.inner.logger,
             "!! PolyContext callback to is_firing by {:?}! returning {:?}",
             self.m_ctx.my_subtree_id,
             val,
         );
         val
+    }
     fn read_msg(&mut self, ekey: Key) -> Option<&Payload> {
         assert!(self.ekeys.contains(&ekey));
         let val = self.inbox.get(&ekey);
     fn read_msg(&mut self, port: Port) -> Option<&Payload> {
         assert!(self.ports.contains(&port));
         let val = self.inbox.get(&port);
         log!(
             &mut self.m_ctx.inner.logger,
             "!! PolyContext callback to read_msg by {:?}! returning {:?}",
             self.m_ctx.my_subtree_id,
             val,
         );

src/runtime/connector.rs

➞

Show inline comments

@@ @@ -110,41 +110,41 @@ impl Connector { @@
     pub fn put(&mut self, native_port_index: usize, payload: Payload) -> Result<(), PortOpErr> {
         use PortOpErr::*;
         let connected = match self {
             Connector::Connected(connected) => connected,
             _ => return Err(NotConnected),
         };
-        let (ekey, native_polarity) =
+        let (port, native_polarity) =
             *connected.native_interface.get(native_port_index).ok_or(IndexOutOfBounds)?;
         if native_polarity != Putter {
             return Err(WrongPolarity);
+        }
         let sync_batch = connected.sync_batches.iter_mut().last().expect("no sync batch!");
-        if sync_batch.puts.contains_key(&ekey) {
+        if sync_batch.puts.contains_key(&port) {
             return Err(DuplicateOperation);
+        }
-        sync_batch.puts.insert(ekey, payload);
+        sync_batch.puts.insert(port, payload);
         Ok(())
+    }
     pub fn get(&mut self, native_port_index: usize) -> Result<(), PortOpErr> {
         use PortOpErr::*;
         let connected = match self {
             Connector::Connected(connected) => connected,
             _ => return Err(NotConnected),
         };
-        let (ekey, native_polarity) =
+        let (port, native_polarity) =
             *connected.native_interface.get(native_port_index).ok_or(IndexOutOfBounds)?;
         if native_polarity != Getter {
             return Err(WrongPolarity);
+        }
         let sync_batch = connected.sync_batches.iter_mut().last().expect("no sync batch!");
-        if sync_batch.gets.contains(&ekey) {
+        if sync_batch.gets.contains(&port) {
             return Err(DuplicateOperation);
+        }
-        sync_batch.gets.insert(ekey);
+        sync_batch.gets.insert(port);
         Ok(())
+    }
     pub fn next_batch(&mut self) -> Result<usize, ()> {
         let connected = match self {
             Connector::Connected(connected) => connected,
             _ => return Err(()),

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -31,14 +31,14 @@ pub(crate) enum CommonSatResult { @@
 pub(crate) struct Predicate {
     pub assigned: BTreeMap<ChannelId, bool>,
+}
 #[derive(Debug, Default)]
 struct SyncBatch {
     puts: HashMap<Key, Payload>,
     gets: HashSet<Key>,
     puts: HashMap<Port, Payload>,
     gets: HashSet<Port>,
+}
 #[derive(Debug)]
 pub enum Connector {
     Unconfigured(Unconfigured),
     Configured(Configured),
@@ @@ -56,13 +56,13 @@ pub struct Configured { @@
     protocol_description: Arc<ProtocolD>,
     main_component: Vec<u8>,
     logger: String,
+}
 #[derive(Debug)]
 pub struct Connected {
-    native_interface: Vec<(Key, Polarity)>,
+    native_interface: Vec<(Port, Polarity)>,
     sync_batches: Vec<SyncBatch>,
     controller: Controller,
+}
 #[derive(Debug, Copy, Clone)]
 pub enum PortBinding {
@@ @@ -75,23 +75,23 @@ pub enum PortBinding { @@
 struct Arena<T> {
     storage: Vec<T>,
+}
 #[derive(Debug)]
 struct ReceivedMsg {
-    recipient: Key,
+    recipient: Port,
     msg: Msg,
+}
 #[derive(Debug)]
 struct MessengerState {
     poll: Poll,
     events: Events,
     delayed: Vec<ReceivedMsg>,
     undelayed: Vec<ReceivedMsg>,
-    polled_undrained: IndexSet<Key>,
+    polled_undrained: IndexSet<Port>,
+}
 #[derive(Debug)]
 struct ChannelIdStream {
     controller_id: ControllerId,
     next_channel_index: ChannelIndex,
+}
@@ @@ -119,19 +119,19 @@ struct ControllerInner { @@
 #[derive(Debug, Default)]
 struct ControllerEphemeral {
     solution_storage: SolutionStorage,
     poly_n: Option<PolyN>,
     poly_ps: Vec<PolyP>,
     mono_ps: Vec<MonoP>,
-    ekey_to_holder: HashMap<Key, PolyId>,
+    port_to_holder: HashMap<Port, PolyId>,
+}
 #[derive(Debug)]
 struct ControllerFamily {
     parent_ekey: Option<Key>,
     children_ekeys: Vec<Key>,
     parent_port: Option<Port>,
     children_ports: Vec<Port>,
+}
 #[derive(Debug)]
 pub(crate) enum SyncRunResult {
     BlockingForRecv,
     AllBranchesComplete,
@@ @@ -146,18 +146,18 @@ enum PolyId { @@
+}
 #[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
 pub(crate) enum SubtreeId {
     PolyN,
     PolyP { index: usize },
-    ChildController { ekey: Key },
+    ChildController { port: Port },
+}
 pub(crate) struct MonoPContext<'a> {
     inner: &'a mut ControllerInner,
-    ekeys: &'a mut HashSet<Key>,
+    ports: &'a mut HashSet<Port>,
     mono_ps: &'a mut Vec<MonoP>,
+}
 pub(crate) struct PolyPContext<'a> {
     my_subtree_id: SubtreeId,
     inner: &'a mut ControllerInner,
     solution_storage: &'a mut SolutionStorage,
@@ @@ -168,15 +168,15 @@ impl PolyPContext<'_> { @@
         let Self { solution_storage, my_subtree_id, inner } = self;
         PolyPContext { solution_storage, my_subtree_id: *my_subtree_id, inner }
+    }
+}
 struct BranchPContext<'m, 'r> {
     m_ctx: PolyPContext<'m>,
-    ekeys: &'r HashSet<Key>,
+    ports: &'r HashSet<Port>,
     predicate: &'r Predicate,
-    inbox: &'r HashMap<Key, Payload>,
+    inbox: &'r HashMap<Port, Payload>,
+}
 #[derive(Default)]
 pub(crate) struct SolutionStorage {
     old_local: HashSet<Predicate>,
     new_local: HashSet<Predicate>,
@@ @@ -184,52 +184,52 @@ pub(crate) struct SolutionStorage { @@
     subtree_solutions: Vec<HashSet<Predicate>>,
     subtree_id_to_index: HashMap<SubtreeId, usize>,
+}
 trait Messengerlike {
     fn get_state_mut(&mut self) -> &mut MessengerState;
-    fn get_endpoint_mut(&mut self, eekey: Key) -> &mut Endpoint;
+    fn get_endpoint_mut(&mut self, eport: Port) -> &mut Endpoint;
     fn delay(&mut self, received: ReceivedMsg) {
         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> {
+    fn send(&mut self, to: Port, 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() {
+            while let Some(eport) = self.get_state_mut().polled_undrained.pop() {
                 if let Some(msg) = self
-                    .get_endpoint_mut(eekey)
+                    .get_endpoint_mut(eport)
                     .recv()
-                    .map_err(|e| MessengerRecvErr::EndpointErr(eekey, e))?
+                    .map_err(|e| MessengerRecvErr::EndpointErr(eport, e))?
+                {
                     // 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 }));
                     self.get_state_mut().polled_undrained.insert(eport);
                     return Ok(Some(ReceivedMsg { recipient: eport, 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()));
+                        state.polled_undrained.insert(Port::from_token(e.token()));
+                    }
+                }
                 Err(PollDeadlineErr::PollingFailed) => return Err(MessengerRecvErr::PollingFailed),
                 Err(PollDeadlineErr::Timeout) => return Ok(None),
+            }
+        }
@@ @@ -239,32 +239,32 @@ trait Messengerlike { @@
         if let Some(x) = self.get_state_mut().undelayed.pop() {
             return Ok(x);
+        }
         loop {
             // polled_undrained may not be empty
-            while let Some(eekey) = self.get_state_mut().polled_undrained.pop() {
+            while let Some(eport) = self.get_state_mut().polled_undrained.pop() {
                 if let Some(msg) = self
-                    .get_endpoint_mut(eekey)
+                    .get_endpoint_mut(eport)
                     .recv()
-                    .map_err(|e| MessengerRecvErr::EndpointErr(eekey, e))?
+                    .map_err(|e| MessengerRecvErr::EndpointErr(eport, e))?
+                {
                     // this endpoint MAY still have messages! check again in future
                     self.get_state_mut().polled_undrained.insert(eekey);
                     return Ok(ReceivedMsg { recipient: eekey, msg });
                     self.get_state_mut().polled_undrained.insert(eport);
                     return Ok(ReceivedMsg { recipient: eport, msg });
+                }
+            }
             let state = self.get_state_mut();
             state
                 .poll
                 .poll(&mut state.events, None)
                 .map_err(|_| MessengerRecvErr::PollingFailed)?;
             for e in state.events.iter() {
-                state.polled_undrained.insert(Key::from_token(e.token()));
+                state.polled_undrained.insert(Port::from_token(e.token()));
+            }
+        }
+    }
+}
 /////////////////////////////////
@@ @@ -290,44 +290,39 @@ impl From<MessengerRecvErr> for SyncErr { @@
+}
 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 {
+    pub fn alloc(&mut self, t: T) -> Port {
         self.storage.push(t);
-        Key::from_raw(self.storage.len() - 1)
+        Port::from_raw(self.storage.len() - 1)
+    }
-    pub fn get(&self, key: Key) -> Option<&T> {
+    pub fn get(&self, key: Port) -> Option<&T> {
         self.storage.get(key.to_raw() as usize)
+    }
-    pub fn get_mut(&mut self, key: Key) -> Option<&mut T> {
+    pub fn get_mut(&mut self, key: Port) -> 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> {
+    pub fn type_convert<X>(self, f: impl FnMut((Port, 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)> {
+    pub fn iter(&self) -> impl Iterator<Item = (Port, &T)> {
         self.keyspace().zip(self.storage.iter())
+    }
     pub fn len(&self) -> usize {
         self.storage.len()
+    }
     pub fn keyspace(&self) -> impl Iterator<Item = Key> {
         (0..self.storage.len()).map(Key::from_raw)
     pub fn keyspace(&self) -> impl Iterator<Item = Port> {
         (0..self.storage.len()).map(Port::from_raw)
+    }
+}
 impl ChannelIdStream {
     fn new(controller_id: ControllerId) -> Self {
         Self { controller_id, next_channel_index: 0 }

src/runtime/setup.rs

➞

Show inline comments

@@ @@ -22,100 +22,100 @@ impl Controller { @@
         major: ControllerId,
         main_component: &[u8],
         protocol_description: Arc<ProtocolD>,
         bound_proto_interface: &[(PortBinding, Polarity)],
         logger: &mut String,
         deadline: Instant,
-    ) -> Result<(Self, Vec<(Key, Polarity)>), ConnectErr> {
+    ) -> Result<(Self, Vec<(Port, Polarity)>), ConnectErr> {
         use ConnectErr::*;
         log!(logger, "CONNECT PHASE START! MY CID={:?} STARTING LOGGER ~", major);
         let mut channel_id_stream = ChannelIdStream::new(major);
         let mut endpoint_ext_todos = Arena::default();
         let mut ekeys_native = vec![];
         let mut ekeys_proto = vec![];
         let mut ekeys_network = vec![];
         let mut ports_native = vec![];
         let mut ports_proto = vec![];
         let mut ports_network = vec![];
         let mut native_interface = vec![];
         /*
 .  - allocate an EndpointExtTodo for every native and interface port
-            - store all the resulting keys in two keylists for the interfaces of the native and proto components
+            - store all the resulting ports in two portlists for the interfaces of the native and proto components
                 native: [a, c,    f]
                          |  |     |
                          |  |     |
                 proto:  [b, d, e, g]
                                ^todo
                 arena: <A,B,C,D,E,F,G>
         */
         for &(binding, polarity) in bound_proto_interface.iter() {
             match binding {
                 PortBinding::Native => {
                     let channel_id = channel_id_stream.next();
-                    let ([ekey_native, ekey_proto], native_polarity) = {
+                    let ([port_native, port_proto], native_polarity) = {
                         let [p, g] = Endpoint::new_memory_pair();
-                        let mut endpoint_to_key = |endpoint, polarity| {
+                        let mut endpoint_to_port = |endpoint, polarity| {
                             endpoint_ext_todos.alloc(EndpointExtTodo::Finished(EndpointExt {
                                 endpoint,
                                 info: EndpointInfo { polarity, channel_id },
                             }))
                         };
                         let pkey = endpoint_to_key(p, Putter);
                         let gkey = endpoint_to_key(g, Getter);
                         let key_pair = match polarity {
                             Putter => [gkey, pkey],
                             Getter => [pkey, gkey],
                         let pport = endpoint_to_port(p, Putter);
                         let gport = endpoint_to_port(g, Getter);
                         let port_pair = match polarity {
                             Putter => [gport, pport],
                             Getter => [pport, gport],
                         };
-                        (key_pair, !polarity)
+                        (port_pair, !polarity)
                     };
                     native_interface.push((ekey_native, native_polarity));
                     ekeys_native.push(ekey_native);
                     ekeys_proto.push(ekey_proto);
                     native_interface.push((port_native, native_polarity));
                     ports_native.push(port_native);
                     ports_proto.push(port_proto);
+                }
                 PortBinding::Passive(addr) => {
                     let channel_id = channel_id_stream.next();
-                    let ekey_proto = endpoint_ext_todos.alloc(EndpointExtTodo::PassiveAccepting {
+                    let port_proto = endpoint_ext_todos.alloc(EndpointExtTodo::PassiveAccepting {
                         addr,
                         info: EndpointInfo { polarity, channel_id },
                         listener: TcpListener::bind(&addr).map_err(|_| BindFailed(addr))?,
                     });
                     ekeys_network.push(ekey_proto);
                     ekeys_proto.push(ekey_proto);
                     ports_network.push(port_proto);
                     ports_proto.push(port_proto);
+                }
                 PortBinding::Active(addr) => {
-                    let ekey_proto = endpoint_ext_todos.alloc(EndpointExtTodo::ActiveConnecting {
+                    let port_proto = endpoint_ext_todos.alloc(EndpointExtTodo::ActiveConnecting {
                         addr,
                         polarity,
                         stream: TcpStream::connect(&addr).unwrap(),
                     });
                     ekeys_network.push(ekey_proto);
                     ekeys_proto.push(ekey_proto);
                     ports_network.push(port_proto);
                     ports_proto.push(port_proto);
+                }
+            }
+        }
         log!(logger, "{:03?} setup todos...", major);
         // 2. convert the arena to Arena<EndpointExt>  and return the
         let (mut messenger_state, mut endpoint_exts) =
             Self::finish_endpoint_ext_todos(major, logger, endpoint_ext_todos, deadline)?;
-        let n_mono = MonoN { ekeys: ekeys_native.into_iter().collect(), result: None };
+        let n_mono = MonoN { ports: ports_native.into_iter().collect(), result: None };
         let p_monos = vec![MonoP {
             state: protocol_description.new_main_component(main_component, &ekeys_proto),
             ekeys: ekeys_proto.into_iter().collect(),
             state: protocol_description.new_main_component(main_component, &ports_proto),
             ports: ports_proto.into_iter().collect(),
         }];
         // 6. Become a node in a sink tree, computing {PARENT, CHILDREN} from {NEIGHBORS}
         let family = Self::setup_sink_tree_family(
             major,
             logger,
             &mut endpoint_exts,
             &mut messenger_state,
-            ekeys_network,
+            ports_network,
             deadline,
         )?;
         log!(logger, "CONNECT PHASE END! ~");
         let inner = ControllerInner {
             family,
@@ @@ -176,28 +176,28 @@ impl Controller { @@
             delayed: vec![],
             undelayed: vec![],
             polled_undrained: Default::default(),
         };
         // 2. Register all EndpointExtTodos with ms.poll. each has one of {Endpoint, TcpStream, TcpListener}
-        // 3. store the keyset of EndpointExtTodos which are not Finished in `to_finish`.
+        // 3. store the portset of EndpointExtTodos which are not Finished in `to_finish`.
         let mut to_finish = HashSet::<_>::default();
         log!(logger, "endpoint_ext_todos len {:?}", endpoint_ext_todos.len());
         for (key, t) in endpoint_ext_todos.iter() {
             let token = key.to_token();
         for (port, t) in endpoint_ext_todos.iter() {
             let token = port.to_token();
             match t {
                 ActiveRecving { .. } | PassiveConnecting { .. } => unreachable!(),
                 Finished(EndpointExt { endpoint, .. }) => {
                     ms.poll.register(endpoint, token, ready_r, edge)
+                }
                 ActiveConnecting { stream, .. } => {
-                    to_finish.insert(key);
+                    to_finish.insert(port);
                     ms.poll.register(stream, token, ready_w, edge)
+                }
                 PassiveAccepting { listener, .. } => {
-                    to_finish.insert(key);
+                    to_finish.insert(port);
                     ms.poll.register(listener, token, ready_r, edge)
+                }
+            }
             .expect("register first");
+        }
         // invariant: every EndpointExtTodo has one thing registered with mio
@@ @@ -212,17 +212,17 @@ impl Controller { @@
                 log!(logger, "{:03?} timing out", major);
+                e
             })?;
             for event in ms.events.iter() {
                 log!(logger, "event {:#?}", event);
                 let token = event.token();
                 let ekey = Key::from_token(token);
                 let entry = endpoint_ext_todos.get_mut(ekey).unwrap();
                 let port = Port::from_token(token);
                 let entry = endpoint_ext_todos.get_mut(port).unwrap();
                 match entry {
                     Finished(_) => {
-                        polled_undrained_later.insert(ekey);
+                        polled_undrained_later.insert(port);
+                    }
                     PassiveAccepting { addr, listener, .. } => {
                         log!(logger, "{:03?} start PassiveAccepting...", major);
                         assert!(event.readiness().is_readable());
                         let (stream, _peer_addr) =
                             listener.accept().map_err(|_| AcceptFailed(*addr))?;
@@ @@ -252,13 +252,13 @@ impl Controller { @@
                                 res = endpoint.send(msg);
                                 Finished(EndpointExt { info, endpoint })
                             }]
                         });
                         res?;
                         log!(logger, "{:03?} ... end PassiveConnecting", major);
-                        assert!(to_finish.remove(&ekey));
+                        assert!(to_finish.remove(&port));
+                    }
                     ActiveConnecting { addr, stream, .. } => {
                         log!(logger, "{:03?} start ActiveConnecting...", major);
                         assert!(event.readiness().is_writable());
                         if Self::connection_testing_read(stream, &mut next_inbox).is_ok() {
                             // connect successful
@@ @@ -295,26 +295,26 @@ impl Controller { @@
                                 take_mut::take(entry, |e| {
                                     assert_let![ActiveRecving { polarity, endpoint, .. } = e => {
                                         let info = EndpointInfo { polarity, channel_id: info.channel_id };
                                         Finished(EndpointExt { info, endpoint })
                                     }]
                                 });
                                 ms.polled_undrained.insert(ekey);
                                 assert!(to_finish.remove(&ekey));
                                 ms.polled_undrained.insert(port);
                                 assert!(to_finish.remove(&port));
                                 break 'recv_loop;
                             } else {
-                                ms.delayed.push(ReceivedMsg { recipient: ekey, msg });
+                                ms.delayed.push(ReceivedMsg { recipient: port, msg });
+                            }
+                        }
                         log!(logger, "{:03?} ... end ActiveRecving", major);
+                    }
+                }
+            }
+        }
         for ekey in polled_undrained_later {
             ms.polled_undrained.insert(ekey);
         for port in polled_undrained_later {
             ms.polled_undrained.insert(port);
+        }
         let endpoint_exts = endpoint_ext_todos.type_convert(|(_, todo)| match todo {
             Finished(endpoint_ext) => endpoint_ext,
             _ => unreachable!(),
         });
         Ok((ms, endpoint_exts))
@@ @@ -322,41 +322,41 @@ impl Controller { @@
     fn setup_sink_tree_family(
         major: ControllerId,
         logger: &mut String,
         endpoint_exts: &mut Arena<EndpointExt>,
         messenger_state: &mut MessengerState,
-        neighbors: Vec<Key>,
+        neighbors: Vec<Port>,
         deadline: Instant,
     ) -> Result<ControllerFamily, ConnectErr> {
         use {ConnectErr::*, Msg::SetupMsg as S, SetupMsg::*};
         log!(logger, "neighbors {:?}", &neighbors);
         let mut messenger = (messenger_state, endpoint_exts);
         impl Messengerlike for (&mut MessengerState, &mut Arena<EndpointExt>) {
             fn get_state_mut(&mut self) -> &mut MessengerState {
                 self.0
+            }
             fn get_endpoint_mut(&mut self, ekey: Key) -> &mut Endpoint {
                 &mut self.1.get_mut(ekey).expect("OUT OF BOUNDS").endpoint
             fn get_endpoint_mut(&mut self, port: Port) -> &mut Endpoint {
                 &mut self.1.get_mut(port).expect("OUT OF BOUNDS").endpoint
+            }
+        }
-        // 1. broadcast my ID as the first echo. await reply from all in net_keylist
+        // 1. broadcast my ID as the first echo. await reply from all in net_portlist
         let echo = S(LeaderEcho { maybe_leader: major });
         let mut awaiting = IndexSet::with_capacity(neighbors.len());
         for &n in neighbors.iter() {
             log!(logger, "{:?}'s initial echo to {:?}, {:?}", major, n, &echo);
             messenger.send(n, echo.clone())?;
             awaiting.insert(n);
+        }
         // 2. Receive incoming replies. whenever a higher-id echo arrives,
         //    adopt it as leader, sender as parent, and reset the await set.
-        let mut parent: Option<Key> = None;
+        let mut parent: Option<Port> = None;
         let mut my_leader = major;
         messenger.undelay_all();
         'echo_loop: while !awaiting.is_empty() || parent.is_some() {
             let ReceivedMsg { recipient, msg } = messenger.recv(deadline)?.ok_or(Timeout)?;
             log!(logger, "{:?} GOT {:?} {:?}", major, &recipient, &msg);
             match msg {
@@ @@ -467,18 +467,18 @@ impl Controller { @@
                     assert!(Some(recipient) != parent);
                     // they wouldn't send me this if they considered me their parent
+                }
                 _ => messenger.delay(ReceivedMsg { recipient, msg }),
+            }
+        }
-        Ok(ControllerFamily { parent_ekey: parent, children_ekeys: children })
+        Ok(ControllerFamily { parent_port: parent, children_ports: children })
+    }
+}
 impl Messengerlike for Controller {
     fn get_state_mut(&mut self) -> &mut MessengerState {
         &mut self.inner.messenger_state
+    }
     fn get_endpoint_mut(&mut self, ekey: Key) -> &mut Endpoint {
         &mut self.inner.endpoint_exts.get_mut(ekey).expect("OUT OF BOUNDS").endpoint
     fn get_endpoint_mut(&mut self, port: Port) -> &mut Endpoint {
         &mut self.inner.endpoint_exts.get_mut(port).expect("OUT OF BOUNDS").endpoint
+    }
+}

0 comments (0 inline, 0 general)