From 823695be0b8f0c3c40dccefc880b36de184becb2 Mon Sep 17 00:00:00 2001 From: Nathan Cooper Date: Fri, 17 Jul 2026 09:33:16 -0400 Subject: [PATCH 1/4] add debounce and throttle --- fastcore/_modidx.py | 12 ++ fastcore/aio.py | 77 +++++++- nbs/03c_aio.ipynb | 463 +++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 546 insertions(+), 6 deletions(-) diff --git a/fastcore/_modidx.py b/fastcore/_modidx.py index 2083be03..1b88eb6f 100644 --- a/fastcore/_modidx.py +++ b/fastcore/_modidx.py @@ -8,9 +8,20 @@ 'syms': { 'fastcore.aio': { 'fastcore.aio.CachedAwaitable': ('aio.html#cachedawaitable', 'fastcore/aio.py'), 'fastcore.aio.CachedAwaitable.__await__': ('aio.html#cachedawaitable.__await__', 'fastcore/aio.py'), 'fastcore.aio.CachedAwaitable.__init__': ('aio.html#cachedawaitable.__init__', 'fastcore/aio.py'), + 'fastcore.aio.Debounce': ('aio.html#debounce', 'fastcore/aio.py'), + 'fastcore.aio.Debounce.__call__': ('aio.html#debounce.__call__', 'fastcore/aio.py'), + 'fastcore.aio.Debounce.__get__': ('aio.html#debounce.__get__', 'fastcore/aio.py'), + 'fastcore.aio.Debounce.__init__': ('aio.html#debounce.__init__', 'fastcore/aio.py'), + 'fastcore.aio.Debounce.__set_name__': ('aio.html#debounce.__set_name__', 'fastcore/aio.py'), + 'fastcore.aio.Debounce._fire': ('aio.html#debounce._fire', 'fastcore/aio.py'), + 'fastcore.aio.Debounce._put_latest': ('aio.html#debounce._put_latest', 'fastcore/aio.py'), + 'fastcore.aio.Debounce._run': ('aio.html#debounce._run', 'fastcore/aio.py'), + 'fastcore.aio.Debounce.cancel': ('aio.html#debounce.cancel', 'fastcore/aio.py'), + 'fastcore.aio.Debounce.flush': ('aio.html#debounce.flush', 'fastcore/aio.py'), 'fastcore.aio._get_loop': ('aio.html#_get_loop', 'fastcore/aio.py'), 'fastcore.aio.acache': ('aio.html#acache', 'fastcore/aio.py'), 'fastcore.aio.ctx_sync': ('aio.html#ctx_sync', 'fastcore/aio.py'), + 'fastcore.aio.debounced': ('aio.html#debounced', 'fastcore/aio.py'), 'fastcore.aio.is_async_callable': ('aio.html#is_async_callable', 'fastcore/aio.py'), 'fastcore.aio.iter_sync': ('aio.html#iter_sync', 'fastcore/aio.py'), 'fastcore.aio.mapa': ('aio.html#mapa', 'fastcore/aio.py'), @@ -20,6 +31,7 @@ 'fastcore.aio.reawaitable': ('aio.html#reawaitable', 'fastcore/aio.py'), 'fastcore.aio.run_sync': ('aio.html#run_sync', 'fastcore/aio.py'), 'fastcore.aio.then': ('aio.html#then', 'fastcore/aio.py'), + 'fastcore.aio.throttled': ('aio.html#throttled', 'fastcore/aio.py'), 'fastcore.aio.to_aiter': ('aio.html#to_aiter', 'fastcore/aio.py')}, 'fastcore.all': {}, 'fastcore.ansi': {}, diff --git a/fastcore/aio.py b/fastcore/aio.py index 16033b58..08d8da86 100644 --- a/fastcore/aio.py +++ b/fastcore/aio.py @@ -1,4 +1,4 @@ -"""Bridging async and sync code: `run_sync`, `iter_sync`, `ctx_sync`, `maybe_await`, and `then` +"""Bridging async and sync code: `run_sync`, `iter_sync`, `ctx_sync`, `maybe_await`, and `then`, plus `Debounce` for coalescing bursts of calls Docs: https://fastcore.fast.ai/aio.html.md""" @@ -6,7 +6,7 @@ # %% auto #0 __all__ = ['run_sync', 'iter_sync', 'ctx_sync', 'maybe_await', 'then', 'acache', 'CachedAwaitable', 'reawaitable', - 'is_async_callable', 'to_aiter', 'maybe_aiter', 'mapa', 'noopa'] + 'is_async_callable', 'to_aiter', 'maybe_aiter', 'mapa', 'noopa', 'Debounce', 'debounced', 'throttled'] # %% ../nbs/03c_aio.ipynb #7e2193be import asyncio,threading @@ -136,3 +136,76 @@ async def mapa(f, items): async def noopa(x=None, *args, **kwargs): "Do nothing (async)" return x + +# %% ../nbs/03c_aio.ipynb #314d1e40 +class Debounce: + "Coalesce calls to `f` into one, made `wait` secs after calls stop (or `max_wait` secs after they start)" + def __init__(self, f, wait, max_wait=None, leading=False, trailing=True): + assert leading or trailing,"one of `leading`/`trailing` must be set" + store_attr() + self._task,self._q = None,None + try: self._loop = asyncio.get_running_loop() + except RuntimeError: self._loop = None + + def _put_latest(self, q, call): + if q.full(): q.get_nowait() + q.put_nowait(call) + + def __call__(self, *args, **kw): + try: loop = asyncio.get_running_loop() + except RuntimeError: + if self._loop is None: raise RuntimeError('Debounce must first be called from an async event loop') from None + return self._loop.call_soon_threadsafe(partial(self.__call__, *args, **kw)) + + if self._loop is None: self._loop = loop + elif loop is not self._loop: return self._loop.call_soon_threadsafe(partial(self.__call__, *args, **kw)) + + self._deadline = self._loop.time()+self.wait + call = (args, kw) + if self._task and not self._task.done(): self._put_latest(self._q, call) + else: + self._q = asyncio.Queue(maxsize=1) + self._cap = self._loop.time()+self.max_wait if self.max_wait else float('inf') + if not self.leading: self._put_latest(self._q, call) + self._task = self._loop.create_task(self._run(self._q, call if self.leading else None)) + + async def _fire(self, p): await maybe_await(self.f(*p[0], **p[1])) + async def _run(self, q, first): + if first is not None: await self._fire(first) + + while True: + delay = min(self._deadline,self._cap) - self._loop.time() + if delay>0: await asyncio.sleep(delay); continue + if not self.trailing: + if not q.empty(): q.get_nowait() # drop the rest of the burst + return + if q.empty(): return + await self._fire(q.get_nowait()) + if q.empty(): return + if self.max_wait: self._cap = self._loop.time()+self.max_wait + + def cancel(self): + "Discard any pending call" + if self._task and not self._task.done(): self._task.cancel() + self._task,self._q = None,None + + async def flush(self): + "Run any pending call now instead of waiting" + p = self._q.get_nowait() if self._q and not self._q.empty() else None + self.cancel() + if p is not None: await self._fire(p) + + def __set_name__(self, owner, name): self._name = name + def __get__(self, obj, objtype=None): + if obj is None: return self + res = obj.__dict__[self._name] = Debounce(partial(self.f, obj), self.wait, self.max_wait, self.leading, self.trailing) + return res + +# %% ../nbs/03c_aio.ipynb #7d9f6bcf +def debounced(wait, max_wait=None, leading=False, trailing=True): + "Decorator: `Debounce` a function or method" + return partial(Debounce, wait=wait, max_wait=max_wait, leading=leading, trailing=trailing) + +def throttled(wait, leading=False): + "Decorator: fire at most once per `wait` secs (`Debounce` with `max_wait=wait`)" + return debounced(wait, max_wait=wait, leading=leading) diff --git a/nbs/03c_aio.ipynb b/nbs/03c_aio.ipynb index 46f2fefc..509c7484 100644 --- a/nbs/03c_aio.ipynb +++ b/nbs/03c_aio.ipynb @@ -17,7 +17,7 @@ "source": [ "# Async helpers\n", "\n", - "> Bridging async and sync code: `run_sync`, `iter_sync`, `ctx_sync`, `maybe_await`, and `then`" + "> Bridging async and sync code: `run_sync`, `iter_sync`, `ctx_sync`, `maybe_await`, and `then`, plus `Debounce` for coalescing bursts of calls\n" ] }, { @@ -133,7 +133,35 @@ "execution_count": null, "id": "7c6fa8cc", "metadata": {}, - "outputs": [], + "outputs": [ + { + "ename": "RuntimeError", + "evalue": "asyncio.run() cannot be called from a running event loop", + "output_type": "error", + "traceback": [ + "\u001b[31m---------------------------------------------------------------------------\u001b[39m", + "\u001b[31mRuntimeError\u001b[39m Traceback (most recent call last)", + "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[12]\u001b[39m\u001b[32m, line 2\u001b[39m", + "\u001b[32m 1\u001b[39m \u001b[38;5;28;01masync\u001b[39;00m \u001b[38;5;28;01mdef\u001b[39;00m _outer(): \u001b[38;5;28;01mreturn\u001b[39;00m run_sync(_double(\u001b[32m4\u001b[39m))", + "\u001b[32m----> \u001b[39m\u001b[32m2\u001b[39m test_eq(asyncio.run(_outer()), \u001b[32m8\u001b[39m)", + "", + "\u001b[36mFile \u001b[39m\u001b[32m~/.local/share/uv/python/cpython-3.13.5-linux-x86_64-gnu/lib/python3.13/asyncio/runners.py:191\u001b[39m, in \u001b[36mrun\u001b[39m\u001b[34m(main, debug, loop_factory)\u001b[39m", + "\u001b[32m 161\u001b[39m \u001b[38;5;250m\u001b[39m\u001b[33;03m\"\"\"Execute the coroutine and return the result.\u001b[39;00m", + "\u001b[32m 162\u001b[39m ", + "\u001b[32m 163\u001b[39m \u001b[33;03mThis function runs the passed coroutine, taking care of\u001b[39;00m", + "\u001b[32m (...)\u001b[39m\u001b[32m 187\u001b[39m \u001b[33;03m asyncio.run(main())\u001b[39;00m", + "\u001b[32m 188\u001b[39m \u001b[33;03m\"\"\"\u001b[39;00m", + "\u001b[32m 189\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m events._get_running_loop() \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m:", + "\u001b[32m 190\u001b[39m \u001b[38;5;66;03m# fail fast with short traceback\u001b[39;00m", + "\u001b[32m--> \u001b[39m\u001b[32m191\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mRuntimeError\u001b[39;00m(", + "\u001b[32m 192\u001b[39m \u001b[33m\"\u001b[39m\u001b[33masyncio.run() cannot be called from a running event loop\u001b[39m\u001b[33m\"\u001b[39m)", + "\u001b[32m 194\u001b[39m \u001b[38;5;28;01mwith\u001b[39;00m Runner(debug=debug, loop_factory=loop_factory) \u001b[38;5;28;01mas\u001b[39;00m runner:", + "\u001b[32m 195\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m runner.run(main)", + "", + "\u001b[31mRuntimeError\u001b[39m: asyncio.run() cannot be called from a running event loop" + ] + } + ], "source": [ "async def _outer(): return run_sync(_double(4))\n", "test_eq(asyncio.run(_outer()), 8)" @@ -402,7 +430,22 @@ "execution_count": null, "id": "aae0fa98", "metadata": {}, - "outputs": [], + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "data\n" + ] + }, + { + "name": "stdout", + "output_type": "stream", + "text": [ + "data\n" + ] + } + ], "source": [ "@reawaitable\n", "async def fetch_data():\n", @@ -564,9 +607,421 @@ " \"Do nothing (async)\"\n", " return x" ] + }, + { + "cell_type": "markdown", + "id": "8e8075b2", + "metadata": {}, + "source": [ + "## Debounce and throttle\n", + "\n", + "Coalesce bursts of calls into fewer invocations of a function. A typical use is auto-saving shortly after edits stop, rather than on every keystroke.\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "314d1e40", + "metadata": {}, + "outputs": [], + "source": [ + "#| export\n", + "class Debounce:\n", + " \"Coalesce calls to `f` into one, made `wait` secs after calls stop (or `max_wait` secs after they start)\"\n", + " def __init__(self, f, wait, max_wait=None, leading=False, trailing=True):\n", + " assert leading or trailing,\"one of `leading`/`trailing` must be set\"\n", + " store_attr()\n", + " self._task,self._q = None,None\n", + " try: self._loop = asyncio.get_running_loop()\n", + " except RuntimeError: self._loop = None\n", + "\n", + " def _put_latest(self, q, call):\n", + " if q.full(): q.get_nowait()\n", + " q.put_nowait(call)\n", + "\n", + " def __call__(self, *args, **kw):\n", + " try: loop = asyncio.get_running_loop()\n", + " except RuntimeError:\n", + " if self._loop is None: raise RuntimeError('Debounce must first be called from an async event loop') from None\n", + " return self._loop.call_soon_threadsafe(partial(self.__call__, *args, **kw))\n", + "\n", + " if self._loop is None: self._loop = loop\n", + " elif loop is not self._loop: return self._loop.call_soon_threadsafe(partial(self.__call__, *args, **kw))\n", + "\n", + " self._deadline = self._loop.time()+self.wait\n", + " call = (args, kw)\n", + " if self._task and not self._task.done(): self._put_latest(self._q, call)\n", + " else:\n", + " self._q = asyncio.Queue(maxsize=1)\n", + " self._cap = self._loop.time()+self.max_wait if self.max_wait else float('inf')\n", + " if not self.leading: self._put_latest(self._q, call)\n", + " self._task = self._loop.create_task(self._run(self._q, call if self.leading else None))\n", + "\n", + " async def _fire(self, p): await maybe_await(self.f(*p[0], **p[1]))\n", + " async def _run(self, q, first):\n", + " if first is not None: await self._fire(first)\n", + "\n", + " while True:\n", + " delay = min(self._deadline,self._cap) - self._loop.time()\n", + " if delay>0: await asyncio.sleep(delay); continue\n", + " if not self.trailing:\n", + " if not q.empty(): q.get_nowait() # drop the rest of the burst\n", + " return\n", + " if q.empty(): return\n", + " await self._fire(q.get_nowait())\n", + " if q.empty(): return\n", + " if self.max_wait: self._cap = self._loop.time()+self.max_wait\n", + "\n", + " def cancel(self):\n", + " \"Discard any pending call\"\n", + " if self._task and not self._task.done(): self._task.cancel()\n", + " self._task,self._q = None,None\n", + "\n", + " async def flush(self):\n", + " \"Run any pending call now instead of waiting\"\n", + " p = self._q.get_nowait() if self._q and not self._q.empty() else None\n", + " self.cancel()\n", + " if p is not None: await self._fire(p)\n", + "\n", + " def __set_name__(self, owner, name): self._name = name\n", + " def __get__(self, obj, objtype=None):\n", + " if obj is None: return self\n", + " res = obj.__dict__[self._name] = Debounce(partial(self.f, obj), self.wait, self.max_wait, self.leading, self.trailing)\n", + " return res" + ] + }, + { + "cell_type": "markdown", + "id": "097348cf", + "metadata": {}, + "source": [ + "Calling a `Debounce` is fire-and-forget. The call records its args (last call wins) and returns at once, and `f` runs later in a worker task on the event loop. The instance binds to the loop it was constructed on, or else the first loop it is called from, and keeps that loop for life. `f` can be sync or async, but a blocking `f` will stall the loop, so pass something like `partial(run_in_threadpool, f)` instead. Each call resets the `wait` timer, so a steady stream of calls fires nothing until they pause:\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "3fed94dd", + "metadata": {}, + "outputs": [], + "source": [ + "calls = []\n", + "d = Debounce(calls.append, 0.1)\n", + "for i in range(5): d(i)\n", + "test_eq(calls, [])\n", + "await asyncio.sleep(0.25)\n", + "test_eq(calls, [4])" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "387ce245", + "metadata": {}, + "outputs": [], + "source": [ + "calls.clear()\n", + "d = Debounce(calls.append, 0.1)\n", + "for i in range(4):\n", + " d(i)\n", + " await asyncio.sleep(0.06) # gaps shorter than `wait`, so the timer keeps resetting\n", + "test_eq(calls, [])\n", + "await asyncio.sleep(0.15)\n", + "test_eq(calls, [3])" + ] + }, + { + "cell_type": "markdown", + "id": "6a9f9f11", + "metadata": {}, + "source": [ + "With `max_wait` set, a burst cannot postpone the fire forever. It comes no later than `max_wait` after the burst began.\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "48a0f677", + "metadata": {}, + "outputs": [], + "source": [ + "calls.clear()\n", + "d = Debounce(calls.append, 0.1, max_wait=0.2)\n", + "for i in range(8):\n", + " d(i)\n", + " await asyncio.sleep(0.06) # a steady stream: plain debounce would never fire...\n", + "assert calls # ...but `max_wait` forced a fire mid-stream\n", + "await asyncio.sleep(0.25)\n", + "test_eq(calls[-1], 7) # and the trailing fire still delivered the last args" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "7d9f6bcf", + "metadata": {}, + "outputs": [], + "source": [ + "#| export\n", + "def debounced(wait, max_wait=None, leading=False, trailing=True):\n", + " \"Decorator: `Debounce` a function or method\"\n", + " return partial(Debounce, wait=wait, max_wait=max_wait, leading=leading, trailing=trailing)\n", + "\n", + "def throttled(wait, leading=False):\n", + " \"Decorator: fire at most once per `wait` secs (`Debounce` with `max_wait=wait`)\"\n", + " return debounced(wait, max_wait=wait, leading=leading)" + ] + }, + { + "cell_type": "markdown", + "id": "5a8cf58a", + "metadata": {}, + "source": [ + "A throttle is a debounce whose cap equals its wait. The first call of a burst fixes the fire time, so the per-call resets never win.\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "601cbb32", + "metadata": {}, + "outputs": [], + "source": [ + "calls.clear()\n", + "@throttled(0.1)\n", + "def tsave(x): calls.append(x)\n", + "\n", + "for i in range(5): tsave(i)\n", + "await asyncio.sleep(0.15)\n", + "test_eq(calls, [4])" + ] + }, + { + "cell_type": "markdown", + "id": "f72091a9", + "metadata": {}, + "source": [ + "On a method, `Debounce` acts as a descriptor. Each instance lazily gets its own independent debouncer, cached in the instance dict like `functools.cached_property` (so `__slots__` classes are not supported).\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "42336d6e", + "metadata": {}, + "outputs": [], + "source": [ + "class _Doc:\n", + " def __init__(self): self.saved = []\n", + " @debounced(0.1)\n", + " def save(self, x): self.saved.append(x)\n", + "\n", + "a,b = _Doc(),_Doc()\n", + "a.save(1); a.save(2); b.save(3)\n", + "await asyncio.sleep(0.15)\n", + "test_eq((a.saved,b.saved), ([2],[3]))\n", + "assert a.save is a.save" + ] + }, + { + "cell_type": "markdown", + "id": "0235bedb", + "metadata": {}, + "source": [ + "`leading=True` fires the first call of a burst right away. The trailing fire then covers the rest of the burst, and is skipped if there was no rest. `trailing=False` drops the rest instead, which gives a classic rate-limiter:\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "8956d934", + "metadata": {}, + "outputs": [], + "source": [ + "calls.clear()\n", + "d = Debounce(calls.append, 0.1, leading=True)\n", + "d(1); await asyncio.sleep(0.02)\n", + "test_eq(calls, [1]) # leading fire, almost immediate\n", + "await asyncio.sleep(0.15)\n", + "test_eq(calls, [1]) # burst had no other calls, so no trailing fire\n", + "d(2); d(3); await asyncio.sleep(0.25)\n", + "test_eq(calls, [1,2,3]) # leading fired 2; trailing coalesced the rest" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "338feb0d", + "metadata": {}, + "outputs": [], + "source": [ + "calls.clear()\n", + "d = Debounce(calls.append, 0.1, leading=True, trailing=False)\n", + "for i in range(5): d(i)\n", + "await asyncio.sleep(0.25)\n", + "test_eq(calls, [0]) # later calls in the burst were dropped, not queued\n", + "await d.flush()\n", + "test_eq(calls, [0]) # dropped calls are gone, so there is nothing for `flush` to run" + ] + }, + { + "cell_type": "markdown", + "id": "3f167794", + "metadata": {}, + "source": [ + "An async `f` is awaited on the loop. `flush` runs any pending call immediately, which is handy at shutdown, and `cancel` discards it. Use both from the owner loop only. Plain calls work from threads with no loop at all, and are forwarded:\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "9633ef89", + "metadata": {}, + "outputs": [], + "source": [ + "acalls = []\n", + "async def _asave(x): acalls.append(x)\n", + "\n", + "d = Debounce(_asave, 0.1)\n", + "d(1); d(2)\n", + "await asyncio.sleep(0.15)\n", + "test_eq(acalls, [2])" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "79dbbbc7", + "metadata": {}, + "outputs": [], + "source": [ + "calls.clear()\n", + "d = Debounce(calls.append, 5)\n", + "d(1); await d.flush()\n", + "test_eq(calls, [1]) # pending call ran now, not 5s later\n", + "await d.flush() # nothing pending: a no-op\n", + "d(2); d.cancel()\n", + "await asyncio.sleep(0.05)\n", + "test_eq(calls, [1]) # cancelled call never ran" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "e209db2b", + "metadata": {}, + "outputs": [], + "source": [ + "calls.clear()\n", + "d = Debounce(calls.append, 0.1)\n", + "threading.Thread(target=d, args=(7,)).start()\n", + "await asyncio.sleep(0.2)\n", + "test_eq(calls, [7])" + ] + }, + { + "cell_type": "markdown", + "id": "a8083335", + "metadata": {}, + "source": [ + "Calls that arrive while an async `f` is still running do not overlap it. They wait for it to finish, and only the latest is kept:\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "8cf24a13", + "metadata": {}, + "outputs": [], + "source": [ + "calls = []\n", + "started,release = asyncio.Event(),asyncio.Event()\n", + "\n", + "async def _slow(x):\n", + " calls.append(x)\n", + " if x == 1:\n", + " started.set()\n", + " await release.wait()\n", + "\n", + "d = Debounce(_slow, 0.05)\n", + "d(1); await started.wait()\n", + "d(2); d(3)\n", + "await asyncio.sleep(0.1)\n", + "test_eq(calls, [1]) # no overlapping call\n", + "\n", + "release.set()\n", + "await asyncio.sleep(0.1)\n", + "test_eq(calls, [1,3]) # latest queued call won" + ] + }, + { + "cell_type": "markdown", + "id": "502ffb93", + "metadata": {}, + "source": [ + "A call from a thread running a different event loop is forwarded to the loop that owns the `Debounce`. The instance never migrates between loops:\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "8d1427f3", + "metadata": {}, + "outputs": [], + "source": [ + "calls = []\n", + "owner = asyncio.get_running_loop()\n", + "\n", + "async def _record(x): calls.append((x, asyncio.get_running_loop()))\n", + "d = Debounce(_record, 0.05)\n", + "\n", + "async def _foreign(): d(7)\n", + "t = threading.Thread(target=lambda: asyncio.run(_foreign()))\n", + "t.start(); t.join()\n", + "await asyncio.sleep(0.1)\n", + "\n", + "test_eq(calls, [(7,owner)])\n", + "assert d._loop is owner" + ] + }, + { + "cell_type": "markdown", + "id": "a81dabf3", + "metadata": {}, + "source": [ + "If `f` raises, the exception ends that burst's worker task, and asyncio logs it as an unretrieved task exception. Later bursts are unaffected:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "f93dd78a", + "metadata": {}, + "outputs": [], + "source": [ + "calls = []\n", + "def _boom(x):\n", + " if x==1: raise ValueError('boom')\n", + " calls.append(x)\n", + "\n", + "d = Debounce(_boom, 0.1)\n", + "d(1); await asyncio.sleep(0.15)\n", + "test_fail(d._task.result, contains='boom') # the error ended up on the worker task\n", + "d(2); await asyncio.sleep(0.15)\n", + "test_eq(calls, [2])" + ] } ], - "metadata": {}, + "metadata": { + "solveit": { + "autosave": false, + "default_code": false, + "mode": "learning", + "use_thinking": true, + "use_tools": true, + "ver": 2 + } + }, "nbformat": 4, "nbformat_minor": 5 } From 11c2ea4811c1f8ee2b3ef5d003274bbade2e63b3 Mon Sep 17 00:00:00 2001 From: Nathan Cooper Date: Fri, 17 Jul 2026 09:41:28 -0400 Subject: [PATCH 2/4] expect_fail --- nbs/03c_aio.ipynb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nbs/03c_aio.ipynb b/nbs/03c_aio.ipynb index 509c7484..090d7cba 100644 --- a/nbs/03c_aio.ipynb +++ b/nbs/03c_aio.ipynb @@ -1006,7 +1006,7 @@ "\n", "d = Debounce(_boom, 0.1)\n", "d(1); await asyncio.sleep(0.15)\n", - "test_fail(d._task.result, contains='boom') # the error ended up on the worker task\n", + "with expect_fail(contains='boom'): d._task.result() # the error ended up on the worker task\n", "d(2); await asyncio.sleep(0.15)\n", "test_eq(calls, [2])" ] From d236bb4cbb4bce1f1d31751020e75877621d8e67 Mon Sep 17 00:00:00 2001 From: Nathan Cooper Date: Fri, 17 Jul 2026 09:45:45 -0400 Subject: [PATCH 3/4] fix --- nbs/03c_aio.ipynb | 32 ++------------------------------ 1 file changed, 2 insertions(+), 30 deletions(-) diff --git a/nbs/03c_aio.ipynb b/nbs/03c_aio.ipynb index 090d7cba..38c5108a 100644 --- a/nbs/03c_aio.ipynb +++ b/nbs/03c_aio.ipynb @@ -133,38 +133,10 @@ "execution_count": null, "id": "7c6fa8cc", "metadata": {}, - "outputs": [ - { - "ename": "RuntimeError", - "evalue": "asyncio.run() cannot be called from a running event loop", - "output_type": "error", - "traceback": [ - "\u001b[31m---------------------------------------------------------------------------\u001b[39m", - "\u001b[31mRuntimeError\u001b[39m Traceback (most recent call last)", - "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[12]\u001b[39m\u001b[32m, line 2\u001b[39m", - "\u001b[32m 1\u001b[39m \u001b[38;5;28;01masync\u001b[39;00m \u001b[38;5;28;01mdef\u001b[39;00m _outer(): \u001b[38;5;28;01mreturn\u001b[39;00m run_sync(_double(\u001b[32m4\u001b[39m))", - "\u001b[32m----> \u001b[39m\u001b[32m2\u001b[39m test_eq(asyncio.run(_outer()), \u001b[32m8\u001b[39m)", - "", - "\u001b[36mFile \u001b[39m\u001b[32m~/.local/share/uv/python/cpython-3.13.5-linux-x86_64-gnu/lib/python3.13/asyncio/runners.py:191\u001b[39m, in \u001b[36mrun\u001b[39m\u001b[34m(main, debug, loop_factory)\u001b[39m", - "\u001b[32m 161\u001b[39m \u001b[38;5;250m\u001b[39m\u001b[33;03m\"\"\"Execute the coroutine and return the result.\u001b[39;00m", - "\u001b[32m 162\u001b[39m ", - "\u001b[32m 163\u001b[39m \u001b[33;03mThis function runs the passed coroutine, taking care of\u001b[39;00m", - "\u001b[32m (...)\u001b[39m\u001b[32m 187\u001b[39m \u001b[33;03m asyncio.run(main())\u001b[39;00m", - "\u001b[32m 188\u001b[39m \u001b[33;03m\"\"\"\u001b[39;00m", - "\u001b[32m 189\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m events._get_running_loop() \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m:", - "\u001b[32m 190\u001b[39m \u001b[38;5;66;03m# fail fast with short traceback\u001b[39;00m", - "\u001b[32m--> \u001b[39m\u001b[32m191\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mRuntimeError\u001b[39;00m(", - "\u001b[32m 192\u001b[39m \u001b[33m\"\u001b[39m\u001b[33masyncio.run() cannot be called from a running event loop\u001b[39m\u001b[33m\"\u001b[39m)", - "\u001b[32m 194\u001b[39m \u001b[38;5;28;01mwith\u001b[39;00m Runner(debug=debug, loop_factory=loop_factory) \u001b[38;5;28;01mas\u001b[39;00m runner:", - "\u001b[32m 195\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m runner.run(main)", - "", - "\u001b[31mRuntimeError\u001b[39m: asyncio.run() cannot be called from a running event loop" - ] - } - ], + "outputs": [], "source": [ "async def _outer(): return run_sync(_double(4))\n", - "test_eq(asyncio.run(_outer()), 8)" + "test_eq(await _outer(), 8)" ] }, { From 1766765841a7a69ffcaceb9bd92ca2cc78df48d9 Mon Sep 17 00:00:00 2001 From: Nathan Cooper Date: Fri, 17 Jul 2026 12:11:50 -0400 Subject: [PATCH 4/4] work with patch --- fastcore/aio.py | 5 +++-- fastcore/basics.py | 3 ++- nbs/01_basics.ipynb | 6 ++++-- nbs/03c_aio.ipynb | 31 +++++++++++++++++++++++++++++-- 4 files changed, 38 insertions(+), 7 deletions(-) diff --git a/fastcore/aio.py b/fastcore/aio.py index 08d8da86..be64d714 100644 --- a/fastcore/aio.py +++ b/fastcore/aio.py @@ -11,7 +11,7 @@ # %% ../nbs/03c_aio.ipynb #7e2193be import asyncio,threading from contextlib import contextmanager -from functools import wraps +from functools import wraps,update_wrapper from .imports import * from .xtras import UNSET from .basics import * @@ -143,7 +143,8 @@ class Debounce: def __init__(self, f, wait, max_wait=None, leading=False, trailing=True): assert leading or trailing,"one of `leading`/`trailing` must be set" store_attr() - self._task,self._q = None,None + update_wrapper(self, f, updated=()) + self._task,self._q,self._name = None,None,getattr(f, '__name__', None) try: self._loop = asyncio.get_running_loop() except RuntimeError: self._loop = None diff --git a/fastcore/basics.py b/fastcore/basics.py index 531c403a..a320785e 100644 --- a/fastcore/basics.py +++ b/fastcore/basics.py @@ -1135,7 +1135,8 @@ def _inner(f): for o in functools.WRAPPER_ASSIGNMENTS: setattr(nf, o, getattr(f,o)) nf.__name__ = _nm nf.__qualname__ = f"{c_.__name__}.{_nm}" - if hasattr(nf.__code__, 'co_qualname'): nf.__code__ = nf.__code__.replace(co_qualname=nf.__qualname__) + if hasattr(nf, '__code__') and hasattr(nf.__code__, 'co_qualname'): + nf.__code__ = nf.__code__.replace(co_qualname=nf.__qualname__) if hasattr(c_, _nm) and not hasattr(c_, onm): setattr(c_, onm, getattr(c_, _nm)) if cls_method: attr = _clsmethod(nf) elif static_method: attr = staticmethod(nf) diff --git a/nbs/01_basics.ipynb b/nbs/01_basics.ipynb index bf5deacf..5545249c 100644 --- a/nbs/01_basics.ipynb +++ b/nbs/01_basics.ipynb @@ -6541,7 +6541,8 @@ " for o in functools.WRAPPER_ASSIGNMENTS: setattr(nf, o, getattr(f,o))\n", " nf.__name__ = _nm\n", " nf.__qualname__ = f\"{c_.__name__}.{_nm}\"\n", - " if hasattr(nf.__code__, 'co_qualname'): nf.__code__ = nf.__code__.replace(co_qualname=nf.__qualname__)\n", + " if hasattr(nf, '__code__') and hasattr(nf.__code__, 'co_qualname'):\n", + " nf.__code__ = nf.__code__.replace(co_qualname=nf.__qualname__)\n", " if hasattr(c_, _nm) and not hasattr(c_, onm): setattr(c_, onm, getattr(c_, _nm))\n", " if cls_method: attr = _clsmethod(nf)\n", " elif static_method: attr = staticmethod(nf)\n", @@ -8327,7 +8328,7 @@ " def __init__(self, name): self.name = name\n", " def __json__(self): return dict(name=self.name)\n", "\n", - "with expect_fail(TypeError, 'Person is not JSON serializable'): dumps(Person('Alyssa'))\n" + "with expect_fail(TypeError, 'Person is not JSON serializable'): json.dumps(Person('Alyssa'))\n" ] }, { @@ -8606,6 +8607,7 @@ ], "metadata": { "solveit": { + "autosave": false, "default_code": false, "mode": "learning", "use_fence": false, diff --git a/nbs/03c_aio.ipynb b/nbs/03c_aio.ipynb index 38c5108a..590f8e7e 100644 --- a/nbs/03c_aio.ipynb +++ b/nbs/03c_aio.ipynb @@ -30,7 +30,7 @@ "#| export\n", "import asyncio,threading\n", "from contextlib import contextmanager\n", - "from functools import wraps\n", + "from functools import wraps,update_wrapper\n", "from fastcore.imports import *\n", "from fastcore.xtras import UNSET\n", "from fastcore.basics import *" @@ -603,7 +603,8 @@ " def __init__(self, f, wait, max_wait=None, leading=False, trailing=True):\n", " assert leading or trailing,\"one of `leading`/`trailing` must be set\"\n", " store_attr()\n", - " self._task,self._q = None,None\n", + " update_wrapper(self, f, updated=())\n", + " self._task,self._q,self._name = None,None,getattr(f, '__name__', None)\n", " try: self._loop = asyncio.get_running_loop()\n", " except RuntimeError: self._loop = None\n", "\n", @@ -795,6 +796,32 @@ "assert a.save is a.save" ] }, + { + "cell_type": "markdown", + "id": "85c62eff", + "metadata": {}, + "source": [ + "`Debounce` copies its function's metadata, `functools.wraps` style, so it also composes with `patch`:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "e50a0170", + "metadata": {}, + "outputs": [], + "source": [ + "@patch\n", + "@debounced(0.1)\n", + "def dsave(self:_Doc, x): self.saved.append(x)\n", + "\n", + "c = _Doc()\n", + "c.dsave(1); c.dsave(2)\n", + "await asyncio.sleep(0.15)\n", + "test_eq(c.saved, [2])\n", + "test_eq(_Doc.dsave.__name__, 'dsave')" + ] + }, { "cell_type": "markdown", "id": "0235bedb",