Почему чтение нескольких файлов происходит в то же время медленнее, чем чтение последовательно?

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

# Calling my parsing function from Client.
L = getParsedFiles('/home/tony/Lab/slicedFiles') <--- 1000 .txt files found here.
                                                       combined ~100MB

Следуя этому примеру из документации python:

from multiprocessing import Pool

def f(x):
    return x*x

if __name__ == '__main__':
    p = Pool(5)
    print(p.map(f, [1, 2, 3]))

Я написал этот фрагмент кода:

from multiprocessing import Pool
from api.ttypes import *

import gc
import os

def _parse(pathToFile):
    myList = []
    with open(pathToFile) as f:
        for line in f:
            s = line.split()
            x, y = [int(v) for v in s]
            obj = CoresetPoint(x, y)
            gc.disable()
            myList.append(obj)
            gc.enable()
    return Points(myList)

def getParsedFiles(pathToFile):
    myList = []
    p = Pool(2)
    for filename in os.listdir(pathToFile):
        if filename.endswith(".txt"):
            myList.append(filename)
    return p.map(_pars, , myList)

Я последовал примеру, поместил все имена файлов, заканчивающиеся на .txt в списке, затем создал пулы и сопоставил их с моей функцией. Затем я хочу вернуть список объектов. Каждый объект содержит анализируемые данные файла. Однако меня поражает, что я получил следующие результаты:

#Pool 32  ---> ~162(s)
#Pool 16 ---> ~150(s)
#Pool 12 ---> ~142(s)
#Pool 2 ---> ~130(s)

График:

Технические характеристики машины:

62.8 GiB RAM
Intel® Core™ i7-6850K CPU @ 3.60GHz × 12

Что мне здесь не хватает?
Спасибо заранее!

Ответ 1

Похоже, вы привязка ввода/вывода:

В информатике привязка ввода-вывода относится к условию, в котором время, необходимое для завершения вычисления, определяется главным образом периодом, ожидающим завершения операций ввода-вывода. Это противоположность задачи, связанной с ЦП. Это обстоятельство возникает, когда скорость, с которой запрашиваются данные, медленнее, чем скорость, которую она потребляет, или, другими словами, больше времени тратится на обработку данных, чем на обработку.

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

Когда вы обрабатываете строку за раз, а входы разделяются, вы можете использовать fileinput перебирать строки из нескольких файлов и сопоставлять строки обработки функций вместо файлов:

Передача одной строки за раз может быть слишком медленной, поэтому мы можем попросить карту передать куски и настроить, пока не найдем сладость. Наша функция анализирует фрагменты строк:

def _parse_coreset_points(lines):
    return Points([_parse_coreset_point(line) for line in lines])

def _parse_coreset_point(line):
    s = line.split()
    x, y = [int(v) for v in s]
    return CoresetPoint(x, y)

И наша основная функция:

import fileinput

def getParsedFiles(directory):
    pool = Pool(2)

    txts = [filename for filename in os.listdir(directory):
            if filename.endswith(".txt")]

    return pool.imap(_parse_coreset_points, fileinput.input(txts), chunksize=100)

Ответ 2

В общем, никогда не рекомендуется читать с одного и того же физического (вращающегося) жесткого диска из разных потоков одновременно, потому что каждый коммутатор вызывает дополнительную задержку около 10 мс для позиционирования считывающей головки жесткого диска (будет отличаться на SSD).

Как уже говорилось в @peter-wood, лучше иметь один поток, считывающий данные, и обрабатывать эти данные другими потоками.

Кроме того, чтобы действительно проверить разницу, я думаю, вы должны сделать тест с некоторыми более крупными файлами. Например: текущие жесткие диски должны иметь возможность читать около 100 МБ/сек. Таким образом, чтение данных в 100 КБ файле за один раз займет 1 мс, а позиционирование головы чтения в начале этого файла займет 10 мс.

С другой стороны, глядя на ваши цифры (если они предназначены для одного цикла), трудно поверить, что единственная проблема здесь связана с ограничением ввода-вывода. Общее количество данных - 100 Мбайт, что занимает 1 секунду для чтения с диска плюс некоторые накладные расходы, но ваша программа занимает 130 секунд. Я не знаю, является ли это число с файлами, холодными на диске, или в среднем несколькими тестами, в которых данные уже кэшируются ОС (с 62 ГБ или ОЗУ все эти данные должны быть кэшированы во второй раз) - это было бы Интересно видеть оба числа.

Итак, должно быть что-то еще. Давайте более подробно рассмотрим ваш цикл:

for line in f:
    s = line.split()
    x, y = [int(v) for v in s]
    obj = CoresetPoint(x, y)
    gc.disable()
    myList.append(obj)
    gc.enable()

Пока я не знаю Python, я предполагаю, что вызов gc является проблемой здесь. Они вызываются для каждой строки, считанной с диска. Я не знаю, насколько дороги эти вызовы (или что, если gc.enable() запускает сборку мусора, например) и почему они нужны только для append(obj), но могут быть и другие проблемы, потому что это многопоточность:

Предполагая, что объект gc является глобальным (т.е. не является локальным потоком), вы можете иметь что-то вроде этого:

thread 1 : gc.disable()
# switch to thread 2
thread 2 : gc.disable()
thread 2 : myList.append(obj)
thread 2 : gc.enable()
# gc now enabled!
# switch back to thread 1 (or one of the other threads)
thread 1 : myList.append(obj)
thread 1 : gc.enable()

И если число потоков <= число ядер, то даже не было бы переключения, все они будут одновременно называть это.

Кроме того, если объект gc является потокобезопасным (это было бы хуже, если бы оно не было), ему пришлось бы сделать некоторую блокировку, чтобы безопасно изменить его внутреннее состояние, которое заставит все остальные потоки ждать.

Например, gc.disable() будет выглядеть примерно так:

def disable()
    lock()  # all other threads are blocked for gc calls now
    alter internal data
    unlock()

И поскольку gc.disable() и gc.enable() вызываются в замкнутом цикле, это может повредить производительность при использовании нескольких потоков.

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

В зависимости от того, как Python копирует или перемещает объекты, также может быть немного лучше использовать myList.append(CoresetPoint(x, y)).

Итак, было бы интересно протестировать одно и то же на одном 100-мегабайтном файле с одним потоком и без вызовов gc.

Если обработка занимает больше времени, чем чтение (т.е. не привязана к вводу/выводу), используйте один поток для чтения данных в буфере (требуется 1 или 2 секунды на один 100 МБ файл, если он еще не кэширован) и несколько потоков для обработки данных (но все же без этих вызовов gc в этом узком цикле).

Вам не нужно разделить данные на несколько файлов, чтобы иметь возможность использовать потоки. Просто позвольте им обрабатывать разные части одного и того же файла (даже с 14-гигабайтным файлом).