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

Changeset - c3cc95b4ef76

Parent rev.

Child rev.

[Not reviewed]

0 1 0

Christopher Esterhuyse - 5 years ago 2020-02-21 13:08:01
christopher.esterhuyse@gmail.com

connecting nice

1 file changed with 18 insertions and 8 deletions:

src/runtime/experimental/api.rs

0 comments (0 inline, 0 general)

src/runtime/experimental/api.rs

➞

Show inline comments

@@ @@ -77,25 +77,25 @@ impl Binds<OutPort> for Connecting { @@
     fn bind(&mut self, coupling: Coupling, addr: SocketAddr) -> OutPort {
         self.bindings.push(Binding { coupling, polarity: Polarity::Putter, addr });
         OutPort(Port(self.bindings.len() - 1))
+    }
+}
 #[derive(Debug, Clone)]
 pub enum ConnectErr {
     BindErr(SocketAddr),
     NewSocketErr(SocketAddr),
     AcceptErr(SocketAddr),
     ConnectionShutdown(SocketAddr),
     PortKindMismatch(SocketAddr),
+    PortKindMismatch(Port, SocketAddr),
     EndpointErr(Port, EndpointErr),
     PollInitFailed,
     PollingFailed,
     Timeout,
+}
 #[derive(Debug)]
 struct Component {
     protocol: Arc<ProtocolD>,
     port_set: HashSet<Port>,
     identifier: Arc<[u8]>,
     state: ProtocolS,
@@ @@ -231,31 +231,31 @@ impl Connecting { @@
+                        }
                         ms.poll.reregister(&stream, token, ready_r, edge).expect("55");
                         let polarity = binding.polarity;
                         let info = EndpointInfo { polarity, channel_id };
                         let msg = Msg::SetupMsg(SetupMsg::ChannelSetup { info });
                         let mut endpoint = Endpoint::from_fresh_stream(stream);
                         endpoint.send(msg).map_err(|e| EndpointErr(Port(index), e))?;
                         let endpoint_ext = EndpointExt { endpoint, info };
                         endpoint_exts.occupy_reserved(index, endpoint_ext);
                         num_todos_remaining -= 1;
+                    }
                     Some(Todo::ActiveRecving { mut endpoint }) => {
-                        let ekey = Key::from_raw(index);
+                        let ekey = Port(index);
                         'recv_loop: while let Some(msg) =
-                            endpoint.recv().map_err(|e| EndpointErr(Port(index), e))?
+                            endpoint.recv().map_err(|e| EndpointErr(ekey, e))?
+                        {
                             if let Msg::SetupMsg(SetupMsg::ChannelSetup { info }) = msg {
                                 if info.polarity == binding.polarity {
                                     return Err(PortKindMismatch(binding.addr));
+                                    return Err(PortKindMismatch(ekey, binding.addr));
+                                }
                                 let channel_id = info.channel_id;
                                 let info = EndpointInfo { polarity: binding.polarity, channel_id };
                                 ms.polled_undrained.insert(ekey);
                                 let endpoint_ext = EndpointExt { endpoint, info };
                                 endpoint_exts.occupy_reserved(index, endpoint_ext);
                                 num_todos_remaining -= 1;
                                 break 'recv_loop;
                             } else {
                                 ms.delayed.push(ReceivedMsg { recipient: ekey, msg });
+                            }
+                        }
+}
 unsafe fn as_const_slice<'a, T>(len: usize, ptr: *const T) -> &'a [T] {
     std::slice::from_raw_parts(ptr, len)
+}
 #[test]
 fn api_connecting() {
     let addrs: [SocketAddr; 3] = [
         "127.0.0.1:8888".parse().unwrap(),
         "127.0.0.1:8889".parse().unwrap(),
         "127.0.0.1:8890".parse().unwrap(),
     ];
     const TIMEOUT: Option<Duration> = Some(Duration::from_secs(1));
     let handles = vec![
         std::thread::spawn(move || {
             let mut connecting = Connecting::default();
             let _a: OutPort = connecting.bind(Coupling::Active, addrs[0]);
             let connected = connecting.connect(None);
             let _a: OutPort = connecting.bind(Coupling::Passive, addrs[0]);
             let _b: OutPort = connecting.bind(Coupling::Active, addrs[1]);
             let connected = connecting.connect(TIMEOUT);
             println!("A: {:#?}", connected);
         }),
         std::thread::spawn(move || {
             let mut connecting = Connecting::default();
             let _a: OutPort = connecting.bind(Coupling::Passive, addrs[0]);
             let connected = connecting.connect(Some(Duration::from_secs(2)));
             let _a: InPort = connecting.bind(Coupling::Active, addrs[0]);
             let _b: InPort = connecting.bind(Coupling::Passive, addrs[1]);
             let _c: InPort = connecting.bind(Coupling::Active, addrs[2]);
             let connected = connecting.connect(TIMEOUT);
             println!("B: {:#?}", connected);
         }),
         std::thread::spawn(move || {
             let mut connecting = Connecting::default();
             let _a: OutPort = connecting.bind(Coupling::Passive, addrs[2]);
             let connected = connecting.connect(TIMEOUT);
             println!("C: {:#?}", connected);
         }),
     ];
     for h in handles {
         h.join().unwrap();
+    }
+}

0 comments (0 inline, 0 general)