More notes

Author: mikeshardmind

misc_info.md

@@ -0,0 +1,32 @@
+# What?
+
+This is just some assorted info that may be useful to anyone considering why
+certain things in this library are done the way there are and also
+as a place to record some important information that I may not always remember
+all of, but may need to revisit in the future related to it.
+
+## Task switching semantic changes in the standard library
+
+### Before eager tasks
+
+- await
+- async with (both enter and exit)
+- async for (per loop)
+- return (from a coroutine)
+- yield (from a coroutine)
+- Raising an exception (that leaves a coroutine)
+
+### As of eager tasks, the below are included
+
+- Constructing an asyncio.Task with eager_start=True while the associated event loop is running
+- asyncio.create_task
+- asyncio.ensure_future (when wrapping a non-future awaitable)
+- asyncio.wait_for\* (when any non-future awaitables are waited on) (ensure_future)
+- asyncio.as_completed\* (when any non-future awaitables are waited on) (ensure_future)
+- asyncio.gather\* (when any non-future awaitables are gathered) (ensure_future)
+- asyncio.IocpProactor.accept (unlikely to impact most)  (ensure_future on a coro)
+- asyncio.StreamReaderProtocol.connection_made (library use, perhaps?) (create task)
+- Constructing asyncio.BaseSubprocessTransport (create task)
+    - Note: extremely unlikely to impact anyone, asyncio.create_subprocess_* doesn't construct this prior to an await.
+
+\* Likely not a meaningfully new place, awaiting these immediately is the most common use.

src/async_utils/_nm_building_blocks.py

@@ -80,6 +80,9 @@ def __init__(self, lock: threading.RLock) -> None:
     async def __aenter__(self) -> None:
         acquired = False
         while not acquired:
+            # TODO: This kindof sucks, and we can avoid this
+            # if we're also the scheduler, find a way not to use this in
+            # any async scheduler that is threading-aware when I get to that.
             acquired = self._lock.acquire(blocking=False)
             await asyncio.sleep(0)
 

src/async_utils/waterfall.py

@@ -127,7 +127,7 @@ async def _loop(self) -> None:
 
                 t = asyncio.create_task(self.callback(queue_items))
                 tasks.add(t)
-                t.add_done_callback(tasks.remove)
+                t.add_done_callback(tasks.discard)
 
                 for _ in range(num_items):
                     self.queue.task_done()
@@ -137,7 +137,10 @@ async def _loop(self) -> None:
             await asyncio.wait_for(f, timeout=self.max_wait_finalize)
 
     async def _finalize(self) -> None:
-        # WARNING: Do not allow an async context switch before the gather below
+        # WARNING: Do not allow an async context switch before the queue is drained
+        # This can be changed to utilize queue.Shutdown in 3.13+
+        # or when more of asyncio queues have been replaced here
+        # as part of freethreading efforts.
 
         self._alive = False
         remaining_items: Sequence[T] = []
@@ -152,27 +155,28 @@ async def _finalize(self) -> None:
 
             remaining_items.append(ev)
 
+        # Context switches are safe again.
+
         if not remaining_items:
             return
 
         num_remaining = len(remaining_items)
 
-        pending_futures: list[asyncio.Task[object]] = []
+        remaining_tasks: set[asyncio.Task[object]] = set()
 
         for chunk in (
             remaining_items[p : p + self.max_quantity]
             for p in range(0, num_remaining, self.max_quantity)
         ):
             fut = asyncio.create_task(self.callback(chunk))
-            pending_futures.append(fut)
+            fut.add_done_callback(remaining_tasks.discard)
+            remaining_tasks.add(fut)
 
-        gathered = asyncio.gather(*pending_futures)
+        timeout = self.max_wait_finalize
+        _done, pending = await asyncio.wait(remaining_tasks, timeout=timeout)
 
-        try:
-            await asyncio.wait_for(gathered, timeout=self.max_wait_finalize)
-        except TimeoutError:
-            for task in pending_futures:
-                task.cancel()
+        for task in pending:
+            task.cancel()
 
         for _ in range(num_remaining):
             self.queue.task_done()