Repository: tornadoweb/tornado

Issue:
EBADF in accept() handler if TCPServer.stop() was called in the meantime
We've had sporadic issues where a EBADF error when calling accept() was being logged. See example here:
https://travis-ci.org/dask/distributed/jobs/238031120#L1596-L1607

After some debugging, I've come to the conclusion that this happens whenever a IO handler calls TCPServer.stop(), while another handler is pending for a connection request on that TCPServer's listening socket(s). TCPServer.stop() closes the listening socket immediately and, when the queued accept handler is called, it finds a closed socket fd.

Discussion:

Patch:
diff --git a/tornado/netutil.py b/tornado/netutil.py
index 31bbe803..e58db35d 100644
--- a/tornado/netutil.py
+++ b/tornado/netutil.py
@@ -241,9 +241,18 @@ def add_accept_handler(sock, callback):
     is different from the ``callback(fd, events)`` signature used for
     `.IOLoop` handlers.
 
+    A callable is returned which, when called, will remove the `.IOLoop`
+    event handler and stop processing further incoming connections.
+
     .. versionchanged:: 5.0
        The ``io_loop`` argument (deprecated since version 4.1) has been removed.
+
+    .. versionchanged:: 5.0
+       A callable is returned (``None`` was returned before).
     """
+    io_loop = IOLoop.current()
+    removed = [False]
+
     def accept_handler(fd, events):
         # More connections may come in while we're handling callbacks;
         # to prevent starvation of other tasks we must limit the number
@@ -257,6 +266,9 @@ def add_accept_handler(sock, callback):
         # heuristic for the number of connections we can reasonably
         # accept at once.
         for i in xrange(_DEFAULT_BACKLOG):
+            if removed[0]:
+                # The socket was probably closed
+                return
             try:
                 connection, address = sock.accept()
             except socket.error as e:
@@ -272,7 +284,13 @@ def add_accept_handler(sock, callback):
                 raise
             set_close_exec(connection.fileno())
             callback(connection, address)
-    IOLoop.current().add_handler(sock, accept_handler, IOLoop.READ)
+
+    def remove_handler():
+        io_loop.remove_handler(sock)
+        removed[0] = True
+
+    io_loop.add_handler(sock, accept_handler, IOLoop.READ)
+    return remove_handler
 
 
 def is_valid_ip(ip):
diff --git a/tornado/tcpserver.py b/tornado/tcpserver.py
index b3cdaa2c..fea215f5 100644
--- a/tornado/tcpserver.py
+++ b/tornado/tcpserver.py
@@ -110,7 +110,8 @@ class TCPServer(object):
                  read_chunk_size=None):
         self.io_loop = IOLoop.current()
         self.ssl_options = ssl_options
-        self._sockets = {}  # fd -> socket object
+        self._sockets = {}   # fd -> socket object
+        self._handlers = {}  # fd -> remove_handler callable
         self._pending_sockets = []
         self._started = False
         self._stopped = False
@@ -156,7 +157,8 @@ class TCPServer(object):
         """
         for sock in sockets:
             self._sockets[sock.fileno()] = sock
-            add_accept_handler(sock, self._handle_connection)
+            self._handlers[sock.fileno()] = add_accept_handler(
+                sock, self._handle_connection)
 
     def add_socket(self, socket):
         """Singular version of `add_sockets`.  Takes a single socket object."""
@@ -233,7 +235,8 @@ class TCPServer(object):
         self._stopped = True
         for fd, sock in self._sockets.items():
             assert sock.fileno() == fd
-            self.io_loop.remove_handler(fd)
+            # Unregister socket from IOLoop
+            self._handlers.pop(fd)()
             sock.close()
 
     def handle_stream(self, stream, address):


Test patch:
diff --git a/tornado/test/tcpserver_test.py b/tornado/test/tcpserver_test.py
index 9afb5420..2417992c 100644
--- a/tornado/test/tcpserver_test.py
+++ b/tornado/test/tcpserver_test.py
@@ -68,3 +68,48 @@ class TCPServerTest(AsyncTestCase):
         server.add_socket(sock)
         server.stop()
         server.stop()
+
+    @gen_test
+    def test_stop_in_callback(self):
+        # Issue #2069: calling server.stop() in a loop callback should not
+        # raise EBADF when the loop handles other server connection
+        # requests in the same loop iteration
+
+        class TestServer(TCPServer):
+            @gen.coroutine
+            def handle_stream(self, stream, address):
+                server.stop()
+                yield stream.read_until_close()
+
+        sock, port = bind_unused_port()
+        server = TestServer()
+        server.add_socket(sock)
+        server_addr = ('localhost', port)
+        N = 40
+        clients = [IOStream(socket.socket()) for i in range(N)]
+        connected_clients = []
+
+        @gen.coroutine
+        def connect(c):
+            try:
+                yield c.connect(server_addr)
+            except EnvironmentError:
+                pass
+            else:
+                connected_clients.append(c)
+
+        yield [connect(c) for c in clients]
+
+        self.assertGreater(len(connected_clients), 0,
+                           "all clients failed connecting")
+        try:
+            if len(connected_clients) == N:
+                # Ideally we'd make the test deterministic, but we're testing
+                # for a race condition in combination with the system's TCP stack...
+                self.skipTest("at least one client should fail connecting "
+                              "for the test to be meaningful")
+        finally:
+            for c in connected_clients:
+                c.close()
+
+        # Here tearDown() would re-raise the EBADF encountered in the IO loop