Làm thế nào để chuỗi tương lai theo cách không bị chặn? Đó là, làm thế nào để sử dụng một tương lai như một đầu vào trong tương lai khác mà không ngăn chặn?

Question

Sử dụng ví dụ dưới đây, cách future2 sử dụng kết quả của future1 sau khi future1 hoàn tất (không chặn future3 gửi được)?

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

Điều này có thể đạt được bằng cách cẩn thận tạo một cuộc gọi lại để gửi thao tác thứ hai sau khi thao tác đầu tiên hoàn thành. Đáng buồn thay, không thể vượt qua một tương lai tùy ý để pool.submit do đó cần thêm một bước nữa để ràng buộc hai tương lai lại với nhau.

Đây là một thực thể:

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 

Lưu ý rằng copy_future_state là một phiên bản sửa đổi nhẹ của asyncio.futures._set_concurrent_future_state.

Cách sử dụng:

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)