auto merge of #13688 : alexcrichton/rust/accept-timeout, r=brson

This adds experimental support for timeouts when accepting sockets through
`TcpAcceptor::accept`. This does not add a separate `accept_timeout` function,
but rather it adds a `set_timeout` function instead. This second function is
intended to be used as a hard deadline after which all accepts will never block
and fail immediately.

This idea was derived from Go's SetDeadline() methods. We do not currently have
a robust time abstraction in the standard library, so I opted to have the
argument be a relative time in millseconds into the future. I believe a more
appropriate argument type is an absolute time, but this concept does not exist
yet (this is also why the function is marked #[experimental]).

The native support is built on select(), similarly to connect_timeout(), and the
green support is based on channel select and a timer.

cc #13523

2 files changed, 86 insertions, 1 deletions

diff --git a/src/libstd/io/net/tcp.rs b/src/libstd/io/net/tcp.rs
index 4f1e6bd7418..0619c89aac1 100644
--- a/src/libstd/io/net/tcp.rs
+++ b/src/libstd/io/net/tcp.rs
@@ -22,7 +22,7 @@ use io::IoResult;
 use io::net::ip::SocketAddr;
 use io::{Reader, Writer, Listener, Acceptor};
 use kinds::Send;
-use option::{None, Some};
+use option::{None, Some, Option};
 use rt::rtio::{IoFactory, LocalIo, RtioSocket, RtioTcpListener};
 use rt::rtio::{RtioTcpAcceptor, RtioTcpStream};
 
@@ -184,6 +184,56 @@ pub struct TcpAcceptor {
     obj: ~RtioTcpAcceptor:Send
 }
 
+impl TcpAcceptor {
+    /// Prevents blocking on all future accepts after `ms` milliseconds have
+    /// elapsed.
+    ///
+    /// This function is used to set a deadline after which this acceptor will
+    /// time out accepting any connections. The argument is the relative
+    /// distance, in milliseconds, to a point in the future after which all
+    /// accepts will fail.
+    ///
+    /// If the argument specified is `None`, then any previously registered
+    /// timeout is cleared.
+    ///
+    /// A timeout of `0` can be used to "poll" this acceptor to see if it has
+    /// any pending connections. All pending connections will be accepted,
+    /// regardless of whether the timeout has expired or not (the accept will
+    /// not block in this case).
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// # #![allow(experimental)]
+    /// use std::io::net::tcp::TcpListener;
+    /// use std::io::net::ip::{SocketAddr, Ipv4Addr};
+    /// use std::io::{Listener, Acceptor, TimedOut};
+    ///
+    /// let addr = SocketAddr { ip: Ipv4Addr(127, 0, 0, 1), port: 8482 };
+    /// let mut a = TcpListener::bind(addr).listen().unwrap();
+    ///
+    /// // After 100ms have passed, all accepts will fail
+    /// a.set_timeout(Some(100));
+    ///
+    /// match a.accept() {
+    ///     Ok(..) => println!("accepted a socket"),
+    ///     Err(ref e) if e.kind == TimedOut => { println!("timed out!"); }
+    ///     Err(e) => println!("err: {}", e),
+    /// }
+    ///
+    /// // Reset the timeout and try again
+    /// a.set_timeout(Some(100));
+    /// let socket = a.accept();
+    ///
+    /// // Clear the timeout and block indefinitely waiting for a connection
+    /// a.set_timeout(None);
+    /// let socket = a.accept();
+    /// ```
+    #[experimental = "the type of the argument and name of this function are \
+                      subject to change"]
+    pub fn set_timeout(&mut self, ms: Option<u64>) { self.obj.set_timeout(ms); }
+}
+
 impl Acceptor<TcpStream> for TcpAcceptor {
     fn accept(&mut self) -> IoResult<TcpStream> {
         self.obj.accept().map(TcpStream::new)
@@ -191,6 +241,7 @@ impl Acceptor<TcpStream> for TcpAcceptor {
 }
 
 #[cfg(test)]
+#[allow(experimental)]
 mod test {
     use super::*;
     use io::net::ip::SocketAddr;
@@ -749,4 +800,37 @@ mod test {
         assert!(s.write([1]).is_err());
         assert_eq!(s.read_to_end(), Ok(vec!(1)));
     })
+
+    iotest!(fn accept_timeout() {
+        let addr = next_test_ip4();
+        let mut a = TcpListener::bind(addr).unwrap().listen().unwrap();
+
+        a.set_timeout(Some(10));
+
+        // Make sure we time out once and future invocations also time out
+        let err = a.accept().err().unwrap();
+        assert_eq!(err.kind, TimedOut);
+        let err = a.accept().err().unwrap();
+        assert_eq!(err.kind, TimedOut);
+
+        // Also make sure that even though the timeout is expired that we will
+        // continue to receive any pending connections.
+        let l = TcpStream::connect(addr).unwrap();
+        for i in range(0, 1001) {
+            match a.accept() {
+                Ok(..) => break,
+                Err(ref e) if e.kind == TimedOut => {}
+                Err(e) => fail!("error: {}", e),
+            }
+            if i == 1000 { fail!("should have a pending connection") }
+        }
+        drop(l);
+
+        // Unset the timeout and make sure that this always blocks.
+        a.set_timeout(None);
+        spawn(proc() {
+            drop(TcpStream::connect(addr));
+        });
+        a.accept().unwrap();
+    })
 }
diff --git a/src/libstd/rt/rtio.rs b/src/libstd/rt/rtio.rs
index 0f3fc9c21ce..5dd14834669 100644
--- a/src/libstd/rt/rtio.rs
+++ b/src/libstd/rt/rtio.rs
@@ -200,6 +200,7 @@ pub trait RtioTcpAcceptor : RtioSocket {
     fn accept(&mut self) -> IoResult<~RtioTcpStream:Send>;
     fn accept_simultaneously(&mut self) -> IoResult<()>;
     fn dont_accept_simultaneously(&mut self) -> IoResult<()>;
+    fn set_timeout(&mut self, timeout: Option<u64>);
 }
 
 pub trait RtioTcpStream : RtioSocket {