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

Changeset - 02eb59c6fd66

Parent rev.

Child rev.

[Not reviewed]

0 7 0

Christopher Esterhuyse - 5 years ago 2020-05-18 08:59:28
christopher.esterhuyse@gmail.com

introduced new Payload type

7 files changed with 77 insertions and 50 deletions:

src/common.rs

src/protocol/eval.rs

src/protocol/mod.rs

src/runtime/connector.rs

src/runtime/ffi.rs

src/runtime/serde.rs

src/test/connector.rs

0 comments (0 inline, 0 general)

src/common.rs

➞

Show inline comments

@@ @@ -3,52 +3,55 @@ @@
 pub use core::{
     cmp::Ordering,
     fmt::{Debug, Formatter},
     hash::{Hash, Hasher},
     ops::{Range, RangeFrom},
     time::Duration,
 };
 pub use indexmap::{IndexMap, IndexSet};
 pub use maplit::{hashmap, hashset};
 pub use mio::{
     net::{TcpListener, TcpStream},
     Event, Evented, Events, Poll, PollOpt, Ready, Token,
 };
 pub use std::{
     collections::{hash_map::Entry, BTreeMap, HashMap, HashSet},
     convert::TryInto,
     net::SocketAddr,
     sync::Arc,
     time::Instant,
 };
 pub use Polarity::*;
 ///////////////////// DEFS /////////////////////
 pub type Payload = Vec<u8>;
+// pub type Payload = Vec<u8>;
 pub type ControllerId = u32;
 pub type ChannelIndex = u32;
 #[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd)]
 pub struct Payload(Vec<u8>);
 /// This is a unique identifier for a channel (i.e., port).
 #[derive(Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd)]
 pub struct ChannelId {
     pub(crate) controller_id: ControllerId,
     pub(crate) channel_index: ChannelIndex,
+}
 #[derive(Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd)]
 pub enum Polarity {
     Putter, // output port (from the perspective of the component)
     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),
     WrongNumberOfParamaters { expected: usize },
@@ @@ -90,43 +93,71 @@ pub enum MonoBlocker { @@
 pub enum PolyBlocker {
     Inconsistent,
     SyncBlockEnd,
     CouldntReadMsg(Key),
     CouldntCheckFiring(Key),
     PutMsg(Key, 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_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>;
+}
 ///////////////////// IMPL /////////////////////
 impl Payload {
     pub fn new(len: usize) -> Payload {
         let mut v = Vec::with_capacity(len);
         unsafe {
             v.set_len(len);
+        }
         Payload(v)
+    }
     pub fn len(&self) -> usize {
         self.0.len()
+    }
     pub fn as_slice(&self) -> &[u8] {
         &self.0
+    }
     pub fn as_mut_slice(&mut self) -> &mut [u8] {
         &mut self.0
+    }
+}
 impl std::iter::FromIterator<u8> for Payload {
     fn from_iter<I: IntoIterator<Item = u8>>(it: I) -> Self {
         Self(it.into_iter().collect())
+    }
+}
 impl From<Vec<u8>> for Payload {
     fn from(s: Vec<u8>) -> Self {
         Self(s.into())
+    }
+}
 impl Debug for Port {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "Port({})", self.0)
+    }
+}
 impl Key {
     pub fn from_raw(raw: usize) -> Self {
         Self(raw)
+    }
     pub fn to_raw(self) -> usize {
         self.0
+    }
     pub fn to_token(self) -> mio::Token {
         mio::Token(self.0.try_into().unwrap())
+    }
     pub fn from_token(t: mio::Token) -> Self {
         Self(t.0.try_into().unwrap())
+    }
+}

src/protocol/eval.rs

➞

Show inline comments

@@ @@ -27,167 +27,167 @@ trait ValueImpl { @@
     fn exact_type(&self) -> Type;
     fn is_type_compatible(&self, t: &Type) -> bool;
+}
 #[derive(Debug, Clone)]
 pub enum Value {
     Input(InputValue),
     Output(OutputValue),
     Message(MessageValue),
     Boolean(BooleanValue),
     Byte(ByteValue),
     Short(ShortValue),
     Int(IntValue),
     Long(LongValue),
     InputArray(InputArrayValue),
     OutputArray(OutputArrayValue),
     MessageArray(MessageArrayValue),
     BooleanArray(BooleanArrayValue),
     ByteArray(ByteArrayValue),
     ShortArray(ShortArrayValue),
     IntArray(IntArrayValue),
     LongArray(LongArrayValue),
+}
 impl Value {
-    pub fn receive_message(buffer: &Vec<u8>) -> Value {
+    pub fn receive_message(buffer: &Payload) -> Value {
         Value::Message(MessageValue(Some(buffer.clone())))
+    }
     fn create_message(length: Value) -> Value {
         match length {
             Value::Byte(_) | Value::Short(_) | Value::Int(_) | Value::Long(_) => {
                 let length: i64 = i64::from(length);
                 if length < 0 || length > MESSAGE_MAX_LENGTH {
                     // Only messages within the expected length are allowed
                     Value::Message(MessageValue(None))
                 } else {
-                    Value::Message(MessageValue(Some(vec![0; length.try_into().unwrap()])))
+                    Value::Message(MessageValue(Some(Payload::new(0))))
+                }
+            }
             _ => unimplemented!(),
+        }
+    }
     fn from_constant(constant: &Constant) -> Value {
         match constant {
             Constant::Null => Value::Message(MessageValue(None)),
             Constant::True => Value::Boolean(BooleanValue(true)),
             Constant::False => Value::Boolean(BooleanValue(false)),
             Constant::Integer(data) => {
                 // Convert raw ASCII data to UTF-8 string
                 let raw = String::from_utf8_lossy(data);
                 let val = raw.parse::<i64>().unwrap();
                 if val >= BYTE_MIN && val <= BYTE_MAX {
                     Value::Byte(ByteValue(val as i8))
                 } else if val >= SHORT_MIN && val <= SHORT_MAX {
                     Value::Short(ShortValue(val as i16))
                 } else if val >= INT_MIN && val <= INT_MAX {
                     Value::Int(IntValue(val as i32))
                 } else {
                     Value::Long(LongValue(val))
+                }
+            }
             Constant::Character(data) => unimplemented!(),
+        }
+    }
     fn set(&mut self, index: &Value, value: &Value) -> Option<Value> {
         // The index must be of integer type, and non-negative
         let the_index: usize;
         match index {
             Value::Byte(_) | Value::Short(_) | Value::Int(_) | Value::Long(_) => {
                 let index = i64::from(index);
                 if index < 0 || index >= MESSAGE_MAX_LENGTH {
                     // It is inconsistent to update out of bounds
                     return None;
+                }
                 the_index = index.try_into().unwrap();
+            }
             _ => unreachable!(),
+        }
         // The subject must be either a message or an array
         // And the value and the subject must be compatible
         match (self, value) {
             (Value::Message(MessageValue(None)), _) => {
                 // It is inconsistent to update the null message
                 None
+            }
-            (Value::Message(MessageValue(Some(buffer))), Value::Byte(ByteValue(b))) => {
+            (Value::Message(MessageValue(Some(payload))), Value::Byte(ByteValue(b))) => {
                 if *b < 0 {
                     // It is inconsistent to update with a negative value
                     return None;
+                }
-                if let Some(slot) = buffer.get_mut(the_index) {
+                if let Some(slot) = payload.as_mut_slice().get_mut(the_index) {
                     *slot = (*b).try_into().unwrap();
                     Some(value.clone())
                 } else {
                     // It is inconsistent to update out of bounds
                     None
+                }
+            }
-            (Value::Message(MessageValue(Some(buffer))), Value::Short(ShortValue(b))) => {
+            (Value::Message(MessageValue(Some(payload))), Value::Short(ShortValue(b))) => {
                 if *b < 0 || *b > BYTE_MAX as i16 {
                     // It is inconsistent to update with a negative value or a too large value
                     return None;
+                }
-                if let Some(slot) = buffer.get_mut(the_index) {
+                if let Some(slot) = payload.as_mut_slice().get_mut(the_index) {
                     *slot = (*b).try_into().unwrap();
                     Some(value.clone())
                 } else {
                     // It is inconsistent to update out of bounds
                     None
+                }
+            }
             (Value::InputArray(_), Value::Input(_)) => todo!(),
             (Value::OutputArray(_), Value::Output(_)) => todo!(),
             (Value::MessageArray(_), Value::Message(_)) => todo!(),
             (Value::BooleanArray(_), Value::Boolean(_)) => todo!(),
             (Value::ByteArray(_), Value::Byte(_)) => todo!(),
             (Value::ShortArray(_), Value::Short(_)) => todo!(),
             (Value::IntArray(_), Value::Int(_)) => todo!(),
             (Value::LongArray(_), Value::Long(_)) => todo!(),
             _ => unreachable!(),
+        }
+    }
     fn get(&self, index: &Value) -> Option<Value> {
         // The index must be of integer type, and non-negative
         let the_index: usize;
         match index {
             Value::Byte(_) | Value::Short(_) | Value::Int(_) | Value::Long(_) => {
                 let index = i64::from(index);
                 if index < 0 || index >= MESSAGE_MAX_LENGTH {
                     // It is inconsistent to update out of bounds
                     return None;
+                }
                 the_index = index.try_into().unwrap();
+            }
             _ => unreachable!(),
+        }
         // The subject must be either a message or an array
         match self {
             Value::Message(MessageValue(None)) => {
                 // It is inconsistent to read from the null message
                 None
+            }
             Value::Message(MessageValue(Some(buffer))) => {
                 if let Some(slot) = buffer.get(the_index) {
             Value::Message(MessageValue(Some(payload))) => {
                 if let Some(slot) = payload.as_slice().get(the_index) {
                     Some(Value::Short(ShortValue((*slot).try_into().unwrap())))
                 } else {
                     // It is inconsistent to update out of bounds
                     None
+                }
+            }
             Value::InputArray(_) => todo!(),
             Value::OutputArray(_) => todo!(),
             Value::MessageArray(_) => todo!(),
             Value::BooleanArray(_) => todo!(),
             Value::ByteArray(_) => todo!(),
             Value::ShortArray(_) => todo!(),
             Value::IntArray(_) => todo!(),
             Value::LongArray(_) => todo!(),
             _ => unreachable!(),
+        }
+    }
     fn length(&self) -> Option<Value> {
         // The subject must be either a message or an array
         match self {
             Value::Message(MessageValue(None)) => {
                 // It is inconsistent to get length from the null message
                 None
+            }
@@ @@ -915,69 +915,61 @@ pub struct OutputValue(pub Key); @@
 impl Display for OutputValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "#out")
+    }
+}
 impl ValueImpl for OutputValue {
     fn exact_type(&self) -> Type {
         Type::OUTPUT
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Output => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone)]
-pub struct MessageValue(pub Option<Vec<u8>>);
+pub struct MessageValue(pub Option<Payload>);
 impl Display for MessageValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         match &self.0 {
             None => write!(f, "null"),
             Some(vec) => {
                 write!(f, "#msg({};", vec.len())?;
                 let mut i = 0;
                 for v in vec.iter() {
                     if i > 0 {
                         write!(f, ",")?;
+                    }
                     write!(f, "{}", v)?;
                     i += 1;
                     if i >= 10 {
                         write!(f, ",...")?;
                         break;
+                    }
             Some(payload) => {
                 // format print up to 10 bytes
                 let mut slice = payload.as_slice();
                 if slice.len() > 10 {
                     slice = &slice[..10];
+                }
-                write!(f, ")")
+                f.debug_list().entries(slice.iter().copied()).finish()
+            }
+        }
+    }
+}
 impl ValueImpl for MessageValue {
     fn exact_type(&self) -> Type {
         Type::MESSAGE
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Message => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone)]
 pub struct BooleanValue(bool);

src/protocol/mod.rs

➞

Show inline comments

@@ @@ -207,68 +207,72 @@ 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 new_component(&mut self, args: &[Value], init_state: ComponentStateImpl) -> () {
         match self {
             EvalContext::None => unreachable!(),
             EvalContext::Mono(context) => {
                 let mut moved_keys = HashSet::new();
                 for arg in args.iter() {
                     match arg {
                         Value::Output(OutputValue(key)) => { moved_keys.insert(*key); }
                         Value::Input(InputValue(key)) => { moved_keys.insert(*key); }
                         Value::Output(OutputValue(key)) => {
                             moved_keys.insert(*key);
+                        }
                         Value::Input(InputValue(key)) => {
                             moved_keys.insert(*key);
+                        }
                         _ => {}
+                    }
+                }
                 context.new_component(moved_keys, init_state)
+            }
             EvalContext::Poly(_) => unreachable!(),
+        }
+    }
     fn new_channel(&mut self) -> [Value; 2] {
         match self {
             EvalContext::None => unreachable!(),
             EvalContext::Mono(context) => {
                 let [from, to] = context.new_channel();
                 let from = Value::Output(OutputValue(from));
                 let to = Value::Input(InputValue(to));
                 return [from, to];
+            }
             EvalContext::Poly(_) => unreachable!()
+            EvalContext::Poly(_) => unreachable!(),
+        }
+    }
     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),
                 _ => 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::Input(InputValue(key)) => context.read_msg(key).map(Value::receive_message),
                 _ => unreachable!(),
             },

src/runtime/connector.rs

➞

Show inline comments

@@ @@ -161,27 +161,27 @@ impl Connector { @@
             _ => return Err(NotConnected),
         };
         // do the synchronous round!
         let res =
             connected.controller.sync_round(Some(deadline), Some(connected.sync_batches.drain(..)));
         connected.sync_batches.push(SyncBatch::default());
         res?;
         Ok(connected.controller.inner.mono_n.result.as_mut().expect("qqqs").0)
+    }
     pub fn read_gotten(&self, native_port_index: usize) -> Result<&[u8], ReadGottenErr> {
         use ReadGottenErr::*;
         let connected = match self {
             Connector::Connected(connected) => connected,
             _ => return Err(NotConnected),
         };
         let &(key, polarity) =
             connected.native_interface.get(native_port_index).ok_or(IndexOutOfBounds)?;
         if polarity != Getter {
             return Err(WrongPolarity);
+        }
         let result = connected.controller.inner.mono_n.result.as_ref().ok_or(NoPreviousRound)?;
         let payload = result.1.get(&key).ok_or(DidNotGet)?;
         Ok(payload)
+        Ok(payload.as_slice())
+    }
+}

src/runtime/ffi.rs

➞

Show inline comments

@@ @@ -237,51 +237,51 @@ pub unsafe extern "C" fn connector_connect( @@
     Box::into_raw(b); // don't drop!
     ret
+}
 /// Destroys the given connector, freeing its underlying resources.
 /// # Safety
 /// TODO
 #[no_mangle]
 pub unsafe extern "C" fn connector_destroy(connector: *mut Connector) {
     let c = Box::from_raw(connector); // unsafe!
     drop(c); // for readability
+}
 /// Prepares to synchronously put a message at the given port, reading it from the given buffer.
 /// # Safety
 /// TODO
 #[no_mangle]
 pub unsafe extern "C" fn connector_put(
     connector: *mut Connector,
     proto_port_index: c_uint,
     buf_ptr: *mut c_uchar,
     msg_len: c_uint,
 ) -> c_int {
     let buf = std::slice::from_raw_parts_mut(buf_ptr, msg_len.try_into().unwrap());
-    let payload = buf.to_vec(); // unsafe
+    let vec: Vec<u8> = buf.to_vec(); // unsafe
     let mut b = Box::from_raw(connector); // unsafe!
-    let ret = b.put(proto_port_index.try_into().unwrap(), payload);
+    let ret = b.put(proto_port_index.try_into().unwrap(), vec.into());
     Box::into_raw(b); // don't drop!
     match ret {
         Ok(()) => 0,
         Err(e) => {
             overwrite_last_error(format!("{:?}", e).as_bytes());
             -1
+        }
+    }
+}
 /// Prepares to synchronously put a message at the given port, writing it to the given buffer.
 /// - 0 SUCCESS
 /// - 1 this port has the wrong direction
 /// - 2 this port is already marked to get
 /// # Safety
 /// TODO
 #[no_mangle]
 pub unsafe extern "C" fn connector_get(
     connector: *mut Connector,
     proto_port_index: c_uint,
 ) -> c_int {
     let mut b = Box::from_raw(connector); // unsafe!
     let ret = b.get(proto_port_index.try_into().unwrap());
     Box::into_raw(b); // don't drop!

src/runtime/serde.rs

➞

Show inline comments

@@ @@ -98,62 +98,62 @@ impl<W: Write> Ser<u32> for W { @@
     fn ser(&mut self, t: &u32) -> Result<(), std::io::Error> {
         self.write_u32::<BigEndian>(*t)
+    }
+}
 impl<R: Read> De<u32> for R {
     fn de(&mut self) -> Result<u32, std::io::Error> {
         self.read_u32::<BigEndian>()
+    }
+}
 impl<W: Write> Ser<u64> for W {
     fn ser(&mut self, t: &u64) -> Result<(), std::io::Error> {
         self.write_u64::<BigEndian>(*t)
+    }
+}
 impl<R: Read> De<u64> for R {
     fn de(&mut self) -> Result<u64, std::io::Error> {
         self.read_u64::<BigEndian>()
+    }
+}
 impl<W: Write> Ser<Payload> for W {
     fn ser(&mut self, t: &Payload) -> Result<(), std::io::Error> {
         self.ser(&VarLenInt(t.len() as u64))?;
         for byte in t {
+        for byte in t.as_slice() {
             self.ser(byte)?;
+        }
         Ok(())
+    }
+}
 impl<R: Read> De<Payload> for R {
     fn de(&mut self) -> Result<Payload, std::io::Error> {
         let VarLenInt(len) = self.de()?;
         let mut x = Vec::with_capacity(len as usize);
         for _ in 0..len {
             x.push(self.de()?);
+        }
         Ok(x)
+        Ok(x.into())
+    }
+}
 impl<W: Write> Ser<VarLenInt> for W {
     fn ser(&mut self, t: &VarLenInt) -> Result<(), std::io::Error> {
         integer_encoding::VarIntWriter::write_varint(self, t.0).map(|_| ())
+    }
+}
 impl<R: Read> De<VarLenInt> for R {
     fn de(&mut self) -> Result<VarLenInt, std::io::Error> {
         integer_encoding::VarIntReader::read_varint(self).map(VarLenInt)
+    }
+}
 impl<W: Write> Ser<ChannelId> for W {
     fn ser(&mut self, t: &ChannelId) -> Result<(), std::io::Error> {
         self.ser(&t.controller_id)?;
         self.ser(&VarLenInt(t.channel_index as u64))
+    }
+}
 impl<R: Read> De<ChannelId> for R {
     fn de(&mut self) -> Result<ChannelId, std::io::Error> {
         Ok(ChannelId {
             controller_id: self.de()?,

src/test/connector.rs

➞

Show inline comments

@@ @@ -157,49 +157,49 @@ fn connector_connected_but_silent_natives() { @@
+}
 #[test]
 fn connector_self_forward_ok() {
     // Test a deterministic system
     // where a native has no network bindings
     // and sends messages to itself
     /*
         /-->\
     Alice   forward
         \<--/
     */
     let timeout = Duration::from_millis(1_500);
     const N: usize = 5;
     static MSG: &[u8] = b"Echo!";
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"forward").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 x.put(0, MSG.to_vec()).unwrap();
+                x.put(0, MSG.to_vec().into()).unwrap();
                 x.get(1).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(MSG), x.read_gotten(1));
+            }
         },
     ]));
+}
 #[test]
 fn connector_token_spout_ok() {
     // Test a deterministic system where the proto
     // creates token messages
     /*
     Alice<--token_spout
     */
     let timeout = Duration::from_millis(1_500);
     const N: usize = 5;
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"token_spout").unwrap();
             x.bind_port(0, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
@@ @@ -236,235 +236,235 @@ fn connector_waiter_ok() { @@
         },
     ]));
+}
 #[test]
 fn connector_self_forward_timeout() {
     // Test a deterministic system
     // where a native has no network bindings
     // and sends messages to itself
     /*
         /-->\
     Alice   forward
         \<--/
     */
     let timeout = Duration::from_millis(500);
     static MSG: &[u8] = b"Echo!";
     assert!(run_connector_set(&[
         //
         &|x| {
             // Sender
             x.configure(PDL, b"forward").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             x.put(0, MSG.to_vec()).unwrap();
+            x.put(0, MSG.to_vec().into()).unwrap();
             // native and forward components cannot find a solution
             assert_eq!(Err(SyncErr::Timeout), x.sync(timeout));
         },
     ]));
+}
 #[test]
 fn connector_forward_det() {
     // Test if a deterministic protocol and natives can pass one message
     /*
     Alice -->forward--P|A-->forward--> Bob
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr()];
     const N: usize = 5;
     static MSG: &[u8] = b"Hello!";
     assert!(run_connector_set(&[
         &|x| {
             x.configure(PDL, b"forward").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Passive(addrs[0])).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 x.put(0, MSG.to_vec()).unwrap();
+                x.put(0, MSG.to_vec().into()).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
+            }
         },
         &|x| {
             x.configure(PDL, b"forward").unwrap();
             x.bind_port(0, Active(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 x.get(0).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(MSG), x.read_gotten(0));
+            }
         },
     ]));
+}
 #[test]
 fn connector_nondet_proto_det_natives() {
     // Test the use of a nondeterministic protocol
     // where Alice decides the choice and the others conform
     /*
     Alice -->sync--A|P-->sync--> Bob
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr()];
     const N: usize = 5;
     static MSG: &[u8] = b"Message, here!";
     assert!(run_connector_set(&[
         &|x| {
             // Alice
             x.configure(PDL, b"sync").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Active(addrs[0])).unwrap();
             x.connect(timeout).unwrap();
             for _i in 0..N {
                 x.put(0, MSG.to_vec()).unwrap();
+                x.put(0, MSG.to_vec().into()).unwrap();
                 assert_eq!(0, x.sync(timeout).unwrap());
+            }
         },
         &|x| {
             // Bob
             x.configure(PDL, b"sync").unwrap();
             x.bind_port(0, Passive(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _i in 0..N {
                 x.get(0).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(MSG), x.read_gotten(0));
+            }
         },
     ]));
+}
 #[test]
 fn connector_putter_determines() {
     // putter and getter
     /*
     Alice -->sync--A|P-->sync--> Bob
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr()];
     const N: usize = 3;
     static MSG: &[u8] = b"Hidey ho!";
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"sync").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Active(addrs[0])).unwrap();
             x.connect(timeout).unwrap();
             for _i in 0..N {
                 x.put(0, MSG.to_vec()).unwrap();
+                x.put(0, MSG.to_vec().into()).unwrap();
                 assert_eq!(0, x.sync(timeout).unwrap());
+            }
         },
         &|x| {
             // Bob
             x.configure(PDL, b"sync").unwrap();
             x.bind_port(0, Passive(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _i in 0..N {
                 // batches [{0=>*}, {0=>?}]
                 x.get(0).unwrap();
                 x.next_batch().unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(MSG), x.read_gotten(0));
+            }
         },
     ]));
+}
 #[test]
 fn connector_getter_determines() {
     // putter and getter
     /*
     Alice -->sync--A|P-->sync--> Bob
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr()];
     const N: usize = 5;
     static MSG: &[u8] = b"Hidey ho!";
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"sync").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Active(addrs[0])).unwrap();
             x.connect(timeout).unwrap();
             for _i in 0..N {
                 // batches [{0=>?}, {0=>*}]
                 x.put(0, MSG.to_vec()).unwrap();
+                x.put(0, MSG.to_vec().into()).unwrap();
                 x.next_batch().unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
+            }
         },
         &|x| {
             // Bob
             x.configure(PDL, b"sync").unwrap();
             x.bind_port(0, Passive(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _i in 0..N {
                 x.get(0).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(MSG), x.read_gotten(0));
+            }
         },
     ]));
+}
 #[test]
 fn connector_alternator_2() {
     // Test a deterministic system which
     // alternates sending Sender's messages to A or B
     /*                    /--|-->A
     Sender -->alternator_2
                           \--|-->B
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr(), next_addr()];
     const N: usize = 5;
     static MSG: &[u8] = b"message";
     assert!(run_connector_set(&[
         //
         &|x| {
             // Sender
             x.configure(PDL, b"alternator_2").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Passive(addrs[0])).unwrap();
             x.bind_port(2, Passive(addrs[1])).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 for _ in 0..2 {
                     x.put(0, MSG.to_vec()).unwrap();
+                    x.put(0, MSG.to_vec().into()).unwrap();
                     assert_eq!(0, x.sync(timeout).unwrap());
+                }
+            }
         },
         &|x| {
             // A
             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 {
                 // get msg round
                 x.get(0).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout)); // GET ONE
                 assert_eq!(Ok(MSG), x.read_gotten(0));
                 // silent round
                 assert_eq!(Ok(0), x.sync(timeout)); // MISS ONE
                 assert_eq!(Err(ReadGottenErr::DidNotGet), x.read_gotten(0));
+            }
         },
         &|x| {
             // B
             x.configure(PDL, b"sync").unwrap();
@@ @@ -484,174 +484,174 @@ fn connector_alternator_2() { @@
+            }
         },
     ]));
+}
 #[test]
 fn connector_composite_chain_a() {
     // Check if composition works. Forward messages through long chains
     /*
     Alice -->sync-->sync-->A|P-->sync--> Bob
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr(), next_addr()];
     const N: usize = 1;
     static MSG: &[u8] = b"SSS";
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"sync_2").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Active(addrs[0])).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 x.put(0, MSG.to_vec()).unwrap();
+                x.put(0, MSG.to_vec().into()).unwrap();
                 assert_eq!(0, x.sync(timeout).unwrap());
+            }
         },
         &|x| {
             // Bob
             x.configure(PDL, b"forward").unwrap();
             x.bind_port(0, Passive(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 // get msg round
                 x.get(0).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(MSG), x.read_gotten(0));
+            }
         },
     ]));
+}
 #[test]
 fn connector_composite_chain_b() {
     // Check if composition works. Forward messages through long chains
     /*
     Alice -->sync-->sync-->A|P-->sync-->sync--> Bob
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr(), next_addr()];
     const N: usize = 1;
     static MSG: &[u8] = b"SSS";
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"sync_2").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Active(addrs[0])).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 x.put(0, MSG.to_vec()).unwrap();
+                x.put(0, MSG.to_vec().into()).unwrap();
                 assert_eq!(0, x.sync(timeout).unwrap());
+            }
         },
         &|x| {
             // Bob
             x.configure(PDL, b"sync_2").unwrap();
             x.bind_port(0, Passive(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 // get msg round
                 x.get(0).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(MSG), x.read_gotten(0));
+            }
         },
     ]));
+}
 #[test]
 fn connector_exchange() {
     /*
         /-->\      /-->P|A-->\      /-->\
     Alice   exchange         exchange   Bob
         \<--/      \<--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.put(0, b"A->B".to_vec().into()));
                 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.put(0, b"B->A".to_vec().into()));
                 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 connector_both() {
     /* ------->   -----P|A---->   ------->
       / /--->\ \ / /---P|A-->\ \ / /--->\ \
     Alice    exchange       exchange     Bob
     */
     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 a
             x.bind_port(1, Passive(addrs[0])).unwrap(); // peer out a
             x.bind_port(2, Native).unwrap(); // native in b
             x.bind_port(3, Passive(addrs[1])).unwrap(); // peer out b
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 assert_eq!(Ok(()), x.put(0, b"one".to_vec()));
                 assert_eq!(Ok(()), x.put(1, b"two".to_vec()));
                 assert_eq!(Ok(()), x.put(0, b"one".to_vec().into()));
                 assert_eq!(Ok(()), x.put(1, b"two".to_vec().into()));
                 assert_eq!(Ok(0), x.sync(timeout));
+            }
         },
         &|x| {
             // Alice
             x.configure(PDL, b"exchange").unwrap();
             x.bind_port(0, Active(addrs[0])).unwrap(); // peer in a
             x.bind_port(1, Native).unwrap(); // native out b
             x.bind_port(2, Active(addrs[1])).unwrap(); // peer in b
             x.bind_port(3, Native).unwrap(); // native out a
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 assert_eq!(Ok(()), x.get(0));
                 assert_eq!(Ok(()), x.get(1));
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(b"one" as &[u8]), x.read_gotten(0));
                 assert_eq!(Ok(b"two" as &[u8]), x.read_gotten(1));
+            }
         },
     ]));
+}
 #[test]
 fn connector_routing_filter() {
@@ @@ -713,164 +713,164 @@ fn connector_routing_filter() { @@
         },
     ]));
+}
 #[test]
 fn connector_fifo_1_e() {
     /*
         /-->\
     Alice   fifo_1
         \<--/
     */
     let timeout = Duration::from_millis(1_500);
     const N: usize = 10;
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"fifo_1_e").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 // put
                 assert_eq!(Ok(()), x.put(0, b"message~".to_vec()));
+                assert_eq!(Ok(()), x.put(0, b"message~".to_vec().into()));
                 assert_eq!(Ok(0), x.sync(timeout));
                 // get
                 assert_eq!(Ok(()), x.get(1));
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(b"message~" as &[u8]), x.read_gotten(1));
+            }
         },
     ]));
+}
 #[test]
 #[should_panic]
 fn connector_causal_loop() {
     /*
         /-->\      /-->P|A-->\      /-->\
     Alice   exchange         exchange   Bob
         \<--/      \<--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, Passive(addrs[0])).unwrap(); // peer out
             x.bind_port(1, Passive(addrs[1])).unwrap(); // peer in
             x.bind_port(2, Native).unwrap(); // native in
             x.bind_port(3, Native).unwrap(); // native out
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 assert_eq!(Ok(()), x.put(0, b"A->B".to_vec()));
+                assert_eq!(Ok(()), x.put(0, b"A->B".to_vec().into()));
                 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, Active(addrs[1])).unwrap(); // peer out
             x.bind_port(1, Active(addrs[0])).unwrap(); // peer in
             x.bind_port(2, Native).unwrap(); // native in
             x.bind_port(3, Native).unwrap(); // native out
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 assert_eq!(Ok(()), x.put(0, b"B->A".to_vec()));
+                assert_eq!(Ok(()), x.put(0, b"B->A".to_vec().into()));
                 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]
 #[should_panic]
 fn connector_causal_loop2() {
     /*
         /-->\     /<---\
     Alice   samelen-->repl
         \<-------------/
     */
     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"samelen_repl").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 assert_eq!(Ok(()), x.put(0, b"foo".to_vec()));
+                assert_eq!(Ok(()), x.put(0, b"foo".to_vec().into()));
                 assert_eq!(Ok(()), x.get(1));
                 assert_eq!(Ok(0), x.sync(timeout));
+            }
         },
     ]));
+}
 #[test]
 fn connector_recover() {
     let connect_timeout = Duration::from_millis(1500);
     let comm_timeout = Duration::from_millis(300);
     let addrs = [next_addr()];
     fn putter_does(i: usize) -> bool {
         i % 3 == 0
+    }
     fn getter_does(i: usize) -> bool {
         i % 2 == 0
+    }
     fn expect_res(i: usize) -> Result<usize, SyncErr> {
         if putter_does(i) && getter_does(i) {
             Ok(0)
         } else {
             Err(SyncErr::Timeout)
+        }
+    }
     const N: usize = 11;
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"forward").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Passive(addrs[0])).unwrap();
             x.connect(connect_timeout).unwrap();
             for i in 0..N {
                 if putter_does(i) {
                     assert_eq!(Ok(()), x.put(0, b"msg".to_vec()));
+                    assert_eq!(Ok(()), x.put(0, b"msg".to_vec().into()));
+                }
                 assert_eq!(expect_res(i), x.sync(comm_timeout));
+            }
         },
         &|x| {
             // Bob
             x.configure(PDL, b"forward").unwrap();
             x.bind_port(0, Active(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(connect_timeout).unwrap();
             for i in 0..N {
                 if getter_does(i) {
                     assert_eq!(Ok(()), x.get(0));
+                }
                 assert_eq!(expect_res(i), x.sync(comm_timeout));
                 if expect_res(i).is_ok() {
                     assert_eq!(Ok(b"msg" as &[u8]), x.read_gotten(0));
+                }
+            }
         },
     ]));
+}

0 comments (0 inline, 0 general)