Process.join() ve kuyruk çok sayıda

Question

Ben multiprocessing.Process kullanarak döngü yani

N = 1000000 
for i in xrange(N): 
    #do something 

için bölmek çalışıyorum

çalışmaz ve N. küçük değerleri için iyi çalışır Sorun ne zaman ortaya Daha büyük N değerlerini kullanıyorum. P.join() 'den önce veya sırasında garip bir şeyler oluyor ve program cevap vermiyor. Eğer baskıyı koyarsam, fonksiyonun tanımında q.put (i) yerine her şey iyi çalışır.

Herhangi bir yardım için minnettarım. İşte kod.

from multiprocessing import Process, Queue 

def f(q,nMin, nMax): # function for multiprocessing 
    for i in xrange(nMin,nMax): 
     q.put(i) 

if __name__ == '__main__': 

    nEntries = 1000000 

    nCpu = 10 
    nEventsPerCpu = nEntries/nCpu 
    processes = [] 

    q = Queue() 

    for i in xrange(nCpu): 
     processes.append(Process(target=f, args=(q,i*nEventsPerCpu,(i+1)*nEventsPerCpu))) 

    for p in processes: 
     p.start() 

    for p in processes: 
     p.join() 

    print q.qsize() 

Answer 1

bunu asla Sen sınırları olmadan kuyruğunu büyümeye çalışıyoruz ve ana süreç bu bir tamamlamak için bekleyen durdu yani senin, kuyrukta odasında bekleyen bir alt sürecine kadar katılıyor ve irade.

Veri birleştirmeden önce sıradan çıkarırsanız, iyi çalışır. , P.is_alive(), bir örtülü katılmak gerçekleştirmelidir belgelere göre

while 1: 
    running = any(p.is_alive() for p in processes) 
    while not queue.empty(): 
     process_queue_data() 
    if not running: 
     break 

ama aynı zamanda en iyi uygulama olabileceğini ima görünmektedir: Eğer kullanabilirsiniz

tekniklerden biri böyle bir şeydir bundan sonra tüm iş parçacıklarında katılımları açıkça gerçekleştirmek için.

Düzenleme: Bu oldukça açık olsa da, tüm bu performans olmayabilir. Bunu nasıl daha iyi performans gösterdiğiniz, yüksek oranda görev ve makineye özgü olacaktır (ve genel olarak, birçoğunda G/Ç'de engellenmeyecekse, bir seferde pek çok işlem oluşturmamalısınız).

CPU sayısı ile işlem sayısını azaltarak yanında bazı kolay düzeltmeler biraz daha hızlı (yine koşullara bağlı olarak) aşağıdaki gibi görünebilir hale getirmek için:

liveprocs = list(processes) 
while liveprocs: 
    try: 
     while 1: 
      process_queue_data(q.get(False)) 
    except Queue.Empty: 
     pass 

    time.sleep(0.5) # Give tasks a chance to put more data in 
    if not q.empty(): 
     continue 
    liveprocs = [p for p in liveprocs if p.is_alive()]