arthur-tacca

Trio event loop notes

These are the steps of the Trio event loop:

Compute timeout (0 if still tasks in runq i.e. the list of tasks ready to resume, else clock.deadline_to_sleep_time(deadlines.next_deadline()))
Wait for IO events or timeout (events = io_manager.get_events(timeout)) – this is the actual select()-like call.
Events are "processed" i.e. scheduled into the relevant task (io_manager.process_events(events) which calls _core.reschedule(self._registered[key], outcome.Value(event)) i.e. added to runq)
Deadlines are expired (deadlines.expire(clock.current_time()), which effectively calls task.reschedule(Cancelled) (via the task's abort function) i.e. adds the task to the runq)

This is now superceded by the proof of concept code:


	"""Implementation of Trio event and chnanel types where cancellation always succeeds.

	See: https://github.com/python-trio/trio/issues/2973
	"""

	import attr
	from collections import deque
	import trio
	from typing import Generic, Optional, TypeVar

	# Adapted with permission from the EdgeDB project;
	# license: PSFL.


	__all__ = ("TaskGroup",)

	from . import events
	from . import exceptions
	from . import tasks

	# Question: if we cancel a scope, and another task is active in that scope, is it guaranteed that
	# it will raise CancelledError when it next wakes? (Even if it is already queued to wake for some
	# other reason.)
	#
	# The answer, at least in practice, is no: in this example, send_and_cancel() sends some data to
	# a memory channel then cancels a scope, and then receive() (which is running in that scope) wakes
	# with the first value that was sent.
	#
	# On my computer, this code produces the following output:
	#

	# Illustration of missed cleanup in an asyncio task group due to the way it
	# cancels a task that has not started by never running it at all. See issue:
	# https://github.com/python/cpython/issues/116048
	#
	# When run on my computer, I usually get the output:
	#
	# Total connections created: 3300
	# Connections left unclosed: 0
	#
	# i.e., all connections that were created were cleaned up. But sometimes I get:

	# aioresult variant of StreamResultsNursery
	# Original idea by smurfix: https://gist.github.com/smurfix/0130817fa5ba6d3bb4a0f00e4d93cf86
	# Fixed non-aioresult version: https://gist.github.com/arthur-tacca/6c676a21d0dcc0582edb50c9c2aa3e3c


	from collections import deque
	import math

	from aioresult import ResultCapture
	import trio

	// Moved to:
	// https://github.com/arthur-tacca/dart-cancel-examples/blob/main/lib/abort.dart

	# Original idea by smurfix: https://gist.github.com/smurfix/0130817fa5ba6d3bb4a0f00e4d93cf86
	# aioresult variant: https://gist.github.com/arthur-tacca/5c717ae68ac037e72ae45fd1e9ca1345

	from collections import deque
	import math
	import trio


	class StreamResultsNursery:
	def __init__(self, max_running_tasks=math.inf):

	#
	# -- An improved version of this code is now in the aioresult library --
	#
	# https://github.com/arthur-tacca/aioresult
	#
	#
	# - aioresult has a ResultCapture class similar to the one below
	# - It also has a Future class that allows manually setting the result, which shares a base class with ResultCapture
	# - There is a utility function with a similar effect to StartableResultCapture below (but much simpler)
	# - There are utility functions wait_any(), wait_all() and to_channel() (which also work with Future instances)