Useless lock

Author: blast-hardcheese

src/replit_river/client_session.py

@@ -147,8 +147,7 @@ async def _handle_messages_from_ws(self) -> None:
 
                     if msg.controlFlags & ACK_BIT != 0:
                         continue
-                    async with self._stream_lock:
-                        stream = self._streams.get(msg.streamId, None)
+                    stream = self._streams.get(msg.streamId, None)
                     if msg.controlFlags & STREAM_OPEN_BIT == 0:
                         if not stream:
                             logger.warning("no stream for %s", msg.streamId)
@@ -178,8 +177,7 @@ async def _handle_messages_from_ws(self) -> None:
                     if msg.controlFlags & STREAM_CLOSED_BIT != 0:
                         if stream:
                             stream.close()
-                        async with self._stream_lock:
-                            del self._streams[msg.streamId]
+                        del self._streams[msg.streamId]
                 except OutOfOrderMessageException:
                     logger.exception("Out of order message, closing connection")
                     await ws_wrapper.close()

src/replit_river/server_session.py

@@ -32,6 +32,7 @@
 
 logger = logging.getLogger(__name__)
 
+
 trace_propagator = TraceContextTextMapPropagator()
 trace_setter = TransportMessageTracingSetter()
 
@@ -142,8 +143,7 @@ async def _handle_messages_from_ws(self, tg: asyncio.TaskGroup) -> None:
 
                     if msg.controlFlags & ACK_BIT != 0:
                         continue
-                    async with self._stream_lock:
-                        stream = self._streams.get(msg.streamId)
+                    stream = self._streams.get(msg.streamId)
                     if msg.controlFlags & STREAM_OPEN_BIT == 0:
                         if not stream:
                             logger.warning("no stream for %s", msg.streamId)
@@ -170,15 +170,13 @@ async def _handle_messages_from_ws(self, tg: asyncio.TaskGroup) -> None:
                         if isinstance(_stream, IgnoreMessage):
                             continue
                         if not stream:
-                            async with self._stream_lock:
-                                self._streams[msg.streamId] = _stream
+                            self._streams[msg.streamId] = _stream
                         stream = _stream
 
                     if msg.controlFlags & STREAM_CLOSED_BIT != 0:
                         if stream:
                             stream.close()
-                        async with self._stream_lock:
-                            del self._streams[msg.streamId]
+                        del self._streams[msg.streamId]
                 except OutOfOrderMessageException:
                     logger.exception("Out of order message, closing connection")
                     await ws_wrapper.close()

src/replit_river/session.py

@@ -75,7 +75,6 @@ def __init__(
         self._retry_connection_callback = retry_connection_callback
 
         # stream for tasks
-        self._stream_lock = asyncio.Lock()
         self._streams: dict[str, Channel[Any]] = {}
 
         # book keeping
@@ -287,8 +286,7 @@ async def close(self) -> None:
             # throw exception correctly.
             for stream in self._streams.values():
                 stream.close()
-            async with self._stream_lock:
-                self._streams.clear()
+            self._streams.clear()
 
             self._state = SessionState.CLOSED