async on_ws_frame hooks

[leftys]

Hi, do you have any idea how to make reading the data asynchronous? Eg. making the on_ws_frame function async? I tried asyncio.create_task and pushing the data into asyncio.Queue, but both is pretty slow.

[tarasko]

No, I have no idea.

That was the reason why I decided not to provide async iteration interface like aiohttp and websockets do.
Because both create_task and Queue are slow and introduce significant delay and a lot of churn for the event loop.

Which is very undesirable when there is a spike in market activity.

Internally that's what aiohttp and websockets do. They have same non-async on_ws_frame or similar, and they set result on a future object. And later event loop picks it up and wakes up user coroutine.

[tarasko]

On a second thought,
what is potentially possible to do is to create and run coroutine directly and when a frame arrives just call coroutine.send(ws_frame).
This will eliminate event loop and all asyncio.Future machinery from the equation and make it faster. How faster I don't know.

class MyListener(picows.WSListener):
    async def on_ws_frame_async(self, ws_frame: picows.WSFrame):
         ...

    def on_ws_frame(self, ws_frame: picows.WSFrame):
         self.on_ws_frame_async(ws_frame).send()

[leftys]

On a second thought, what is potentially possible to do is to create and run coroutine directly and when a frame arrives just call coroutine.send(ws_frame). This will eliminate event loop and all asyncio.Future machinery from the equation and make it faster.

I think this will only run the coroutine until its first await statement, so it doesn't work in general. Not like calling await some_coro()

[tarasko]

I found this code snippet from the old time when I had to execute tasks eagerly.
It is not perfect but it works in general (if coroutine yields multiple times):

    def create_eager_task(self, coro) -> asyncio.Future:
        # caveat: This is not a proper task starter, cancelling returned future doesn't cancel operation.
        # The operation will carry on in background

        final_fut = asyncio.get_running_loop().create_future()

        def continuation(fut: Optional[asyncio.Future]):
            try:
                # In asyncio framework, this either return a future or throws a user exception
                # or throws StopIteration with return value
                next_fut: asyncio.Future = coro.send(fut)
                if next_fut is None:
                    print("PANIC!!! This should never happen! Stop before we get rekt")
                    exit(-1)
                next_fut.add_done_callback(continuation)
            except StopIteration as ex:
                if not final_fut.done():
                    final_fut.set_result(ex.value)
            except CancelledError as ex:
                if not final_fut.done():
                    final_fut.set_exception(ex)
            except Exception as ex:
                if not final_fut.done():
                    final_fut.set_exception(ex)

        # When first (just started) coro.send() is called, None must be passed as an argument,
        # otherwise a future that was passed as a continuation argument
        continuation(None)
        return final_fut

[leftys]

Thats interesting, in python3.11 the next_fut is always None.

[leftys]

i think something like this is needed for coroutine callbacks:

async def listen():
    await loop.sock_connect(sock_client, addr)
    while True:
        data = await loop.sock_recv(sock_client, n)
        await self._callback_coro(data)