kwcoco.util.util_futures module

Deprecated and functionality moved to ubelt

class kwcoco.util.util_futures.Executor(mode='thread', max_workers=0)

Bases: object

A concrete asynchronous executor with a configurable backend.

The type of parallelism (or lack thereof) is configured via the mode parameter, which can be: “process”, “thread”, or “serial”. This allows the user to easily enable / disable parallelism or switch between processes and threads without modifying the surrounding logic.

SeeAlso:

• concurrent.futures.ThreadPoolExecutor

• concurrent.futures.ProcessPoolExecutor

• SerialExecutor

• JobPool

In the case where you cant or dont want to use ubelt.Executor you can get similar behavior with the following pure-python snippet:

def Executor(max_workers):
    # Stdlib-only "ubelt.Executor"-like behavior
    if max_workers == 1:
        import contextlib
        def submit_partial(func, *args, **kwargs):
            def wrapper():
                return func(*args, **kwargs)
            wrapper.result = wrapper
            return wrapper
        executor = contextlib.nullcontext()
        executor.submit = submit_partial
    else:
        from concurrent.futures import ThreadPoolExecutor
        executor = ThreadPoolExecutor(max_workers=max_workers)
    return executor

executor = Executor(0)
with executor:
    jobs = []

    for arg in range(1000):
        job = executor.submit(chr, arg)
        jobs.append(job)

    results = []
    for job in jobs:
        result = job.result()
        results.append(result)

print('results = {}'.format(ub.urepr(results, nl=1)))

Variables:

backend (SerialExecutor | ThreadPoolExecutor | ProcessPoolExecutor) –

Example

>>> import ubelt as ub
>>> # Prototype code using simple serial processing
>>> executor = ub.Executor(mode='serial', max_workers=0)
>>> jobs = [executor.submit(sum, [i + 1, i]) for i in range(10)]
>>> print([job.result() for job in jobs])
[1, 3, 5, 7, 9, 11, 13, 15, 17, 19]

>>> # Enable parallelism by only changing one parameter
>>> executor = ub.Executor(mode='process', max_workers=0)
>>> jobs = [executor.submit(sum, [i + 1, i]) for i in range(10)]
>>> print([job.result() for job in jobs])
[1, 3, 5, 7, 9, 11, 13, 15, 17, 19]

Parameters:

• mode (str) – The backend parallelism mechanism. Can be either thread, serial, or process. Defaults to ‘thread’.

• max_workers (int) – number of workers. If 0, serial is forced. Defaults to 0.

submit(func, *args, **kw)

Calls the submit function of the underlying backend.

Returns:

a future representing the job

Return type:

concurrent.futures.Future

shutdown()

Calls the shutdown function of the underlying backend.

map(fn, *iterables, **kwargs)

Calls the map function of the underlying backend.

CommandLine

xdoctest -m ubelt.util_futures Executor.map

Example

>>> import ubelt as ub
>>> import concurrent.futures
>>> import string
>>> with ub.Executor(mode='serial') as executor:
...     result_iter = executor.map(int, string.digits)
...     results = list(result_iter)
>>> print('results = {!r}'.format(results))
results = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
>>> with ub.Executor(mode='thread', max_workers=2) as executor:
...     result_iter = executor.map(int, string.digits)
...     results = list(result_iter)
>>> # xdoctest: +IGNORE_WANT
>>> print('results = {!r}'.format(results))
results = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

class kwcoco.util.util_futures.JobPool(mode='thread', max_workers=0, transient=False)

Bases: object

Abstracts away boilerplate of submitting and collecting jobs

This is a basic wrapper around ubelt.util_futures.Executor that simplifies the most basic case by 1. keeping track of references to submitted futures for you and 2. providing an as_completed method to consume those futures as they are ready.

Variables:

• executor (Executor) – internal executor object

• jobs (List[Future]) – internal job list. Note: do not rely on this attribute, it may change in the future.

Example

>>> import ubelt as ub
>>> def worker(data):
>>>     return data + 1
>>> pool = ub.JobPool('thread', max_workers=16)
>>> for data in ub.ProgIter(range(10), desc='submit jobs'):
>>>     pool.submit(worker, data)
>>> final = []
>>> for job in pool.as_completed(desc='collect jobs'):
>>>     info = job.result()
>>>     final.append(info)
>>> print('final = {!r}'.format(final))

Parameters:

• mode (str) – The backend parallelism mechanism. Can be either thread, serial, or process. Defaults to ‘thread’.

• max_workers (int) – number of workers. If 0, serial is forced. Defaults to 0.

• transient (bool) – if True, references to jobs will be discarded as they are returned by as_completed(). Otherwise the jobs attribute holds a reference to all jobs ever submitted. Default to False.

submit(func, *args, **kwargs)

Submit a job managed by the pool

Parameters:

• func (Callable[…, Any]) – A callable that will take as many arguments as there are passed iterables.

• *args – positional arguments to pass to the function

• *kwargs – keyword arguments to pass to the function

Returns:

a future representing the job

Return type:

concurrent.futures.Future

shutdown()

_clear_completed()

as_completed(timeout=None, desc=None, progkw=None)

Generates completed jobs in an arbitrary order

Parameters:

• timeout (float | None) – Specify the the maximum number of seconds to wait for a job. Note: this is ignored in serial mode.

• desc (str | None) – if specified, reports progress with a ubelt.progiter.ProgIter object.

• progkw (dict | None) – extra keyword arguments to ubelt.progiter.ProgIter.

Yields:

concurrent.futures.Future – The completed future object containing the results of a job.

CommandLine

xdoctest -m ubelt.util_futures JobPool.as_completed

Example

>>> import ubelt as ub
>>> pool = ub.JobPool('thread', max_workers=8)
>>> text = ub.paragraph(
...     '''
...     UDP is a cool protocol, check out the wiki:
...
...     UDP-based Data Transfer Protocol (UDT), is a high-performance
...     data transfer protocol designed for transferring large
...     volumetric datasets over high-speed wide area networks. Such
...     settings are typically disadvantageous for the more common TCP
...     protocol.
...     ''')
>>> for word in text.split(' '):
...     pool.submit(print, word)
>>> for _ in pool.as_completed():
...     pass
>>> pool.shutdown()

join(**kwargs)

Like JobPool.as_completed(), but executes the result method of each future and returns only after all processes are complete. This allows for lower-boilerplate prototyping.

Parameters:

**kwargs – passed to JobPool.as_completed()

Returns:

list of results

Return type:

List[Any]

Example

>>> import ubelt as ub
>>> # We just want to try replacing our simple iterative algorithm
>>> # with the embarrassingly parallel version
>>> arglist = list(zip(range(1000), range(1000)))
>>> func = ub.identity
>>> #
>>> # Original version
>>> for args in arglist:
>>>     func(*args)
>>> #
>>> # Potentially parallel version
>>> jobs = ub.JobPool(max_workers=0)
>>> for args in arglist:
>>>     jobs.submit(func, *args)
>>> _ = jobs.join(desc='running')