Queue.Queue vs collections.deque

J'ai besoin d'une file d'attente dans laquelle plusieurs threads peuvent mettre des éléments et à partir de laquelle plusieurs threads peuvent lire.

Python a au moins deux classes de files d'attente, Queue.Queue et collections.deque, la première utilisant apparemment la seconde en interne. Les deux prétendent être thread-safe dans la documentation.

Cependant, la documentation de la file d'attente indique également:

collections.deque est une implémentation alternative de files d'attente illimitées avec des opérations atomiques rapides append () et popleft () qui ne nécessitent pas de verrouillage.

Ce que je suppose que je ne comprends pas tout à fait: cela signifie-t-il que deque n'est pas entièrement thread-safe après tout?

Si c'est le cas, je ne comprends peut-être pas entièrement la différence entre les deux classes. Je peux voir que Queue ajoute une fonctionnalité de blocage. D'autre part, il perd certaines fonctionnalités deque comme la prise en charge de l'opérateur interne.

Accéder directement à l'objet deque interne est

x dans la file d'attente (). deque

thread-safe?

Aussi, pourquoi Queue utilise-t-il un mutex pour ses opérations alors que deque est déjà thread-safe?

Queue.Queueet collections.dequeservent à des fins différentes. Queue.Queue est destiné à permettre à différents threads de communiquer à l'aide de messages / données en file d'attente, alors qu'il collections.dequeest simplement conçu comme une structure de données. C'est pourquoi Queue.Queuea des méthodes telles que put_nowait(), get_nowait()et join(), alors collections.dequene fonctionne pas. Queue.Queuen'est pas destiné à être utilisé comme une collection, c'est pourquoi il n'a pas les goûts de l' inopérateur.

Cela se résume à ceci: si vous avez plusieurs threads et que vous voulez qu'ils puissent communiquer sans avoir besoin de verrous, vous recherchez Queue.Queue; si vous voulez juste une file d'attente ou une file d'attente double comme structure de données, utilisez collections.deque.

Enfin, accéder et manipuler le deque interne d'un Queue.Queueest jouer avec le feu - vous ne voulez vraiment pas faire cela.

Si tout ce que vous recherchez est un moyen sûr pour les threads de transférer des objets entre les threads , les deux fonctionneraient (à la fois pour FIFO et LIFO). Pour FIFO:

• Queue.put()et Queue.get()sont thread-safe
• deque.append()et deque.popleft()sont thread-safe

Remarque:

• D'autres opérations sur dequepeuvent ne pas être thread-safe, je ne suis pas sûr.
• dequene bloque pas pop()ou popleft()donc vous ne pouvez pas baser votre flux de thread consommateur sur le blocage jusqu'à ce qu'un nouvel élément arrive.

Cependant, il semble que deque présente un avantage d'efficacité significatif . Voici quelques résultats de référence en quelques secondes en utilisant CPython 2.7.3 pour insérer et supprimer 100 000 éléments

deque 0.0747888759791
Queue 1.60079066852

Voici le code de référence:

import time
import Queue
import collections

q = collections.deque()
t0 = time.clock()
for i in xrange(100000):
    q.append(1)
for i in xrange(100000):
    q.popleft()
print 'deque', time.clock() - t0

q = Queue.Queue(200000)
t0 = time.clock()
for i in xrange(100000):
    q.put(1)
for i in xrange(100000):
    q.get()
print 'Queue', time.clock() - t0

Pour plus d'informations, il existe un ticket Python référencé pour deque thread-safety ( https://bugs.python.org/issue15329 ). Titre "clarifier quelles méthodes deque sont thread-safe"

Bottom line ici: https://bugs.python.org/issue15329#msg199368

Les opérations append (), appendleft (), pop (), popleft () et len ​​(d) de deque sont thread-safe dans CPython. Les méthodes d'ajout ont un DECREF à la fin (pour les cas où maxlen a été défini), mais cela se produit après que toutes les mises à jour de structure ont été effectuées et que les invariants ont été restaurés, il est donc normal de traiter ces opérations comme atomiques.

Quoi qu'il en soit, si vous n'êtes pas sûr à 100% et que vous préférez la fiabilité à la performance, mettez simplement un verrou similaire;)

Toutes les méthodes à un seul élément sur dequesont atomiques et thread-safe. Toutes les autres méthodes sont également thread-safe. Des choses comme len(dq), dq[4]donnent des valeurs correctes momentanées. Mais pensez par exemple à dq.extend(mylist)ceci: vous n'obtenez pas la garantie que tous les éléments dans mylistsont classés dans une ligne lorsque d'autres threads ajoutent également des éléments du même côté - mais ce n'est généralement pas une exigence dans la communication inter-thread et pour la tâche en question.

Donc, a dequeest ~ 20x plus rapide que Queue(qui utilise a dequesous le capot) et à moins que vous n'ayez pas besoin de l'API de synchronisation "confortable" (blocage / timeout), de l' maxsizeobéissance stricte ou du "Override these methods (_put, _get, ..) ) pour implémenter le comportement de sous-classification d' autres organisations de files d'attente , ou lorsque vous vous occupez de telles choses vous-même, alors un bare dequeest une bonne et efficace affaire pour la communication inter-thread à grande vitesse.

En fait, l'utilisation intensive d'un mutex supplémentaire et d' ._get()appels de méthode supplémentaires, etc. Queue.pyest due à des contraintes de compatibilité ascendante, à une sur-conception passée et au manque de soin pour fournir une solution efficace à cet important problème de goulot d'étranglement de vitesse dans la communication inter-thread. Une liste était utilisée dans les anciennes versions de Python - mais même list.append () /. Pop (0) était et est atomique et threadsafe ...

En ajoutant notify_all()à chacun d'eux deque append, des popleftrésultats bien pires dequeque l'amélioration de 20x obtenue par le dequecomportement par défaut :

deque + notify_all: 0.469802
Queue:              0.667279

@Jonathan modifie un peu son code et j'obtiens le benchmark en utilisant cPython 3.6.2 et j'ajoute une condition dans la boucle deque pour simuler le comportement de Queue.

import time
from queue import Queue
import threading
import collections

mutex = threading.Lock()
condition = threading.Condition(mutex)
q = collections.deque()
t0 = time.clock()
for i in range(100000):
    with condition:
        q.append(1)
        condition.notify_all()
for _ in range(100000):
    with condition:
        q.popleft()
        condition.notify_all()
print('deque', time.clock() - t0)

q = Queue(200000)
t0 = time.clock()
for _ in range(100000):
    q.put(1)
for _ in range(100000):
    q.get()
print('Queue', time.clock() - t0)

Et il semble que les performances soient limitées par cette fonction condition.notify_all()

collections.deque est une implémentation alternative de files d'attente illimitées avec des opérations rapides atomiques append () et popleft () qui ne nécessitent pas de verrouillage. file d'attente de documents

dequeest thread-safe. «opérations qui ne nécessitent pas de verrouillage» signifie que vous n'avez pas à faire le verrouillage vous-même, le deques'en charge.

En regardant la Queuesource, le deque interne est appelé self.queueet utilise un mutex pour les accesseurs et les mutations, il Queue().queuen'est donc pas sûr à utiliser pour les threads.

Si vous recherchez un opérateur "in", alors une deque ou une file d'attente n'est peut-être pas la structure de données la plus appropriée pour votre problème.

(il semble que je n'ai pas la réputation de commenter ...) Vous devez faire attention aux méthodes de deque que vous utilisez à partir de différents threads.

deque.get () semble être threadsafe, mais j'ai trouvé que faire

for item in a_deque:
   process(item)

peut échouer si un autre thread ajoute des éléments en même temps. J'ai eu une RuntimeException qui se plaignait de "deque muté pendant l'itération".

Vérifiez collectionsmodule.c pour voir quelles opérations sont affectées par cela