bpo-30931: Asyncore alternative fix

Lib/asyncore.py

@@ -131,33 +131,30 @@ def poll(timeout=0.0, map=None):
             is_r = obj.readable()
             is_w = obj.writable()
             if is_r:
-                r.append(fd)
+                r.append(obj)
             # accepting sockets should not be writable
             if is_w and not obj.accepting:
-                w.append(fd)
+                w.append(obj)
             if is_r or is_w:
-                e.append(fd)
+                e.append(obj)
         if [] == r == w == e:
             time.sleep(timeout)
             return
 
         r, w, e = select.select(r, w, e, timeout)
 
-        for fd in r:
-            obj = map.get(fd)
-            if obj is None:
+        for obj in r:
+            if obj.closing:
                 continue
             read(obj)
 
-        for fd in w:
-            obj = map.get(fd)
-            if obj is None:
+        for obj in w:
+            if obj.closing:
                 continue
             write(obj)
 
-        for fd in e:
-            obj = map.get(fd)
-            if obj is None:
+        for obj in e:
+            if obj.closing:
                 continue
             _exception(obj)
 
@@ -181,11 +178,12 @@ def poll2(timeout=0.0, map=None):
                 pollster.register(fd, flags)
 
         r = pollster.poll(timeout)
-        for fd, flags in r:
-            obj = map.get(fd)
-            if obj is None:
-                continue
-            readwrite(obj, flags)
+        ready = [(map[fd], flags) for fd, flags in r]
+
+        for obj, flags in ready:
+            if not obj.closing:
+                readwrite(obj, flags)
+
 
 poll3 = poll2                           # Alias for backward compatibility
 
@@ -388,6 +386,10 @@ def recv(self, buffer_size):
                 raise
 
     def close(self):
+        if self.closing:
+            return
+        self.closing = True
+
         self.connected = False
         self.accepting = False
         self.connecting = False
@@ -399,6 +401,11 @@ def close(self):
                 if why.args[0] not in (ENOTCONN, EBADF):
                     raise
 
+    def fileno(self):
+        if self.socket is None:
+            raise socket.error(EBADF, 'Bad file descriptor')
+        return self.socket.fileno()
+
     # log and log_info may be overridden to provide more sophisticated
     # logging and warning methods. In general, log is for 'hit' logging
     # and 'log_info' is for informational, warning and error logging.

Lib/test/test_asyncore.py

@@ -499,6 +499,12 @@ def handle_connect(self):
         pass
 
 
+class SendHandler(BaseTestHandler):
+    def __init__(self, conn):
+        BaseTestHandler.__init__(self, conn)
+        self.send(b'x' * 1024)
+
+
 class BaseTestAPI:
 
     def tearDown(self):
@@ -576,12 +582,7 @@ class TestClient(BaseClient):
             def handle_read(self):
                 self.flag = True
 
-        class TestHandler(BaseTestHandler):
-            def __init__(self, conn):
-                BaseTestHandler.__init__(self, conn)
-                self.send(b'x' * 1024)
-
-        server = BaseServer(self.family, self.addr, TestHandler)
+        server = BaseServer(self.family, self.addr, SendHandler)
         client = TestClient(self.family, server.address)
         self.loop_waiting_for_flag(client)
 
@@ -800,6 +801,44 @@ def test_quick_connect(self):
             if t.is_alive():
                 self.fail("join() timed out")
 
+    def test_map_altered_in_loop(self):
+        family = self.family
+        fail = self.fail
+
+        class PoisonedClient(BaseClient):
+            """
+            This dispatcher is created after closing a writable one
+            """
+            def handle_write(self):
+                fail("Attempt to call handle_write on the wrong client")
+
+            # The dispatcher is not writable and therefore handle_write shouldn't be called
+            def writable(self):
+                return False
+
+        class ManagerClient(BaseClient):
+
+            def __init__(self, family, address):
+                BaseClient.__init__(self, family, address)
+                self.old_client = BaseClient(family, address)
+
+            def handle_write(self):
+                old_fd = self.old_client._fileno
+                self.old_client.close()
+                # This trusts that the fd of this client new
+                # will be the same to the one just closed
+                new_client = PoisonedClient(family, server.address)
+                if new_client._fileno != old_fd:
+                    raise unittest.SkipTest("The test is meaningful only if the fd for the old and "
+                                            "the new dispatcher are the same")
+
+                self.flag = True
+
+        server = BaseServer(self.family, self.addr, SendHandler)
+        manager = ManagerClient(self.family, server.address)
+        self.loop_waiting_for_flag(manager)
+
+
 class TestAPI_UseIPv4Sockets(BaseTestAPI):
     family = socket.AF_INET
     addr = (support.HOST, 0)