66'wait' ,'wait_for' ,'as_completed' ,'sleep' ,
77'gather' ,'shield' ,'ensure_future' ,'run_coroutine_threadsafe' ,
88'current_task' ,'all_tasks' ,
9+ 'create_eager_task_factory' ,'eager_task_factory' ,
910'_register_task' ,'_unregister_task' ,'_enter_task' ,'_leave_task' ,
1011)
1112
@@ -43,22 +44,26 @@ def all_tasks(loop=None):
4344"""Return a set of all tasks for the loop."""
4445if loop is None :
4546loop = events .get_running_loop ()
46- # Looping over a WeakSet (_all_tasks) isn't safe as it can be updated from another
47- # thread while we do so. Therefore we cast it to list prior to filtering. The list
48- # cast itself requires iteration, so we repeat it several times ignoring
49- # RuntimeErrors (which are not very likely to occur). See issues 34970 and 36607 for
50- # details.
47+ # capturing the set of eager tasks first, so if an eager task "graduates"
48+ # to a regular task in another thread, we don't risk missing it.
49+ eager_tasks = list (_eager_tasks )
50+ # Looping over the WeakSet isn't safe as it can be updated from another
51+ # thread, therefore we cast it to list prior to filtering. The list cast
52+ # itself requires iteration, so we repeat it several times ignoring
53+ # RuntimeErrors (which are not very likely to occur).
54+ # See issues 34970 and 36607 for details.
55+ scheduled_tasks = None
5156i = 0
5257while True :
5358try :
54- tasks = list (_all_tasks )
59+ scheduled_tasks = list (_scheduled_tasks )
5560except RuntimeError :
5661i += 1
5762if i >= 1000 :
5863raise
5964else :
6065break
61- return {t for t in tasks
66+ return {t for t in itertools . chain ( scheduled_tasks , eager_tasks )
6267if futures ._get_loop (t )is loop and not t .done ()}
6368
6469
@@ -93,7 +98,8 @@ class Task(futures._PyFuture): # Inherit Python Task implementation
9398# status is still pending
9499_log_destroy_pending = True
95100
96- def __init__ (self ,coro ,* ,loop = None ,name = None ,context = None ):
101+ def __init__ (self ,coro ,* ,loop = None ,name = None ,context = None ,
102+ eager_start = False ):
97103super ().__init__ (loop = loop )
98104if self ._source_traceback :
99105del self ._source_traceback [- 1 ]
@@ -117,8 +123,11 @@ def __init__(self, coro, *, loop=None, name=None, context=None):
117123else :
118124self ._context = context
119125
120- self ._loop .call_soon (self .__step ,context = self ._context )
121- _register_task (self )
126+ if eager_start and self ._loop .is_running ():
127+ self .__eager_start ()
128+ else :
129+ self ._loop .call_soon (self .__step ,context = self ._context )
130+ _register_task (self )
122131
123132def __del__ (self ):
124133if self ._state == futures ._PENDING and self ._log_destroy_pending :
@@ -250,6 +259,25 @@ def uncancel(self):
250259self ._num_cancels_requested -= 1
251260return self ._num_cancels_requested
252261
262+ def __eager_start (self ):
263+ prev_task = _swap_current_task (self ._loop ,self )
264+ try :
265+ _register_eager_task (self )
266+ try :
267+ self ._context .run (self .__step_run_and_handle_result ,None )
268+ finally :
269+ _unregister_eager_task (self )
270+ finally :
271+ try :
272+ curtask = _swap_current_task (self ._loop ,prev_task )
273+ assert curtask is self
274+ finally :
275+ if self .done ():
276+ self ._coro = None
277+ self = None # Needed to break cycles when an exception occurs.
278+ else :
279+ _register_task (self )
280+
253281def __step (self ,exc = None ):
254282if self .done ():
255283raise exceptions .InvalidStateError (
@@ -258,11 +286,17 @@ def __step(self, exc=None):
258286if not isinstance (exc ,exceptions .CancelledError ):
259287exc = self ._make_cancelled_error ()
260288self ._must_cancel = False
261- coro = self ._coro
262289self ._fut_waiter = None
263290
264291_enter_task (self ._loop ,self )
265- # Call either coro.throw(exc) or coro.send(None).
292+ try :
293+ self .__step_run_and_handle_result (exc )
294+ finally :
295+ _leave_task (self ._loop ,self )
296+ self = None # Needed to break cycles when an exception occurs.
297+
298+ def __step_run_and_handle_result (self ,exc ):
299+ coro = self ._coro
266300try :
267301if exc is None :
268302# We use the `send` method directly, because coroutines
@@ -334,7 +368,6 @@ def __step(self, exc=None):
334368self ._loop .call_soon (
335369self .__step ,new_exc ,context = self ._context )
336370finally :
337- _leave_task (self ._loop ,self )
338371self = None # Needed to break cycles when an exception occurs.
339372
340373def __wakeup (self ,future ):
@@ -897,17 +930,41 @@ def callback():
897930return future
898931
899932
900- # WeakSet containing all alive tasks.
901- _all_tasks = weakref .WeakSet ()
933+ def create_eager_task_factory (custom_task_constructor ):
934+
935+ if "eager_start" not in inspect .signature (custom_task_constructor ).parameters :
936+ raise TypeError (
937+ "Provided constructor does not support eager task execution" )
938+
939+ def factory (loop ,coro ,* ,name = None ,context = None ):
940+ return custom_task_constructor (
941+ coro ,loop = loop ,name = name ,context = context ,eager_start = True )
942+
943+
944+ return factory
945+
946+ eager_task_factory = create_eager_task_factory (Task )
947+
948+
949+ # Collectively these two sets hold references to the complete set of active
950+ # tasks. Eagerly executed tasks use a faster regular set as an optimization
951+ # but may graduate to a WeakSet if the task blocks on IO.
952+ _scheduled_tasks = weakref .WeakSet ()
953+ _eager_tasks = set ()
902954
903955# Dictionary containing tasks that are currently active in
904956# all running event loops. {EventLoop: Task}
905957_current_tasks = {}
906958
907959
908960def _register_task (task ):
909- """Register a new task in asyncio as executed by loop."""
910- _all_tasks .add (task )
961+ """Register an asyncio Task scheduled to run on an event loop."""
962+ _scheduled_tasks .add (task )
963+
964+
965+ def _register_eager_task (task ):
966+ """Register an asyncio Task about to be eagerly executed."""
967+ _eager_tasks .add (task )
911968
912969
913970def _enter_task (loop ,task ):
@@ -926,28 +983,49 @@ def _leave_task(loop, task):
926983del _current_tasks [loop ]
927984
928985
986+ def _swap_current_task (loop ,task ):
987+ prev_task = _current_tasks .get (loop )
988+ if task is None :
989+ del _current_tasks [loop ]
990+ else :
991+ _current_tasks [loop ]= task
992+ return prev_task
993+
994+
929995def _unregister_task (task ):
930- """Unregister a task."""
931- _all_tasks .discard (task )
996+ """Unregister a completed, scheduled Task."""
997+ _scheduled_tasks .discard (task )
998+
999+
1000+ def _unregister_eager_task (task ):
1001+ """Unregister a task which finished its first eager step."""
1002+ _eager_tasks .discard (task )
9321003
9331004
9341005_py_current_task = current_task
9351006_py_register_task = _register_task
1007+ _py_register_eager_task = _register_eager_task
9361008_py_unregister_task = _unregister_task
1009+ _py_unregister_eager_task = _unregister_eager_task
9371010_py_enter_task = _enter_task
9381011_py_leave_task = _leave_task
1012+ _py_swap_current_task = _swap_current_task
9391013
9401014
9411015try :
942- from _asyncio import (_register_task ,_unregister_task ,
943- _enter_task ,_leave_task ,
944- _all_tasks ,_current_tasks ,
1016+ from _asyncio import (_register_task ,_register_eager_task ,
1017+ _unregister_task ,_unregister_eager_task ,
1018+ _enter_task ,_leave_task ,_swap_current_task ,
1019+ _scheduled_tasks ,_eager_tasks ,_current_tasks ,
9451020current_task )
9461021except ImportError :
9471022pass
9481023else :
9491024_c_current_task = current_task
9501025_c_register_task = _register_task
1026+ _c_register_eager_task = _register_eager_task
9511027_c_unregister_task = _unregister_task
1028+ _c_unregister_eager_task = _unregister_eager_task
9521029_c_enter_task = _enter_task
9531030_c_leave_task = _leave_task
1031+ _c_swap_current_task = _swap_current_task