Python Multiprocessing imap chunksize

Question

Я хотел бы сравнить две папки с одинаковой структурой пути и одинаковыми файлами во всех подпапках. Папка довольно большая с размером около 80 ГБ и номерами файлов 8000.

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

Когда я перебирал списки, мне было очень много времени, чтобы выполнить все сравнение, а также скорость использования ЦП была низкой.

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

from multiprocessing import Pool, cpu_count 
import hashlib 

def calc_md5(item): 
    m = hashlib.md5() 
    with open(item, 'rb') as f: 
     for chunk in iter(lambda: f.read(4096), b""): 
      m.update(chunk) 
    return m.hexdigest() 

def worker(args): 
    a, b = args 
    return calc_md5(a) == calc_md5(b) 

def multi_compare(queue_a, queue_b, thread): 
    pool = Pool(processes = cpu_count() - 1) 
    # Task iterable 
    task = zip(queue_a, queue_b) 
    # Multiprocessing 
    for retval in pool.imap_unordered(worker, task, chunksize = 5): 
     if not retval: 
      print "Bad Detected" 

Здесь queue_a и queue_b являются путями для Меряться файла, отсортированных по размеру файла. Я ожидаю более высокую загрузку процессора и лучшую производительность из этих многопроцессорных методов, но, похоже, это не так. Хотя простая последовательная итерация занимает около 3200 секунд, метод многопроцессорности занимает около 4600 секунд.

Мне интересно, почему это так? Это хороший момент для многопроцессорности? Что является узким местом этой плохой производительности в моем коде? Есть ли способ его улучшить?

Редактировать: Я установил chunksize в соответствии с моим чувством кишки. Думаю, я могу изменить его на длину queue_a или queue_b, разделенную номером потока, и отсортировать очередь задач как первую 1/4 из нее, содержащую элементы queue_a [0 :: thread] или queue_b [0 :: thread] и наоборот , Это будет поддерживать одинаковый размер задач для всего потока и постоянно поддерживать весь поток. Я не знаю, является ли это хорошим способом получить дополнительную производительность, и я все еще тестирую это.

Редактировать: Тест в приведенном выше редактировании занимает 4000 секунд. Чуть лучше, чем chunksize = 5. Еще хуже, чем серийный метод. Итак, я хотел бы спросить, как определить узкое место этой многопроцессорной программы.

Спасибо!

Answer 1

Это IO, ограничивающее производительность. Алгоритм MD5 теперь является слишком простой задачей для процессоров. Следующий код вычисляет производительность MD5 в GB/s.

import time 
import hashlib 
from multiprocessing import Pool 

def worker(x): 
    data = bytearray(xrange(256)) * 4 * 1024 
    md5 = hashlib.md5() 
    for x in xrange(1024): 
     md5.update(data) 

if __name__ == '__main__': 
    num_workers = 4 
    pool = Pool(num_workers) 
    start = time.time() 
    pool.map(worker, xrange(num_workers)) 
    print num_workers/(time.time() - start), 'Gb/s' 

современный мобильный i3 процессора относительно слабой компании Intel (2 ядра, 4 нити) имеет возможность хэш со скоростью 1 Гбит в секунду. Сравните это с SATA3 bandwidth, который составляет 600 Мбит/с. Таким образом, даже при использовании SSD интерфейс диска ограничивает скорость хеширования.
На жестких дисках ситуация еще хуже. Несколько считывающих устройств вынуждают диск перемещать свои считывающие головки, что вызывает больше задержек, чем если бы использовался только один поток считывателей. Это как чтение сильно фрагментированного файла.
Когда набор данных не настолько большой кэш файлов ОС, может очень помочь. Это не ваше дело.