Параллельная организация очередей - Многопроцессорный пул, Python - PullRequest
Новая
       7

Параллельная организация очередей - Многопроцессорный пул, Python

0 голосов
/ 19 декабря 2011

Моя цель - пройти по каталогу и вычислить MD5 всех файлов в нем.Я использовал код для решения аналогичной проблемы

Параллельное сопоставление файлов, Python 

import os
import re
import sys
import time
import md5

from stat import S_ISREG

import multiprocessing

global queue
size_limit = 500000

target = sys.argv[1]



############Analysis and Multiprocessing####################
############Analysis and Multiprocessing####################
############Analysis and Multiprocessing####################
############Analysis and Multiprocessing####################
############Analysis and Multiprocessing####################
############Analysis and Multiprocessing####################
############Analysis and Multiprocessing####################
############Analysis and Multiprocessing####################
############Analysis and Multiprocessing####################
############Analysis and Multiprocessing####################

def walk_files(topdir):
     """yield up full pathname for each file in tree under topdir"""
     for dirpath, dirnames, filenames in os.walk(topdir):
         for fname in filenames:
             pathname = os.path.join(dirpath, fname)
             yield pathname

 def files_to_search(topdir):
     """yield up full pathname for only files we want to search"""
     for fname in walk_files(topdir):
         try:
             # if it is a regular file and big enough, we want to search it
             sr = os.stat(fname)
             if S_ISREG(sr.st_mode) and sr.st_size <= size_limit:
                 yield fname
         except OSError:
             pass

def worker_search_fn(fname):
     fp = open(fname, 'rt')
     # read one line at a time from file
     contents = fp.read()
     hash = md5.md5(contents)
     global queue
     print "enqueue"
     queue.put(fname+'-'+hash.hexdigest())

################MAIN MAIN MAIN#######################
################MAIN MAIN MAIN#######################
################MAIN MAIN MAIN#######################
################MAIN MAIN MAIN#######################
################MAIN MAIN MAIN#######################

#kick of processes to md5 the files and wait till completeion

queue = multiprocessing.Queue()
pool = multiprocessing.Pool()
pool.map(worker_search_fn, files_to_search(target))
pool.close()
pool.join()

#Should be done, now lets send do our analysis
while not queue.empty():
    print queue.get()

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

Возможно, есть лучший способ сделать это?Структура не должна быть очередью, но должна быть свободна от блокировки, чтобы в полной мере использовать многопроцессорность.Я хочу, чтобы recurs и md5 каталог параллельно, и когда это будет сделано, сделайте что-нибудь со списком в целом.Для отладки я просто печатаю завершенную очередь.Есть предложения?

Ответы [ 2 ]

3 голосов
/ 19 декабря 2011

Неясно, связана ли ваша программа с вводом-выводом или с ЦП, т. Е. Один процесс может работать лучше, чем несколько процессов, если задача связана с вводом-выводом, например, путем минимизации числа обращений к диску.Вы можете проверить это, указав различные значения nprocesses (ниже) и посмотреть, что дает лучшие результаты в вашем случае.

Вам не нужна очередь в этом случае: 

#!/usr/bin/env python
import os
import sys

from hashlib         import md5
from multiprocessing import Pool, freeze_support
from stat            import S_ISREG

def walk_files(topdir):
     """yield up full pathname for each file in tree under topdir"""
     for dirpath, dirnames, filenames in os.walk(topdir):
         for fname in filenames:
             pathname = os.path.join(dirpath, fname)
             yield pathname

def files_to_process(topdir, size_limit):
    """yield up full pathname for only files we want to process"""
    for fname in walk_files(topdir):
        try: sr = os.stat(fname)
        except OSError: pass
        else:
            # if it is a regular file and small enough, we want to process it
            if S_ISREG(sr.st_mode) and sr.st_size <= size_limit:
                yield fname

def md5sum(fname):
    with open(fname, 'rb') as fp:
        # read all file at once
        contents = fp.read()
        hash = md5(contents)
        return fname, hash.hexdigest()

def main(argv=None):
    if argv is None:
        argv = sys.argv
    topdir = argv[1]
    size_limit = 500000
    nprocesses = 1

    pool = Pool(processes=nprocesses)
    files = files_to_process(topdir, size_limit)
    for fname, hexdigest in pool.imap_unordered(md5sum, files):
        print("%s\t%s" % (fname, hexdigest))

if __name__=="__main__":
    freeze_support()
    main()

Пример

$ python md5sum.py .
./md5sum.py 9db44d3117673790f1061d4b8f00e8ce
1 голос
/ 19 декабря 2011

Поскольку большой каталог требует большого времени для выполнения walk_files() Это не тупик

И ...

удалить pool.join()

multiprocessing.Pool().map() блоков дорезультат готов, поэтому вам не нужно pool.join()

...