Python-Multiprozessor-Deadlock beim Aufruf von Condition.notify(), bevor der andere Prozess bereit ist

Question

Ich habe einen Produzenten und Verbraucher-Prozess und eine gemeinsame Speicherbereich zwischen ihnen. Ich verwende das Condition-Objekt, um den Konsumentenprozess aus dem Produzentenprozess zu signalisieren, wenn neue Daten verfügbar sind.

Aktuelles Problem ist die ArrayConsumer Deadlocks auf cond.wait() mit der folgenden Spur:

Producer: (3, 3, 30, 100) 
Producer Done 
Consumer: (3, 3, 30, 100) 
Consumer Waiting... 

Was aus meinem Verständnis geschieht, ist, dass der Hersteller die Verbraucher informiert, bevor die Verbraucher bereit ist, die Benachrichtigung zu erhalten, und wartet daher auf die Benachrichtigung des Herstellers. Was ist der beste Weg, um sicherzustellen, dass der Verbraucher nur dann signalisiert wird, wenn er bereit ist?

def ArrayConsumer(aArray, cond): 
    print('Consumer:', aArray.shape) 

    with cond: 
     print('Consumer Waiting...') 
     cond.wait() 
     print('Waiting finished..') 
     print('Consumer:', aArray[1, 1, 1, 1]) 

def ArrayProducer(aArray, cond): 
    print('Producer:', aArray.shape) 
    with cond: 
     aArray.fill(1) 
     cond.notify() 
    print('Producer Done') 

if __name__ == '__main__': 

    # sharing numpy arrays 
    csi = csiMatrix(3, 3, 30, 100) 
    shared = sm.sharedmem.empty(csi.shape, dtype=complex) 
    shared[:] = csi 

    cond = Condition() 

    p = Process(target=ArrayProducer, args=(shared, cond,)) 
    c = Process(target=ArrayConsumer, args=(shared, cond,)) 
    p.start() 
    c.start() 

    p.join() 
    c.join() 

Answer 1

Die Condition.notify Methode ist ein noop wenn niemand wartet. Sie müssen Ihre Logik in einer while-Schleife kapseln und regelmäßig prüfen, ob die Daten bereit sind. Die Python-Dokumentation hat eine P&C example. Sie müssten eine Zeitüberschreitung für die Methode wait festlegen und die Daten in Ihrem Array bei jeder Iteration überprüfen.

Es ist viel einfacher, eine Lock und eine Event zu verwenden, die einen Long-Polling-Mechanismus vermeiden würde.

def ArrayConsumer(aArray, lock, event): 
    print('Consumer:', aArray.shape) 
    print('Consumer Waiting...') 
    event.wait() 
    print('Waiting finished..') 
    with lock: 
     print('Consumer:', aArray[1, 1, 1, 1]) 
     event.clear() 

def ArrayProducer(aArray, lock, event): 
    print('Producer:', aArray.shape) 
    with lock: 
     aArray.fill(1) 
     event.set() 
    print('Producer Done')` 

Trotzdem Ich würde empfehlen, ein Pipe oder zu verwenden Queue das Erzeuger- und Verbrauchermuster zu implementieren. Es ist ein viel robuster Mechanismus.