njsmith: First impression: there is more code here than I was expecting :-)

Why do we need a dedicated reaper task with its own entry queue, etc.? Is there some reason the finalizer hook can't do

```python
async def aclose_forcefully_with_error_log(obj):
    try:
        await aclose_forcefully(obj)
    except:
        log_unraiseable()

token.run_sync_soon(spawn_system_task, aclose_forcefully_with_error_log, obj)
```

?

Do we need to reimplement the fallback finalizer? Won't the VM run that automatically if we don't call `aclose` ourselves?

I'm not sure we need the `conditional_asyncgen_hooks` machinery, but if we do, I feel like we can avoid the frame locals hacking :-).
oremanj: > there is more code here than I was expecting :-)

More than I was expecting too, but I think there's a good reason for most of it in this case.

> Why do we need a dedicated reaper task with its own entry queue, etc.?

The entry queue is the runner's entry queue, not an additional one. The only reason I used the entry queue rather than the TrioToken was because trio.run() has some logic for not creating the TrioToken until it's needed, and I figured there was some reason for that that I was missing. If the asyncgen manager used a TrioToken, the TrioToken would always exist.

The reaper is its own task mostly because it needs to finalize the async generators that are still alive when the main task exits. (If we just let these dangle, then eventually they'll be GC'ed and will try to call our finalizer after the run has finished.) I guess that could be done directly at the place in `task_exited()` where we cancel the `system_nursery`, but this way seemed better-encapsulated. Also, really we want to start the shutdown process when all other system tasks have exited (except the run_sync_soon task and the asyncgen reaper), not just when the main task has exited, so that system tasks can safely use async generators in the same way that regular tasks can (rather than having a brief window after the main task has exited where newly iterated asyncgens won't be cleaned up). The current code will do that once we implement #1521.

> Do we need to reimplement the fallback finalizer? Won't the VM run that automatically if we don't call aclose ourselves?

Nope -- objects are only finalized once, per PEP 442.

```
In [1]: async def example():
   ...:     try:
   ...:         yield 1
   ...:     finally:
   ...:         await trio.lowlevel.cancel_shielded_checkpoint()
   ...:

In [2]: import trio

In [3]: example().asend(None).send(None)
Exception ignored in: <async_generator object example at 0x7f2468229280>
Traceback (most recent call last):
  File "<ipython-input-3-ac27a8ac6851>", line 1, in <module>
RuntimeError: async generator ignored GeneratorExit
---------------------------------------------------------------------------
StopIteration                             Traceback (most recent call last)
<ipython-input-3-ac27a8ac6851> in <module>
----> 1 example().asend(None).send(None)

StopIteration: 1

In [5]: import sys; sys.set_asyncgen_hooks(lambda g: None, lambda g: None)

In [6]: example().asend(None).send(None)
---------------------------------------------------------------------------
StopIteration                             Traceback (most recent call last)
<ipython-input-6-ac27a8ac6851> in <module>
----> 1 example().asend(None).send(None)

StopIteration: 1
```

> I'm not sure we need the conditional_asyncgen_hooks machinery

Trio itself maybe doesn't strictly need it (although it's important if you're doing guest mode on top of asyncio -- otherwise asyncgens created in asyncio will be finalized in Trio). But trio-asyncio will need it, and tricycle will need it, and having each library copy/paste most of `conditional_asyncgen_hooks` and claim its own locals key seems worse than doing it centrally.

> but if we do, I feel like we can avoid the frame locals hacking :-).

If you have a better way to track "this asyncgen is mine and that one is not", I'd love to hear it!
* You can't use a WeakSet of asyncgens-that-are-mine because weakrefs are broken before finalizers are called.
* You can't use a normal set because then the finalizer will never be called.
* You can't use a set of `id`s because the finalizer hook is only called for asyncgens that haven't been fully exhausted, leaving the opportunity for the set to grow without bound.
oremanj: > ...and use the system nursery for finalize calls in general. That would let us get rid of all the machinery around pending_finalize_calls, inflight_starts, etc., wouldn't it?

That would be starting more system tasks after the system nursery has begun shutting down and cancelling the existing ones, which we both said in #1521 shouldn't be allowed for service tasks in general.

Now that I think about it, though, we could remove pending_finalize_calls by letting start_close use run_sync_soon to schedule another call to itself if it's invoked before the nursery is open, since that's only a one-tick window. And similarly we could have the reaper task use run_sync_soon to schedule its own wakeup, which would remove the need for inflight_starts.

> it assumes that the context managers will be opened in strict nested order, but that's not guaranteed. If two invocations get interleaved, then you'll get a big mess.

That's a very good point -- OK, no context manager :-)

> trio-asyncio needs to store more per-generator state, because there can be multiple loops, and it needs to track which one each async generator belongs to.

We could support this by having *our* interface be that the firstiter hook returns its corresponding finalizer hook. trio-asyncio could return `loop._do_finalizer` after determining in firstiter what loop to use.

Or we could use a TreeVar for the hooks-to-use and not have multiple layers of hook delegation going on at all. Trio would install thread-level asyncgen hooks that delegate to the hook in the TreeVar, or call the asyncgen_manager hook if the TreeVar has no value here, or call the previously installed thread-level hook if the TreeVar is inaccessible since we're not in a task.

In either case, we would only need to store one datum in the locals dict -- "which finalizer should we use 
for this asyncgen?" -- and it could go away if we got firstiter-returns-finalizer upstreamed.

> I bet we could sell Yury on a change where firstiter is allowed to return the finalizer hook, or if it returns None then it uses the global one. But that wouldn't help before 3.9 at the earliest. Might be worth it anyway; it'd be a pretty simple change, and at least it'd put a time limit on this frame hacking stuff.

This seems worthwhile, especially if we still can make 3.9 even though we missed the first beta by a couple weeks. The change is small and likely to be backwards-compatible in all practical cases, although it's possible to write code whose behavior would change (if you're returning something from the firstiter hook even though that return value is currently ignored). What would the process be here? Submit a bug against Python first, or just a PR directly?

> Speaking of CPython remembering the finalizer hook when the async gen is created: doesn't that mean we don't need the fallback finalizer logic, since our finalizer hook will know how to handle everything that it gets given?

If we get firstiter-returns-finalizer upstreamed, then yes, we could remove this bit too: install the thread-level finalizer as None and have firstiter return None if we're not in a task. With current interpreter capabilities, we still need the fallback object, because the asyncgen object has remembered our thread-level finalizer unconditionally by the time our firstiter hook realizes "this isn't one of mine".
njsmith: > We could support this by having our interface be that the firstiter hook returns its corresponding finalizer hook. [...] In either case, we would only need to store one datum in the locals dict -- "which finalizer should we use for this asyncgen?" -- and it could go away if we got firstiter-returns-finalizer upstreamed.

OK yeah if we settle on making our asyncgen hooks extensible then this seems like a much better way to do that.

> What would the process be here? Submit a bug against Python first, or just a PR directly?

I think open an issue on BPO.

Maybe we should also see about changing the behavior so that if the firstiter hook raises, then the async gen is *not* marked as having started iterating?

(Oh wait, here's another thing to check: can a firstiter hook call `aclose()` on the not-quite-started-yet agen? I *think* it can, which would mean that we would actually have the ability to stop the production of new agens.)

> That would be starting more system tasks after the system nursery has begun shutting down and cancelling the existing ones, which we both said in #1521 shouldn't be allowed for service tasks in general.

Well, maybe we were wrong, if the first time we try to use them we run into problems with it :-)

In general though I'm having trouble keeping track of the combination of asyncgen tricky details + multiple layers of event loop composition + future service task semantics all at once.

----

Minimal useful thing: for plain `trio.run`, ignoring trio-asyncio etc. for the moment, I guess what we need is:

- a `finalize` hook that tosses each async generator into the system nursery to clean up
- a `firstiter` hook that adds agens to a `WeakSet`
  - and I guess should be very careful about how it does this, because we know it's on the same thread (thank goodness), but it could in theory be called from signal/GC context, so it needs to be reentrant wrt itself + all the rest of Trio
- during shutdown, a plan for finalizing all the agens in that `WeakSet`

That last step is the really tricky one.

In theory, we'd better clean up the agens after like... every other piece of user code has finished. If some task is still iterating an agen when we try to `aclose` it, then that will either cause a weird exception in the task that's trying to iterate it, *or* fail entirely (because you can't call `aclose` while another task is in the middle of `__anext__`. You get `RuntimeError: aclose(): asynchronous generator is already running` – though at least you can detect this ahead of time by checking `.ag_running`). Neither of these is good.

Of course, during the cleanup those agens might do arbitrary things, like create new agens or try to spawn new system tasks.

There's also some tricky sequencing between shutting down `run_sync_soon` and agen cleanup. An agen can disappear from our `WeakSet` and have its `finalize` hook called at any moment. If we shut down `run_sync_soon` and *then* start processing our `WeakSet`, then any agens that get GCed in between those two events will get lost. OTOH, it seems like you might *want* to shut down `run_sync_soon` before you start processing agens, to reduce the chance of new user code coming in and expecting stuff to work?

I guess a useful trick is that we could convert our `WeakSet` into regular `set` before we shut down `run_sync_soon`. Once it's a regular `set` we know that no more `finalize` calls will happen.

So I'm thinking of a sequence like:

1. Wait for all user tasks to exit (→ we are totally quiescent except for what might arrive through `run_sync_soon`)
2. Disable the creation of new user tasks (i.e., turn off `spawn_system_task`)
3. Convert our agen `WeakSet` into a regular `set`
4. Disable new calls to `run_sync_soon`
5. Drain the `run_sync_soon` queue
6. Start killing async generators, and iterate until they're all gone
7. Turn off the autojump clock
8. Remove our `SIGINT` handler
9. Check for pending `SIGINT` one last time
10. Exit

(This will also fix some other bugs, e.g. right now if a user system task uses `trio.to_thread` they can deadlock during shutdown b/c of the `run_sync_soon` task disappearing.)

Then there are a bunch of questions about how to implement this and how to extend it to cover more complex scenarios (guest mode, trio-asyncio, etc.). But first, does that sound like the right semantics to be aiming for?
oremanj: > (Oh wait, here's another thing to check: can a firstiter hook call aclose() on the not-quite-started-yet agen? I think it can, which would mean that we would actually have the ability to stop the production of new agens.)

This does work, although it's awkward because firstiter is synchronous (you have to say `agen.aclose().send(None)` and catch the StopIteration.)

An async generator can still be iterated after aclose(), though; it just raises StopAsyncIteration immediately. This seems confusing enough that we shouldn't bother -- if a user is actively trying to subvert our firstiter hook by catching the exception and continuing to iterate the generator, I think they deserve what they get.

> and I guess should be very careful about how it does this, because we know it's on the same thread (thank goodness), but it could in theory be called from signal/GC context, so it needs to be reentrant wrt itself + all the rest of Trio

Signal/GC context isn't async-colored, and firstiter-ing an async generator is. Function color is an illusion, sure, but do you think we need to worry about people saying `some_agen.asend(None).send(None)` from signal/GC context? I wouldn't expect to see that in real code ever. Or put another way: if we're worried about people implementing their own coroutine runner to iterate an asyncgen from signal/GC context, we should probably be just as worried about them doing that to every other Trio function. And we're not, so...?

> OTOH, it seems like you might want to shut down run_sync_soon before you start processing agens, to reduce the chance of new user code coming in and expecting stuff to work?

That seems dangerous to me -- imagine if an async generator had yielded inside `async with await trio.open_file(...):` such that its cleanup had to exit that context. Closing an async file uses a thread, and you need run_sync_soon to get the result of that thread back into Trio. Bang, you're deadlocked.

I think continuing to run the run_sync_soon task until after asyncgens are finalized is reasonably safe. If we disable spawn_system_task before starting asyncgen finalization, then all newly submitted user code will have to be synchronous, which tends to reduce the chance of it using an async generator.

> So I'm thinking of a sequence like: [...]

The sequence you've outlined sounds good to me, except for the order of shutting down the run_sync_soon task as mentioned above. It's worth noting that we can get this order pretty easily given #1521. We would set up the system nursery as:
* First service system task: autojumper (if applicable)
* Second service system task: run_sync_soon task
* Third service system task: sleep forever, then finalize remaining asyncgens in parallel
* Maybe more service system tasks created by the user or by other parts of Trio
* `from_thread.run`, GC'ed async generator finalizers, etc run as non-service system tasks
* And of course the main task is a non-service system task, technically

Shutdown sequence: after the main task exits...
* Turn our WeakSet of async generators into a regular set. No more finalizer calls will occur after this point, but we might have some existing ones running.
* Cancel the system nursery's cancel scope. This cancellation is initially only visible to the non-service system tasks.
* Once all non-service system tasks have exited (includes already-started finalizers), the system nursery is considered closed (no more spawning system tasks).
* Now the service system tasks get cancelled in opposite order of the order they were started. The two or three Trio-internal ones mentioned above were started first, so they get cancelled last.
* By the time the cancellation reaches the asyncgen canary task, there is no more user code running, except for the asyncgen finalizers we're running.
* Once all asyncgens are finalized, the run_sync_soon task does one last batch of running and exits.
* Then turn off the autojump clock
* Then the system nursery has all tasks completed, so the init task can exit
* Then remove SIGINT handler and check for missed KeyboardInterrupts

Basically, by implementing #1521 and starting our system tasks in the right order, we get the right shutdown order without having to do much at all by way of careful shutdown orchestration. And can still simplify this PR #1564 a good deal relative to what I originally uploaded -- for example, the asyncgen system task no longer does anything while user code is running, so there's no need to manage startup/shutdown quite so carefully.
njsmith: > This does work, although it's awkward because firstiter is synchronous (you have to say agen.aclose().send(None) and catch the StopIteration.)

Compared to everything else we're doing here, that level of awkwardness barely registers :-)

It sounds like we may not need to shut down asyncgen production though, so that's nice.

> Signal/GC context isn't async-colored, and firstiter-ing an async generator is. Function color is an illusion, sure, but do you think we need to worry about people saying some_agen.asend(None).send(None) from signal/GC context?

Oh yeah, good point. (Also I guess `firstiter` will likely be reentrant-safe anyway.)

> That seems dangerous to me -- imagine if an async generator had yielded inside async with await trio.open_file(...): such that its cleanup had to exit that context. Closing an async file uses a thread, and you need run_sync_soon to get the result of that thread back into Trio. Bang, you're deadlocked.

A-ha! Fantastic example, thank you.

> It's worth noting that we can get this order pretty easily given #1521.

No doubt, but I think for right now I'd rather try to get something landed without blocking on other reworks :-). It's fewer things to keep in our heads at once, and we can always refactor later.

So how about something like this:

```python
async def teardown_asyncgen(self, asyncgen):
    with CancelScope() as cscope:
        cscope.cancel()
    # If another asyncgen is waiting on it, put it back in the queue for later
    if asyncgen.ag_running:
        self.asyncgens.add(asyncgen)
    else:
        try:
            await asyncgen.aclose()
        except BaseException as exc:
            asyncgen_logger.exception(exc)

async def init(self):
    # This is also where we put the autojump clock, somehow
    async with open_nursery() as self.real_system_nursery:
        real_system_nursery.start_soon(entry_queue_task)
        # This is where spawn_system_task puts things, and it's auto-cancelled when main exits.
        # When it's closed, spawn_system_task stops working.
        async with open_nursery() as self.user_system_nursery:
            self.main_task = spawn(self.user_system_nursery, ...)
        # At this point we know all user tasks are gone.
        self.asyncgens = set(self.asyncgens)
        # No more asyncgen finalizers will be called. But there might be some working their
        # way through the run_sync_soon queue. So flush it.
        task = current_task()
        self.entry_queue.run_sync_soon(reschedule, task)
        await wait_task_rescheduled(lambda _: Abort.FAILED)
        # All pending finalizers have run now. Add any trailing asyncgens to our force-cleanup set.
        self.asyncgens.update(self.trailing_finalizer_asyncgens)
        self.trailing_finalizer_asyncgens = None
        # Loop needed in case some asyncgen's cleanup path firstiters a new asyncgen, or manages
        # to block our closing another asyncgen.
        # I guess ideally we'd do some dance involving the GC referrers graph to try to close
        # asyncgens from the top down...
        while self.asyncgens:
            batch = self.asyncgens
            self.asyncgens = set()
            async with open_nursery() as kill_them_all:
                for asyncgen in batch:
                    kill_them_all.start_soon(self.teardown_asyncgen, asyncgen)
        # There are no more asyncgens, woohoo.
        self.real_system_nursery.cancel()

def finalizer(self, asyncgen):
    def finalizer_cb():
        try:
            self.user_system_nursery.start_soon(self.teardown_asyncgen, asyncgen)
        except RuntimeError:  # nursery is closed, final shutdown is starting
            self.trailing_finalizer_asyncgens.add(asyncgen)

    self.entry_queue.run_sync_soon(finalizer_cb)
```
oremanj: That sketch seems reasonable to me. (Nice touch with the loop during final teardown -- my previous attempts just denied asyncgen firstiter'ing after that point.)

I do still want to at least try to get #1521 in first, because I think doing the teardown dance on top of it will be cleaner. I have a mostly-complete implementation which just needs #1579 to be workable; if it winds up raising design questions that aren't quickly resolvable then I'll come back and implement the interim non-service-task-based approach to asyncgen teardown here.
njsmith: My issue with blocking on #1521 is that I don't think we're confident of all the details of that API yet, and I'd rather focus on that on its own so it gets the attention it deserves, without trying to juggle this at the same time.
oremanj: > I think open an issue on BPO.

https://bugs.python.org/issue40916

> My issue with blocking on #1521 is that I don't think we're confident of all the details of that API yet, and I'd rather focus on that on its own so it gets the attention it deserves, without trying to juggle this at the same time.

Fair enough - I'll rework this to not depend on that. I do think we at least need to resolve #1579 first though.
oremanj: After #1588, there will be only one task that would go in the "real" system nursery: the run_sync_soon task. Should we just... unroll that into some things that `init` can call directly? Like different functions for "process callbacks until cancelled", "process callbacks once", and "close and process all remaining callbacks". Then we wouldn't need a "real" vs "user" system nursery at all; the only system nursery would be the one that gets shut down before the last stage of async generator finalization.
njsmith: > After #1588, there will be only one task that would go in the "real" system nursery: the run_sync_soon task. Should we just... unroll that into some things that init can call directly?

Maybe? But let's something landed first here before worrying about refactoring `run_sync_soon`. (We're going to need to go back to refactoring the init task immediately as part of the `KeyboardInterrupt` work anyway, so we'll have plenty of time to think about it...)
diff --git a/docs/source/conf.py b/docs/source/conf.py
old mode 100644
new mode 100755
diff --git a/docs/source/reference-core.rst b/docs/source/reference-core.rst
index 24f9ab52..664a8d96 100644
--- a/docs/source/reference-core.rst
+++ b/docs/source/reference-core.rst
@@ -1465,6 +1465,182 @@ don't have any special access to Trio's internals.)
    :members:
 
 
+.. _async-generators:
+
+Notes on async generators
+-------------------------
+
+Python 3.6 added support for *async generators*, which can use
+``await``, ``async for``, and ``async with`` in between their ``yield``
+statements. As you might expect, you use ``async for`` to iterate
+over them. :pep:`525` has many more details if you want them.
+
+For example, the following is a roundabout way to print
+the numbers 0 through 9 with a 1-second delay before each one::
+
+    async def range_slowly(*args):
+        """Like range(), but adds a 1-second sleep before each value."""
+        for value in range(*args):
+            await trio.sleep(1)
+            yield value
+
+    async def use_it():
+        async for value in range_slowly(10):
+            print(value)
+
+    trio.run(use_it)
+
+Trio supports async generators, with some caveats described in this section.
+
+Finalization
+~~~~~~~~~~~~
+
+If you iterate over an async generator in its entirety, like the
+example above does, then the execution of the async generator will
+occur completely in the context of the code that's iterating over it,
+and there aren't too many surprises.
+
+If you abandon a partially-completed async generator, though, such as
+by ``break``\ing out of the iteration, things aren't so simple.  The
+async generator iterator object is still alive, waiting for you to
+resume iterating it so it can produce more values. At some point,
+Python will realize that you've dropped all references to the
+iterator, and will call on Trio to throw in a `GeneratorExit` exception
+so that any remaining cleanup code inside the generator has a chance
+to run: ``finally`` blocks, ``__aexit__`` handlers, and so on.
+
+So far, so good. Unfortunately, Python provides no guarantees about
+*when* this happens. It could be as soon as you break out of the
+``async for`` loop, or an arbitrary amount of time later. It could
+even be after the entire Trio run has finished! Just about the only
+guarantee is that it *won't* happen in the task that was using the
+generator. That task will continue on with whatever else it's doing,
+and the async generator cleanup will happen "sometime later,
+somewhere else": potentially with different context variables,
+not subject to timeouts, and/or after any nurseries you're using have
+been closed.
+
+If you don't like that ambiguity, and you want to ensure that a
+generator's ``finally`` blocks and ``__aexit__`` handlers execute as
+soon as you're done using it, then you'll need to wrap your use of the
+generator in something like `async_generator.aclosing()
+<https://async-generator.readthedocs.io/en/latest/reference.html#context-managers>`__::
+
+    # Instead of this:
+    async for value in my_generator():
+        if value == 42:
+            break
+
+    # Do this:
+    async with aclosing(my_generator()) as aiter:
+        async for value in aiter:
+            if value == 42:
+                break
+
+This is cumbersome, but Python unfortunately doesn't provide any other
+reliable options. If you use ``aclosing()``, then
+your generator's cleanup code executes in the same context as the
+rest of its iterations, so timeouts, exceptions, and context
+variables work like you'd expect.
+
+If you don't use ``aclosing()``, then Trio will do
+its best anyway, but you'll have to contend with the following semantics:
+
+* The cleanup of the generator occurs in a cancelled context, i.e.,
+  all blocking calls executed during cleanup will raise `Cancelled`.
+  This is to compensate for the fact that any timeouts surrounding
+  the original use of the generator have been long since forgotten.
+
+* The cleanup runs without access to any :ref:`context variables
+  <task-local-storage>` that may have been present when the generator
+  was originally being used.
+
+* If the generator raises an exception during cleanup, then it's
+  printed to the ``trio.async_generator_errors`` logger and otherwise
+  ignored.
+
+* If an async generator is still alive at the end of the whole
+  call to :func:`trio.run`, then it will be cleaned up after all
+  tasks have exited and before :func:`trio.run` returns.
+  Since the "system nursery" has already been closed at this point,
+  Trio isn't able to support any new calls to
+  :func:`trio.lowlevel.spawn_system_task`.
+
+If you plan to run your code on PyPy to take advantage of its better
+performance, you should be aware that PyPy is *far more likely* than
+CPython to perform async generator cleanup at a time well after the
+last use of the generator. (This is a consequence of the fact that
+PyPy does not use reference counting to manage memory.)  To help catch
+issues like this, Trio will issue a `ResourceWarning` (ignored by
+default, but enabled when running under ``python -X dev`` for example)
+for each async generator that needs to be handled through the fallback
+finalization path.
+
+Cancel scopes and nurseries
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+.. warning:: You may not write a ``yield`` statement that suspends an async generator
+   inside a `CancelScope` or `Nursery` that was entered within the generator.
+
+That is, this is OK::
+
+    async def some_agen():
+        with trio.move_on_after(1):
+            await long_operation()
+        yield "first"
+        async with trio.open_nursery() as nursery:
+            nursery.start_soon(task1)
+            nursery.start_soon(task2)
+        yield "second"
+        ...
+
+But this is not::
+
+    async def some_agen():
+        with trio.move_on_after(1):
+            yield "first"
+        async with trio.open_nursery() as nursery:
+            yield "second"
+        ...
+
+Async generators decorated with ``@asynccontextmanager`` to serve as
+the template for an async context manager are *not* subject to this
+constraint, because ``@asynccontextmanager`` uses them in a limited
+way that doesn't create problems.
+
+Violating the rule described in this section will sometimes get you a
+useful error message, but Trio is not able to detect all such cases,
+so sometimes you'll get an unhelpful `TrioInternalError`. (And
+sometimes it will seem to work, which is probably the worst outcome of
+all, since then you might not notice the issue until you perform some
+minor refactoring of the generator or the code that's iterating it, or
+just get unlucky. There is a `proposed Python enhancement
+<https://discuss.python.org/t/preventing-yield-inside-certain-context-managers/1091>`__
+that would at least make it fail consistently.)
+
+The reason for the restriction on cancel scopes has to do with the
+difficulty of noticing when a generator gets suspended and
+resumed. The cancel scopes inside the generator shouldn't affect code
+running outside the generator, but Trio isn't involved in the process
+of exiting and reentering the generator, so it would be hard pressed
+to keep its cancellation plumbing in the correct state. Nurseries
+use a cancel scope internally, so they have all the problems of cancel
+scopes plus a number of problems of their own: for example, when
+the generator is suspended, what should the background tasks do?
+There's no good way to suspend them, but if they keep running and throw
+an exception, where can that exception be reraised?
+
+If you have an async generator that wants to ``yield`` from within a nursery
+or cancel scope, your best bet is to refactor it to be a separate task
+that communicates over memory channels.
+
+For more discussion and some experimental partial workarounds, see
+Trio issues `264 <https://github.com/python-trio/trio/issues/264>`__
+(especially `this comment
+<https://github.com/python-trio/trio/issues/264#issuecomment-418989328>`__)
+and `638 <https://github.com/python-trio/trio/issues/638>`__.
+
+
 .. _threads:
 
 Threads (if you must)
diff --git a/newsfragments/265.headline.rst b/newsfragments/265.headline.rst
new file mode 100644
index 00000000..57a1ae30
--- /dev/null
+++ b/newsfragments/265.headline.rst
@@ -0,0 +1,6 @@
+Trio now supports automatic :ref:`async generator finalization
+<async-generators>`, so more async generators will work even if you
+don't wrap them in ``async with async_generator.aclosing():``
+blocks. Please see the documentation for important caveats; in
+particular, yielding within a nursery or cancel scope remains
+unsupported.
diff --git a/trio/_core/_asyncgens.py b/trio/_core/_asyncgens.py
new file mode 100644
index 00000000..cb88a1d5
--- /dev/null
+++ b/trio/_core/_asyncgens.py
@@ -0,0 +1,193 @@
+import attr
+import logging
+import sys
+import warnings
+import weakref
+
+from .._util import name_asyncgen
+from . import _run
+from .. import _core
+
+# Used to log exceptions in async generator finalizers
+ASYNCGEN_LOGGER = logging.getLogger("trio.async_generator_errors")
+
+
+@attr.s(eq=False, slots=True)
+class AsyncGenerators:
+    # Async generators are added to this set when first iterated. Any
+    # left after the main task exits will be closed before trio.run()
+    # returns.  During most of the run, this is a WeakSet so GC works.
+    # During shutdown, when we're finalizing all the remaining
+    # asyncgens after the system nursery has been closed, it's a
+    # regular set so we don't have to deal with GC firing at
+    # unexpected times.
+    alive = attr.ib(factory=weakref.WeakSet)
+
+    # This collects async generators that get garbage collected during
+    # the one-tick window between the system nursery closing and the
+    # init task starting end-of-run asyncgen finalization.
+    trailing_needs_finalize = attr.ib(factory=set)
+
+    prev_hooks = attr.ib(init=False)
+
+    def install_hooks(self, runner):
+        def firstiter(agen):
+            if hasattr(_run.GLOBAL_RUN_CONTEXT, "task"):
+                self.alive.add(agen)
+            else:
+                # An async generator first iterated outside of a Trio
+                # task doesn't belong to Trio. Probably we're in guest
+                # mode and the async generator belongs to our host.
+                # The locals dictionary is the only good place to
+                # remember this fact, at least until
+                # https://bugs.python.org/issue40916 is implemented.
+                agen.ag_frame.f_locals["@trio_foreign_asyncgen"] = True
+                if self.prev_hooks.firstiter is not None:
+                    self.prev_hooks.firstiter(agen)
+
+        def finalize_in_trio_context(agen, agen_name):
+            try:
+                runner.spawn_system_task(
+                    self._finalize_one,
+                    agen,
+                    agen_name,
+                    name=f"close asyncgen {agen_name} (abandoned)",
+                )
+            except RuntimeError:
+                # There is a one-tick window where the system nursery
+                # is closed but the init task hasn't yet made
+                # self.asyncgens a strong set to disable GC. We seem to
+                # have hit it.
+                self.trailing_needs_finalize.add(agen)
+
+        def finalizer(agen):
+            agen_name = name_asyncgen(agen)
+            try:
+                is_ours = not agen.ag_frame.f_locals.get("@trio_foreign_asyncgen")
+            except AttributeError:  # pragma: no cover
+                is_ours = True
+
+            if is_ours:
+                runner.entry_queue.run_sync_soon(
+                    finalize_in_trio_context, agen, agen_name
+                )
+
+                # Do this last, because it might raise an exception
+                # depending on the user's warnings filter. (That
+                # exception will be printed to the terminal and
+                # ignored, since we're running in GC context.)
+                warnings.warn(
+                    f"Async generator {agen_name!r} was garbage collected before it "
+                    f"had been exhausted. Surround its use in 'async with "
+                    f"aclosing(...):' to ensure that it gets cleaned up as soon as "
+                    f"you're done using it.",
+                    ResourceWarning,
+                    stacklevel=2,
+                    source=agen,
+                )
+            else:
+                # Not ours -> forward to the host loop's async generator finalizer
+                if self.prev_hooks.finalizer is not None:
+                    self.prev_hooks.finalizer(agen)
+                else:
+                    # Host has no finalizer.  Reimplement the default
+                    # Python behavior with no hooks installed: throw in
+                    # GeneratorExit, step once, raise RuntimeError if
+                    # it doesn't exit.
+                    closer = agen.aclose()
+                    try:
+                        # If the next thing is a yield, this will raise RuntimeError
+                        # which we allow to propagate
+                        closer.send(None)
+                    except StopIteration:
+                        pass
+                    else:
+                        # If the next thing is an await, we get here. Give a nicer
+                        # error than the default "async generator ignored GeneratorExit"
+                        raise RuntimeError(
+                            f"Non-Trio async generator {agen_name!r} awaited something "
+                            f"during finalization; install a finalization hook to "
+                            f"support this, or wrap it in 'async with aclosing(...):'"
+                        )
+
+        self.prev_hooks = sys.get_asyncgen_hooks()
+        sys.set_asyncgen_hooks(firstiter=firstiter, finalizer=finalizer)
+
+    async def finalize_remaining(self, runner):
+        # This is called from init after shutting down the system nursery.
+        # The only tasks running at this point are init and
+        # the run_sync_soon task, and since the system nursery is closed,
+        # there's no way for user code to spawn more.
+        assert _core.current_task() is runner.init_task
+        assert len(runner.tasks) == 2
+
+        # To make async generator finalization easier to reason
+        # about, we'll shut down asyncgen garbage collection by turning
+        # the alive WeakSet into a regular set.
+        self.alive = set(self.alive)
+
+        # Process all pending run_sync_soon callbacks, in case one of
+        # them was an asyncgen finalizer that snuck in under the wire.
+        runner.entry_queue.run_sync_soon(runner.reschedule, runner.init_task)
+        await _core.wait_task_rescheduled(
+            lambda _: _core.Abort.FAILED  # pragma: no cover
+        )
+        self.alive.update(self.trailing_needs_finalize)
+        self.trailing_needs_finalize.clear()
+
+        # None of the still-living tasks use async generators, so
+        # every async generator must be suspended at a yield point --
+        # there's no one to be doing the iteration. That's good,
+        # because aclose() only works on an asyncgen that's suspended
+        # at a yield point.  (If it's suspended at an event loop trap,
+        # because someone is in the middle of iterating it, then you
+        # get a RuntimeError on 3.8+, and a nasty surprise on earlier
+        # versions due to https://bugs.python.org/issue32526.)
+        #
+        # However, once we start aclose() of one async generator, it
+        # might start fetching the next value from another, thus
+        # preventing us from closing that other (at least until
+        # aclose() of the first one is complete).  This constraint
+        # effectively requires us to finalize the remaining asyncgens
+        # in arbitrary order, rather than doing all of them at the
+        # same time. On 3.8+ we could defer any generator with
+        # ag_running=True to a later batch, but that only catches
+        # the case where our aclose() starts after the user's
+        # asend()/etc. If our aclose() starts first, then the
+        # user's asend()/etc will raise RuntimeError, since they're
+        # probably not checking ag_running.
+        #
+        # It might be possible to allow some parallelized cleanup if
+        # we can determine that a certain set of asyncgens have no
+        # interdependencies, using gc.get_referents() and such.
+        # But just doing one at a time will typically work well enough
+        # (since each aclose() executes in a cancelled scope) and
+        # is much easier to reason about.
+
+        # It's possible that that cleanup code will itself create
+        # more async generators, so we iterate repeatedly until
+        # all are gone.
+        while self.alive:
+            batch = self.alive
+            self.alive = set()
+            for agen in batch:
+                await self._finalize_one(agen, name_asyncgen(agen))
+
+    def close(self):
+        sys.set_asyncgen_hooks(*self.prev_hooks)
+
+    async def _finalize_one(self, agen, name):
+        try:
+            # This shield ensures that finalize_asyncgen never exits
+            # with an exception, not even a Cancelled. The inside
+            # is cancelled so there's no deadlock risk.
+            with _core.CancelScope(shield=True) as cancel_scope:
+                cancel_scope.cancel()
+                await agen.aclose()
+        except BaseException:
+            ASYNCGEN_LOGGER.exception(
+                "Exception ignored during finalization of async generator %r -- "
+                "surround your use of the generator in 'async with aclosing(...):' "
+                "to raise exceptions like this in the context where they're generated",
+                name,
+            )
diff --git a/trio/_core/_entry_queue.py b/trio/_core/_entry_queue.py
index 791ab8ca..a1587a18 100644
--- a/trio/_core/_entry_queue.py
+++ b/trio/_core/_entry_queue.py
@@ -56,7 +56,15 @@ class EntryQueue:
                 async def kill_everything(exc):
                     raise exc
 
-                _core.spawn_system_task(kill_everything, exc)
+                try:
+                    _core.spawn_system_task(kill_everything, exc)
+                except RuntimeError:
+                    # We're quite late in the shutdown process and the
+                    # system nursery is already closed.
+                    # TODO(2020-06): this is a gross hack and should
+                    # be fixed soon when we address #1607.
+                    _core.current_task().parent_nursery.start_soon(kill_everything, exc)
+
             return True
 
         # This has to be carefully written to be safe in the face of new items
@@ -102,10 +110,6 @@ class EntryQueue:
     def size(self):
         return len(self.queue) + len(self.idempotent_queue)
 
-    def spawn(self):
-        name = "<TrioToken.run_sync_soon task>"
-        _core.spawn_system_task(self.task, name=name)
-
     def run_sync_soon(self, sync_fn, *args, idempotent=False):
         with self.lock:
             if self.done:
diff --git a/trio/_core/_run.py b/trio/_core/_run.py
index 6e909785..55cc4a15 100644
--- a/trio/_core/_run.py
+++ b/trio/_core/_run.py
@@ -9,8 +9,10 @@ import select
 import sys
 import threading
 from collections import deque
+import collections.abc
 from contextlib import contextmanager
 import warnings
+import weakref
 import enum
 
 from contextvars import copy_context
@@ -41,6 +43,7 @@ from ._traps import (
     PermanentlyDetachCoroutineObject,
     WaitTaskRescheduled,
 )
+from ._asyncgens import AsyncGenerators
 from ._thread_cache import start_thread_soon
 from ._instrumentation import Instruments
 from .. import _core
@@ -1289,6 +1292,7 @@ class Runner:
 
     entry_queue = attr.ib(factory=EntryQueue)
     trio_token = attr.ib(default=None)
+    asyncgens = attr.ib(factory=AsyncGenerators)
 
     # If everything goes idle for this long, we call clock._autojump()
     clock_autojump_threshold = attr.ib(default=inf)
@@ -1306,6 +1310,7 @@ class Runner:
     def close(self):
         self.io_manager.close()
         self.entry_queue.close()
+        self.asyncgens.close()
         if "after_run" in self.instruments:
             self.instruments.call("after_run")
         # This is where KI protection gets disabled, so we do it last
@@ -1499,11 +1504,7 @@ class Runner:
 
         task._activate_cancel_status(None)
         self.tasks.remove(task)
-        if task is self.main_task:
-            self.main_task_outcome = outcome
-            self.system_nursery.cancel_scope.cancel()
-            self.system_nursery._child_finished(task, Value(None))
-        elif task is self.init_task:
+        if task is self.init_task:
             # If the init task crashed, then something is very wrong and we
             # let the error propagate. (It'll eventually be wrapped in a
             # TrioInternalError.)
@@ -1513,6 +1514,9 @@ class Runner:
             if self.tasks:  # pragma: no cover
                 raise TrioInternalError
         else:
+            if task is self.main_task:
+                self.main_task_outcome = outcome
+                outcome = Value(None)
             task._parent_nursery._child_finished(task, outcome)
 
         if "task_exited" in self.instruments:
@@ -1567,14 +1571,37 @@ class Runner:
         )
 
     async def init(self, async_fn, args):
-        async with open_nursery() as system_nursery:
-            self.system_nursery = system_nursery
-            try:
-                self.main_task = self.spawn_impl(async_fn, args, system_nursery, None)
-            except BaseException as exc:
-                self.main_task_outcome = Error(exc)
-                system_nursery.cancel_scope.cancel()
-            self.entry_queue.spawn()
+        # run_sync_soon task runs here:
+        async with open_nursery() as run_sync_soon_nursery:
+            # All other system tasks run here:
+            async with open_nursery() as self.system_nursery:
+                # Only the main task runs here:
+                async with open_nursery() as main_task_nursery:
+                    try:
+                        self.main_task = self.spawn_impl(
+                            async_fn, args, main_task_nursery, None
+                        )
+                    except BaseException as exc:
+                        self.main_task_outcome = Error(exc)
+                        return
+                    self.spawn_impl(
+                        self.entry_queue.task,
+                        (),
+                        run_sync_soon_nursery,
+                        "<TrioToken.run_sync_soon task>",
+                        system_task=True,
+                    )
+
+                # Main task is done; start shutting down system tasks
+                self.system_nursery.cancel_scope.cancel()
+
+            # System nursery is closed; finalize remaining async generators
+            await self.asyncgens.finalize_remaining(self)
+
+            # There are no more asyncgens, which means no more user-provided
+            # code except possibly run_sync_soon callbacks. It's finally safe
+            # to stop the run_sync_soon task and exit run().
+            run_sync_soon_nursery.cancel_scope.cancel()
 
     ################
     # Outside context problems
@@ -1799,6 +1826,7 @@ def setup_runner(clock, instruments, restrict_keyboard_interrupt_to_checkpoints)
         system_context=system_context,
         ki_manager=ki_manager,
     )
+    runner.asyncgens.install_hooks(runner)
 
     # This is where KI protection gets enabled, so we want to do it early - in
     # particular before we start modifying global state like GLOBAL_RUN_CONTEXT
diff --git a/trio/_util.py b/trio/_util.py
index 4f5475ef..f1e8f9eb 100644
--- a/trio/_util.py
+++ b/trio/_util.py
@@ -354,3 +354,20 @@ class NoPublicConstructor(Final):
 
     def _create(cls: t.Type[T], *args: t.Any, **kwargs: t.Any) -> T:
         return super().__call__(*args, **kwargs)  # type: ignore
+
+
+def name_asyncgen(agen):
+    """Return the fully-qualified name of the async generator function
+    that produced the async generator iterator *agen*.
+    """
+    if not hasattr(agen, "ag_code"):  # pragma: no cover
+        return repr(agen)
+    try:
+        module = agen.ag_frame.f_globals["__name__"]
+    except (AttributeError, KeyError):
+        module = "<{}>".format(agen.ag_code.co_filename)
+    try:
+        qualname = agen.__qualname__
+    except AttributeError:
+        qualname = agen.ag_code.co_name
+    return f"{module}.{qualname}"

Implement a strategy for handling (async) generator cleanup
There's a problem with generators and async generators: they don't necessarily iterate to the end. Sometimes they get suspended in the middle, and then garbage collected. When this happens, their `__del__` method `throw`s in a `GeneratorExit` exception, so that `finally` blocks and `__exit__` methods get a chance to run. BUT, `__del__` is very weird: it can be run in the middle of any arbitrary python bytecode instruction. It's like a signal handler but *even more so*.

So for example, this code:
```python
def generator_fn():
    try:
        yield
    finally:
        print(trio.current_task())

async def caller():
    for _ in generator_fn():
        break
```
Here the generator's `finally` block gets run at some totally arbitrary time, so it could in principle print *any* task, or crash because no task is currently active, or because it doesn't even run in the trio thread. And similar problems arise for pretty much any access to trio APIs. (Note that even without trio, the same thing happens for `threading.current_thread()`, accessing thread-local storage, etc. But it trio the use of this kind of implicit thread-bound state is much more common, since the run loop itself is accessed via thread-local storage.)

[PEP 533](https://www.python.org/dev/peps/pep-0533/) would be the real fix, but that seems to be stalled for now.

For regular generators, we just need to document this; I can't see what else we can do. There are some related issues in #264 that also need documentation; probably they should go in the same section of the manual.

For async generators, this problem is both better and worse. The thing about async generators is that the gc can't actually clean them up directly: it would like to do the same thing as it does for regular generators and have `__del__` throw in `GeneratorExit`, but this doesn't really work, because for an async generator you should use `athrow` and let the coroutine runner iterate it, but `__del__` doesn't have access to a coroutine runner. So PEP 525 defines some hooks that trio can set, to get (a) notification that an async generator has been started, and (b) notification when an async generator is ready for garbage collection.

The intention was that these hooks are used to let the event loop take care of doing `gen.athrow(GeneratorExit)` to emulate how regular generators work. But as we saw above, regular generators are already broken so... we don't want to emulate them. Being able to re-enter trio's run loop doesn't help, because we need a task context. And this is even more true for async generators, since the whole point of an async generator is that it can do async-y things, and if you want to `await` something you need to make sure you have a timeout set, which means you *really really really* need to make sure that this happens in the original task context where the user was expecting it to happen. Letting an `await` sneak outside of its `with fail_after` is definitely not OK.

So we aren't going to use these hooks in the way they were originally intended. But, until PEP 533 becomes a thing, maybe we can at least use these hooks for harm reduction.

Some options:

* Don't define the hooks at all. In this case I'm not 100% sure what Python does, but I think it does something like: throw in `GeneratorExit` and iterate once, and if it yields again (e.g. because there's an `async with` or a `finally: await ...`) then it gives up and prints an error message. This is basically like synchronous generators, but maybe a little less friendly. It's not *awful*, but maybe we can do better.

* Use the `firstiter` hook to forbid the use of async generators, unless we have some explicit evidence that someone is taking responsibility for cleaning them up. For example, we could make it so that our @acontextmanager marks its async generators as being cleaned up (though 3.7's `contextlib.asynccontextmanager` is going to be a problem...), and provide an `aclosing` context manager that does the same. This approach was pioneered by Curio, which goes a step further and introspects the generator's bytecode, and if it doesn't see any dangerous operations (`yield` or `yield from` inside an `except`, `finally`, `with`, or `async with`), then it allows those async generators as well. I really don't want to have this kind of bytecode introspection in trio. OTOH without it then async comprehensions like `(2 * i async for i in aiter)` become essentially unusable.

* Use the `finalizer` hook to catch when an async generator is cleaned up, and run it in a context that (a) is carefully set up so that any attempt to access trio state generates a clear error message, and (b) any attempt to `yield` causes a clear error / crashes the program (as opposed to just being printed to stderr, like regular `__del__` does). (If doing this we'd probably also want a `firstiter` hook like asyncio's, so we can force all async generators to be finalized when the run loop exits. This is just a minor refinement, though.)

The good thing about the `finalizer` approach is that it has no false negatives: all async generators that can be run safely are allowed to run. The downside is that it has false positives: if an async generator is written poorly, then it's possible that it will be allowed to run and everything will look fine in tests, but then in production it blows up (e.g. because in tests it's always iterated to the end, and in production someone breaks out of a loop so the generator gets collected while only half-finished). The `firstiter` approach is the opposite: it only allows generators to run if it can be sure that they're safe in all circumstances.

(If PEP 533 is too much for 3.7, then maybe a smaller ask would be set_gen_hooks() that work the same way but for synchronous generators?)
diff --git a/trio/_core/tests/test_asyncgen.py b/trio/_core/tests/test_asyncgen.py
new file mode 100644
index 00000000..dd15da9b
--- /dev/null
+++ b/trio/_core/tests/test_asyncgen.py
@@ -0,0 +1,316 @@
+import sys
+import weakref
+import pytest
+from math import inf
+from functools import partial
+from async_generator import aclosing
+from ... import _core
+from .tutil import gc_collect_harder, buggy_pypy_asyncgens
+
+
+def test_asyncgen_basics():
+    collected = []
+
+    async def example(cause):
+        try:
+            try:
+                yield 42
+            except GeneratorExit:
+                pass
+            await _core.checkpoint()
+        except _core.Cancelled:
+            assert "exhausted" not in cause
+            task_name = _core.current_task().name
+            assert cause in task_name or task_name == "<init>"
+            assert _core.current_effective_deadline() == -inf
+            with pytest.raises(_core.Cancelled):
+                await _core.checkpoint()
+            collected.append(cause)
+        else:
+            assert "async_main" in _core.current_task().name
+            assert "exhausted" in cause
+            assert _core.current_effective_deadline() == inf
+            await _core.checkpoint()
+            collected.append(cause)
+
+    saved = []
+
+    async def async_main():
+        # GC'ed before exhausted
+        with pytest.warns(
+            ResourceWarning, match="Async generator.*collected before.*exhausted",
+        ):
+            assert 42 == await example("abandoned").asend(None)
+            gc_collect_harder()
+        await _core.wait_all_tasks_blocked()
+        assert collected.pop() == "abandoned"
+
+        # aclosing() ensures it's cleaned up at point of use
+        async with aclosing(example("exhausted 1")) as aiter:
+            assert 42 == await aiter.asend(None)
+        assert collected.pop() == "exhausted 1"
+
+        # Also fine if you exhaust it at point of use
+        async for val in example("exhausted 2"):
+            assert val == 42
+        assert collected.pop() == "exhausted 2"
+
+        gc_collect_harder()
+
+        # No problems saving the geniter when using either of these patterns
+        async with aclosing(example("exhausted 3")) as aiter:
+            saved.append(aiter)
+            assert 42 == await aiter.asend(None)
+        assert collected.pop() == "exhausted 3"
+
+        # Also fine if you exhaust it at point of use
+        saved.append(example("exhausted 4"))
+        async for val in saved[-1]:
+            assert val == 42
+        assert collected.pop() == "exhausted 4"
+
+        # Leave one referenced-but-unexhausted and make sure it gets cleaned up
+        if buggy_pypy_asyncgens:
+            collected.append("outlived run")
+        else:
+            saved.append(example("outlived run"))
+            assert 42 == await saved[-1].asend(None)
+            assert collected == []
+
+    _core.run(async_main)
+    assert collected.pop() == "outlived run"
+    for agen in saved:
+        assert agen.ag_frame is None  # all should now be exhausted
+
+
+async def test_asyncgen_throws_during_finalization(caplog):
+    record = []
+
+    async def agen():
+        try:
+            yield 1
+        finally:
+            await _core.cancel_shielded_checkpoint()
+            record.append("crashing")
+            raise ValueError("oops")
+
+    await agen().asend(None)
+    gc_collect_harder()
+    await _core.wait_all_tasks_blocked()
+    assert record == ["crashing"]
+    exc_type, exc_value, exc_traceback = caplog.records[0].exc_info
+    assert exc_type is ValueError
+    assert str(exc_value) == "oops"
+    assert "during finalization of async generator" in caplog.records[0].message
+
+
+@pytest.mark.skipif(buggy_pypy_asyncgens, reason="pypy 7.2.0 is buggy")
+def test_firstiter_after_closing():
+    saved = []
+    record = []
+
+    async def funky_agen():
+        try:
+            yield 1
+        except GeneratorExit:
+            record.append("cleanup 1")
+            raise
+        try:
+            yield 2
+        finally:
+            record.append("cleanup 2")
+            await funky_agen().asend(None)
+
+    async def async_main():
+        aiter = funky_agen()
+        saved.append(aiter)
+        assert 1 == await aiter.asend(None)
+        assert 2 == await aiter.asend(None)
+
+    _core.run(async_main)
+    assert record == ["cleanup 2", "cleanup 1"]
+
+
+@pytest.mark.skipif(buggy_pypy_asyncgens, reason="pypy 7.2.0 is buggy")
+def test_interdependent_asyncgen_cleanup_order():
+    saved = []
+    record = []
+
+    async def innermost():
+        try:
+            yield 1
+        finally:
+            await _core.cancel_shielded_checkpoint()
+            record.append("innermost")
+
+    async def agen(label, inner):
+        try:
+            yield await inner.asend(None)
+        finally:
+            # Either `inner` has already been cleaned up, or
+            # we're about to exhaust it. Either way, we wind
+            # up with `record` containing the labels in
+            # innermost-to-outermost order.
+            with pytest.raises(StopAsyncIteration):
+                await inner.asend(None)
+            record.append(label)
+
+    async def async_main():
+        # This makes a chain of 101 interdependent asyncgens:
+        # agen(99)'s cleanup will iterate agen(98)'s will iterate
+        # ... agen(0)'s will iterate innermost()'s
+        ag_chain = innermost()
+        for idx in range(100):
+            ag_chain = agen(idx, ag_chain)
+        saved.append(ag_chain)
+        assert 1 == await ag_chain.asend(None)
+        assert record == []
+
+    _core.run(async_main)
+    assert record == ["innermost"] + list(range(100))
+
+
+def test_last_minute_gc_edge_case():
+    saved = []
+    record = []
+    needs_retry = True
+
+    async def agen():
+        try:
+            yield 1
+        finally:
+            record.append("cleaned up")
+
+    def collect_at_opportune_moment(token):
+        runner = _core._run.GLOBAL_RUN_CONTEXT.runner
+        if runner.system_nursery._closed and isinstance(
+            runner.asyncgens.alive, weakref.WeakSet
+        ):
+            saved.clear()
+            record.append("final collection")
+            gc_collect_harder()
+            record.append("done")
+        else:
+            try:
+                token.run_sync_soon(collect_at_opportune_moment, token)
+            except _core.RunFinishedError:  # pragma: no cover
+                nonlocal needs_retry
+                needs_retry = True
+
+    async def async_main():
+        token = _core.current_trio_token()
+        token.run_sync_soon(collect_at_opportune_moment, token)
+        saved.append(agen())
+        await saved[-1].asend(None)
+
+    # Actually running into the edge case requires that the run_sync_soon task
+    # execute in between the system nursery's closure and the strong-ification
+    # of runner.asyncgens. There's about a 25% chance that it doesn't
+    # (if the run_sync_soon task runs before init on one tick and after init
+    # on the next tick); if we try enough times, we can make the chance of
+    # failure as small as we want.
+    for attempt in range(50):
+        needs_retry = False
+        del record[:]
+        del saved[:]
+        _core.run(async_main)
+        if needs_retry:  # pragma: no cover
+            if not buggy_pypy_asyncgens:
+                assert record == ["cleaned up"]
+        else:
+            assert record == ["final collection", "done", "cleaned up"]
+            break
+    else:  # pragma: no cover
+        pytest.fail(
+            f"Didn't manage to hit the trailing_finalizer_asyncgens case "
+            f"despite trying {attempt} times"
+        )
+
+
+async def step_outside_async_context(aiter):
+    # abort_fns run outside of task context, at least if they're
+    # triggered by a deadline expiry rather than a direct
+    # cancellation.  Thus, an asyncgen first iterated inside one
+    # will appear non-Trio, and since no other hooks were installed,
+    # will use the last-ditch fallback handling (that tries to mimic
+    # CPython's behavior with no hooks).
+    #
+    # NB: the strangeness with aiter being an attribute of abort_fn is
+    # to make it as easy as possible to ensure we don't hang onto a
+    # reference to aiter inside the guts of the run loop.
+    def abort_fn(_):
+        with pytest.raises(StopIteration, match="42"):
+            abort_fn.aiter.asend(None).send(None)
+        del abort_fn.aiter
+        return _core.Abort.SUCCEEDED
+
+    abort_fn.aiter = aiter
+
+    async with _core.open_nursery() as nursery:
+        nursery.start_soon(_core.wait_task_rescheduled, abort_fn)
+        await _core.wait_all_tasks_blocked()
+        nursery.cancel_scope.deadline = _core.current_time()
+
+
+@pytest.mark.skipif(buggy_pypy_asyncgens, reason="pypy 7.2.0 is buggy")
+async def test_fallback_when_no_hook_claims_it(capsys):
+    async def well_behaved():
+        yield 42
+
+    async def yields_after_yield():
+        with pytest.raises(GeneratorExit):
+            yield 42
+        yield 100
+
+    async def awaits_after_yield():
+        with pytest.raises(GeneratorExit):
+            yield 42
+        await _core.cancel_shielded_checkpoint()
+
+    await step_outside_async_context(well_behaved())
+    gc_collect_harder()
+    assert capsys.readouterr().err == ""
+
+    await step_outside_async_context(yields_after_yield())
+    gc_collect_harder()
+    assert "ignored GeneratorExit" in capsys.readouterr().err
+
+    await step_outside_async_context(awaits_after_yield())
+    gc_collect_harder()
+    assert "awaited something during finalization" in capsys.readouterr().err
+
+
+@pytest.mark.skipif(buggy_pypy_asyncgens, reason="pypy 7.2.0 is buggy")
+def test_delegation_to_existing_hooks():
+    record = []
+
+    def my_firstiter(agen):
+        record.append("firstiter " + agen.ag_frame.f_locals["arg"])
+
+    def my_finalizer(agen):
+        record.append("finalizer " + agen.ag_frame.f_locals["arg"])
+
+    async def example(arg):
+        try:
+            yield 42
+        finally:
+            with pytest.raises(_core.Cancelled):
+                await _core.checkpoint()
+            record.append("trio collected " + arg)
+
+    async def async_main():
+        await step_outside_async_context(example("theirs"))
+        assert 42 == await example("ours").asend(None)
+        gc_collect_harder()
+        assert record == ["firstiter theirs", "finalizer theirs"]
+        record[:] = []
+        await _core.wait_all_tasks_blocked()
+        assert record == ["trio collected ours"]
+
+    old_hooks = sys.get_asyncgen_hooks()
+    sys.set_asyncgen_hooks(my_firstiter, my_finalizer)
+    try:
+        _core.run(async_main)
+    finally:
+        assert sys.get_asyncgen_hooks() == (my_firstiter, my_finalizer)
+        sys.set_asyncgen_hooks(*old_hooks)
diff --git a/trio/_core/tests/test_guest_mode.py b/trio/_core/tests/test_guest_mode.py
index e58ec3eb..381d966f 100644
--- a/trio/_core/tests/test_guest_mode.py
+++ b/trio/_core/tests/test_guest_mode.py
@@ -1,5 +1,7 @@
 import pytest
 import asyncio
+import contextvars
+import sys
 import traceback
 import queue
 from functools import partial
@@ -11,7 +13,7 @@ import time
 
 import trio
 import trio.testing
-from .tutil import gc_collect_harder
+from .tutil import gc_collect_harder, buggy_pypy_asyncgens
 from ..._util import signal_raise
 
 # The simplest possible "host" loop.
@@ -497,3 +499,45 @@ def test_guest_mode_autojump_clock_threshold_changing():
     # Should be basically instantaneous, but we'll leave a generous buffer to
     # account for any CI weirdness
     assert end - start < DURATION / 2
+
+
+@pytest.mark.skipif(buggy_pypy_asyncgens, reason="PyPy 7.2 is buggy")
+def test_guest_mode_asyncgens():
+    import sniffio
+
+    record = set()
+
+    async def agen(label):
+        assert sniffio.current_async_library() == label
+        try:
+            yield 1
+        finally:
+            library = sniffio.current_async_library()
+            try:
+                await sys.modules[library].sleep(0)
+            except trio.Cancelled:
+                pass
+            record.add((label, library))
+
+    async def iterate_in_aio():
+        # "trio" gets inherited from our Trio caller if we don't set this
+        sniffio.current_async_library_cvar.set("asyncio")
+        await agen("asyncio").asend(None)
+
+    async def trio_main():
+        task = asyncio.ensure_future(iterate_in_aio())
+        done_evt = trio.Event()
+        task.add_done_callback(lambda _: done_evt.set())
+        with trio.fail_after(1):
+            await done_evt.wait()
+
+        await agen("trio").asend(None)
+
+        gc_collect_harder()
+
+    # Ensure we don't pollute the thread-level context if run under
+    # an asyncio without contextvars support (3.6)
+    context = contextvars.copy_context()
+    context.run(aiotrio_run, trio_main, host_uses_signal_set_wakeup_fd=True)
+
+    assert record == {("asyncio", "asyncio"), ("trio", "trio")}
diff --git a/trio/_core/tests/test_instrumentation.py b/trio/_core/tests/test_instrumentation.py
index 946caf79..57d3461d 100644
--- a/trio/_core/tests/test_instrumentation.py
+++ b/trio/_core/tests/test_instrumentation.py
@@ -73,7 +73,6 @@ def test_instruments(recwarn):
         + [("before", task), ("schedule", task), ("after", task)] * 5
         + [("before", task), ("after", task), ("after_run",)]
     )
-    assert len(r1.record) > len(r2.record) > len(r3.record)
     assert r1.record == r2.record + r3.record
     assert list(r1.filter_tasks([task])) == expected
 
diff --git a/trio/_core/tests/test_run.py b/trio/_core/tests/test_run.py
index 04512f7b..1591983d 100644
--- a/trio/_core/tests/test_run.py
+++ b/trio/_core/tests/test_run.py
@@ -20,6 +20,7 @@ from .tutil import (
     check_sequence_matches,
     gc_collect_harder,
     ignore_coroutine_never_awaited_warnings,
+    buggy_pypy_asyncgens,
 )
 
 from ... import _core
@@ -1331,6 +1332,33 @@ async def test_TrioToken_run_sync_soon_massive_queue():
     assert counter[0] == COUNT
 
 
+@pytest.mark.skipif(buggy_pypy_asyncgens, reason="PyPy 7.2 is buggy")
+def test_TrioToken_run_sync_soon_late_crash():
+    # Crash after system nursery is closed -- easiest way to do that is
+    # from an async generator finalizer.
+    record = []
+    saved = []
+
+    async def agen():
+        token = _core.current_trio_token()
+        try:
+            yield 1
+        finally:
+            token.run_sync_soon(lambda: {}["nope"])
+            token.run_sync_soon(lambda: record.append("2nd ran"))
+
+    async def main():
+        saved.append(agen())
+        await saved[-1].asend(None)
+        record.append("main exiting")
+
+    with pytest.raises(_core.TrioInternalError) as excinfo:
+        _core.run(main)
+
+    assert type(excinfo.value.__cause__) is KeyError
+    assert record == ["main exiting", "2nd ran"]
+
+
 async def test_slow_abort_basic():
     with _core.CancelScope() as scope:
         scope.cancel()
diff --git a/trio/_core/tests/tutil.py b/trio/_core/tests/tutil.py
index 9fcf3e2e..43673386 100644
--- a/trio/_core/tests/tutil.py
+++ b/trio/_core/tests/tutil.py
@@ -2,6 +2,7 @@
 import socket as stdlib_socket
 import os
 import sys
+from typing import TYPE_CHECKING
 
 import pytest
 import warnings
@@ -14,6 +15,18 @@ from trio.tests.conftest import RUN_SLOW
 
 slow = pytest.mark.skipif(not RUN_SLOW, reason="use --run-slow to run slow tests",)
 
+# PyPy 7.2 was released with a bug that just never called the async
+# generator 'firstiter' hook at all.  This impacts tests of end-of-run
+# finalization (nothing gets added to runner.asyncgens) and tests of
+# "foreign" async generator behavior (since the firstiter hook is what
+# marks the asyncgen as foreign), but most tests of GC-mediated
+# finalization still work.
+buggy_pypy_asyncgens = (
+    not TYPE_CHECKING
+    and sys.implementation.name == "pypy"
+    and sys.pypy_version_info < (7, 3)
+)
+
 try:
     s = stdlib_socket.socket(stdlib_socket.AF_INET6, stdlib_socket.SOCK_STREAM, 0)
 except OSError:  # pragma: no cover

pip install -e .[dev]
parse_log_pytest
requirements.txt
3.9
test-requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
appdirs==1.4.4
astor==0.8.1
astroid==2.4.2
async-generator==1.10
attrs==19.3.0
backcall==0.2.0
black==19.10b0
cffi==1.14.1
click==7.1.2
coverage==5.2.1
cryptography==3.0
decorator==4.4.2
flake8==3.8.3
idna==2.10
immutables==0.14
iniconfig==1.0.1
ipython==7.16.1
ipython-genutils==0.2.0
isort==4.3.21
jedi==0.17.2
lazy-object-proxy==1.4.3
mccabe==0.6.1
more-itertools==8.4.0
mypy==0.782
mypy-extensions==0.4.3
outcome==1.0.1
packaging==20.4
parso==0.7.1
pathspec==0.8.0
pexpect==4.8.0
pickleshare==0.7.5
pluggy==0.13.1
prompt-toolkit==3.0.5
ptyprocess==0.6.0
py==1.9.0
pycodestyle==2.6.0
pycparser==2.20
pyflakes==2.2.0
Pygments==2.6.1
pylint==2.5.3
pyOpenSSL==19.1.0
pyparsing==2.4.7
pytest==6.0.1
pytest-cov==2.10.0
regex==2020.7.14
six==1.15.0
sniffio==1.1.0
sortedcontainers==2.2.2
toml==0.10.1
traitlets==4.3.3
-e git+https://github.com/python-trio/trio.git@163cfaac6d560d029a6d9950ccd0c347acddffa6#egg=trio
trustme==0.6.0
typed-ast==1.4.1
typing-extensions==3.7.4.2
wcwidth==0.2.5
wrapt==1.12.1

name: trio
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=3.0.16=h5eee18b_0
  - pip=25.0=py39h06a4308_0
  - python=3.9.21=he870216_1
  - readline=8.2=h5eee18b_0
  - setuptools=75.8.0=py39h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - tzdata=2025a=h04d1e81_0
  - wheel=0.45.1=py39h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - appdirs==1.4.4
      - astor==0.8.1
      - astroid==2.4.2
      - async-generator==1.10
      - attrs==19.3.0
      - backcall==0.2.0
      - black==19.10b0
      - cffi==1.14.1
      - click==7.1.2
      - coverage==5.2.1
      - cryptography==3.0
      - decorator==4.4.2
      - flake8==3.8.3
      - idna==2.10
      - immutables==0.14
      - iniconfig==1.0.1
      - ipython==7.16.1
      - ipython-genutils==0.2.0
      - isort==4.3.21
      - jedi==0.17.2
      - lazy-object-proxy==1.4.3
      - mccabe==0.6.1
      - more-itertools==8.4.0
      - mypy==0.782
      - mypy-extensions==0.4.3
      - outcome==1.0.1
      - packaging==20.4
      - parso==0.7.1
      - pathspec==0.8.0
      - pexpect==4.8.0
      - pickleshare==0.7.5
      - pluggy==0.13.1
      - prompt-toolkit==3.0.5
      - ptyprocess==0.6.0
      - py==1.9.0
      - pycodestyle==2.6.0
      - pycparser==2.20
      - pyflakes==2.2.0
      - pygments==2.6.1
      - pylint==2.5.3
      - pyopenssl==19.1.0
      - pyparsing==2.4.7
      - pytest==6.0.1
      - pytest-cov==2.10.0
      - regex==2020.7.14
      - six==1.15.0
      - sniffio==1.1.0
      - sortedcontainers==2.2.2
      - toml==0.10.1
      - traitlets==4.3.3
      - trio==0.16.0+dev
      - trustme==0.6.0
      - typed-ast==1.4.1
      - typing-extensions==3.7.4.2
      - wcwidth==0.2.5
      - wrapt==1.12.1
prefix: /opt/conda/envs/trio

trio/_core/tests/test_asyncgen.py::test_asyncgen_basics trio/_core/tests/test_asyncgen.py::test_asyncgen_throws_during_finalization trio/_core/tests/test_asyncgen.py::test_firstiter_after_closing trio/_core/tests/test_asyncgen.py::test_interdependent_asyncgen_cleanup_order trio/_core/tests/test_asyncgen.py::test_last_minute_gc_edge_case trio/_core/tests/test_asyncgen.py::test_fallback_when_no_hook_claims_it trio/_core/tests/test_asyncgen.py::test_delegation_to_existing_hooks trio/_core/tests/test_guest_mode.py::test_guest_mode_asyncgens trio/_core/tests/test_run.py::test_TrioToken_run_sync_soon_late_crash
trio/_core/tests/test_guest_mode.py::test_guest_trivial trio/_core/tests/test_guest_mode.py::test_guest_can_do_io trio/_core/tests/test_guest_mode.py::test_host_can_directly_wake_trio_task trio/_core/tests/test_guest_mode.py::test_host_altering_deadlines_wakes_trio_up trio/_core/tests/test_guest_mode.py::test_warn_set_wakeup_fd_overwrite trio/_core/tests/test_guest_mode.py::test_host_wakeup_doesnt_trigger_wait_all_tasks_blocked trio/_core/tests/test_guest_mode.py::test_guest_warns_if_abandoned trio/_core/tests/test_guest_mode.py::test_guest_mode_on_asyncio trio/_core/tests/test_guest_mode.py::test_guest_mode_internal_errors trio/_core/tests/test_guest_mode.py::test_guest_mode_ki trio/_core/tests/test_guest_mode.py::test_guest_mode_autojump_clock_threshold_changing trio/_core/tests/test_instrumentation.py::test_instruments trio/_core/tests/test_instrumentation.py::test_instruments_interleave trio/_core/tests/test_instrumentation.py::test_null_instrument trio/_core/tests/test_instrumentation.py::test_instrument_before_after_run trio/_core/tests/test_instrumentation.py::test_instrument_task_spawn_exit trio/_core/tests/test_instrumentation.py::test_instruments_crash trio/_core/tests/test_instrumentation.py::test_instruments_monkeypatch trio/_core/tests/test_instrumentation.py::test_instrument_that_raises_on_getattr trio/_core/tests/test_run.py::test_basic trio/_core/tests/test_run.py::test_initial_task_error trio/_core/tests/test_run.py::test_run_nesting trio/_core/tests/test_run.py::test_nursery_warn_use_async_with trio/_core/tests/test_run.py::test_nursery_main_block_error_basic trio/_core/tests/test_run.py::test_child_crash_basic trio/_core/tests/test_run.py::test_basic_interleave trio/_core/tests/test_run.py::test_task_crash_propagation trio/_core/tests/test_run.py::test_main_and_task_both_crash trio/_core/tests/test_run.py::test_two_child_crashes trio/_core/tests/test_run.py::test_child_crash_wakes_parent trio/_core/tests/test_run.py::test_reschedule trio/_core/tests/test_run.py::test_current_time trio/_core/tests/test_run.py::test_current_time_with_mock_clock trio/_core/tests/test_run.py::test_current_clock trio/_core/tests/test_run.py::test_current_task trio/_core/tests/test_run.py::test_root_task trio/_core/tests/test_run.py::test_out_of_context trio/_core/tests/test_run.py::test_current_statistics trio/_core/tests/test_run.py::test_cancel_scope_repr trio/_core/tests/test_run.py::test_cancel_points trio/_core/tests/test_run.py::test_cancel_edge_cases trio/_core/tests/test_run.py::test_cancel_scope_multierror_filtering trio/_core/tests/test_run.py::test_precancelled_task trio/_core/tests/test_run.py::test_cancel_shielding trio/_core/tests/test_run.py::test_cancel_inheritance trio/_core/tests/test_run.py::test_cancel_shield_abort trio/_core/tests/test_run.py::test_basic_timeout trio/_core/tests/test_run.py::test_cancel_scope_nesting trio/_core/tests/test_run.py::test_unshield_while_cancel_propagating trio/_core/tests/test_run.py::test_cancel_unbound trio/_core/tests/test_run.py::test_cancel_scope_misnesting trio/_core/tests/test_run.py::test_timekeeping trio/_core/tests/test_run.py::test_failed_abort trio/_core/tests/test_run.py::test_broken_abort trio/_core/tests/test_run.py::test_error_in_run_loop trio/_core/tests/test_run.py::test_spawn_system_task trio/_core/tests/test_run.py::test_system_task_crash trio/_core/tests/test_run.py::test_system_task_crash_MultiError trio/_core/tests/test_run.py::test_system_task_crash_plus_Cancelled trio/_core/tests/test_run.py::test_system_task_crash_KeyboardInterrupt trio/_core/tests/test_run.py::test_yield_briefly_checks_for_timeout trio/_core/tests/test_run.py::test_exc_info trio/_core/tests/test_run.py::test_exc_info_after_yield_error trio/_core/tests/test_run.py::test_exception_chaining_after_yield_error trio/_core/tests/test_run.py::test_nursery_exception_chaining_doesnt_make_context_loops trio/_core/tests/test_run.py::test_TrioToken_identity trio/_core/tests/test_run.py::test_TrioToken_run_sync_soon_basic trio/_core/tests/test_run.py::test_TrioToken_run_sync_soon_too_late trio/_core/tests/test_run.py::test_TrioToken_run_sync_soon_idempotent trio/_core/tests/test_run.py::test_TrioToken_run_sync_soon_idempotent_requeue trio/_core/tests/test_run.py::test_TrioToken_run_sync_soon_after_main_crash trio/_core/tests/test_run.py::test_TrioToken_run_sync_soon_crashes trio/_core/tests/test_run.py::test_TrioToken_run_sync_soon_FIFO trio/_core/tests/test_run.py::test_TrioToken_run_sync_soon_starvation_resistance trio/_core/tests/test_run.py::test_TrioToken_run_sync_soon_threaded_stress_test trio/_core/tests/test_run.py::test_TrioToken_run_sync_soon_massive_queue trio/_core/tests/test_run.py::test_slow_abort_basic trio/_core/tests/test_run.py::test_slow_abort_edge_cases trio/_core/tests/test_run.py::test_task_tree_introspection trio/_core/tests/test_run.py::test_nursery_closure trio/_core/tests/test_run.py::test_spawn_name trio/_core/tests/test_run.py::test_current_effective_deadline trio/_core/tests/test_run.py::test_nice_error_on_bad_calls_to_run_or_spawn trio/_core/tests/test_run.py::test_calling_asyncio_function_gives_nice_error trio/_core/tests/test_run.py::test_asyncio_function_inside_nursery_does_not_explode trio/_core/tests/test_run.py::test_trivial_yields trio/_core/tests/test_run.py::test_nursery_start trio/_core/tests/test_run.py::test_task_nursery_stack trio/_core/tests/test_run.py::test_nursery_start_with_cancelled_nursery trio/_core/tests/test_run.py::test_nursery_start_keeps_nursery_open trio/_core/tests/test_run.py::test_nursery_explicit_exception trio/_core/tests/test_run.py::test_nursery_stop_iteration trio/_core/tests/test_run.py::test_nursery_stop_async_iteration trio/_core/tests/test_run.py::test_traceback_frame_removal trio/_core/tests/test_run.py::test_contextvar_support trio/_core/tests/test_run.py::test_contextvar_multitask trio/_core/tests/test_run.py::test_system_task_contexts trio/_core/tests/test_run.py::test_Nursery_init trio/_core/tests/test_run.py::test_Nursery_private_init trio/_core/tests/test_run.py::test_Nursery_subclass trio/_core/tests/test_run.py::test_Cancelled_init trio/_core/tests/test_run.py::test_Cancelled_str trio/_core/tests/test_run.py::test_Cancelled_subclass trio/_core/tests/test_run.py::test_CancelScope_subclass trio/_core/tests/test_run.py::test_sniffio_integration trio/_core/tests/test_run.py::test_Task_custom_sleep_data trio/_core/tests/test_run.py::test_permanently_detach_coroutine_object trio/_core/tests/test_run.py::test_detach_and_reattach_coroutine_object trio/_core/tests/test_run.py::test_detached_coroutine_cancellation trio/_core/tests/test_run.py::test_async_function_implemented_in_C trio/_core/tests/test_run.py::test_very_deep_cancel_scope_nesting trio/_core/tests/test_run.py::test_cancel_scope_deadline_duplicates
trio/_util.py trio/_core/_entry_queue.py docs/source/reference-core.rst trio/_core/_asyncgens.py newsfragments/265.headline.rst trio/_core/_run.py docs/source/conf.py
trio/_util.py trio/_core/_entry_queue.py docs/source/reference-core.rst trio/_core/_asyncgens.py newsfragments/265.headline.rst trio/_core/_run.py docs/source/conf.py