Questo approccio lockless consumer-consumer Python è thread-safe?

Question

Ho recentemente scritto un programma che utilizzava un semplice pattern produttore/consumatore. Inizialmente aveva un bug relativo all'utilizzo improprio del threading.Lock che alla fine ho risolto. Ma mi ha fatto pensare se è possibile implementare il modello produttore/consumatore in un modo senza blocco.

Requisiti nel mio caso erano semplici:

• Un thread produttore.
• Un thread di consumo.
• La coda ha posto per un solo articolo.
• Il produttore può produrre l'elemento successivo prima che venga consumato quello corrente. L'oggetto attuale è quindi perso, ma va bene per me.
• Il consumatore può consumare l'articolo corrente prima che venga prodotto quello successivo. L'oggetto corrente viene quindi consumato due volte (o più), ma per me va bene.

Così ho scritto questo:

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) 

La mia domanda è: E 'questo codice thread-safe?

Commento immediato: questo codice non è veramente privo di blocco - Io uso CPython e ha GIL.

Ho testato il codice un po 'e sembra funzionare. Si traduce in alcune operazioni LOAD e STORE che sono atomiche a causa di GIL. Ma so anche che l'operazione del x non è atomica quando x implementa il metodo __del__. Quindi, se il mio articolo ha un metodo __del__ e accade qualche brutto scheduling, le cose potrebbero rompersi. O no?

Un'altra domanda è: che tipo di restrizioni (ad esempio sul tipo di articoli prodotti) devo imporre per far funzionare correttamente il codice sopra?

Le mie domande riguardano solo la possibilità teorica di sfruttare le stranezze di CPython e GIL per ottenere soluzioni senza lock (cioè senza serrature come threading.Lock esplicitamente nel codice).

Answer 1

Sì questo lavoro nel modo in cui hai descritto:

1. che il produttore può produrre un elemento skippable.
2. Che il consumatore possa consumare lo stesso elemento.

Ma so anche che del funzionamento x non è atomica quando x attrezzi del metodo. Quindi, se il mio articolo ha un metodo del e accade qualche brutto scheduling, le cose potrebbero rompersi.

Non vedo un "del" qui. Se si verifica un delume in consume_item, è possibile che del si verifichi nel thread del produttore. Non penso che questo sarebbe un "problema".

Non preoccupatevi di usare questo però. Finirai per usare la CPU su cicli di polling inutili, e non sarà più veloce dell'uso di una coda con blocchi dato che Python ha già un blocco globale.

Answer 2

Questo non è davvero thread-safe perché produttore potrebbe sovrascrivere QUEUE_ITEM prima consumatore ha consumato e consumatore potrebbe consumare QUEUE_ITEM due volte. Come hai detto, sei d'accordo, ma la maggior parte delle persone non lo è.

Qualcuno con una maggiore conoscenza degli interni di cpython dovrà rispondere a più domande teoriche.

Answer 3

Penso che sia possibile che un thread venga interrotto durante la produzione/consumo, specialmente se gli oggetti sono oggetti grandi. Modifica: questa è solo un'ipotesi. Non sono esperto

Anche i thread possono produrre/consumare qualsiasi numero di elementi prima che l'altro inizi a funzionare.

Answer 4

È possibile utilizzare un elenco come coda fintanto che si tiene premuto append/pop poiché entrambi sono atomici.

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) 

In un ambito di classe come di seguito, è anche possibile cancellare la coda.

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 = [] 

Dovrai scoprire quali operazioni di elenco sono atomiche osservando il bytecode generato.

Answer 5

Trickery ti morderà. Basta usare la coda per comunicare tra i thread.

Answer 6

Il __del__ potrebbe essere un problema come hai detto tu. Potrebbe essere evitato, se solo ci fosse un modo per impedire al garbage collector di invocare il metodo __del__ sul vecchio oggetto prima di completare l'assegnazione di quello nuovo allo QUEUE_ITEM. Avremmo bisogno di qualcosa del tipo:

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

Ho paura, non so se è possibile, però.