Futures in Python

Introduction

There are situations when we want to asynchronously execute some task(s) in the background and at a later point in time check if those task(s) are complete and get the results from the execution.

Python has the concurrent.futures module that provides a high-level interface for asynchronously executing functions that represent future executions.

The following is a simple python program named FuturesOne.py that demonstrates asynchronous future execution using a ThreadPoolExecutor:

#
# Name: FuturesOne.py
#

import concurrent.futures

# ----- Method factorial -----

def factorial(n):
    if n <= 0:
        return 1
    return n * factorial(n-1)

# ----- Main -----

if __name__ == '__main__':
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=2)

    future_one = executor.submit(factorial, 3)
    future_two = executor.submit(factorial, 5)

    result_one = None
    result_two = None

    try:
        result_one = future_one.result()
        result_two = future_two.result()
    except Exception as ex:
        print("FuturesOne:: <main> :: Exception: %s" % ex)

    print("FuturesOne:: <main> :: result from future_one = " + str(result_one))
    print("FuturesOne:: <main> :: result from future_two = " + str(result_two))

    executor.shutdown()

Executing FuturesOne.py produces the following output:

Output.1

FuturesOne:: <main> :: result from future_one = 6
FuturesOne:: <main> :: result from future_two = 120

Some of the code in the Python program FuturesOne.py needs a a little explanation:

• Future :: Represents an asynchronous callable task and provides methods for tracking the progress of the task. The method result() returns the result produced by the asynchronous task. If the asynchronous task has not yet completed then this method will block and wait

  Ex: result_one = future_one.result()

• ThreadPoolExecutor :: Implements the abstract class Executor. It manages a pool of threads and produces a Future for each of the asynchronous task submitted for execution. The method submit(func, *args) schedules the execution of the function func(*args) using a thread from the executor pool and returns a Future representing the execution of the asynchronous task

  Ex: future_one = executor.submit(factorial, 3)

  The method shutdown() signals the executor to free all the threads and its resources from the executor pool when all the currently executing future tasks have completed

  Ex: executor.shutdown()

The following is a simple python program named FuturesTwo.py that demonstrates the asynchronous future execution using a callback function:

#
# Name: FuturesTwo.py
#

import concurrent.futures

# ----- Method factorial -----

def factorial(n):
    if n <= 0:
        return 1
    return n * factorial(n-1)

# ----- Method done_callback -----

def done_callback(fut):
    result = None

    try:
        result = fut.result()
    except Exception as exc:
        print("FuturesTwo:: <future_callback> :: Exception: %s" % exc)

    print("FuturesTwo:: <future_callback> :: result = " + str(result))

# ----- Main -----

if __name__ == '__main__':
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=2)

    future_one = executor.submit(factorial, 3)
    future_one.add_done_callback(done_callback)

    future_two = executor.submit(factorial, 5)
    future_two.add_done_callback(done_callback)

    executor.shutdown()

Executing FuturesTwo.py results in the following output:

Output.2

FuturesTwo:: <future_callback> :: result = 6
FuturesTwo:: <future_callback> :: result = 120

The code in the Python program FuturesTwo.py is similar to the previous Python program FuturesOne.py except for the callback.

The method add_done_callback(func) allows one to attach a callback function on a Future. The specified callback function (func) is invoked when the asynchronous future task either completes execution or is cancelled.

The callback function has the format:

def func(future):
{ [body] }

where the passed argument is of type Future.

Ex: future_one.add_done_callback(done_callback)

The following is a simple python program named FuturesThree.py that demonstrates the asynchronous future execution of a collection of iterables:

#
# Name: FuturesThree.py
#

import concurrent.futures

# ----- Method factorial -----

def factorial(n):
    if n <= 0:
        return 1
    return n * factorial(n-1)

# ----- Main -----

if __name__ == '__main__':
    numbers = [3, 4, 5, 6, 7]

    print("FuturesThree:: <main> :: numbers for factorial = " + str(numbers))

    executor = concurrent.futures.ThreadPoolExecutor(max_workers=3)

    results = [x for x in executor.map(factorial, numbers)]

    print("FuturesThree:: <main> :: results from map execution = " + str(results))

    executor.shutdown()

Executing FuturesThree.py results in the following output:

Output.3

FuturesThree:: <main> :: numbers for factorial = [3, 4, 5, 6, 7]
FuturesThree:: <main> :: results from map execution = [6, 24, 120, 720, 5040]

The method map(func, iterables) asynchronously executes the specified function (func) for each of the values from the specified iterables (collection) using the available threads from the executor pool. The results are returned as an iterator. If one or more of the asynchronous future task raises an exception then those exceptions are raised when the result is fetched from the iterator.

Ex: results = [x for x in executor.map(factorial, numbers)]

References

Python Quick Notes :: Part - 1

Python Quick Notes :: Part - 2

Python Quick Notes :: Part - 3

Python Quick Notes :: Part - 4

Python Quick Notes :: Part - 5

Python Quick Notes :: Part - 6

Introspection in Python

Python 'with' Statement