Pool Multiprocessing Python

Question

В основном проблема заключается в следующем: у меня есть группа работников, у которых есть функция, предписанная каждому (функция is worker (alist)), и я пытаюсь обрабатывать 35 рабочих одновременно. Каждый рабочий считывает свою строку из файла (по модулю) и должен обрабатывать строку с помощью функции «работник». Я проверил проверку пера и обнаружил, что сырые манипуляции и удаление бесполезных индексов работают на 100% по назначению.

Элемент args функции pool.apply_async не передает список «raw» в него и запускает процесс. Сырье полностью корректно и нормально функционирует, рабочий сам по себе работает нормально, функция pool.apply_async - единственное место, где, похоже, проблема, и я понятия не имею, как ее исправить. Любая помощь, пожалуйста?

Соответствующий код здесь:

NUM_WORKERS=35 
f=open("test.csv") 
pool=multiprocessing.Pool() 
open("final.csv",'w') 
for workernumber in range(1, NUM_WORKERS): 
    for i,line in enumerate(f): 
     if i==0: 
      print "Skipping first line" #dont do anything 
     elif i%workernumber==0: 
      raw = line.split(',')[0][1:-1].split() 
      uselessindices=[-2,-3,-4,-5,-6] 
      counter=0 
      for ui in uselessindices: 
       del raw[ui+counter] 
       counter+=1 
      print raw 
      pool.apply_async(worker, args=(raw,)) 
pool.close() 
pool.join() 

Answer 1

import multiprocessing 

def worker(arg): 
    print 'doing work "%s"' % arg 
    return 

NUM_WORKERS=35 
with open('test.csv', 'w') as test: 
    for i in xrange(100): 
     if i % 10 == 0: 
      test.write('\n') 
     test.write('"%s 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23",' % i) 

f=open("test.csv") 
pool=multiprocessing.Pool(processes=NUM_WORKERS) 
open("final.csv",'w') 

for i, line in enumerate(f): 
    if i == 0: 
     continue 
    raw = line.split(',')[0][1:-1].split() 
    uselessindices=[-2,-3,-4,-5,-6] 
    counter=0 
    for ui in uselessindices: 
     del raw[ui+counter] 
     counter+=1 
    pool.apply_async(worker, args=(raw,)) 
pool.close() 
pool.join() 
print 'last raw len: %s' % len(raw) 
print 'last raw value: %s' % raw 

Выход:

doing work "['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10', '11', '12', '13', '14', '15', '16', '17', '23']" 
doing work "['10', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10', '11', '12', '13', '14', '15', '16', '17', '23']" 
doing work "['20', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10', '11', '12', '13', '14', '15', '16', '17', '23']" 
doing work "['30', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10', '11', '12', '13', '14', '15', '16', '17', '23']" 
doing work "['40', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10', '11', '12', '13', '14', '15', '16', '17', '23']" 
doing work "['50', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10', '11', '12', '13', '14', '15', '16', '17', '23']" 
doing work "['60', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10', '11', '12', '13', '14', '15', '16', '17', '23']" 
doing work "['70', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10', '11', '12', '13', '14', '15', '16', '17', '23']" 
doing work "['80', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10', '11', '12', '13', '14', '15', '16', '17', '23']" 
doing work "['90', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10', '11', '12', '13', '14', '15', '16', '17', '23']" 
last raw len: 19 
last raw value: ['90', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10', '11', '12', '13', '14', '15', '16', '17', '23'] 

Answer 2

Я предлагаю вам поставить расчет raw в функцию генератора, а затем использовать Pool.imap_unordered() или Pool.map() для запуска worker() по всем элементам в генераторе.

Что-то вроде этого непроверенного кода:

def get_raw(): 
    with open("test.csv", 'rU') as f: 
     for i, line in enumerate(f): 
      if i == 0: 
       # skip header 
       continue 
      raw = line.split(',')[0][1:-1].split() 
      uselessindices=[-2,-3,-4,-5,-6] 
      counter=0 
      for ui in uselessindices: 
       del raw[ui+counter] 
       counter+=1 
      yield raw 

pool=multiprocessing.Pool(processes=NUM_WORKERS) 

pool.map(worker, get_raw()) 
pool.close() 
pool.join() 

Answer 3

Так я узнал, что это не бросает ошибку, что происходило внутри работника в результате несогласованного числа входов в функцию ребенка (AKA работник вызывал другую функцию dosomething (a1, a2, ... a20) и предоставлял только 19 входов). Кажется, async не будет вызывать ошибки в проблемах, происходящих внутри рабочего, что очень раздражает, но теперь я понимаю. Спасибо за помощь!