Python - Многопроцессорность - больше заданий, которые выполняются cpus.

Question

Я понимаю, что могу использовать класс пула и, вероятно, получить то, что мне нужно, но я хочу немного более тонкого контроля над моей проблемой. У меня больше рабочих мест, чем у процессоров, поэтому я не хочу, чтобы они запускали все за один раз.

Например:

from multiprocessing import Process,cpu_count 
for dir_name in directories: 
    src_dir = os.path.join(top_level,dir_name) 
    dst_dir = src_dir.replace(args.src_dir,args.target_dir) 
    p = Process(target=transfer_directory, args=(src_dir, dst_dir,)) 
    p.start() 

Однако, если у меня есть более чем 16 каталогов, я тогда начнет больше рабочих мест, чем у меня есть процессоры. Вот мое решение, которое действительно взломало.

from multiprocessing import Process,cpu_count 
jobs = [] 
for dir_name in directories: 
    src_dir = os.path.join(top_level,dir_name) 
    dst_dir = src_dir.replace(args.src_dir,args.target_dir) 
    p = Process(target=transfer_directory, args=(src_dir, dst_dir,)) 
    jobs.append(p) 

alive_jobs = [] 
while jobs: 
    if len(alive_jobs) >= cpu_count(): 
     time.sleep(5) 
     print alive_jobs 
     for aj in alive_jobs: 
      if aj.is_alive(): 
       continue 
      else: 
       print "job {} removed".format(aj) 
       alive_jobs.remove(aj) 

     continue 

    for job in jobs: 
     if job.is_alive(): 
      continue 
     job.start() 
     alive_jobs.append(job) 
     print alive_jobs 
     jobs.remove(job) 
     if len(alive_jobs) >= cpu_count(): 
      break 

Есть ли лучшее решение с использованием встроенных инструментов?

Answer 1

Я хочу поделиться своей идеей здесь: создать число процессов равно cpu_count(), используйте очередь сохраняет все свои каталоги и передать очереди в ваш метод transfer_directory, занимает dir_name из Очереди когда процесс завершает свой Работа. Проект выглядит следующим образом:

NUM_OF_PROCESSES = multiprocessing.cpu_count() 
TIME_OUT_IN_SECONDS = 60 

for dir_name in directories: 
    my_queue.put(dir_name) 

# creates processes that equals to number of CPU 
processes = [multiprocessing.Process(target=transfer_directory, args=(my_queue,)) for x in range(NUM_OF_PROCESSES)] 

# starts processes 
for p in processes: 
    p.start() 

# blocks the calling thread 
for p in processes: 
    p.join() 



def transfer_directory(my_queue): 
    """processes element of directory queue if queue is not empty""" 
    while my_queue is not empty: 
     dir_name = my_queue.get(timeout=TIME_OUT_IN_SECONDS) 
     src_dir = os.path.join(top_level,dir_name) 
     dst_dir = src_dir.replace(args.src_dir,args.target_dir) 

Edit: Он также работает эффективно для чтения большого файла. Я изо всех сил пытался читать огромный файл (это было более 10 миллионов строк), используя multiprocessing, и, наконец, я использую один процесс, начинающийся producer(a_queue), который просто читает и помещает строки в очередь, а затем запускает несколько consumers(a_queue) брать строки от a_queue и выполнять трудоемкую работу, и она работает правильно для меня.