support for tasks coroutines and the scheduler all task createtask firstcompleted firstexception allcompleted wait waitfor ascompleted sleep gather shield ensurefuture runcoroutinethreadsafe currenttask alltasks createeagertaskfactory eagertaskfactory registertask unregistertask entertask leavetask import concurrent futures import contextvars import functools import inspect import itertools import math import types import weakref from types import genericalias from import basetasks from import coroutines from import events from import exceptions from import futures from import timeouts helper to generate new task names this uses itertools count instead of a 1 operation because the latter is not thread safe see bpo11866 for a longer explanation tasknamecounter itertools count1 next def currenttaskloopnone return a set of all tasks for the loop if loop is none loop events getrunningloop capturing the set of eager tasks first so if an eager task graduates to a regular task in another thread we don t risk missing it eagertasks listeagertasks looping over the weakset isn t safe as it can be updated from another thread therefore we cast it to list prior to filtering the list cast itself requires iteration so we repeat it several times ignoring runtimeerrors which are not very likely to occur see issues 34970 and 36607 for details scheduledtasks none i 0 while true try scheduledtasks listscheduledtasks except runtimeerror i 1 if i 1000 raise else break return t for t in itertools chainscheduledtasks eagertasks if futures getloopt is loop and not t done class taskfutures pyfuture inherit python task implementation from a python future implementation an important invariant maintained while a task not done futwaiter is either none or a future the future can be either done or not done the task can be in any of 3 states 1 futwaiter is not none and not futwaiter done step is not scheduled and the task is waiting for futwaiter 2 futwaiter is none or futwaiter done and step is scheduled the task is waiting for step to be executed 3 futwaiter is none and step is not scheduled the task is currently executing in step in state 1 one of the callbacks of futwaiter must be wakeup the transition from 1 to 2 happens when futwaiter becomes done as it schedules wakeup to be called which calls step so we way that step is scheduled it transitions from 2 to 3 when step is executed and it clears futwaiter to none if false don t log a message if the task is destroyed while its status is still pending raise after future init attrs are required for del prevent logging for pending task in del return the list of stack frames for this task s coroutine if the coroutine is not done this returns the stack where it is suspended if the coroutine has completed successfully or was cancelled this returns an empty list if the coroutine was terminated by an exception this returns the list of traceback frames the frames are always ordered from oldest to newest the optional limit gives the maximum number of frames to return by default all available frames are returned its meaning differs depending on whether a stack or a traceback is returned the newest frames of a stack are returned but the oldest frames of a traceback are returned this matches the behavior of the traceback module for reasons beyond our control only one stack frame is returned for a suspended coroutine print the stack or traceback for this task s coroutine this produces output similar to that of the traceback module for the frames retrieved by getstack the limit argument is passed to getstack the file argument is an io stream to which the output is written by default output is written to sys stderr request that this task cancel itself this arranges for a cancellederror to be thrown into the wrapped coroutine on the next cycle through the event loop the coroutine then has a chance to clean up or even deny the request using tryexceptfinally unlike future cancel this does not guarantee that the task will be cancelled the exception might be caught and acted upon delaying cancellation of the task or preventing cancellation completely the task may also return a value or raise a different exception immediately after this method is called task cancelled will not return true unless the task was already cancelled a task will be marked as cancelled when the wrapped coroutine terminates with a cancellederror exception even if cancel was not called this also increases the task s count of cancellation requests these two lines are controversial see discussion starting at https github compythoncpythonpull31394issuecomment1053545331 also remember that this is duplicated in asynciomodule c if self numcancelsrequested 1 return false leave self futwaiter it may be a task that catches and ignores the cancellation so we may have to cancel it again later it must be the case that self step is already scheduled return the count of the task s cancellation requests this count is incremented when cancel is called and may be decremented using uncancel decrement the task s count of cancellation requests this should be called by the party that called cancel on the task beforehand returns the remaining number of cancellation requests we use the send method directly because coroutines don t have iter and next methods task is cancelled right before coro stops save the original exception so we can chain it later yielded future must come from future iter bare yield relinquishes control for one event loop iteration yielding a generator is just wrong yielding something else is an error this may also be a cancellation don t pass the value of future result explicitly as future iter and future await don t need it if we call stepvalue none instead of step python eval loop would use sendvalue method call instead of next which is slower for futures that return nongenerator iterators from their iter ctask is needed for tests schedule the execution of a coroutine object in a spawn task return a task object use legacy api if context is not needed wait and ascompleted similar to those in pep 3148 wait for the futures or tasks given by fs to complete the fs iterable must not be empty returns two sets of future done pending usage done pending await asyncio waitfs note this does not raise timeouterror futures that aren t done when the timeout occurs are returned in the second set wait for the single future or coroutine to complete with timeout coroutine will be wrapped in task returns result of the future or coroutine when a timeout occurs it cancels the task and raises timeouterror to avoid the task cancellation wrap it in shield if the wait is cancelled the task is also cancelled if the task supresses the cancellation and returns a value instead that value is returned this function is a coroutine the special case for timeout 0 is for the following case async def testwaitfor funcstarted false async def func nonlocal funcstarted funcstarted true try await asyncio waitforfunc 0 except asyncio timeouterror assert not funcstarted else assert false asyncio runtestwaitfor internal helper for wait the fs argument must be a collection of futures cancel the fut future or task and wait until it completes loop events getrunningloop waiter loop createfuture cb functools partialreleasewaiter waiter fut adddonecallbackcb try fut cancel we cannot wait on fut directly to make sure cancelandwait itself is reliably cancellable await waiter finally fut removedonecallbackcb this is not a coroutine it is just an iterator yielding futures def ascompletedfs timeoutnone use result if futures isfuturefs or coroutines iscoroutinefs raise typeerrorfexpect an iterable of futures not typefs name from queues import queue import here to avoid circular import problem done queue loop events geteventloop todo ensurefuturef looploop for f in setfs timeouthandle none def ontimeout for f in todo f removedonecallbackoncompletion done putnowaitnone queue a dummy value for waitforone todo clear can t do todo removef in the loop def oncompletionf if not todo return ontimeout was here first todo removef done putnowaitf if not todo and timeouthandle is not none timeouthandle cancel async def waitforone f await done get if f is none dummy value from ontimeout raise exceptions timeouterror return f result may raise f exception for f in todo f adddonecallbackoncompletion if todo and timeout is not none timeouthandle loop calllatertimeout ontimeout for in rangelentodo yield waitforone types coroutine def sleep0 yield async def sleepdelay resultnone wrap a coroutine or an awaitable in a future if the argument is a future it is returned directly helper for gather this overrides cancel to cancel all the children and act more like task cancel which doesn t immediately mark itself as cancelled if any child tasks were actually cancelled we should propagate the cancellation request regardless of returnexceptions argument see issue 32684 return a future aggregating results from the given coroutinesfutures coroutines will be wrapped in a future and scheduled in the event loop they will not necessarily be scheduled in the same order as passed in all futures must share the same event loop if all the tasks are done successfully the returned future s result is the list of results in the order of the original sequence not necessarily the order of results arrival if returnexceptions is true exceptions in the tasks are treated the same as successful results and gathered in the result list otherwise the first raised exception will be immediately propagated to the returned future cancellation if the outer future is cancelled all children that have not completed yet are also cancelled if any child is cancelled this is treated as if it raised cancellederror the outer future is not cancelled in this case this is to prevent the cancellation of one child to cause other children to be cancelled if returnexceptions is false cancelling gather after it has been marked done won t cancel any submitted awaitables for instance gather can be marked done after propagating an exception to the caller therefore calling gather cancel after catching an exception raised by one of the awaitables from gather won t cancel any other awaitables mark exception retrieved check if fut is cancelled first as fut exception will raise a cancellederror instead of returning it all futures are done create a list of results and set it to the outer future check if fut is cancelled first as fut exception will raise a cancellederror instead of returning it also since we re adding the exception return value to results instead of raising it don t bother setting context this also lets us preserve calling makecancellederror at most once if gather is being cancelled we must propagate the cancellation regardless of returnexceptions argument see issue 32684 arg was not a future therefore fut is a new future created specifically for arg since the caller can t control it disable the destroy pending task warning there s a duplicate future object in corosorfutures run done callbacks after gatheringfuture created so any postprocessing can be performed at this point optimization in the special case that all futures finished eagerly this will effectively complete the gather eagerly with the last callback setting the result or exception on outer before returning it wait for a future shielding it from cancellation the statement task asyncio createtasksomething res await shieldtask is exactly equivalent to the statement res await something except that if the coroutine containing it is cancelled the task running in something is not cancelled from the pov of something the cancellation did not happen but its caller is still cancelled so the yieldfrom expression still raises cancellederror note if something is cancelled by other means this will still cancel shield if you want to completely ignore cancellation not recommended you can combine shield with a tryexcept clause as follows task asyncio createtasksomething try res await shieldtask except cancellederror res none save a reference to tasks passed to this function to avoid a task disappearing midexecution the event loop only keeps weak references to tasks a task that isn t referenced elsewhere may get garbage collected at any time even before it s done shortcut mark inner s result as retrieved submit a coroutine object to a given event loop return a concurrent futures future to access the result create a function suitable for use as a task factory on an eventloop example usage loop settaskfactory asyncio createeagertaskfactorymytaskconstructor now tasks created will be started immediately rather than being first scheduled to an event loop the constructor argument can be any callable that returns a taskcompatible object and has a signature compatible with task init it must have the eagerstart keyword argument most applications will use task for customtaskconstructor and in this case there s no need to call createeagertaskfactory directly instead the global eagertaskfactory instance can be used e g loop settaskfactoryasyncio eagertaskfactory collectively these two sets hold references to the complete set of active tasks eagerly executed tasks use a faster regular set as an optimization but may graduate to a weakset if the task blocks on io dictionary containing tasks that are currently active in all running event loops eventloop task register an asyncio task scheduled to run on an event loop scheduledtasks addtask def registereagertasktask unregister a completed scheduled task scheduledtasks discardtask def unregistereagertasktask helper to generate new task names this uses itertools count instead of a 1 operation because the latter is not thread safe see bpo 11866 for a longer explanation return a currently executed task return a set of all tasks for the loop capturing the set of eager tasks first so if an eager task graduates to a regular task in another thread we don t risk missing it looping over the weakset isn t safe as it can be updated from another thread therefore we cast it to list prior to filtering the list cast itself requires iteration so we repeat it several times ignoring runtimeerrors which are not very likely to occur see issues 34970 and 36607 for details inherit python task implementation from a python future implementation a coroutine wrapped in a future an important invariant maintained while a task not done _fut_waiter is either none or a future the future can be either done or not done the task can be in any of 3 states 1 _fut_waiter is not none and not _fut_waiter done __step is not scheduled and the task is waiting for _fut_waiter 2 _fut_waiter is none or _fut_waiter done and __step is scheduled the task is waiting for __step to be executed 3 _fut_waiter is none and __step is not scheduled the task is currently executing in __step in state 1 one of the callbacks of __fut_waiter must be __wakeup the transition from 1 to 2 happens when _fut_waiter becomes done as it schedules __wakeup to be called which calls __step so we way that __step is scheduled it transitions from 2 to 3 when __step is executed and it clears _fut_waiter to none if false don t log a message if the task is destroyed while its status is still pending raise after future __init__ attrs are required for __del__ prevent logging for pending task in __del__ return the list of stack frames for this task s coroutine if the coroutine is not done this returns the stack where it is suspended if the coroutine has completed successfully or was cancelled this returns an empty list if the coroutine was terminated by an exception this returns the list of traceback frames the frames are always ordered from oldest to newest the optional limit gives the maximum number of frames to return by default all available frames are returned its meaning differs depending on whether a stack or a traceback is returned the newest frames of a stack are returned but the oldest frames of a traceback are returned this matches the behavior of the traceback module for reasons beyond our control only one stack frame is returned for a suspended coroutine print the stack or traceback for this task s coroutine this produces output similar to that of the traceback module for the frames retrieved by get_stack the limit argument is passed to get_stack the file argument is an i o stream to which the output is written by default output is written to sys stderr request that this task cancel itself this arranges for a cancellederror to be thrown into the wrapped coroutine on the next cycle through the event loop the coroutine then has a chance to clean up or even deny the request using try except finally unlike future cancel this does not guarantee that the task will be cancelled the exception might be caught and acted upon delaying cancellation of the task or preventing cancellation completely the task may also return a value or raise a different exception immediately after this method is called task cancelled will not return true unless the task was already cancelled a task will be marked as cancelled when the wrapped coroutine terminates with a cancellederror exception even if cancel was not called this also increases the task s count of cancellation requests these two lines are controversial see discussion starting at https github com python cpython pull 31394 issuecomment 1053545331 also remember that this is duplicated in _asynciomodule c if self _num_cancels_requested 1 return false leave self _fut_waiter it may be a task that catches and ignores the cancellation so we may have to cancel it again later it must be the case that self __step is already scheduled return the count of the task s cancellation requests this count is incremented when cancel is called and may be decremented using uncancel decrement the task s count of cancellation requests this should be called by the party that called cancel on the task beforehand returns the remaining number of cancellation requests needed to break cycles when an exception occurs needed to break cycles when an exception occurs we use the send method directly because coroutines don t have __iter__ and __next__ methods task is cancelled right before coro stops save the original exception so we can chain it later i e future cancel self yielded future must come from future __iter__ bare yield relinquishes control for one event loop iteration yielding a generator is just wrong yielding something else is an error needed to break cycles when an exception occurs this may also be a cancellation don t pass the value of future result explicitly as future __iter__ and future __await__ don t need it if we call _step value none instead of _step python eval loop would use send value method call instead of __next__ which is slower for futures that return non generator iterators from their __iter__ needed to break cycles when an exception occurs _ctask is needed for tests schedule the execution of a coroutine object in a spawn task return a task object use legacy api if context is not needed wait and as_completed similar to those in pep 3148 wait for the futures or tasks given by fs to complete the fs iterable must not be empty returns two sets of future done pending usage done pending await asyncio wait fs note this does not raise timeouterror futures that aren t done when the timeout occurs are returned in the second set wait for the single future or coroutine to complete with timeout coroutine will be wrapped in task returns result of the future or coroutine when a timeout occurs it cancels the task and raises timeouterror to avoid the task cancellation wrap it in shield if the wait is cancelled the task is also cancelled if the task supresses the cancellation and returns a value instead that value is returned this function is a coroutine the special case for timeout 0 is for the following case async def test_waitfor func_started false async def func nonlocal func_started func_started true try await asyncio wait_for func 0 except asyncio timeouterror assert not func_started else assert false asyncio run test_waitfor internal helper for wait the fs argument must be a collection of futures cancel the fut future or task and wait until it completes we cannot wait on fut directly to make sure _cancel_and_wait itself is reliably cancellable this is not a coroutine it is just an iterator yielding futures return an iterator whose values are coroutines when waiting for the yielded coroutines you ll get the results or exceptions of the original futures or coroutines in the order in which and as soon as they complete this differs from pep 3148 the proper way to use this is for f in as_completed fs result await f the await may raise use result if a timeout is specified the await will raise timeouterror when the timeout occurs before all futures are done note the futures f are not necessarily members of fs import here to avoid circular import problem queue a dummy value for _wait_for_one can t do todo remove f in the loop _on_timeout was here first dummy value from _on_timeout may raise f exception skip one event loop run cycle this is a private helper for asyncio sleep used when the delay is set to 0 it uses a bare yield expression which task __step knows how to handle instead of creating a future object coroutine that completes after a given time in seconds wrap a coroutine or an awaitable in a future if the argument is a future it is returned directly helper for gather this overrides cancel to cancel all the children and act more like task cancel which doesn t immediately mark itself as cancelled if any child tasks were actually cancelled we should propagate the cancellation request regardless of return_exceptions argument see issue 32684 return a future aggregating results from the given coroutines futures coroutines will be wrapped in a future and scheduled in the event loop they will not necessarily be scheduled in the same order as passed in all futures must share the same event loop if all the tasks are done successfully the returned future s result is the list of results in the order of the original sequence not necessarily the order of results arrival if return_exceptions is true exceptions in the tasks are treated the same as successful results and gathered in the result list otherwise the first raised exception will be immediately propagated to the returned future cancellation if the outer future is cancelled all children that have not completed yet are also cancelled if any child is cancelled this is treated as if it raised cancellederror the outer future is not cancelled in this case this is to prevent the cancellation of one child to cause other children to be cancelled if return_exceptions is false cancelling gather after it has been marked done won t cancel any submitted awaitables for instance gather can be marked done after propagating an exception to the caller therefore calling gather cancel after catching an exception raised by one of the awaitables from gather won t cancel any other awaitables mark exception retrieved check if fut is cancelled first as fut exception will raise a cancellederror instead of returning it all futures are done create a list of results and set it to the outer future check if fut is cancelled first as fut exception will raise a cancellederror instead of returning it also since we re adding the exception return value to results instead of raising it don t bother setting __context__ this also lets us preserve calling _make_cancelled_error at most once if gather is being cancelled we must propagate the cancellation regardless of return_exceptions argument see issue 32684 bpo 46672 arg was not a future therefore fut is a new future created specifically for arg since the caller can t control it disable the destroy pending task warning there s a duplicate future object in coros_or_futures run done callbacks after gatheringfuture created so any post processing can be performed at this point optimization in the special case that all futures finished eagerly this will effectively complete the gather eagerly with the last callback setting the result or exception on outer before returning it wait for a future shielding it from cancellation the statement task asyncio create_task something res await shield task is exactly equivalent to the statement res await something except that if the coroutine containing it is cancelled the task running in something is not cancelled from the pov of something the cancellation did not happen but its caller is still cancelled so the yield from expression still raises cancellederror note if something is cancelled by other means this will still cancel shield if you want to completely ignore cancellation not recommended you can combine shield with a try except clause as follows task asyncio create_task something try res await shield task except cancellederror res none save a reference to tasks passed to this function to avoid a task disappearing mid execution the event loop only keeps weak references to tasks a task that isn t referenced elsewhere may get garbage collected at any time even before it s done shortcut mark inner s result as retrieved submit a coroutine object to a given event loop return a concurrent futures future to access the result create a function suitable for use as a task factory on an event loop example usage loop set_task_factory asyncio create_eager_task_factory my_task_constructor now tasks created will be started immediately rather than being first scheduled to an event loop the constructor argument can be any callable that returns a task compatible object and has a signature compatible with task __init__ it must have the eager_start keyword argument most applications will use task for custom_task_constructor and in this case there s no need to call create_eager_task_factory directly instead the global eager_task_factory instance can be used e g loop set_task_factory asyncio eager_task_factory collectively these two sets hold references to the complete set of active tasks eagerly executed tasks use a faster regular set as an optimization but may graduate to a weakset if the task blocks on io dictionary containing tasks that are currently active in all running event loops eventloop task register an asyncio task scheduled to run on an event loop register an asyncio task about to be eagerly executed unregister a completed scheduled task unregister a task which finished its first eager step
__all__ = (
    'Task', 'create_task',
    'FIRST_COMPLETED', 'FIRST_EXCEPTION', 'ALL_COMPLETED',
    'wait', 'wait_for', 'as_completed', 'sleep',
    'gather', 'shield', 'ensure_future', 'run_coroutine_threadsafe',
    'current_task', 'all_tasks',
    'create_eager_task_factory', 'eager_task_factory',
    '_register_task', '_unregister_task', '_enter_task', '_leave_task',
)

import concurrent.futures
import contextvars
import functools
import inspect
import itertools
import math
import types
import weakref
from types import GenericAlias

from . import base_tasks
from . import coroutines
from . import events
from . import exceptions
from . import futures
from . import timeouts




_task_name_counter = itertools.count(1).__next__


def current_task(loop=None):
    
    if loop is None:
        loop = events.get_running_loop()
    return _current_tasks.get(loop)


def all_tasks(loop=None):
    
    if loop is None:
        loop = events.get_running_loop()
    
    
    eager_tasks = list(_eager_tasks)
    
    
    
    
    
    scheduled_tasks = None
    i = 0
    while True:
        try:
            scheduled_tasks = list(_scheduled_tasks)
        except RuntimeError:
            i += 1
            if i >= 1000:
                raise
        else:
            break
    return {t for t in itertools.chain(scheduled_tasks, eager_tasks)
            if futures._get_loop(t) is loop and not t.done()}


class Task(futures._PyFuture):  
    
    _log_destroy_pending = True

    def __init__(self, coro, *, loop=None, name=None, context=None,
                 eager_start=False):
        super().__init__(loop=loop)
        if self._source_traceback:
            del self._source_traceback[-1]
        if not coroutines.iscoroutine(coro):
            
            
            self._log_destroy_pending = False
            raise TypeError(f"a coroutine was expected, got {coro!r}")

        if name is None:
            self._name = f'Task-{_task_name_counter()}'
        else:
            self._name = str(name)

        self._num_cancels_requested = 0
        self._must_cancel = False
        self._fut_waiter = None
        self._coro = coro
        if context is None:
            self._context = contextvars.copy_context()
        else:
            self._context = context

        if eager_start and self._loop.is_running():
            self.__eager_start()
        else:
            self._loop.call_soon(self.__step, context=self._context)
            _register_task(self)

    def __del__(self):
        if self._state == futures._PENDING and self._log_destroy_pending:
            context = {
                'task': self,
                'message': 'Task was destroyed but it is pending!',
            }
            if self._source_traceback:
                context['source_traceback'] = self._source_traceback
            self._loop.call_exception_handler(context)
        super().__del__()

    __class_getitem__ = classmethod(GenericAlias)

    def __repr__(self):
        return base_tasks._task_repr(self)

    def get_coro(self):
        return self._coro

    def get_context(self):
        return self._context

    def get_name(self):
        return self._name

    def set_name(self, value):
        self._name = str(value)

    def set_result(self, result):
        raise RuntimeError('Task does not support set_result operation')

    def set_exception(self, exception):
        raise RuntimeError('Task does not support set_exception operation')

    def get_stack(self, *, limit=None):
        
        return base_tasks._task_get_stack(self, limit)

    def print_stack(self, *, limit=None, file=None):
        
        return base_tasks._task_print_stack(self, limit, file)

    def cancel(self, msg=None):
        
        self._log_traceback = False
        if self.done():
            return False
        self._num_cancels_requested += 1
        
        
        
        
        
        if self._fut_waiter is not None:
            if self._fut_waiter.cancel(msg=msg):
                
                
                
                return True
        
        self._must_cancel = True
        self._cancel_message = msg
        return True

    def cancelling(self):
        
        return self._num_cancels_requested

    def uncancel(self):
        
        if self._num_cancels_requested > 0:
            self._num_cancels_requested -= 1
        return self._num_cancels_requested

    def __eager_start(self):
        prev_task = _swap_current_task(self._loop, self)
        try:
            _register_eager_task(self)
            try:
                self._context.run(self.__step_run_and_handle_result, None)
            finally:
                _unregister_eager_task(self)
        finally:
            try:
                curtask = _swap_current_task(self._loop, prev_task)
                assert curtask is self
            finally:
                if self.done():
                    self._coro = None
                    self = None  
                else:
                    _register_task(self)

    def __step(self, exc=None):
        if self.done():
            raise exceptions.InvalidStateError(
                f'_step(): already done: {self!r}, {exc!r}')
        if self._must_cancel:
            if not isinstance(exc, exceptions.CancelledError):
                exc = self._make_cancelled_error()
            self._must_cancel = False
        self._fut_waiter = None

        _enter_task(self._loop, self)
        try:
            self.__step_run_and_handle_result(exc)
        finally:
            _leave_task(self._loop, self)
            self = None  

    def __step_run_and_handle_result(self, exc):
        coro = self._coro
        try:
            if exc is None:
                
                
                result = coro.send(None)
            else:
                result = coro.throw(exc)
        except StopIteration as exc:
            if self._must_cancel:
                
                self._must_cancel = False
                super().cancel(msg=self._cancel_message)
            else:
                super().set_result(exc.value)
        except exceptions.CancelledError as exc:
            
            self._cancelled_exc = exc
            super().cancel()  
        except (KeyboardInterrupt, SystemExit) as exc:
            super().set_exception(exc)
            raise
        except BaseException as exc:
            super().set_exception(exc)
        else:
            blocking = getattr(result, '_asyncio_future_blocking', None)
            if blocking is not None:
                
                if futures._get_loop(result) is not self._loop:
                    new_exc = RuntimeError(
                        f'Task {self!r} got Future '
                        f'{result!r} attached to a different loop')
                    self._loop.call_soon(
                        self.__step, new_exc, context=self._context)
                elif blocking:
                    if result is self:
                        new_exc = RuntimeError(
                            f'Task cannot await on itself: {self!r}')
                        self._loop.call_soon(
                            self.__step, new_exc, context=self._context)
                    else:
                        result._asyncio_future_blocking = False
                        result.add_done_callback(
                            self.__wakeup, context=self._context)
                        self._fut_waiter = result
                        if self._must_cancel:
                            if self._fut_waiter.cancel(
                                    msg=self._cancel_message):
                                self._must_cancel = False
                else:
                    new_exc = RuntimeError(
                        f'yield was used instead of yield from '
                        f'in task {self!r} with {result!r}')
                    self._loop.call_soon(
                        self.__step, new_exc, context=self._context)

            elif result is None:
                
                self._loop.call_soon(self.__step, context=self._context)
            elif inspect.isgenerator(result):
                
                new_exc = RuntimeError(
                    f'yield was used instead of yield from for '
                    f'generator in task {self!r} with {result!r}')
                self._loop.call_soon(
                    self.__step, new_exc, context=self._context)
            else:
                
                new_exc = RuntimeError(f'Task got bad yield: {result!r}')
                self._loop.call_soon(
                    self.__step, new_exc, context=self._context)
        finally:
            self = None  

    def __wakeup(self, future):
        try:
            future.result()
        except BaseException as exc:
            
            self.__step(exc)
        else:
            
            self.__step()
        self = None  


_PyTask = Task

try:
    import _asyncio
except ImportError:
    pass
else:
    
    Task = _CTask = _asyncio.Task


def create_task(coro, *, name=None, context=None):
    
    loop = events.get_running_loop()
    if context is None:
        
        task = loop.create_task(coro)
    else:
        task = loop.create_task(coro, context=context)

    task.set_name(name)
    return task




FIRST_COMPLETED = concurrent.futures.FIRST_COMPLETED
FIRST_EXCEPTION = concurrent.futures.FIRST_EXCEPTION
ALL_COMPLETED = concurrent.futures.ALL_COMPLETED


async def wait(fs, *, timeout=None, return_when=ALL_COMPLETED):
    
    if futures.isfuture(fs) or coroutines.iscoroutine(fs):
        raise TypeError(f"expect a list of futures, not {type(fs).__name__}")
    if not fs:
        raise ValueError('Set of Tasks/Futures is empty.')
    if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
        raise ValueError(f'Invalid return_when value: {return_when}')

    fs = set(fs)

    if any(coroutines.iscoroutine(f) for f in fs):
        raise TypeError("Passing coroutines is forbidden, use tasks explicitly.")

    loop = events.get_running_loop()
    return await _wait(fs, timeout, return_when, loop)


def _release_waiter(waiter, *args):
    if not waiter.done():
        waiter.set_result(None)


async def wait_for(fut, timeout):

    if timeout is not None and timeout <= 0:
        fut = ensure_future(fut)

        if fut.done():
            return fut.result()

        await _cancel_and_wait(fut)
        try:
            return fut.result()
        except exceptions.CancelledError as exc:
            raise TimeoutError from exc

    async with timeouts.timeout(timeout):
        return await fut

async def _wait(fs, timeout, return_when, loop):
    
    assert fs, 'Set of Futures is empty.'
    waiter = loop.create_future()
    timeout_handle = None
    if timeout is not None:
        timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
    counter = len(fs)

    def _on_completion(f):
        nonlocal counter
        counter -= 1
        if (counter <= 0 or
            return_when == FIRST_COMPLETED or
            return_when == FIRST_EXCEPTION and (not f.cancelled() and
                                                f.exception() is not None)):
            if timeout_handle is not None:
                timeout_handle.cancel()
            if not waiter.done():
                waiter.set_result(None)

    for f in fs:
        f.add_done_callback(_on_completion)

    try:
        await waiter
    finally:
        if timeout_handle is not None:
            timeout_handle.cancel()
        for f in fs:
            f.remove_done_callback(_on_completion)

    done, pending = set(), set()
    for f in fs:
        if f.done():
            done.add(f)
        else:
            pending.add(f)
    return done, pending


async def _cancel_and_wait(fut):
    

    loop = events.get_running_loop()
    waiter = loop.create_future()
    cb = functools.partial(_release_waiter, waiter)
    fut.add_done_callback(cb)

    try:
        fut.cancel()
        
        
        await waiter
    finally:
        fut.remove_done_callback(cb)



def as_completed(fs, *, timeout=None):
    
    if futures.isfuture(fs) or coroutines.iscoroutine(fs):
        raise TypeError(f"expect an iterable of futures, not {type(fs).__name__}")

    from .queues import Queue  
    done = Queue()

    loop = events.get_event_loop()
    todo = {ensure_future(f, loop=loop) for f in set(fs)}
    timeout_handle = None

    def _on_timeout():
        for f in todo:
            f.remove_done_callback(_on_completion)
            done.put_nowait(None)  
        todo.clear()  

    def _on_completion(f):
        if not todo:
            return  
        todo.remove(f)
        done.put_nowait(f)
        if not todo and timeout_handle is not None:
            timeout_handle.cancel()

    async def _wait_for_one():
        f = await done.get()
        if f is None:
            
            raise exceptions.TimeoutError
        return f.result()  

    for f in todo:
        f.add_done_callback(_on_completion)
    if todo and timeout is not None:
        timeout_handle = loop.call_later(timeout, _on_timeout)
    for _ in range(len(todo)):
        yield _wait_for_one()


@types.coroutine
def __sleep0():
    
    yield


async def sleep(delay, result=None):
    
    if delay <= 0:
        await __sleep0()
        return result

    if math.isnan(delay):
        raise ValueError("Invalid delay: NaN (not a number)")

    loop = events.get_running_loop()
    future = loop.create_future()
    h = loop.call_later(delay,
                        futures._set_result_unless_cancelled,
                        future, result)
    try:
        return await future
    finally:
        h.cancel()


def ensure_future(coro_or_future, *, loop=None):
    
    if futures.isfuture(coro_or_future):
        if loop is not None and loop is not futures._get_loop(coro_or_future):
            raise ValueError('The future belongs to a different loop than '
                            'the one specified as the loop argument')
        return coro_or_future
    should_close = True
    if not coroutines.iscoroutine(coro_or_future):
        if inspect.isawaitable(coro_or_future):
            async def _wrap_awaitable(awaitable):
                return await awaitable

            coro_or_future = _wrap_awaitable(coro_or_future)
            should_close = False
        else:
            raise TypeError('An asyncio.Future, a coroutine or an awaitable '
                            'is required')

    if loop is None:
        loop = events.get_event_loop()
    try:
        return loop.create_task(coro_or_future)
    except RuntimeError:
        if should_close:
            coro_or_future.close()
        raise


class _GatheringFuture(futures.Future):
    

    def __init__(self, children, *, loop):
        assert loop is not None
        super().__init__(loop=loop)
        self._children = children
        self._cancel_requested = False

    def cancel(self, msg=None):
        if self.done():
            return False
        ret = False
        for child in self._children:
            if child.cancel(msg=msg):
                ret = True
        if ret:
            
            
            
            self._cancel_requested = True
        return ret


def gather(*coros_or_futures, return_exceptions=False):
    
    if not coros_or_futures:
        loop = events.get_event_loop()
        outer = loop.create_future()
        outer.set_result([])
        return outer

    def _done_callback(fut):
        nonlocal nfinished
        nfinished += 1

        if outer is None or outer.done():
            if not fut.cancelled():
                
                fut.exception()
            return

        if not return_exceptions:
            if fut.cancelled():
                
                
                
                exc = fut._make_cancelled_error()
                outer.set_exception(exc)
                return
            else:
                exc = fut.exception()
                if exc is not None:
                    outer.set_exception(exc)
                    return

        if nfinished == nfuts:
            
            
            results = []

            for fut in children:
                if fut.cancelled():
                    
                    
                    
                    
                    
                    
                    res = exceptions.CancelledError(
                        '' if fut._cancel_message is None else
                        fut._cancel_message)
                else:
                    res = fut.exception()
                    if res is None:
                        res = fut.result()
                results.append(res)

            if outer._cancel_requested:
                
                
                
                exc = fut._make_cancelled_error()
                outer.set_exception(exc)
            else:
                outer.set_result(results)

    arg_to_fut = {}
    children = []
    nfuts = 0
    nfinished = 0
    done_futs = []
    loop = None
    outer = None  
    for arg in coros_or_futures:
        if arg not in arg_to_fut:
            fut = ensure_future(arg, loop=loop)
            if loop is None:
                loop = futures._get_loop(fut)
            if fut is not arg:
                
                
                
                
                fut._log_destroy_pending = False

            nfuts += 1
            arg_to_fut[arg] = fut
            if fut.done():
                done_futs.append(fut)
            else:
                fut.add_done_callback(_done_callback)

        else:
            
            fut = arg_to_fut[arg]

        children.append(fut)

    outer = _GatheringFuture(children, loop=loop)
    
    
    
    
    
    for fut in done_futs:
        _done_callback(fut)
    return outer


def shield(arg):
    
    inner = ensure_future(arg)
    if inner.done():
        
        return inner
    loop = futures._get_loop(inner)
    outer = loop.create_future()

    def _inner_done_callback(inner):
        if outer.cancelled():
            if not inner.cancelled():
                
                inner.exception()
            return

        if inner.cancelled():
            outer.cancel()
        else:
            exc = inner.exception()
            if exc is not None:
                outer.set_exception(exc)
            else:
                outer.set_result(inner.result())


    def _outer_done_callback(outer):
        if not inner.done():
            inner.remove_done_callback(_inner_done_callback)

    inner.add_done_callback(_inner_done_callback)
    outer.add_done_callback(_outer_done_callback)
    return outer


def run_coroutine_threadsafe(coro, loop):
    
    if not coroutines.iscoroutine(coro):
        raise TypeError('A coroutine object is required')
    future = concurrent.futures.Future()

    def callback():
        try:
            futures._chain_future(ensure_future(coro, loop=loop), future)
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            if future.set_running_or_notify_cancel():
                future.set_exception(exc)
            raise

    loop.call_soon_threadsafe(callback)
    return future


def create_eager_task_factory(custom_task_constructor):
    

    def factory(loop, coro, *, name=None, context=None):
        return custom_task_constructor(
            coro, loop=loop, name=name, context=context, eager_start=True)

    return factory


eager_task_factory = create_eager_task_factory(Task)





_scheduled_tasks = weakref.WeakSet()
_eager_tasks = set()



_current_tasks = {}


def _register_task(task):
    
    _scheduled_tasks.add(task)


def _register_eager_task(task):
    
    _eager_tasks.add(task)


def _enter_task(loop, task):
    current_task = _current_tasks.get(loop)
    if current_task is not None:
        raise RuntimeError(f"Cannot enter into task {task!r} while another "
                           f"task {current_task!r} is being executed.")
    _current_tasks[loop] = task


def _leave_task(loop, task):
    current_task = _current_tasks.get(loop)
    if current_task is not task:
        raise RuntimeError(f"Leaving task {task!r} does not match "
                           f"the current task {current_task!r}.")
    del _current_tasks[loop]


def _swap_current_task(loop, task):
    prev_task = _current_tasks.get(loop)
    if task is None:
        del _current_tasks[loop]
    else:
        _current_tasks[loop] = task
    return prev_task


def _unregister_task(task):
    
    _scheduled_tasks.discard(task)


def _unregister_eager_task(task):
    
    _eager_tasks.discard(task)


_py_current_task = current_task
_py_register_task = _register_task
_py_register_eager_task = _register_eager_task
_py_unregister_task = _unregister_task
_py_unregister_eager_task = _unregister_eager_task
_py_enter_task = _enter_task
_py_leave_task = _leave_task
_py_swap_current_task = _swap_current_task


try:
    from _asyncio import (_register_task, _register_eager_task,
                          _unregister_task, _unregister_eager_task,
                          _enter_task, _leave_task, _swap_current_task,
                          _scheduled_tasks, _eager_tasks, _current_tasks,
                          current_task)
except ImportError:
    pass
else:
    _c_current_task = current_task
    _c_register_task = _register_task
    _c_register_eager_task = _register_eager_task
    _c_unregister_task = _unregister_task
    _c_unregister_eager_task = _unregister_eager_task
    _c_enter_task = _enter_task
    _c_leave_task = _leave_task
    _c_swap_current_task = _swap_current_task