Merge pull request #857 from hyperium/client-timeout-panic

fix(client): handle when DNS resolves after a timeout triggers

examples/client.rs

@@ -48,7 +48,7 @@ impl hyper::client::Handler<HttpStream> for Dump {
             Err(e) => match e.kind() {
                 io::ErrorKind::WouldBlock => Next::read(),
                 _ => {
-                    println!("ERROR: {}", e);
+                    println!("ERROR:example: {}", e);
                     Next::end()
                 }
             }
@@ -56,7 +56,7 @@ impl hyper::client::Handler<HttpStream> for Dump {
     }
 
     fn on_error(&mut self, err: hyper::Error) -> Next {
-        println!("ERROR: {}", err);
+        println!("ERROR:example: {}", err);
         Next::remove()
     }
 }

src/client/connect.rs

@@ -16,7 +16,7 @@ pub trait Connect {
     /// Type of Transport to create
     type Output: Transport;
     /// The key used to determine if an existing socket can be used.
-    type Key: Eq + Hash + Clone;
+    type Key: Eq + Hash + Clone + fmt::Debug;
     /// Returns the key based off the Url.
     fn key(&self, &Url) -> Option<Self::Key>;
     /// Connect to a remote address.
@@ -96,10 +96,12 @@ impl Connect for HttpConnector {
     }
 
     fn connected(&mut self) -> Option<(Self::Key, io::Result<HttpStream>)> {
-        let (host, addr) = match self.dns.as_ref().expect("dns workers lost").resolved() {
+        let (host, addrs) = match self.dns.as_ref().expect("dns workers lost").resolved() {
             Ok(res) => res,
             Err(_) => return None
         };
+        //TODO: try all addrs
+        let addr = addrs.and_then(|mut addrs| Ok(addrs.next().unwrap()));
         debug!("Http::resolved <- ({:?}, {:?})", host, addr);
         if let Entry::Occupied(mut entry) = self.resolving.entry(host) {
             let resolved = entry.get_mut().remove(0);

src/client/dns.rs

@@ -1,6 +1,7 @@
 use std::io;
-use std::net::{IpAddr, ToSocketAddrs};
+use std::net::{IpAddr, SocketAddr, ToSocketAddrs};
 use std::thread;
+use std::vec;
 
 use ::spmc;
 
@@ -11,7 +12,19 @@ pub struct Dns {
     rx: channel::Receiver<Answer>,
 }
 
-pub type Answer = (String, io::Result<IpAddr>);
+pub type Answer = (String, io::Result<IpAddrs>);
+
+pub struct IpAddrs {
+    iter: vec::IntoIter<SocketAddr>,
+}
+
+impl Iterator for IpAddrs {
+    type Item = IpAddr;
+    #[inline]
+    fn next(&mut self) -> Option<IpAddr> {
+        self.iter.next().map(|addr| addr.ip())
+    }
+}
 
 impl Dns {
     pub fn new(notify: (channel::Sender<Answer>, channel::Receiver<Answer>), threads: usize) -> Dns {
@@ -26,7 +39,7 @@ impl Dns {
     }
 
     pub fn resolve<T: Into<String>>(&self, hostname: T) {
-        self.tx.send(hostname.into()).expect("Workers all died horribly");
+        self.tx.send(hostname.into()).expect("DNS workers all died unexpectedly");
     }
 
     pub fn resolved(&self) -> Result<Answer, channel::TryRecvError> {
@@ -41,9 +54,8 @@ fn work(rx: spmc::Receiver<String>, notify: channel::Sender<Answer>) {
         let notify = worker.notify.as_ref().expect("Worker lost notify");
         while let Ok(host) = rx.recv() {
             debug!("resolve {:?}", host);
-            let res = match (&*host, 80).to_socket_addrs().map(|mut i| i.next()) {
+            let res = match (&*host, 80).to_socket_addrs().map(|i| IpAddrs{ iter: i }) {
-                Ok(Some(addr)) => (host, Ok(addr.ip())),
+                Ok(addrs) => (host, Ok(addrs)),
-                Ok(None) => (host, Err(io::Error::new(io::ErrorKind::Other, "no addresses found"))),
                 Err(e) => (host, Err(e))
             };
 

src/client/mod.rs

@@ -30,7 +30,6 @@ mod response;
 
 /// A Client to make outgoing HTTP requests.
 pub struct Client<H> {
-    //handle: Option<thread::JoinHandle<()>>,
     tx: http::channel::Sender<Notify<H>>,
 }
 
@@ -64,16 +63,6 @@ impl<H> Client<H> {
     pub fn configure() -> Config<DefaultConnector> {
         Config::default()
     }
-
-    /*TODO
-    pub fn http() -> Config<HttpConnector> {
-
-    }
-
-    pub fn https() -> Config<HttpsConnector> {
-
-    }
-    */
 }
 
 impl<H: Handler<<DefaultConnector as Connect>::Output>> Client<H> {
@@ -243,14 +232,6 @@ impl<H> fmt::Display for ClientError<H> {
     }
 }
 
-/*
-impl Drop for Client {
-    fn drop(&mut self) {
-        self.handle.take().map(|handle| handle.join());
-    }
-}
-*/
-
 /// A trait to react to client events that happen for each message.
 ///
 /// Each event handler returns it's desired `Next` action.
@@ -338,15 +319,16 @@ struct Context<K, H> {
 }
 
 impl<K: http::Key, H> Context<K, H> {
-    fn pop_queue(&mut self, key: &K) -> Queued<H> {
+    fn pop_queue(&mut self, key: &K) -> Option<Queued<H>> {
         let mut should_remove = false;
         let queued = {
-            let mut vec = self.queue.get_mut(key).expect("handler not in queue for key");
+            self.queue.get_mut(key).map(|vec| {
-            let queued = vec.remove(0);
+                let queued = vec.remove(0);
-            if vec.is_empty() {
+                if vec.is_empty() {
-                should_remove = true;
+                    should_remove = true;
-            }
+                }
-            queued
+                queued
+            })
         };
         if should_remove {
             self.queue.remove(key);
@@ -379,16 +361,22 @@ impl<K: http::Key, H> Context<K, H> {
 impl<K: http::Key, H: Handler<T>, T: Transport> http::MessageHandlerFactory<K, T> for Context<K, H> {
     type Output = Message<H, T>;
 
-    fn create(&mut self, seed: http::Seed<K>) -> Self::Output {
+    fn create(&mut self, seed: http::Seed<K>) -> Option<Self::Output> {
         let key = seed.key();
-        let queued = self.pop_queue(key);
+        self.pop_queue(key).map(|queued| {
-        let (url, mut handler) = (queued.url, queued.handler);
+            let (url, mut handler) = (queued.url, queued.handler);
-        handler.on_control(seed.control());
+            handler.on_control(seed.control());
-        Message {
+
-            handler: handler,
+            Message {
-            url: Some(url),
+                handler: handler,
-            _marker: PhantomData,
+                url: Some(url),
-        }
+                _marker: PhantomData,
+            }
+        })
+    }
+
+    fn keep_alive_interest(&self) -> Next {
+        Next::wait()
     }
 }
 
@@ -402,6 +390,7 @@ where C: Connect,
       C::Output: Transport,
       H: Handler<C::Output> {
     Connector(C, http::channel::Receiver<Notify<H>>),
+    Connecting((C::Key, C::Output)),
     Socket(http::Conn<C::Key, C::Output, Message<H, C::Output>>)
 }
 
@@ -415,6 +404,7 @@ where
 
 impl<C, H> rotor::Machine for ClientFsm<C, H>
 where C: Connect,
+      C::Key: fmt::Debug,
       C::Output: Transport,
       H: Handler<C::Output> {
     type Context = Context<C::Key, H>;
@@ -422,24 +412,47 @@ where C: Connect,
 
     fn create(seed: Self::Seed, scope: &mut Scope<Self::Context>) -> rotor::Response<Self, rotor::Void> {
         rotor_try!(scope.register(&seed.1, EventSet::writable(), PollOpt::level()));
-        rotor::Response::ok(
+        rotor::Response::ok(ClientFsm::Connecting(seed))
-            ClientFsm::Socket(
-                http::Conn::new(seed.0, seed.1, scope.notifier())
-                    .keep_alive(scope.keep_alive)
-            )
-        )
     }
 
     fn ready(self, events: EventSet, scope: &mut Scope<Self::Context>) -> rotor::Response<Self, Self::Seed> {
         match self {
-            ClientFsm::Connector(..) => {
-                unreachable!("Connector can never be ready")
-            },
             ClientFsm::Socket(conn) => {
                 let res = conn.ready(events, scope);
                 let now = scope.now();
                 scope.conn_response(res, now)
+            },
+            ClientFsm::Connecting(mut seed) => {
+                if events.is_error() || events.is_hup() {
+                    if let Some(err) = seed.1.take_socket_error().err() {
+                        debug!("error while connecting: {:?}", err);
+                        scope.pop_queue(&seed.0).map(move |mut queued| queued.handler.on_error(::Error::Io(err)));
+                        rotor::Response::done()
+                    } else {
+                        trace!("connecting is_error, but no socket error");
+                        rotor::Response::ok(ClientFsm::Connecting(seed))
+                    }
+                } else if events.is_writable() {
+                    if scope.queue.contains_key(&seed.0) {
+                        trace!("connected and writable {:?}", seed.0);
+                        rotor::Response::ok(
+                            ClientFsm::Socket(
+                                http::Conn::new(seed.0, seed.1, Next::write().timeout(scope.connect_timeout), scope.notifier())
+                                    .keep_alive(scope.keep_alive)
+                            )
+                        )
+                    } else {
+                        trace!("connected, but queued handler is gone: {:?}", seed.0); // probably took too long connecting
+                        rotor::Response::done()
+                    }
+                } else {
+                    // spurious?
+                    rotor::Response::ok(ClientFsm::Connecting(seed))
+                }
             }
+            ClientFsm::Connector(..) => {
+                unreachable!("Connector can never be ready")
+            },
         }
     }
 
@@ -477,6 +490,7 @@ where C: Connect,
                     None => rotor::Response::ok(self)
                 }
             }
+            ClientFsm::Connecting(..) => unreachable!(),
             ClientFsm::Socket(conn) => {
                 let res = conn.timeout(scope);
                 let now = scope.now();
@@ -494,13 +508,15 @@ where C: Connect,
                 let res = conn.wakeup(scope);
                 let now = scope.now();
                 scope.conn_response(res, now)
-            }
+            },
+            ClientFsm::Connecting(..) => unreachable!("connecting sockets should not be woken up")
         }
     }
 }
 
 impl<C, H> ClientFsm<C, H>
 where C: Connect,
+      C::Key: fmt::Debug,
       C::Output: Transport,
       H: Handler<C::Output> {
     fn connect(self, scope: &mut rotor::Scope<<Self as rotor::Machine>::Context>) -> rotor::Response<Self, <Self as rotor::Machine>::Seed> {
@@ -509,13 +525,12 @@ where C: Connect,
                 if let Some((key, res)) = connector.connected() {
                     match res {
                         Ok(socket) => {
-                            trace!("connected");
+                            trace!("connecting {:?}", key);
                             return rotor::Response::spawn(ClientFsm::Connector(connector, rx), (key, socket));
                         },
                         Err(e) => {
-                            trace!("connected error = {:?}", e);
+                            trace!("connect error = {:?}", e);
-                            let mut queued = scope.pop_queue(&key);
+                            scope.pop_queue(&key).map(|mut queued| queued.handler.on_error(::Error::Io(e)));
-                            let _ = queued.handler.on_error(::Error::Io(e));
                         }
                     }
                 }

src/http/conn.rs

@@ -163,7 +163,10 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> ConnInner<K, T, H> {
                     Ok(decoder) => {
                         trace!("decoder = {:?}", decoder);
                         let keep_alive = self.keep_alive_enabled && head.should_keep_alive();
-                        let mut handler = scope.create(Seed(&self.key, &self.ctrl.0));
+                        let mut handler = match scope.create(Seed(&self.key, &self.ctrl.0)) {
+                            Some(handler) => handler,
+                            None => unreachable!()
+                        };
                         let next = handler.on_incoming(head, &self.transport);
                         trace!("handler.on_incoming() -> {:?}", next);
 
@@ -276,7 +279,7 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> ConnInner<K, T, H> {
                 };
                 let mut s = State::Http1(http1);
                 if let Some(next) = next {
-                    s.update(next);
+                    s.update(next, &**scope);
                 }
                 trace!("Conn.on_readable State::Http1 completed, new state = State::{:?}", s);
 
@@ -304,7 +307,13 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> ConnInner<K, T, H> {
                 // this could be a Client request, which writes first, so pay
                 // attention to the version written here, which will adjust
                 // our internal state to Http1 or Http2
-                let mut handler = scope.create(Seed(&self.key, &self.ctrl.0));
+                let mut handler = match scope.create(Seed(&self.key, &self.ctrl.0)) {
+                    Some(handler) => handler,
+                    None => {
+                        trace!("could not create handler {:?}", self.key);
+                        return State::Closed;
+                    }
+                };
                 let mut head = http::MessageHead::default();
                 let interest = handler.on_outgoing(&mut head);
                 if head.version == HttpVersion::Http11 {
@@ -427,7 +436,7 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> ConnInner<K, T, H> {
         };
 
         if let Some(next) = next {
-            state.update(next);
+            state.update(next, &**scope);
         }
         state
     }
@@ -439,7 +448,7 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> ConnInner<K, T, H> {
         }
     }
 
-    fn on_error(&mut self, err: ::Error) {
+    fn on_error<F>(&mut self, err: ::Error, factory: &F) where F: MessageHandlerFactory<K, T> {
         debug!("on_error err = {:?}", err);
         trace!("on_error state = {:?}", self.state);
         let next = match self.state {
@@ -447,7 +456,7 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> ConnInner<K, T, H> {
             State::Http1(ref mut http1) => http1.handler.on_error(err),
             State::Closed => Next::remove(),
         };
-        self.state.update(next);
+        self.state.update(next, factory);
     }
 
     fn on_readable<F>(&mut self, scope: &mut Scope<F>)
@@ -477,15 +486,15 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> ConnInner<K, T, H> {
 }
 
 impl<K: Key, T: Transport, H: MessageHandler<T>> Conn<K, T, H> {
-    pub fn new(key: K, transport: T, notify: rotor::Notifier) -> Conn<K, T, H> {
+    pub fn new(key: K, transport: T, next: Next, notify: rotor::Notifier) -> Conn<K, T, H> {
         Conn(Box::new(ConnInner {
             buf: Buffer::new(),
             ctrl: channel::new(notify),
             keep_alive_enabled: true,
             key: key,
             state: State::Init {
-                interest: H::Message::initial_interest().interest,
+                interest: next.interest,
-                timeout: None,
+                timeout: next.timeout,
             },
             transport: transport,
         }))
@@ -506,7 +515,7 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> Conn<K, T, H> {
                     trace!("is_error, but not socket error");
                     // spurious?
                 },
-                Err(e) => self.0.on_error(e.into())
+                Err(e) => self.0.on_error(e.into(), &**scope)
             }
         }
 
@@ -570,7 +579,7 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> Conn<K, T, H> {
             },
             Err(e) => {
                 trace!("error reregistering: {:?}", e);
-                self.0.on_error(e.into());
+                self.0.on_error(e.into(), &**scope);
                 None
             }
         }
@@ -580,7 +589,7 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> Conn<K, T, H> {
     where F: MessageHandlerFactory<K, T, Output=H> {
         while let Ok(next) = self.0.ctrl.1.try_recv() {
             trace!("woke up with {:?}", next);
-            self.0.state.update(next);
+            self.0.state.update(next, &**scope);
         }
         self.ready(EventSet::readable() | EventSet::writable(), scope)
     }
@@ -588,7 +597,7 @@ impl<K: Key, T: Transport, H: MessageHandler<T>> Conn<K, T, H> {
     pub fn timeout<F>(mut self, scope: &mut Scope<F>) -> Option<(Self, Option<Duration>)>
     where F: MessageHandlerFactory<K, T, Output=H> {
         //TODO: check if this was a spurious timeout?
-        self.0.on_error(::Error::Timeout);
+        self.0.on_error(::Error::Timeout, &**scope);
         self.ready(EventSet::none(), scope)
     }
 
@@ -660,7 +669,7 @@ impl<H: MessageHandler<T>, T: Transport> fmt::Debug for State<H, T> {
 }
 
 impl<H: MessageHandler<T>, T: Transport> State<H, T> {
-    fn update(&mut self, next: Next) {
+    fn update<F, K>(&mut self, next: Next, factory: &F) where F: MessageHandlerFactory<K, T>, K: Key {
         let timeout = next.timeout;
         let state = mem::replace(self, State::Closed);
         let new_state = match (state, next.interest) {
@@ -730,10 +739,10 @@ impl<H: MessageHandler<T>, T: Transport> State<H, T> {
                 };
                 match (reading, writing) {
                     (Reading::KeepAlive, Writing::KeepAlive) => {
-                        //XXX keepalive
+                        let next = factory.keep_alive_interest();
                         State::Init {
-                        interest: H::Message::keep_alive_interest().interest,
+                            interest: next.interest,
-                            timeout: None,
+                            timeout: next.timeout,
                         }
                     },
                     (reading, Writing::Chunk(chunk)) => {
@@ -848,6 +857,10 @@ impl<H: MessageHandler<T>, T: Transport> State<H, T> {
             }
         };
         let new_state = match new_state {
+            State::Init { interest, .. } => State::Init {
+                timeout: timeout,
+                interest: interest,
+            },
             State::Http1(mut http1) => {
                 http1.timeout = timeout;
                 State::Http1(http1)
@@ -943,22 +956,13 @@ impl<'a, K: Key + 'a> Seed<'a, K> {
 pub trait MessageHandlerFactory<K: Key, T: Transport> {
     type Output: MessageHandler<T>;
 
-    fn create(&mut self, seed: Seed<K>) -> Self::Output;
+    fn create(&mut self, seed: Seed<K>) -> Option<Self::Output>;
+
+    fn keep_alive_interest(&self) -> Next;
 }
 
-impl<F, K, H, T> MessageHandlerFactory<K, T> for F
+pub trait Key: Eq + Hash + Clone + fmt::Debug {}
-where F: FnMut(Seed<K>) -> H,
+impl<T: Eq + Hash + Clone + fmt::Debug> Key for T {}
-      K: Key,
-      H: MessageHandler<T>,
-      T: Transport {
-    type Output = H;
-    fn create(&mut self, seed: Seed<K>) -> H {
-        self(seed)
-    }
-}
-
-pub trait Key: Eq + Hash + Clone {}
-impl<T: Eq + Hash + Clone> Key for T {}
 
 #[cfg(test)]
 mod tests {

src/http/h1/parse.rs

@@ -4,7 +4,7 @@ use std::io::Write;
 use httparse;
 
 use header::{self, Headers, ContentLength, TransferEncoding};
-use http::{MessageHead, RawStatus, Http1Message, ParseResult, Next, ServerMessage, ClientMessage, Next_, RequestLine};
+use http::{MessageHead, RawStatus, Http1Message, ParseResult, ServerMessage, ClientMessage, RequestLine};
 use http::h1::{Encoder, Decoder};
 use method::Method;
 use status::StatusCode;
@@ -27,14 +27,6 @@ impl Http1Message for ServerMessage {
     type Incoming = RequestLine;
     type Outgoing = StatusCode;
 
-    fn initial_interest() -> Next {
-        Next::new(Next_::Read)
-    }
-
-    fn keep_alive_interest() -> Next {
-        Next::new(Next_::Read)
-    }
-
     fn parse(buf: &[u8]) -> ParseResult<RequestLine> {
         let mut headers = [httparse::EMPTY_HEADER; MAX_HEADERS];
         trace!("Request.parse([Header; {}], [u8; {}])", headers.len(), buf.len());
@@ -113,15 +105,6 @@ impl Http1Message for ClientMessage {
     type Incoming = RawStatus;
     type Outgoing = RequestLine;
 
-
-    fn initial_interest() -> Next {
-        Next::new(Next_::Write)
-    }
-
-    fn keep_alive_interest() -> Next {
-        Next::new(Next_::Wait)
-    }
-
     fn parse(buf: &[u8]) -> ParseResult<RawStatus> {
         let mut headers = [httparse::EMPTY_HEADER; MAX_HEADERS];
         trace!("Response.parse([Header; {}], [u8; {}])", headers.len(), buf.len());

src/http/mod.rs

@@ -280,13 +280,9 @@ pub enum ClientMessage {}
 pub trait Http1Message {
     type Incoming;
     type Outgoing: Default;
-    //TODO: replace with associated const when stable
-    fn initial_interest() -> Next;
-    fn keep_alive_interest() -> Next;
     fn parse(bytes: &[u8]) -> ParseResult<Self::Incoming>;
     fn decoder(head: &MessageHead<Self::Incoming>) -> ::Result<h1::Decoder>;
     fn encode(head: MessageHead<Self::Outgoing>, dst: &mut Vec<u8>) -> h1::Encoder;
-
 }
 
 /// Used to signal desired events when working with asynchronous IO.

src/server/mod.rs

@@ -167,8 +167,12 @@ struct Context<F> {
 impl<F: HandlerFactory<T>, T: Transport> http::MessageHandlerFactory<(), T> for Context<F> {
     type Output = message::Message<F::Output, T>;
 
-    fn create(&mut self, seed: http::Seed<()>) -> Self::Output {
+    fn create(&mut self, seed: http::Seed<()>) -> Option<Self::Output> {
-        message::Message::new(self.factory.create(seed.control()))
+        Some(message::Message::new(self.factory.create(seed.control())))
+    }
+
+    fn keep_alive_interest(&self) -> Next {
+        Next::read()
     }
 }
 
@@ -191,7 +195,7 @@ where A: Accept,
         rotor_try!(scope.register(&seed, EventSet::readable(), PollOpt::level()));
         rotor::Response::ok(
             ServerFsm::Conn(
-                http::Conn::new((), seed, scope.notifier())
+                http::Conn::new((), seed, Next::read(), scope.notifier())
                     .keep_alive(scope.keep_alive)
             )
         )