python concurrent.futures

Avenir

Future est un conteneur qui peut contenir soit le résultat d'un calcul, soit une erreur survenue lors de ce calcul. Lorsqu'un futur est créé, il démarre dans un état PENDING. La bibliothèque n'a pas l'intention de créer cet objet manuellement, sauf peut-être à des fins de test.

import concurrent.futures as futures

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

Le statut PENDING indique qu'un calcul demandé par l'utilisateur a été enregistré dans le pool de threads et placé dans une file d'attente, mais qu'il n'a encore été récupéré par aucun thread pour exécution. Une fois qu'un thread libre prend la tâche (rappel) de la file d'attente, le futur passe à l'état RUNNING. Un futur ne peut être annulé que lorsqu'il est à l'état PENDING. Par conséquent, il existe une fenêtre de temps entre les états PENDING et RUNNING pendant laquelle le calcul demandé peut être annulé.

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

Une opération demandée dans le pool de threads peut soit se terminer avec une valeur calculée, soit entraîner une erreur. Quel que soit le résultat, le futur passe à l’état FINISHED. Le résultat ou l'erreur est ensuite stocké dans les champs correspondants.

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

Pour récupérer le résultat d'un calcul, la méthode result est utilisée. Si le calcul n'est pas encore terminé, cette méthode bloquera le thread actuel (à partir duquel le résultat a été appelé) jusqu'à la fin du calcul ou jusqu'à l'expiration du délai d'attente.

Si le calcul se termine avec succès sans erreur, la méthode de résultat renvoie la valeur calculée.

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

Si une exception s'est produite pendant le calcul, le résultat déclenchera cette exception.

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)

Si la méthode expire pendant l'attente, une TimeoutError est levée.

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)

La tentative d'obtenir le résultat d'un calcul qui a été annulé générera une CancelledError.

import concurrent.futures as futures

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

Stratégie d'attente

Dans le processus de développement, il est assez courant de devoir exécuter N calculs sur un pool de threads et d'attendre leur achèvement. Pour y parvenir, la bibliothèque fournit une fonction d'attente. Il existe plusieurs stratégies d'attente : FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED.

Le point commun à toutes les stratégies d'attente est que si les contrats à terme passés à la méthode d'attente sont déjà terminés, la collecte des contrats à terme passés est renvoyée quelle que soit la stratégie choisie. Peu importe la manière dont ils ont été complétés, qu'il s'agisse d'une erreur, d'un résultat ou s'ils ont été annulés.

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)

Stratégie ALL_COMPLETED

La stratégie ALL_COMPLETED garantit l'attente de l'achèvement de tous les futures passés, ou la sortie après un délai d'attente avec une collection des futures complétées jusqu'à ce moment, qui peuvent être incomplètes.

import concurrent.futures as futures

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

FIRST_COMPLETED

La stratégie FIRST_COMPLETED garantit le retour d'une collection avec au moins un futur complété ou d'une collection vide en cas de timeout. Cette stratégie N'implique PAS que la collection renvoyée ne peut pas contenir plusieurs éléments.

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

FIRST_EXCEPTION

La stratégie FIRST_EXCEPTION interrompt l'attente si l'un des calculs se termine par une erreur. Si aucune exception ne se produit, le comportement est identique au futur ALL_COMPLETED.

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

ThreadPoolExécuteur

L'objet est responsable de la création d'un pool de threads. La méthode principale pour interagir avec cet objet est la méthode Submit. Il permet d'enregistrer un calcul dans le pool de threads. En réponse, un objet Future est renvoyé, qui est utilisé pour surveiller l'état du calcul et obtenir le résultat final.

Propriétés

• De nouveaux fils de discussion sont créés UNIQUEMENT selon les besoins :
  • S'il y a au moins un fil libre lorsqu'un calcul est demandé, aucun nouveau fil n'est créé
  • S'il n'y a pas de threads libres lorsqu'un calcul est demandé, un nouveau thread est créé, à condition que la limite maxWorkers n'ait pas été atteinte.
  • S'il n'y a pas de threads libres et que la limite de maxWorkers a été atteinte, le calcul est placé dans une file d'attente et sera repris par le prochain thread disponible
• Le nombre maximum de threads alloués par défaut aux besoins de calcul est égal au nombre de cœurs de processeur logique
• Une fois créé, un thread n'est pas détruit, même en cas de faible charge

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!