¿Este enfoque de Python productor-consumidor sin bloqueo es seguro para subprocesos?

Question

Recientemente escribí un programa que usaba un patrón simple de productor/consumidor. Inicialmente tenía un error relacionado con el uso incorrecto de subprocesos. Bloqueo que finalmente arreglé. Pero me hizo pensar si es posible implementar el patrón de productor/consumidor de una manera sin cerraduras.

Requisitos en mi caso eran simples:

• Un hilo productor.
• Un hilo de consumo.
• Queue tiene lugar para un solo artículo.
• El productor puede producir el siguiente artículo antes de que se consuma el actual. El artículo actual se pierde, pero eso está bien para mí.
• El consumidor puede consumir el artículo actual antes de que se produzca el siguiente. Por lo tanto, el artículo actual se consume dos veces (o más), pero eso está bien para mí.

Así que escribió esto:

QUEUE_ITEM = None 

# this is executed in one threading.Thread object 
def producer(): 
    global QUEUE_ITEM 
    while True: 
     i = produce_item() 
     QUEUE_ITEM = i 

# this is executed in another threading.Thread object 
def consumer(): 
    global QUEUE_ITEM 
    while True: 
     i = QUEUE_ITEM 
     consume_item(i) 

Mi pregunta es: ¿Es este código flujos seguros?

Comentario inmediato: este código no es realmente sin cerrojo: uso CPython y tiene GIL.

He probado el código un poco y parece funcionar. Se traduce en algunas operaciones LOAD y STORE que son atómicas debido a GIL. Pero también sé que el funcionamiento de del x no es atómico cuando x implementa el método __del__. Entonces, si mi artículo tiene un método __del__ y ocurre una programación desagradable, las cosas pueden romperse. ¿O no?

Otra pregunta es: ¿Qué tipo de restricciones (por ejemplo, en el tipo de elementos producidos) debo imponer para que el código anterior funcione bien?

Mis preguntas son solo acerca de la posibilidad teórica de explotar los caprichos de CPython y GIL para llegar a una solución sin cerraduras (es decir, no bloquea como el enhebrado. Bloqueo explícito en el código).

Answer 1

Si esto va a funcionar de la manera que usted describe:

1. que el productor puede producir un elemento que se puede omitir.
2. Que el consumidor puede consumir el mismo elemento.

Pero también sé que el funcionamiento del X no es atómico cuando x implementos del método. Entonces, si mi artículo tiene un método del y ocurre una programación desagradable, las cosas pueden romperse.

No veo un "del" aquí. Si un del sucede en el elemento de consumo, el del puede aparecer en el hilo del productor. No creo que esto sea un "problema".

No se moleste en usar esto sin embargo. Terminará usando la CPU en ciclos de sondeo inútiles, y no es más rápido que usar una cola con bloqueos ya que Python ya tiene un bloqueo global.

Answer 2

Esto no es realmente hilo seguro porque productor podría sobrescribir QUEUE_ITEM antes consumidor ha consumido y el consumidor podría consumir QUEUE_ITEM dos veces. Como mencionaste, estás de acuerdo con eso, pero la mayoría de la gente no.

Alguien con más conocimiento de las partes internas de cpython tendrá que responderle preguntas más teóricas.

Answer 3

Creo que es posible que un hilo se interrumpa durante la producción/consumo, especialmente si los objetos son grandes. Editar: esto es solo una conjetura salvaje. No soy un experto

También los hilos pueden producir/consumir cualquier cantidad de elementos antes de que el otro empiece a funcionar.

Answer 4

Puede usar una lista como la cola siempre y cuando se adhiera a pop/pop ya que ambos son atómicos.

QUEUE = [] 

# this is executed in one threading.Thread object 
def producer(): 
    global QUEUE 
    while True: 
     i = produce_item() 
     QUEUE.append(i) 

# this is executed in another threading.Thread object 
def consumer(): 
    global QUEUE 
    while True: 
     try: 
      i = QUEUE.pop(0) 
     except IndexError: 
      # queue is empty 
      continue 

     consume_item(i) 

En un alcance de clase como el siguiente, incluso puede borrar la cola.

class Atomic(object): 
    def __init__(self): 
     self.queue = [] 

    # this is executed in one threading.Thread object 
    def producer(self): 
     while True: 
      i = produce_item() 
      self.queue.append(i) 

    # this is executed in another threading.Thread object 
    def consumer(self): 
     while True: 
      try: 
       i = self.queue.pop(0) 
      except IndexError: 
       # queue is empty 
       continue 

      consume_item(i) 

    # There's the possibility producer is still working on it's current item. 
    def clear_queue(self): 
     self.queue = [] 

Vas a tener que averiguar qué operaciones son atómicos lista por mirar el código de bytes generada.

Answer 5

Trickery te morderá. Simplemente use Queue para comunicarse entre hilos.

Answer 6

El __del__ podría ser un problema como usted dijo. Se podría evitar, si hubiera una forma de evitar que el recolector de basura invoque el método __del__ en el objeto anterior antes de que terminemos de asignar el nuevo al QUEUE_ITEM. Necesitaríamos algo como:

increase the reference counter on the old object 
assign a new one to `QUEUE_ITEM` 
decrease the reference counter on the old object 

Me temo, no sé si es posible, sin embargo.