Jak połączyć futures w sposób nieblokujący? To znaczy, jak wykorzystać jedną przyszłość jako dane wejściowe w innej przyszłości bez blokowania?

Question

Korzystając z poniższego przykładu, w jaki sposób future2 można użyć wyniku future1 po zakończeniu future1 (bez blokowania przed wysłaniem future3)?

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

Jest to możliwe do osiągnięcia przez staranne wykonanie wywołania zwrotnego w celu przesłania drugiej operacji po zakończeniu pierwszej operacji. Niestety, nie jest możliwe przekazanie arbitralnej przyszłości pool.submit, więc dodatkowy krok jest wymagany do związania dwóch kontraktów futures.

Tutaj jest możliwa realizacja:

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 

Zauważ, że copy_future_state jest nieco zmodyfikowaną wersją asyncio.futures._set_concurrent_future_state.

Zastosowanie:

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)