Result and friends: various subclassable classes for deferred delivery of values.

A Result is the base class for several callable subclasses which will receive values at a later point in time, and can also be used standalone without subclassing.

A call to a Result will block until the value is received or the Result is cancelled, which will raise an exception in the caller. A Result may be called by multiple users, before or after the value has been delivered; if the value has been delivered the caller returns with it immediately. A Result's state may be inspected (pending, running, ready, cancelled). Callbacks can be registered via a Result's .notify method.

An incomplete Result can be told to call a function to compute its value; the function return will be stored as the value unless the function raises an exception, in which case the exception information is recorded instead. If an exception occurred, it will be reraised for any caller of the Result.

Trite example:

R = Result(name="my demo")

Thread 1:

# this blocks until the Result is ready
value = R()
print(value)
# prints 3 once Thread 2 (below) assigns to it

Thread 2:

R.result = 3

Thread 3:

value = R()
# returns immediately with 3

You can also collect multiple Results in completion order using the report() function:

Rs = [ ... list of Results of whatever type ... ]
...
for R in report(Rs):
    x = R()     # collect result, will return immediately because
                # the Result is complete
    print(x)    # print result

Short summary:

• after: After the completion of Rs call func(*a,**kw) and return its result via R; return the Result object.

• bg: Dispatch a Thread to run func, return a Result to collect its value.

• call_in_thread: Run func(*a,**kw) in a separate Thread via the @in_thread decorator. Return or exception is as for the original function.

• in_thread: Decorator to evaluate func in a separate Thread. Return or exception is as for the original function.

• not_cancelled: A decorator for methods to raise CancellationError if self.cancelled.

• OnDemandFunction: Wrap a callable, run it when required.

• OnDemandResult: Wrap a callable, run it when required.

• report: Generator which yields completed Results.

• Result: Base class for asynchronous collection of a result. This is used to make Result, OnDemandFunctions, LateFunctions and other objects with asynchronous termination.

• ResultSet: A set subclass containing Results, on which one may iterate as Results complete.

Functions

after(Rs, R, func, *a, **kw)

After the completion of Rs call func(*a,**kw) and return its result via R; return the Result object.

Parameters:

• Rs: an iterable of Results.
• R: a Result to collect to result of calling func. If None, one will be created.
• func, a, kw: a callable and its arguments.

bg(func, *a, **kw)

Dispatch a Thread to run func, return a Result to collect its value.

Parameters:

• _name: optional name for the Result, passed to the initialiser
• _extra: optional extra data for the Result, passed to the initialiser

Other parameters are passed to func.

call_in_thread(func, *a, **kw)

Run func(*a,**kw) in a separate Thread via the @in_thread decorator. Return or exception is as for the original function.

in_thread(func)

Decorator to evaluate func in a separate Thread. Return or exception is as for the original function.

This exists to step out of the current Thread's thread local context, such as a database transaction associated with Django's implicit per-Thread database context.

not_cancelled(*da, **dkw)

A decorator for methods to raise CancellationError if self.cancelled.

report(LFs)

Generator which yields completed Results.

This is a generator that yields Results as they complete, useful for waiting for a sequence of Results that may complete in an arbitrary order.

Classes

class OnDemandFunction(Result)

Wrap a callable, run it when required.

class OnDemandResult(Result)

Wrap a callable, run it when required.

class Result(cs.fsm.FSM)

Base class for asynchronous collection of a result. This is used to make Result, OnDemandFunctions, LateFunctions and other objects with asynchronous termination.

In addition to the methods below, for each state value such as self.PENDING there is a corresponding attribute is_pending testing whether the Result is in that state.

Result.__init__(self, name=None, *, lock=None, result=None, state=None, extra=None)

Base initialiser for Result objects and subclasses.

Parameter:

• name: optional parameter naming this object.
• lock: optional locking object, defaults to a new threading.Lock.
• result: if not None, prefill the .result property.
• extra: an optional mapping of extra information to associate with the Result, useful to provide context when collecting the result; the Result has a public attribute .extra which is an AttrableMapping to hold this information.

Result.FSM_TRANSITIONS

{   'CANCELLED': {'cancel': 'CANCELLED'},
    'DONE': {'cancel': 'DONE'},
    'PENDING': {   'cancel': 'CANCELLED',
                   'complete': 'DONE',
                   'dispatch': 'RUNNING'},
    'RUNNING': {'cancel': 'CANCELLED', 'complete': 'DONE'}}

Result.__call__(self, *a, **kw)

Call the Result: wait for it to be ready and then return or raise.

You can optionally supply a callable and arguments, in which case callable(*args,**kwargs) will be called via Result.call and the results applied to this Result.

Basic example:

R = Result()
... hand R to something which will fulfil it later ...
x = R() # wait for fulfilment - value lands in x

Direct call:

R = Result()
... pass R to something which wants the result ...
# call func(1,2,z=3), save result in R
# ready for collection by whatever received R
R(func,1,2,z=3)

Result.bg(self, func, *a, **kw)

Submit a function to compute the result in a separate Thread, returning the Thread.

Keyword arguments for cs.threads.bg may be supplied by prefixing their names with an underscore, for example:

T = R.bg(mainloop, _pre_enter_objects=(S, fs))

This dispatches a Thread to run self.run_func(func,*a,**kw) and as such the Result must be in "pending" state, and transitions to "running".

Result.cancel(self, msg: Optional[str] = None)

Cancel this Result.

Result.cancelled

<property object at 0x10a8b3b00>

Result.empty(self)

Analogue to Queue.empty().

Result.exc_info

<property object at 0x10a822200>

Result.get(self, default=None)

Wait for readiness; return the result if self.exc_info is None, otherwise default.

Result.join(self)

Calling the .join() method waits for the function to run to completion and returns a tuple of (result,exc_info).

On completion the sequence (result,None) is returned. If an exception occurred computing the result the sequence (None,exc_info) is returned where exc_info is a tuple of (exc_type,exc_value,exc_traceback). If the function was cancelled the sequence (None,None) is returned; the .fsm_state will be 'CANCELLED'.

Result.notify(self, notifier: Callable[[ForwardRef('Result')], NoneType])

After the Result completes, call notifier(self).

If the Result has already completed this will happen immediately. If you'd rather self got put on some queue Q, supply Q.put.

Result.pending

<property object at 0x10a821120>

Result.post_notify(self, post_func) -> 'Result'

Return a secondary Result which processes the result of self.

After the self completes, call post_func(retval) where retval is the result of self, and use that to complete the secondary Result.

Important note: because the completion lock object is released after the internal FSM.fsm_event call, the callback used to implement .post_notify is fired before the lock object is released. As such, it would deadlock as it waits for completion of self by using that lock. Therefore the callback dispatches a separate Thread to wait for self and then run post_func.

Example:

# submit packet to data stream
R = submit_packet()
# arrange that when the response is received, decode the response
R2 = R.post_notify(lambda response: decode(response))
# collect decoded response
decoded = R2()

If the Result has already completed this will happen immediately.

Result.put(self, value)

Store the value. Queue-like idiom.

Result.raise_(self, exc=None)

Convenience wrapper for self.exc_info to store an exception result exc. If exc is omitted or None, uses sys.exc_info().

Examples:

# complete the result using the current exception state
R.raise_()

# complete the result with an exception type
R.raise_(RuntimeError)

# complete the result with an exception
R.raise_(ValueError("bad value!"))

Result.ready

<property object at 0x10a8b3010>

Result.result

<property object at 0x10a820180>

Result.run_func(self, func, *a, **kw)

Fulfil the Result by running func(*a,**kw).

Result.run_func_in_thread(self, func, *a, **kw)

Fulfil the Result by running func(*a,**kw) in a separate Thread.

This exists to step out of the current Thread's thread local context, such as a database transaction associated with Django's implicit per-Thread database context.

Result.state

<property object at 0x10a8abdd0>

class ResultSet(builtins.set)

A set subclass containing Results, on which one may iterate as Results complete.

ResultSet.__iter__(self)

Iterating on a ResultSet yields Results as they complete.

ResultSet.wait(self)

Convenience function to wait for all the Results.