Avoid deadlock when closing sync connection with unread messages.

Fix #1336.

docs/project/changelog.rst

@@ -38,6 +38,17 @@ Backwards-incompatible changes
 
     websockets 11.0 is the last version supporting Python 3.7.
 
+11.0.2
+------
+
+*April 18, 2023*
+
+Bug fixes
+.........
+
+* Fixed a deadlock in the :mod:`threading` implementation when closing a
+  connection without reading all messages.
+
 11.0.1
 ------
 

src/websockets/sync/connection.py

@@ -382,8 +382,9 @@ def close(self, code: int = 1000, reason: str = "") -> None:
         """
         Perform the closing handshake.
 
-        :meth:`close` waits for the other end to complete the handshake and
-        for the TCP connection to terminate.
+        :meth:`close` waits for the other end to complete the handshake, for the
+        TCP connection to terminate, and for all incoming messages to be read
+        with :meth:`recv`.
 
         :meth:`close` is idempotent: it doesn't do anything once the
         connection is closed.
@@ -574,9 +575,13 @@ def recv_events(self) -> None:
                 # Given that automatic responses write small amounts of data,
                 # this should be uncommon, so we don't handle the edge case.
 
-                for event in events:
-                    # This isn't expected to raise an exception.
-                    self.process_event(event)
+                try:
+                    for event in events:
+                        # This may raise EOFError if the closing handshake
+                        # times out while a message is waiting to be read.
+                        self.process_event(event)
+                except EOFError:
+                    break
 
             # Breaking out of the while True: ... loop means that we believe
             # that the socket doesn't work anymore.
@@ -600,7 +605,6 @@ def recv_events(self) -> None:
             self.protocol.state = CLOSED
         finally:
             # This isn't expected to raise an exception.
-            self.recv_messages.close()
             self.close_socket()
 
     @contextlib.contextmanager
@@ -745,13 +749,16 @@ def set_recv_events_exc(self, exc: Optional[BaseException]) -> None:
 
     def close_socket(self) -> None:
         """
-        Shutdown and close socket.
+        Shutdown and close socket. Close message assembler.
 
-        shutdown() is required to interrupt recv() on Linux.
+        Calling close_socket() guarantees that recv_events() terminates. Indeed,
+        recv_events() may block only on socket.recv() or on recv_messages.put().
 
         """
+        # shutdown() is required to interrupt recv() on Linux.
         try:
             self.socket.shutdown(socket.SHUT_RDWR)
         except OSError:
             pass  # socket is already closed
         self.socket.close()
+        self.recv_messages.close()

tests/sync/test_connection.py

@@ -458,6 +458,49 @@ def test_close_timeout_waiting_for_connection_closed(self):
         # Remove socket.timeout when dropping Python < 3.10.
         self.assertIsInstance(exc.__cause__, (socket.timeout, TimeoutError))
 
+    def test_close_waits_for_recv(self):
+        self.remote_connection.send("😀")
+
+        close_thread = threading.Thread(target=self.connection.close)
+        close_thread.start()
+
+        # Let close() initiate the closing handshake and send a close frame.
+        time.sleep(MS)
+        self.assertTrue(close_thread.is_alive())
+
+        # Connection isn't closed yet.
+        self.connection.recv()
+
+        # Let close() receive a close frame and finish the closing handshake.
+        time.sleep(MS)
+        self.assertFalse(close_thread.is_alive())
+
+        # Connection is closed now.
+        with self.assertRaises(ConnectionClosedOK) as raised:
+            self.connection.recv()
+
+        exc = raised.exception
+        self.assertEqual(str(exc), "sent 1000 (OK); then received 1000 (OK)")
+        self.assertIsNone(exc.__cause__)
+
+    def test_close_timeout_waiting_for_recv(self):
+        self.remote_connection.send("😀")
+
+        close_thread = threading.Thread(target=self.connection.close)
+        close_thread.start()
+
+        # Let close() time out during the closing handshake.
+        time.sleep(3 * MS)
+        self.assertFalse(close_thread.is_alive())
+
+        # Connection is closed now.
+        with self.assertRaises(ConnectionClosedError) as raised:
+            self.connection.recv()
+
+        exc = raised.exception
+        self.assertEqual(str(exc), "sent 1000 (OK); no close frame received")
+        self.assertIsInstance(exc.__cause__, TimeoutError)
+
     def test_close_idempotency(self):
         """close does nothing if the connection is already closed."""
         self.connection.close()