为什么 readlines() 比 sizehint 读得更多？

Question

背景

我在 Python 2.7.6 中解析非常大的文本文件 (30GB+)。为了加快这个过程，我将文件分成块并使用多处理库将它们分流到子进程。为此，我在我的主进程中迭代文件，记录我想要拆分输入文件的字节位置并将这些字节位置传递给子进程，然后打开输入文件并使用file.readlines(chunk_size)读取它们的块.但是，我发现读入的块似乎比sizehint 参数大得多（4x）。

问题

为什么不注意 sizehint？

示例代码

以下代码演示了我的问题：

import sys

# set test chunk size to 2KB
chunk_size = 1024 * 2

count = 0
chunk_start = 0
chunk_list = []

fi = open('test.txt', 'r')
while True:
    # increment chunk counter
    count += 1

    # calculate new chunk end, advance file pointer
    chunk_end = chunk_start + chunk_size
    fi.seek(chunk_end)

    # advance file pointer to end of current line so chunks don't have broken 
    # lines
    fi.readline() 
    chunk_end = fi.tell()

    # record chunk start and stop positions, chunk number
    chunk_list.append((chunk_start, chunk_end, count))

    # advance start to current end
    chunk_start = chunk_end

    # read a line to confirm we're not past the end of the file
    line = fi.readline()
    if not line:
        break

    # reset file pointer from last line read
    fi.seek(chunk_end, 0)

fi.close()

# This code represents the action taken by subprocesses, but each subprocess
# receives one chunk instead of iterating the list of chunks itself.
with open('test.txt', 'r', 0) as fi:
    # iterate over chunks
    for chunk in chunk_list:
        chunk_start, chunk_end, chunk_num = chunk

        # advance file pointer to chunk start
        fi.seek(chunk_start, 0)

        # print some notes and read in the chunk
        sys.stdout.write("Chunk #{0}: Size: {1} Start {2} Real Start: {3} Stop {4} "
              .format(chunk_num, chunk_end-chunk_start, chunk_start, fi.tell(), chunk_end))
        chunk = fi.readlines(chunk_end - chunk_start)
        print("Real Stop: {0}".format(fi.tell()))

        # write the chunk out to a file for examination
        with open('test_chunk{0}'.format(chunk_num), 'w') as fo:
            fo.writelines(chunk)

结果

我使用大约 23.3KB 的输入文件 (test.txt) 运行此代码，它产生了以下输出：

块 #1：大小：2052 开始 0 实际开始：0 停止 2052 实际停止：8193
块 #2：大小：2051 开始 2052 实际开始：2052 停止 4103 实际停止：10248
块 #3：大小：2050 开始 4103 实际开始：4103 停止 6153 实际停止：12298
块 #4：大小：2050 开始 6153 实际开始：6153 停止 8203 实际停止：14348
块 #5：大小：2050 开始 8203 实际开始：8203 停止 10253 实际停止：16398
块 #6：大小：2050 开始 10253 实际开始：10253 停止 12303 实际停止：18448
块 #7：大小：2050 开始 12303 实际开始：12303 停止 14353 实际停止：20498
块 #8：大小：2050 开始 14353 实际开始：14353 停止 16403 实际停止：22548
块 #9：大小：2050 开始 16403 实际开始：16403 停止 18453 实际停止：23893
块 #10：大小：2050 开始 18453 实际开始：18453 停止 20503 实际停止：23893
块 #11：大小：2050 开始 20503 实际开始：20503 停止 22553 实际停止：23893
块 #12：大小：2048 开始 22553 实际开始：22553 停止 24601 实际停止：23893

报告的每个块大小约为 2KB，所有开始/停止位置都按应有的方式排列，fi.tell() 报告的真实文件位置似乎是正确的，所以我相当确定我的分块算法不错。然而，真正的停止位置显示readlines() 的读数远远超过尺寸提示。此外，输出文件 #1 - #8 为 8.0KB，远大于大小提示。

即使我尝试只在行尾拆分块是错误的，readlines() 仍然不应该读取超过 2KB + 一行的内容。文件 #9 - #12 变得越来越小，这是有道理的，因为块起始点越来越接近文件的末尾，而 readlines() 不会读到文件末尾。

注意事项

1. 我的测试输入文件只是在每一行打印了“\n”，1-5000。
2. 我再次尝试了不同的块和输入文件大小，结果相似。
3. readlines documentation 表示读取大小可能会四舍五入到内部缓冲区的大小，因此我尝试在不使用缓冲的情况下打开文件（如图所示），但没有任何区别。
4. 我使用这个算法来分割文件，因为我需要能够支持*.bz2和*.gz压缩文件，而*.gz文件没有办法让我在不解压文件的情况下识别未压缩文件的大小. *.bz2 文件也没有，但我可以从这些文件末尾寻找 0 个字节并使用 fi.tell() 来获取文件大小。见my related question。
5. 在添加支持压缩文件的要求之前，以前版本的脚本使用 os.path.getsize() 作为分块循环的停止条件，并且 readlines 似乎可以很好地使用该方法。

Answer 1

readlines 文档中提到的缓冲区与 open 调用的第三个参数控制的缓冲区无关。缓冲区是this buffer in file_readlines:

static PyObject *
file_readlines(PyFileObject *f, PyObject *args)
{
    long sizehint = 0;
    PyObject *list = NULL;
    PyObject *line;
    char small_buffer[SMALLCHUNK];

SMALLCHUNK 在前面定义：

#if BUFSIZ < 8192
#define SMALLCHUNK 8192
#else
#define SMALLCHUNK BUFSIZ
#endif

我不知道BUFSIZ 来自哪里，但看起来你得到了#define SMALLCHUNK 8192 案例。在任何情况下，readlines 永远不会使用小于 8 KiB 的缓冲区，所以你应该让你的块更大。

Answer 2

这并不能回答您的问题，但也许会有所帮助...

我感觉可能有更好的方法来分块您的文件，这样可以绕过您当前的问题。只是一个想法，但既然文件可以被迭代，这样的事情会起作用吗？

import bzip2
import gzip
from multiprocessing import Pool, cpu_count


def chunker(filepath):
    """define and yield chunks"""
    if filepath.endswith(".bz"):
        read_open = bzip2.open
    elif filepath.endswith(".gz"):
        read_open = gzip.open

    with read_open(filepath) as in_f:        
        delim = "something"
        chunk = []
        for line in in_f:
            if delim not in line:
                chunk.append(line)
            else:
                current, next_ = line.split(delim)
                chunk.append(current)
                yield chunk
                chunk = [next_]
        if chunk:
            yield chunk

def process_chunk(chunk):
    # do magic
    return 

if __name__ == '__main__':
    filepath = ""
    chunk_iter = chunker(filepath)

    pool = Pool(processes=cpu_count() - 1)
    for result in pool.imap(process_chunk, chunk_iter , chunksize=1)
        print result

或者如果您已经使用 1-pass 来读取和生成块列表，为什么不在读取时将单独的块写为单独的文件（如果您有磁盘空间）。然后你可以给一个工作池一个要处理的文件路径列表。

或者，如果您的工作人员足够快地处理块并且您有内存，您可以在阅读时将整个块传递给Queue。并且工人可以从队列中拉出块。