Path: csiph.com!fu-berlin.de!uni-berlin.de!not-for-mail
From: Ian Kelly <ian.g.kelly@gmail.com>
Newsgroups: comp.lang.python
Subject: Re: Passing data across callbacks in ThreadPoolExecutor
Date: Wed, 17 Feb 2016 09:24:31 -0700
Lines: 42
Message-ID: <mailman.212.1455726313.22075.python-list@python.org>
References: <fe3d0dcf47cb4f378b7a2d2e38d00ed4@activenetwerx.com>
Mime-Version: 1.0
Content-Type: text/plain; charset=UTF-8
In-Reply-To: <fe3d0dcf47cb4f378b7a2d2e38d00ed4@activenetwerx.com>
Precedence: list
Xref: csiph.com comp.lang.python:103056

On Tue, Feb 16, 2016 at 2:48 PM, Joseph L. Casale
<jcasale@activenetwerx.com> wrote:
> What is the pattern for chaining execution of tasks with ThreadPoolExecutor?
> Callbacks is not an adequate facility as each task I have will generate new output.

Can you specify in more detail what your use case is?

If you don't mind having threads sitting around waiting, you can just
submit each chained task at the start and have each task wait on the
futures of its dependencies. The advantage of this is that it's easy
to conceptualize the dependency graph of the tasks. The disadvantage
is that it eats up extra threads. You'll probably want to increase the
size of the thread pool to handle the waiting tasks (or use a separate
ThreadPoolExecutor for each chained task).

E.g.:

def task2(input, f1):
    f1_result = f1.result()  # Wait for f1 to be done.
    result = frobnicate(input, f1_result)
    return result

f1 = executor.submit(task1, input)
f2 = executor.submit(task2, input, f1)

f1.add_done_callback(handle_task1_done)
f2.add_done_callback(handle_task2_done)



Otherwise, is there some reason you can't use multiple callbacks, one
to handle the task's output and one to submit the chained task?

E.g.:

def chain_task2(input, f2):
    f2 = executor.submit(task2, input, f2.result())
    f2.add_done_callback(handle_task2_done)

f1 = executor.submit(task1, input)
f1.add_done_callback(handle_task1_done)
f1.add_done_callback(functools.partial(chain_task2, input))