如何以非阻塞方式链接期货？也就是说，如何在不阻塞的情况下将一个future用作另一个future的输入？

Posted 2023-02-16

【中文标题】如何以非阻塞方式链接期货？也就是说，如何在不阻塞的情况下将一个future用作另一个future的输入？

【英文标题】：How to chain futures in a non-blocking manner? That is, how to use one future as an input in another future without blocking?

【发布时间】：2018-02-05 22:48:42

【问题描述】：

使用下面的例子，一旦future1完成（不阻止future3被提交），future2如何使用future1的结果？

from concurrent.futures import ProcessPoolExecutor
import time

def wait(seconds):
    time.sleep(seconds)
    return seconds

pool = ProcessPoolExecutor()

s = time.time()
future1 = pool.submit(wait, 5)
future2 = pool.submit(wait, future1.result())
future3 = pool.submit(wait, 10)

time_taken = time.time() - s
print(time_taken)

【参考方案1】：

这可以通过精心设计一个回调来实现，以便在第一个操作完成后提交第二个操作。遗憾的是，无法将任意未来传递给pool.submit，因此需要额外的步骤将两个未来绑定在一起。

这是一个可能的实现：

import concurrent.futures

def copy_future_state(source, destination):
    if source.cancelled():
        destination.cancel()
    if not destination.set_running_or_notify_cancel():
        return
    exception = source.exception()
    if exception is not None:
        destination.set_exception(exception)
    else:
        result = source.result()
        destination.set_result(result)


def chain(pool, future, fn):
    result = concurrent.futures.Future()

    def callback(_):
        try:
            temp = pool.submit(fn, future.result())
            copy = lambda _: copy_future_state(temp, result)
            temp.add_done_callback(copy)
        except:
            result.cancel()
            raise

    future.add_done_callback(callback)
    return result

请注意，copy_future_state 是 asyncio.futures._set_concurrent_future_state 的略微修改版本。

用法：

from concurrent.futures import ProcessPoolExecutor

def wait(seconds):
    time.sleep(seconds)
    return seconds

pool = ProcessPoolExecutor()
future1 = pool.submit(wait, 5)
future2 = chain(pool, future1, wait)
future3 = pool.submit(wait, 10)