add @background_with_channel

[jakkdl]

Mostly just shepherding @Zac-HD's implementation from #638 (comment)

I don't understand all the details of it, esp some of the code paths I'm completely failing how to cover, so gonna need some help. And I'm getting an exception that sometimes disappear.. which seems bad?
Feel free to add code suggestions and/or commit directly to the branch.

I don't know if this fully resolves #638, or if there's docs and/or other stuff that should be added.

I will make https://flake8-async.readthedocs.io/en/latest/rules.html#async900 suggest using this, and perhaps make it enabled by default, once released.

[A5rocks]

I'm not too sold on how some of this is done, but at least now it shouldn't fail CI.

edit: I'm also pretty sure the more correct way to fix the race condition would be using nursery.start instead of nursery.start_soon.

[jakkdl]

I'm not too sold on how some of this is done, but at least now it shouldn't fail CI.

edit: I'm also pretty sure the more correct way to fix the race condition would be using nursery.start instead of nursery.start_soon.

I tracked down the race condition to send_chan.aclose() getting cancelled, swallowing the exception, i.e. #1559

We still want to use start instead of start_soon to make sure the iterator gets properly closed.

The reason it works to move the send_chan closure to across the yield is simply because that allows the exception in _move_elems_to_channel to bubble up into the nursery without getting eaten (except if aclosing(aiterable) eats it), and the send channel can raise as much Cancelled as it wants in context_manager.
We do need to manually call send_chan.close() in _move_elems_to_channel though to halt any iterations over recv_chan.

[codecov]

Codecov Report

Attention: Patch coverage is 98.09524% with 2 lines in your changes missing coverage. Please review.

Project coverage is 99.98941%. Comparing base (b92131c) to head (7542973).

Files with missing lines	Patch %	Lines
src/trio/_tests/test_channel.py	97.10145%	1 Missing and 1 partial ⚠️

❌ Your patch check has failed because the patch coverage (98.09524%) is below the target coverage (100.00000%). You can increase the patch coverage or adjust the target coverage.
❌ Your project check has failed because the head coverage (99.98941%) is below the target coverage (100.00000%). You can increase the head coverage or adjust the target coverage.

Additional details and impacted files

@@                 Coverage Diff                  @@
##                 main       #3197         +/-   ##
====================================================
- Coverage   100.00000%   99.98941%   -0.01059%     
====================================================
  Files             124         124                 
  Lines           18787       18889        +102     
  Branches         1269        1274          +5     
====================================================
+ Hits            18787       18887        +100     
- Misses              0           1          +1     
- Partials            0           1          +1     

Files with missing lines	Coverage Δ
src/trio/__init__.py	100.00000% <ø> (ø)
src/trio/_channel.py	100.00000% <100.00000%> (ø)
src/trio/_tests/test_channel.py	99.43503% <97.10145%> (-0.56498%)	⬇️

[jakkdl]

@Zac-HD if you can show why the inner loop is necessary it'd be great, but I'm kinda suspecting it's a remnant of previous implementations or something - because I can't come up with anything that would hit that code path.

[Zac-HD]

I think this'll help? But haven't run many tests yet...

[Zac-HD]

Oh, we should also warn about a subtle change from 'raw' generators:

Suggested change

	"""
	... note::
	With 'raw' async generators, code in the generator will never run
	concurrently with that in the body of the ``async for`` loop - the
	generator is resumed to compute each element on request.
	Even with ``max_buffer_size=0``, a ``@background_with_channel()``
	function will 'precompute' each element in a background task, and
	store it in the channel until requested by the loop.
	This is rarely a problem, so we've avoided the performance cost
	of exactly replicating the behavior of raw generators. If you
	concurent execution would cause problems, we recommend using a
	:class:`trio.Lock` around the critical sections.
	"""

(I've only ever seen a problem from this once, in already-weird code, but worth documenting)

[Zac-HD]

The loop in my prototype evolved from this one by @oremanj, via @belm0 here.

If we could trigger it, it'd have to be a non-Cancelled non-BrokenResourceError, which occurred while waiting in await send_chan.send(value). I think we could in principle get a few things here (e.g. KeyboardInterrupt, GC-related errors, etc), but in each case calling .aclose() on the generator and raising from this function without ever throwing the error into the generator seems like a reasonable response to me.

So... I might be wrong, but this simpler code looks good to me.

[Zac-HD]

Hmm, looking at this again - we want to be sure that we close the channels even if we get an error while starting the generator, which suggests pulling the send channel to the outside:

Suggested change

	await nursery.start(_move_elems_to_channel, ait, send_chan)
	# async with recv_chan could eat exceptions, so use sync cm
	with recv_chan:
	yield recv_chan
	with send_chan, recv_chan:
	nursery.start_soon(_move_elems_to_channel, ait, send_chan)
	try:
	yield recv_chan
	except BaseException:
	with trio.CancelScope(shield=True) as scope:
	scope.cancel()
	await ait.aclose()
	raise
	else:
	await ait.aclose()

I also spent a while looking at the spec for await agen.aclose() too, but in my experiments I couldn't construct a version which misbehaved with our decorator but was OK without it. (although it's easy enough to get errors both ways!)

The tricky part is that I run into basically the same questions as #1559 (comment). I think here my solution gets around that: we aclose_forcefully() if there's any error propagating (and if there's an error during that, it should propagate as usual I think?), while if there's no exception yet we do a bare await ait.aclose() and allow whatever cleanup logic we might have to run.