Is there a method similar to asyncio's call_later or call_at?

[wireboy5]

I am working on a project where I need to delay the execution of a function, while still running the code following and not blocking. I have tried using anyio.to_thread.run_sync to run a synchronous while loop that calls back to anyio like so:

def _wait_sync():
        while time.time() < timer_for:
            pass
        anyio.from_thread.run_async_from_thread(_continue)

This does not work. anyio.to_thread.run_sync still blocks for the specified time. Why would this be? Is there a way to make this work that is not documented yet?

[graingert]

@wireboy5 do you mean something like this?

async def call_later(delay, async_fn):
    await anyio.sleep(delay)
    await async_fn()

async with anyio.create_task_group() as tg:
    tg.start_soon(call_later, 10, async_fn)
    await other_async_fn()

not sure how it relates to a thread

[wireboy5]

Almost. I am currently working on an QUIC implementation using aioquic, and there is a function that requires you to set a timer that executes a specific function sometime in the future. However, my library needs to pass a higher-level control of the connection to the end user, in a form similar to the current anyio TCP client implementation. This means that I must constantly have a timer running, but can not have my library block at all. What do you think of wrapping my library in a context manager, and manually calling the TaskGroup __aenter__ and __aexit__ functions from my own versions of said functions? This feels like a hack, and I am mainly wondering if there was an "official" way to do this.

[agronholm]

It would be easier to suggest solutions if you elaborated on the use case. The code you presented would be better implemented in other ways.

[wireboy5]

Thanks for the reply. I am currently working on an QUIC implementation using aioquic and trying to implement something similar to the re-arm timer section of this function here at this line here. What are some better ways that you recommend? I am considering using my library as a context manager, and manually calling the TaskGroup __aenter__ and __aexit__ functions from my own versions of the same functions. However, this feels slightly like a hack and less of an "official" way to do it. What do you think?

[smurfix]

Whenever you get the idea to manually call __aenter__ and friends you need to restructure your code. Because, you know, experience says that doing that kind of shenanigans almost never turns out well.

The traditional way out is an async context manager. You nest all the contexts you like in there, then yield self (or whatever other useful object you have).

[smurfix]

Also, instead of re-arming a timer task I'd recommend enqueuing the to-be-transmitted data to a writer task, which runs this loop:

async def _writer(self, send_qr, timeout: float):
  while True:
    msg = None
    with anyio.move_on_after(timeout) as tsc:
      async for msg in send_qr:
        await self._transmit(msg)
        msg = None
        tsc.deadline = anyio.current_time() + timeout
      return # send_qr channel was closed
    # timed out
    if msg is None:
      msg = PingMsg()
    await self._transmit(msg)

The "msg / None" dance is there so that if the transmit queue is full and the timeout interrupts _transmit instead of the async for, your message won't get lost.