如何以非阻塞的方式鏈接期貨？也就是說，如何在不阻塞的情況下將未來用作未來的輸入？

Question

使用下面的示例，一旦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) 

Answer 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)