Futuro

Futuro é um contêiner que pode conter o resultado de um cálculo ou um erro que ocorreu durante esse cálculo. Quando um futuro é criado, ele começa no estado PENDENTE. A biblioteca não pretende que este objeto seja criado manualmente, exceto talvez para fins de teste.

import concurrent.futures as futures

f = futures.Future()
assert(f._result is None)
assert(f._exception is None)
assert(f._state == 'PENDING')

O status PENDING indica que uma computação solicitada pelo usuário foi registrada no pool de threads e colocada em uma fila, mas ainda não foi captada por nenhuma thread para execução. Depois que um thread livre retira a tarefa (retorno de chamada) da fila, o futuro passa para o estado RUNNING. Um futuro só pode ser cancelado enquanto estiver no estado PENDENTE. Portanto, há uma janela de tempo entre os estados PENDING e RUNNING durante a qual o cálculo solicitado pode ser cancelado.

import concurrent.futures as futures

def should_cancel_pending_future():
    f = futures.Future()
    assert(f._state == 'PENDING')
    assert(f.cancel())
    assert(f._state == 'CANCELLED')

def should_not_cancel_running_future():
    f = futures.Future()
    f.set_running_or_notify_cancel()
    assert(f._state == 'RUNNING')
    assert(not f.cancel())

def cancel_is_idempotent():
    f = futures.Future()
    assert(f.cancel())
    assert(f.cancel())


should_cancel_pending_future()
should_not_cancel_running_future()
cancel_is_idempotent()

Uma operação solicitada no pool de threads pode ser concluída com um valor computado ou resultar em um erro. Independentemente do resultado, as futuras transições para o estado FINISHED. O resultado ou erro é então armazenado nos campos correspondentes.

import concurrent.futures as futures

def future_completed_with_result():
    f = futures.Future()
    f.set_result('foo')
    assert(f._state == 'FINISHED')
    assert(f._result == 'foo')
    assert(f._exception is None)

def future_completed_with_exception():
    f = futures.Future()
    f.set_exception(NameError())
    assert(f._state == 'FINISHED')
    assert(f._result is None)
    assert(isinstance(f._exception, NameError))

future_completed_with_result()
future_completed_with_exception()

Para recuperar o resultado de um cálculo, o método result é usado. Se o cálculo ainda não estiver concluído, este método bloqueará o thread atual (do qual o resultado foi chamado) até que o cálculo termine ou o tempo de espera expire.

Se o cálculo for concluído com êxito e sem erros, o método de resultado retornará o valor calculado.

import concurrent.futures as futures
import time
import threading

f = futures.Future()
def target():
    time.sleep(1)
    f.set_result('foo')
threading.Thread(target=target).start()
assert(f.result() == 'foo')

Se ocorreu uma exceção durante o cálculo, o resultado gerará essa exceção.

import concurrent.futures as futures
import time
import threading

f = futures.Future()
def target():
    time.sleep(1)
    f.set_exception(NameError)
threading.Thread(target=target).start()
try:
    f.result()
    raise Exception()
except NameError:
    assert(True)

Se o método expirar durante a espera, um TimeoutError será gerado.

import concurrent.futures as futures

f = futures.Future()
try:
    f.result(1)
    raise Exception()
except TimeoutError:
    assert(f._result is None)
    assert(f._exception is None)

A tentativa de obter o resultado de um cálculo que foi cancelado gerará um CancelledError.

import concurrent.futures as futures

f = futures.Future()
assert(f.cancel())
try:
    f.result()
    raise Exception()
except futures.CancelledError:
    assert(True)

Estratégia de espera

No processo de desenvolvimento, é bastante comum precisar executar N cálculos em um pool de threads e aguardar sua conclusão. Para conseguir isso, a biblioteca fornece a função de espera. Existem várias estratégias de espera: FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED.

Comum a todas as estratégias de espera é que se os futuros passados ​​para o método de espera já estiverem concluídos, a coleção dos futuros passados ​​é retornada independentemente da estratégia escolhida. Não importa como foram concluídos, seja com erro, resultado ou se foram cancelados.

import concurrent.futures as futures

def test(return_when):
    f1, f2, f3 = futures.Future(), futures.Future(), futures.Future()
    f1.cancel()
    f1.set_running_or_notify_cancel() # required
    f2.set_result('foo')
    f3.set_exception(NameError)

    r = futures.wait([f1, f2, f3], return_when=return_when)
    assert(len(r.done) == 3)
    assert(len(r.not_done) == 0)

for return_when in [futures.ALL_COMPLETED, futures.FIRST_EXCEPTION, futures.FIRST_COMPLETED]:
    test(return_when)

Estratégia ALL_COMPLETED

A estratégia ALL_COMPLETED garante aguardar a conclusão de todos os futuros passados, ou sair após um tempo limite com uma coleção de futuros concluída até aquele momento, que pode estar incompleta.

import concurrent.futures as futures
import threading
import time

def should_wait_for_all_futures_to_complete():
    f1 = futures.Future()
    f1.set_result('foo')
    f2 = futures.Future()

    def target():
        time.sleep(1)
        f2.set_result('bar')

    threading.Thread(target=target).start()
    r = futures.wait([f1, f2], return_when=futures.ALL_COMPLETED)
    assert(len(r.done) == 2)

def should_exit_on_timeout():
    f1 = futures.Future()
    f1.set_result('foo')
    f2 = futures.Future()
    r = futures.wait(fs=[f1, f2], timeout=1, return_when=futures.ALL_COMPLETED)
    assert(len(r.done) == 1)


should_wait_for_all_futures_to_complete()
should_exit_on_timeout()

FIRST_COMPLETED

A estratégia FIRST_COMPLETED garante o retorno de uma coleção com pelo menos um futuro concluído ou uma coleção vazia em caso de timeout. Esta estratégia NÃO implica que a coleção retornada não possa conter vários elementos.

import concurrent.futures as futures
import threading
import time

f1 = futures.Future()
f2 = futures.Future()

def target():
    time.sleep(1)
    f1.set_result(True)

threading.Thread(target=target).start()

r = futures.wait([f1, f2], return_when=futures.FIRST_COMPLETED)
assert(len(r.done) == 1)
assert(len(r.not_done) == 1)

FIRST_EXCEPTION

A estratégia FIRST_EXCEPTION interrompe a espera se um dos cálculos terminar com erro. Se nenhuma exceção ocorrer, o comportamento será idêntico ao futuro ALL_COMPLETED.

import concurrent.futures as futures
import threading
import time

f1 = futures.Future()
f1.set_result('foo')
f2, f3 = futures.Future(), futures.Future()

def target():
    time.sleep(1)
    f2.set_exception(NameError())

threading.Thread(target=target).start()

r = futures.wait(fs=[f1, f2, f3], return_when=futures.FIRST_EXCEPTION)
assert(len(r.done) == 2)

ThreadPoolExecutor

O objeto é responsável por criar um pool de threads. O principal método para interagir com este objeto é o método Submit. Permite registrar uma computação no pool de threads. Em resposta, é retornado um objeto Future, que é utilizado para monitorar o status do cálculo e obter o resultado final.

Propriedades

• Novos tópicos são criados SOMENTE conforme necessário:
  • Se houver pelo menos um thread livre quando um cálculo for solicitado, nenhum novo thread será criado
  • Se não houver threads livres quando um cálculo for solicitado, um novo thread será criado, desde que o limite de maxWorkers não tenha sido atingido.
  • Se não houver threads livres e o limite maxWorkers tiver sido atingido, o cálculo é colocado em uma fila e será levado pelo próximo thread disponível
• O número máximo de threads alocados para necessidades computacionais por padrão é igual ao número de núcleos de processador lógico
• Uma vez criado, um thread não é destruído, mesmo em caso de carga baixa