Use AnyIO #77

davidbrochart · 2023-06-07T14:29:34Z

This PR has breaking changes, and should probably be released as v1.0.
In particular, WebsocketProvider, WebsocketServer, YRoom and YStore must either be used with an async context manager (this is the preferred way), or using lower-level start() and stop() methods.
Before this PR, some tasks were created implicitly on instanciation. The new approach is more explicit regarding the async nature of these objects, and ensures no task is left running on tear-down. Under the hood, AnyIO's task groups are used, but ypy-websocket can be used in a "pure asyncio" environment, no need to adopt AnyIO outside of this library.
Here is an example with an async context manager:

async def main():
    async with WebsocketServer():
        ...

Which is equivalent to the following, when using the lower-level API (with start()/stop()):

async def main():
    server = WebsocketServer()
    task = asyncio.create_task(server.start())
    await server.started.wait()
    ...
    server.stop()

Closes #76.

fcollonval

Thanks @davidbrochart

I have questions trying to understand all changes. It would also be great to add more documentation to the code especially as it is meant to be extended by other libraries.

ypy_websocket/websocket_server.py

fcollonval · 2023-06-09T14:25:32Z

ypy_websocket/websocket_provider.py

+    async def run(self):
+        async with create_task_group() as self._task_group:
+            self._task_group.start_soon(self._run)
+
+    def stop(self):
+        self._task_group.cancel_scope.cancel()


I'm probably missing something but why not removing all public API (aka run and stop) as this should only be used as async context manager according to the PR description?

I want to allow both an async context manager API (that should be the preferred way) and a lower-level API with start() and stop() equivalent methods. It is a common practice. I'll update the PR description.

fcollonval · 2023-06-09T14:33:34Z

ypy_websocket/websocket_server.py

    async def serve(self, websocket):
-        room = self.get_room(websocket.path)
-        room.clients.append(websocket)
-        await sync(room.ydoc, websocket, self.log)
-        async for message in websocket:
-            # filter messages (e.g. awareness)
-            skip = False
-            if room.on_message:
-                skip = await room.on_message(message)
-            if skip:
-                continue
-            message_type = message[0]
-            if message_type == YMessageType.SYNC:
-                # update our internal state in the background
-                # changes to the internal state are then forwarded to all clients
-                # and stored in the YStore (if any)
-                task = asyncio.create_task(
-                    process_sync_message(message[1:], room.ydoc, websocket, self.log)
-                )
-                self.background_tasks.add(task)
-                task.add_done_callback(self.background_tasks.discard)
-            elif message_type == YMessageType.AWARENESS:
-                # forward awareness messages from this client to all clients,
-                # including itself, because it's used to keep the connection alive
-                self.log.debug(
-                    "Received %s message from endpoint: %s",
-                    YMessageType.AWARENESS.name,
-                    websocket.path,
-                )
-                for client in room.clients:
-                    self.log.debug(
-                        "Sending Y awareness from client with endpoint %s to client with endpoint: %s",
-                        websocket.path,
-                        client.path,
-                    )
-                    task = asyncio.create_task(client.send(message))
-                    self.background_tasks.add(task)
-                    task.add_done_callback(self.background_tasks.discard)
-        # remove this client
-        room.clients = [c for c in room.clients if c != websocket]
-        if self.auto_clean_rooms and not room.clients:
-            self.delete_room(room=room)
+        if self._task_group is None:
+            raise RuntimeError(
+                "The WebsocketServer is not running: use `async with websocket_server:` or `await websocket_server.run()`"
+            )
+
+        await self._task_group.start(self._serve, websocket)


Why not starting to serve directly? The need to do

await websocket_server.run() await websocket_server.serve()

seems error prone to me.

The serve() method serves one WebSocket, but the server can serve multiple WebSockets. The goal of the (now called) start() method is to create the main task group, whose cancel scope is cancelled when stop() is called or when the async context manager exits. This ensures no task is running when the WebsocketServer stops.

fcollonval · 2023-06-09T14:40:49Z

ypy_websocket/yroom.py

+    async def enter(self):
+        if self._entered:
+            return
+
+        async with create_task_group() as self._task_group:
+            self._task_group.start_soon(self._broadcast_updates)
+            self._entered = True
+
+    def exit(self):
+        self._task_group.cancel_scope.cancel()


Would it make sense to align the wording using run/stop?

It seems you implemented a flag to ensure enter is called only once, why not implementing the flag completely and prevent exit if not enter was called or avoiding clashing between using the object in async context manager and public API -- this applies to the other async context manager objects too.

Would it make sense to align the wording using run/stop?

Yes, it's done in f0fa069.

It seems you implemented a flag to ensure enter is called only once, why not implementing the flag completely and prevent exit if not enter was called or avoiding clashing between using the object in async context manager and public API -- this applies to the other async context manager objects too.

Good point, I'll make the code more robust in that regard, thanks.

fcollonval · 2023-06-09T14:41:29Z

ypy_websocket/ystore.py

+        self._task_group.cancel_scope.cancel()
+        return await self._exit_stack.__aexit__(exc_type, exc_value, exc_tb)
+
+    async def start(self):


Same question about naming alignment and protection of the two API usage.

davidbrochart · 2023-06-09T16:18:24Z

I have questions trying to understand all changes. It would also be great to add more documentation to the code especially as it is meant to be extended by other libraries.

Thanks for the review @fcollonval, I updated the PR description. Where do you think I should add comments in the code in particular?

davidbrochart · 2023-06-12T15:29:12Z

The documentation is accessible at https://davidbrochart.github.io/ypy-websocket.

davidbrochart · 2023-06-12T15:56:26Z

See jupyter-server/jupyverse#320 for an example of needed changes.

fcollonval · 2023-06-13T08:33:44Z

Thanks a lot for the comments and the documentation @davidbrochart
The current documentation is a great start. I would say the main missing thing in the doc is in the overview a graph and a description of the roles and interactions of the provided classes: Server / websocket / wsprovider / room / store.

fcollonval · 2023-06-13T08:34:54Z

best would be to use mermaidJS so the graph are nicely display directly in GitHub.

davidbrochart · 2023-06-13T14:14:21Z

Thanks for the suggestion @fcollonval, that's a good idea. Here is how it currently looks (open to feedback):

fcollonval · 2023-06-15T13:31:45Z

Thanks David this looks great.

davidbrochart marked this pull request as draft June 7, 2023 14:29

davidbrochart mentioned this pull request Jun 7, 2023

Refactor room and server tasks cleanup #76

Closed

davidbrochart force-pushed the anyio branch 2 times, most recently from 5f5ef34 to 210f6f4 Compare June 7, 2023 14:52

davidbrochart marked this pull request as ready for review June 7, 2023 14:52

davidbrochart force-pushed the anyio branch from 210f6f4 to ecd4ce2 Compare June 7, 2023 15:23

Use AnyIO's task groups

af36d92

davidbrochart force-pushed the anyio branch from ecd4ce2 to af36d92 Compare June 7, 2023 15:24

davidbrochart added 2 commits June 8, 2023 11:13

Use task groups in WebsocketProvider, use AnyIO's streams

9bfd0d7

Don't close send stream

4517792

davidbrochart force-pushed the anyio branch from 5cf5d28 to 4517792 Compare June 8, 2023 13:01

Add YStore.start async method, use task group in SQLiteYStore

90c798f

davidbrochart force-pushed the anyio branch from 2071e6c to 9d7ad0f Compare June 8, 2023 13:59

davidbrochart changed the title ~~Use AnyIO's task groups~~ Use AnyIO Jun 8, 2023

Use AnyIO in tests

d0ff78d

davidbrochart force-pushed the anyio branch from 9d7ad0f to d0ff78d Compare June 8, 2023 14:10

Update README

de7d8c4

davidbrochart force-pushed the anyio branch from 7be77ca to c702f4c Compare June 8, 2023 14:54

Add more context managers

58cec1a

davidbrochart force-pushed the anyio branch from c702f4c to 58cec1a Compare June 8, 2023 15:07

Make WebsocketServer an async context manager

8713800

davidbrochart force-pushed the anyio branch from 7a3165e to 8713800 Compare June 9, 2023 09:44

fcollonval reviewed Jun 9, 2023

View reviewed changes

Make API uniform

f0fa069

davidbrochart force-pushed the anyio branch from 2de11d9 to f0fa069 Compare June 9, 2023 15:23

Lint

5d544cd

davidbrochart force-pushed the anyio branch from 78bb69f to 5d544cd Compare June 9, 2023 15:39

Check if started when stopping

62cb30a

davidbrochart force-pushed the anyio branch from 7e6aa1e to c27d429 Compare June 12, 2023 15:21

davidbrochart force-pushed the anyio branch from c27d429 to 4ba9c01 Compare June 12, 2023 15:24

davidbrochart force-pushed the anyio branch from 4ba9c01 to 4d7f016 Compare June 12, 2023 15:48

Add documentation

c257459

davidbrochart force-pushed the anyio branch from 4d7f016 to c257459 Compare June 12, 2023 15:51

davidbrochart mentioned this pull request Jun 12, 2023

Support ypy-websocket v0.12 jupyter-server/jupyverse#320

Merged

Add diagram explaining architecture

6b07e59

davidbrochart force-pushed the anyio branch from 594043e to 6b07e59 Compare June 13, 2023 14:11

davidbrochart added 2 commits June 13, 2023 17:56

Add documentation for YRoom and YStore

7b911ed

Point index.md to README.md

fa18405

davidbrochart added 2 commits June 15, 2023 22:18

Lint

326e0c8

Mostly wording

87ab529

davidbrochart force-pushed the anyio branch from b3fee65 to 87ab529 Compare June 15, 2023 21:02

Move Y message handling to YRoom

0d3488b

davidbrochart merged commit 17a2e14 into y-crdt:main Jun 16, 2023

davidbrochart deleted the anyio branch June 16, 2023 13:18

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Use AnyIO #77

Use AnyIO #77

davidbrochart commented Jun 7, 2023 •

edited

Loading

fcollonval left a comment

fcollonval Jun 9, 2023

davidbrochart Jun 9, 2023

fcollonval Jun 9, 2023

davidbrochart Jun 9, 2023

fcollonval Jun 9, 2023

davidbrochart Jun 9, 2023

fcollonval Jun 9, 2023

davidbrochart Jun 9, 2023

davidbrochart commented Jun 9, 2023

davidbrochart commented Jun 12, 2023

davidbrochart commented Jun 12, 2023

fcollonval commented Jun 13, 2023

fcollonval commented Jun 13, 2023

davidbrochart commented Jun 13, 2023

fcollonval commented Jun 15, 2023

Use AnyIO #77

Use AnyIO #77

Conversation

davidbrochart commented Jun 7, 2023 • edited Loading

fcollonval left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davidbrochart commented Jun 9, 2023

davidbrochart commented Jun 12, 2023

davidbrochart commented Jun 12, 2023

fcollonval commented Jun 13, 2023

fcollonval commented Jun 13, 2023

davidbrochart commented Jun 13, 2023

fcollonval commented Jun 15, 2023

davidbrochart commented Jun 7, 2023 •

edited

Loading