为什么同时读取多个文件比按顺序读取要慢？

Question

我试图解析目录中找到的许多文件,但是使用多处理会减慢我的程序.

# 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)

我按照示例,将以a结尾的文件的所有名称放在.txt列表中,然后创建Pools,并将它们映射到我的函数.然后我想返回一个对象列表.每个对象保存文件的已解析数据.然而,我得到以下结果令我感到惊讶:

#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   

我在这里错过了什么？
提前致谢!

Answer 1

看起来你是I/O绑定:

在计算机科学中,I/O bound指的是一种条件,其中完成计算所花费的时间主要由等待输入/输出操作完成所花费的时间来确定.这与CPU绑定的任务相反.当请求数据的速率低于其消耗速率时,或者换句话说,花费更多时间来请求数据而不是处理数据时,就会出现这种情况.

您可能需要让主线程执行读取,并在子流程可用时将数据添加到池中.这与使用不同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)

Answer 2

通常,同时从不同线程读取相同的物理(旋转)硬盘永远不是一个好主意,因为每个开关都会导致大约10ms的额外延迟以定位硬盘的读取头(在SSD上会有所不同) .

正如@ peter-wood已经说过的那样,最好让一个线程读入数据,并让其他线程处理这些数据.

另外,为了真正测试差异,我认为你应该用一些更大的文件进行测试.例如:当前的硬盘应该能够读取大约100MB /秒.因此,一次读取100kB文件的数据需要1ms,而将读取头定位到该文件的开头则需要10ms.

另一方面,查看您的数字(假设这些是针对单个循环),很难相信I/O绑定是这里唯一的问题.总数据为100MB,从磁盘读取需要1秒钟加上一些开销,但您的程序需要130秒.我不知道这个数字是否与磁盘上的文件是冷的,或者是数据已经被OS缓存的多个测试的平均值(62 GB或RAM,所有这些数据应该第二次被缓存) - 它会看到这两个数字很有意思.

所以必须有别的东西.让我们仔细看看你的循环:

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)).

因此,在一个100MB文件上使用一个线程并且没有gc调用来测试相同内容会很有趣.

如果处理时间超过读数(即不是I/O限制),则使用一个线程读取缓冲区中的数据(如果尚未缓存,则应在一个100MB文件上花费1或2秒),并使用多个线程来处理数据(但gc在那个紧密循环中仍然没有那些调用).

您不必将数据拆分为多个文件,以便能够使用线程.只需让它们处理同一文件的不同部分(即使使用14GB文件).