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

@@ @@ -83,13 +83,13 @@ impl Binds<OutPort> for Connecting { @@
 #[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,
+}
@@ @@ -237,19 +237,19 @@ impl Connecting { @@
                         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);
 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)