Python 2.7 Multiprozessing. pool.apply_async in einem anderen pool.apply_async

Question

python 2.7 multiprocessing. pool.apply_async andere Innen pool.apply_async

Ich habe 2-Module A und B. 'A' erklärt Modul einen Pool von Größe 100 und mit pool.apply_async, eine Funktion 'BX' von Modul 'B' bezeichnet wird. Die Funktion 'BX' in Modul 'B' erstellt einen anderen Pool der Größe n, und mit seiner pool.apply_async wird eine andere Funktion aufgerufen.

Dieses Problem, das jetzt im Vordergrund steht, ist, dass der Ausführungsprozess gerade an dem Punkt gestoppt/beendet wird, an dem der Pool in Modul 'B' deklariert wird. dh während der zweite Pool im Modul 'B' deklariert wird, der sich innerhalb des ersten Pools des Moduls 'A' befindet.

Jede Lösung beim Ausführen von pool.apply_async in pool.apply_async?

Answer 1

Sie müssen etwas tun, was ich unten zeige. Beachten Sie, dass ich meine eigene Verzweigung multiprocessing (pathos) verwende, da sie eine bessere Serialisierung bietet, die Ausführung über den Interpreter ermöglicht und effizienter in der Erstellung und Wartung von Pools ist. Allerdings sollte der Workflow in etwa gleich sein wie hier gezeigt:

>>> import pathos as p 
>>> tp1 = p.pools.ThreadPool(100) 
>>> tp2 = p.pools.ThreadPool(50) 
>>> import itertools as it 
>>> 
>>> res = tp1.amap(tp2.amap, it.repeat(lambda x:x**2), [range(10)]*5) 
>>> ans = res.get() 
>>> [v.get() for v in ans] 
[[0, 1, 4, 9, 16, 25, 36, 49, 64, 81], [0, 1, 4, 9, 16, 25, 36, 49, 64, 81], [0, 1, 4, 9, 16, 25, 36, 49, 64, 81], [0, 1, 4, 9, 16, 25, 36, 49, 64, 81], [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]] 

Beachten Sie, dass amap in pathos ist map_async in multiprocessing. Wenn Sie nur wollten apply_async, das ist apipe in pathos. Dann Dinge herunterzufahren:

>>> tp1.close() 
>>> tp2.close() 
>>> tp1.join() 
>>> tp2.join() 
>>> # pathos has an additional shutdown step to clear the pool cache 
>>> tp1.clear() 
>>> tp2.clear() 

Also mit dem Äquivalent von apply_async, es so aussehen würde:

>>> tp1 = p.pools.ThreadPool(100) 
>>> tp2 = p.pools.ThreadPool(50) 
>>> tp1.apipe(tp2.apipe, lambda x:x**2, 10).get().get() 
100 
>>> tp1.close() 
>>> tp2.close() 
>>> tp1.join() 
>>> tp2.join() 
>>> tp1.clear() 
>>> tp2.clear()