_base.py

# Copyright 2009 Brian Quinlan. All Rights Reserved.
# Licensed to PSF under a Contributor Agreement.

__author__='Brian Quinlan (brian@sweetapp.com)'

importcollections
importlogging
importthreading
importtime

FIRST_COMPLETED='FIRST_COMPLETED'
FIRST_EXCEPTION='FIRST_EXCEPTION'
ALL_COMPLETED='ALL_COMPLETED'
_AS_COMPLETED='_AS_COMPLETED'

# Possible future states (for internal use by the futures package).
PENDING='PENDING'
RUNNING='RUNNING'
# The future was cancelled by the user...
CANCELLED='CANCELLED'
# ...and _Waiter.add_cancelled() was called by a worker.
CANCELLED_AND_NOTIFIED='CANCELLED_AND_NOTIFIED'
FINISHED='FINISHED'

_FUTURE_STATES= [
PENDING,
RUNNING,
CANCELLED,
CANCELLED_AND_NOTIFIED,
FINISHED
]

_STATE_TO_DESCRIPTION_MAP= {
PENDING: "pending",
RUNNING: "running",
CANCELLED: "cancelled",
CANCELLED_AND_NOTIFIED: "cancelled",
FINISHED: "finished"
}

# Logger for internal use by the futures package.
LOGGER=logging.getLogger("concurrent.futures")

classError(Exception):
"""Base class for all future-related exceptions."""
pass

classCancelledError(Error):
"""The Future was cancelled."""
pass

classTimeoutError(Error):
"""The operation exceeded the given deadline."""
pass

class_Waiter(object):
"""Provides the event that wait() and as_completed() block on."""
def__init__(self):
self.event=threading.Event()
self.finished_futures= []

defadd_result(self, future):
self.finished_futures.append(future)

defadd_exception(self, future):
self.finished_futures.append(future)

defadd_cancelled(self, future):
self.finished_futures.append(future)

class_AsCompletedWaiter(_Waiter):
"""Used by as_completed()."""

def__init__(self):
super(_AsCompletedWaiter, self).__init__()
self.lock=threading.Lock()

defadd_result(self, future):
withself.lock:
super(_AsCompletedWaiter, self).add_result(future)
self.event.set()

defadd_exception(self, future):
withself.lock:
super(_AsCompletedWaiter, self).add_exception(future)
self.event.set()

defadd_cancelled(self, future):
withself.lock:
super(_AsCompletedWaiter, self).add_cancelled(future)
self.event.set()

class_FirstCompletedWaiter(_Waiter):
"""Used by wait(return_when=FIRST_COMPLETED)."""

defadd_result(self, future):
super().add_result(future)
self.event.set()

defadd_exception(self, future):
super().add_exception(future)
self.event.set()

defadd_cancelled(self, future):
super().add_cancelled(future)
self.event.set()

class_AllCompletedWaiter(_Waiter):
"""Used by wait(return_when=FIRST_EXCEPTION and ALL_COMPLETED)."""

def__init__(self, num_pending_calls, stop_on_exception):
self.num_pending_calls=num_pending_calls
self.stop_on_exception=stop_on_exception
self.lock=threading.Lock()
super().__init__()

def_decrement_pending_calls(self):
withself.lock:
self.num_pending_calls-=1
ifnotself.num_pending_calls:
self.event.set()

defadd_result(self, future):
super().add_result(future)
self._decrement_pending_calls()

defadd_exception(self, future):
super().add_exception(future)
ifself.stop_on_exception:
self.event.set()
else:
self._decrement_pending_calls()

defadd_cancelled(self, future):
super().add_cancelled(future)
self._decrement_pending_calls()

class_AcquireFutures(object):
"""A context manager that does an ordered acquire of Future conditions."""

def__init__(self, futures):
self.futures=sorted(futures, key=id)

def__enter__(self):
forfutureinself.futures:
future._condition.acquire()

def__exit__(self, *args):
forfutureinself.futures:
future._condition.release()

def_create_and_install_waiters(fs, return_when):
ifreturn_when==_AS_COMPLETED:
waiter=_AsCompletedWaiter()
elifreturn_when==FIRST_COMPLETED:
waiter=_FirstCompletedWaiter()
else:
pending_count=sum(
f._statenotin [CANCELLED_AND_NOTIFIED, FINISHED] forfinfs)

ifreturn_when==FIRST_EXCEPTION:
waiter=_AllCompletedWaiter(pending_count, stop_on_exception=True)
elifreturn_when==ALL_COMPLETED:
waiter=_AllCompletedWaiter(pending_count, stop_on_exception=False)
else:
raiseValueError("Invalid return condition: %r"%return_when)

forfinfs:
f._waiters.append(waiter)

returnwaiter


def_yield_finished_futures(fs, waiter, ref_collect):
"""
 Iterate on the list *fs*, yielding finished futures one by one in
 reverse order.
 Before yielding a future, *waiter* is removed from its waiters
 and the future is removed from each set in the collection of sets
 *ref_collect*.

 The aim of this function is to avoid keeping stale references after
 the future is yielded and before the iterator resumes.
 """
whilefs:
f=fs[-1]
forfutures_setinref_collect:
futures_set.remove(f)
withf._condition:
f._waiters.remove(waiter)
delf
# Careful not to keep a reference to the popped value
yieldfs.pop()


defas_completed(fs, timeout=None):
"""An iterator over the given futures that yields each as it completes.

 Args:
 fs: The sequence of Futures (possibly created by different Executors) to
 iterate over.
 timeout: The maximum number of seconds to wait. If None, then there
 is no limit on the wait time.

 Returns:
 An iterator that yields the given Futures as they complete (finished or
 cancelled). If any given Futures are duplicated, they will be returned
 once.

 Raises:
 TimeoutError: If the entire result iterator could not be generated
 before the given timeout.
 """
iftimeoutisnotNone:
end_time=timeout+time.monotonic()

fs=set(fs)
total_futures=len(fs)
with_AcquireFutures(fs):
finished=set(
fforfinfs
iff._statein [CANCELLED_AND_NOTIFIED, FINISHED])
pending=fs-finished
waiter=_create_and_install_waiters(fs, _AS_COMPLETED)
finished=list(finished)
try:
yieldfrom_yield_finished_futures(finished, waiter,
ref_collect=(fs,))

whilepending:
iftimeoutisNone:
wait_timeout=None
else:
wait_timeout=end_time-time.monotonic()
ifwait_timeout<0:
raiseTimeoutError(
'%d (of %d) futures unfinished'% (
len(pending), total_futures))

waiter.event.wait(wait_timeout)

withwaiter.lock:
finished=waiter.finished_futures
waiter.finished_futures= []
waiter.event.clear()

# reverse to keep finishing order
finished.reverse()
yieldfrom_yield_finished_futures(finished, waiter,
ref_collect=(fs, pending))

finally:
# Remove waiter from unfinished futures
forfinfs:
withf._condition:
f._waiters.remove(waiter)

DoneAndNotDoneFutures=collections.namedtuple(
'DoneAndNotDoneFutures', 'done not_done')
defwait(fs, timeout=None, return_when=ALL_COMPLETED):
"""Wait for the futures in the given sequence to complete.

 Args:
 fs: The sequence of Futures (possibly created by different Executors) to
 wait upon.
 timeout: The maximum number of seconds to wait. If None, then there
 is no limit on the wait time.
 return_when: Indicates when this function should return. The options
 are:

 FIRST_COMPLETED - Return when any future finishes or is
 cancelled.
 FIRST_EXCEPTION - Return when any future finishes by raising an
 exception. If no future raises an exception
 then it is equivalent to ALL_COMPLETED.
 ALL_COMPLETED - Return when all futures finish or are cancelled.

 Returns:
 A named 2-tuple of sets. The first set, named 'done', contains the
 futures that completed (is finished or cancelled) before the wait
 completed. The second set, named 'not_done', contains uncompleted
 futures.
 """
with_AcquireFutures(fs):
done=set(fforfinfs
iff._statein [CANCELLED_AND_NOTIFIED, FINISHED])
not_done=set(fs) -done

if (return_when==FIRST_COMPLETED) anddone:
returnDoneAndNotDoneFutures(done, not_done)
elif (return_when==FIRST_EXCEPTION) anddone:
ifany(fforfindone
ifnotf.cancelled() andf.exception() isnotNone):
returnDoneAndNotDoneFutures(done, not_done)

iflen(done) ==len(fs):
returnDoneAndNotDoneFutures(done, not_done)

waiter=_create_and_install_waiters(fs, return_when)

waiter.event.wait(timeout)
forfinfs:
withf._condition:
f._waiters.remove(waiter)

done.update(waiter.finished_futures)
returnDoneAndNotDoneFutures(done, set(fs) -done)

classFuture(object):
"""Represents the result of an asynchronous computation."""

def__init__(self):
"""Initializes the future. Should not be called by clients."""
self._condition=threading.Condition()
self._state=PENDING
self._result=None
self._exception=None
self._waiters= []
self._done_callbacks= []

def_invoke_callbacks(self):
forcallbackinself._done_callbacks:
try:
callback(self)
exceptException:
LOGGER.exception('exception calling callback for %r', self)

def__repr__(self):
withself._condition:
ifself._state==FINISHED:
ifself._exception:
return'<%s at %#x state=%s raised %s>'% (
self.__class__.__name__,
id(self),
_STATE_TO_DESCRIPTION_MAP[self._state],
self._exception.__class__.__name__)
else:
return'<%s at %#x state=%s returned %s>'% (
self.__class__.__name__,
id(self),
_STATE_TO_DESCRIPTION_MAP[self._state],
self._result.__class__.__name__)
return'<%s at %#x state=%s>'% (
self.__class__.__name__,
id(self),
_STATE_TO_DESCRIPTION_MAP[self._state])

defcancel(self):
"""Cancel the future if possible.

 Returns True if the future was cancelled, False otherwise. A future
 cannot be cancelled if it is running or has already completed.
 """
withself._condition:
ifself._statein [RUNNING, FINISHED]:
returnFalse

ifself._statein [CANCELLED, CANCELLED_AND_NOTIFIED]:
returnTrue

self._state=CANCELLED
self._condition.notify_all()

self._invoke_callbacks()
returnTrue

defcancelled(self):
"""Return True if the future was cancelled."""
withself._condition:
returnself._statein [CANCELLED, CANCELLED_AND_NOTIFIED]

defrunning(self):
"""Return True if the future is currently executing."""
withself._condition:
returnself._state==RUNNING

defdone(self):
"""Return True of the future was cancelled or finished executing."""
withself._condition:
returnself._statein [CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED]

def__get_result(self):
ifself._exception:
raiseself._exception
else:
returnself._result

defadd_done_callback(self, fn):
"""Attaches a callable that will be called when the future finishes.

 Args:
 fn: A callable that will be called with this future as its only
 argument when the future completes or is cancelled. The callable
 will always be called by a thread in the same process in which
 it was added. If the future has already completed or been
 cancelled then the callable will be called immediately. These
 callables are called in the order that they were added.
 """
withself._condition:
ifself._statenotin [CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED]:
self._done_callbacks.append(fn)
return
try:
fn(self)
exceptException:
LOGGER.exception('exception calling callback for %r', self)

defresult(self, timeout=None):
"""Return the result of the call that the future represents.

 Args:
 timeout: The number of seconds to wait for the result if the future
 isn't done. If None, then there is no limit on the wait time.

 Returns:
 The result of the call that the future represents.

 Raises:
 CancelledError: If the future was cancelled.
 TimeoutError: If the future didn't finish executing before the given
 timeout.
 Exception: If the call raised then that exception will be raised.
 """
withself._condition:
ifself._statein [CANCELLED, CANCELLED_AND_NOTIFIED]:
raiseCancelledError()
elifself._state==FINISHED:
returnself.__get_result()

self._condition.wait(timeout)

ifself._statein [CANCELLED, CANCELLED_AND_NOTIFIED]:
raiseCancelledError()
elifself._state==FINISHED:
returnself.__get_result()
else:
raiseTimeoutError()

defexception(self, timeout=None):
"""Return the exception raised by the call that the future represents.

 Args:
 timeout: The number of seconds to wait for the exception if the
 future isn't done. If None, then there is no limit on the wait
 time.

 Returns:
 The exception raised by the call that the future represents or None
 if the call completed without raising.

 Raises:
 CancelledError: If the future was cancelled.
 TimeoutError: If the future didn't finish executing before the given
 timeout.
 """

withself._condition:
ifself._statein [CANCELLED, CANCELLED_AND_NOTIFIED]:
raiseCancelledError()
elifself._state==FINISHED:
returnself._exception

self._condition.wait(timeout)

ifself._statein [CANCELLED, CANCELLED_AND_NOTIFIED]:
raiseCancelledError()
elifself._state==FINISHED:
returnself._exception
else:
raiseTimeoutError()

# The following methods should only be used by Executors and in tests.
defset_running_or_notify_cancel(self):
"""Mark the future as running or process any cancel notifications.

 Should only be used by Executor implementations and unit tests.

 If the future has been cancelled (cancel() was called and returned
 True) then any threads waiting on the future completing (though calls
 to as_completed() or wait()) are notified and False is returned.

 If the future was not cancelled then it is put in the running state
 (future calls to running() will return True) and True is returned.

 This method should be called by Executor implementations before
 executing the work associated with this future. If this method returns
 False then the work should not be executed.

 Returns:
 False if the Future was cancelled, True otherwise.

 Raises:
 RuntimeError: if this method was already called or if set_result()
 or set_exception() was called.
 """
withself._condition:
ifself._state==CANCELLED:
self._state=CANCELLED_AND_NOTIFIED
forwaiterinself._waiters:
waiter.add_cancelled(self)
# self._condition.notify_all() is not necessary because
# self.cancel() triggers a notification.
returnFalse
elifself._state==PENDING:
self._state=RUNNING
returnTrue
else:
LOGGER.critical('Future %s in unexpected state: %s',
id(self),
self._state)
raiseRuntimeError('Future in unexpected state')

defset_result(self, result):
"""Sets the return value of work associated with the future.

 Should only be used by Executor implementations and unit tests.
 """
withself._condition:
self._result=result
self._state=FINISHED
forwaiterinself._waiters:
waiter.add_result(self)
self._condition.notify_all()
self._invoke_callbacks()

defset_exception(self, exception):
"""Sets the result of the future as being the given exception.

 Should only be used by Executor implementations and unit tests.
 """
withself._condition:
self._exception=exception
self._state=FINISHED
forwaiterinself._waiters:
waiter.add_exception(self)
self._condition.notify_all()
self._invoke_callbacks()

classExecutor(object):
"""This is an abstract base class for concrete asynchronous executors."""

defsubmit(*args, **kwargs):
"""Submits a callable to be executed with the given arguments.

 Schedules the callable to be executed as fn(*args, **kwargs) and returns
 a Future instance representing the execution of the callable.

 Returns:
 A Future representing the given call.
 """
iflen(args) >=2:
pass
elifnotargs:
raiseTypeError("descriptor 'submit' of 'Executor' object "
"needs an argument")
elif'fn'notinkwargs:
raiseTypeError('submit expected at least 1 positional argument, '
'got %d'% (len(args)-1))

raiseNotImplementedError()

defmap(self, fn, *iterables, timeout=None, chunksize=1):
"""Returns an iterator equivalent to map(fn, iter).

 Args:
 fn: A callable that will take as many arguments as there are
 passed iterables.
 timeout: The maximum number of seconds to wait. If None, then there
 is no limit on the wait time.
 chunksize: The size of the chunks the iterable will be broken into
 before being passed to a child process. This argument is only
 used by ProcessPoolExecutor; it is ignored by
 ThreadPoolExecutor.

 Returns:
 An iterator equivalent to: map(func, *iterables) but the calls may
 be evaluated out-of-order.

 Raises:
 TimeoutError: If the entire result iterator could not be generated
 before the given timeout.
 Exception: If fn(*args) raises for any values.
 """
iftimeoutisnotNone:
end_time=timeout+time.monotonic()

fs= [self.submit(fn, *args) forargsinzip(*iterables)]

# Yield must be hidden in closure so that the futures are submitted
# before the first iterator value is required.
defresult_iterator():
try:
# reverse to keep finishing order
fs.reverse()
whilefs:
# Careful not to keep a reference to the popped future
iftimeoutisNone:
yieldfs.pop().result()
else:
yieldfs.pop().result(end_time-time.monotonic())
finally:
forfutureinfs:
future.cancel()
returnresult_iterator()

defshutdown(self, wait=True):
"""Clean-up the resources associated with the Executor.

 It is safe to call this method several times. Otherwise, no other
 methods can be called after this one.

 Args:
 wait: If True then shutdown will not return until all running
 futures have finished executing and the resources used by the
 executor have been reclaimed.
 """
pass

def__enter__(self):
returnself

def__exit__(self, exc_type, exc_val, exc_tb):
self.shutdown(wait=True)
returnFalse


classBrokenExecutor(RuntimeError):
"""
 Raised when a executor has become non-functional after a severe failure.
 """