Python多进程/多线程加速文件复制

Question

我有一个程序可以将大量文件从一个位置复制到另一个位置 - 我说的是 100,000 多个文件（我正在复制图像序列中的 314g）。他们都在极端的巨大，非常快速的网络存储上。我正在使用shutil按顺序复制文件，这需要一些时间，所以我试图找到优化它的最佳方法。我注意到一些我有效地使用多线程从网络读取文件的软件在加载时间上有很大的提高，所以我想尝试在 python 中执行此操作。

我没有编程多线程/多进程的经验 - 这似乎是正确的领域吗？如果是这样，最好的方法是什么？我查看了其他一些关于在 python 中进行线程文件复制的 SO 帖子，他们似乎都说你没有速度提升，但考虑到我的硬件，我认为情况并非如此。目前，我的 IO 上限还差得很远，资源仅占 1% 左右（我在本地有 40 个内核和 64g 的 RAM）。

Answer 1

这可以通过在 Python 中使用 gevent 来并行化。

我会推荐以下逻辑来实现加速 100k+ 文件复制：

1. 将需要复制的所有 100K+ 文件的名称放入 csv 文件中，例如：'input.csv'。

2. 然后从该 csv 文件创建块。应根据机器中的处理器/内核数来确定块的数量。

3. 将每个块传递给单独的线程。

4. 每个线程按顺序读取该块中的文件名并将其从一个位置复制到另一个位置。

这里是python代码sn-p：

import sys
import os
import multiprocessing

from gevent import monkey
monkey.patch_all()

from gevent.pool import Pool

def _copyFile(file):
    # over here, you can put your own logic of copying a file from source to destination

def _worker(csv_file, chunk):
    f = open(csv_file)
    f.seek(chunk[0])
    for file in f.read(chunk[1]).splitlines():
        _copyFile(file)


def _getChunks(file, size):
    f = open(file)
    while 1:
        start = f.tell()
        f.seek(size, 1)
        s = f.readline()
        yield start, f.tell() - start
        if not s:
            f.close()
            break

if __name__ == "__main__":
    if(len(sys.argv) > 1):
        csv_file_name = sys.argv[1]
    else:
        print "Please provide a csv file as an argument."
        sys.exit()

    no_of_procs = multiprocessing.cpu_count() * 4

    file_size = os.stat(csv_file_name).st_size

    file_size_per_chunk = file_size/no_of_procs

    pool = Pool(no_of_procs)

    for chunk in _getChunks(csv_file_name, file_size_per_chunk):
        pool.apply_async(_worker, (csv_file_name, chunk))

    pool.join()

将文件另存为 file_copier.py。 打开终端并运行：

$ ./file_copier.py input.csv

Answer 2

更新：

我从来没有让 Gevent 工作（第一个答案），因为我无法在没有互联网连接的情况下安装模块，而我的工作站上没有互联网连接。但是，仅使用 python 的内置线程（我已经学会了如何使用它），我就能够将文件复制时间减少 8 倍，我想将其发布为任何有兴趣的人的附加答案！下面是我的代码，可能需要注意的是，由于您的硬件/网络设置，我的 8 倍复制时间很可能因环境而异。

import Queue, threading, os, time
import shutil

fileQueue = Queue.Queue()
destPath = 'path/to/cop'

class ThreadedCopy:
    totalFiles = 0
    copyCount = 0
    lock = threading.Lock()

    def __init__(self):
        with open("filelist.txt", "r") as txt: #txt with a file per line
            fileList = txt.read().splitlines()

        if not os.path.exists(destPath):
            os.mkdir(destPath)

        self.totalFiles = len(fileList)

        print str(self.totalFiles) + " files to copy."
        self.threadWorkerCopy(fileList)


    def CopyWorker(self):
        while True:
            fileName = fileQueue.get()
            shutil.copy(fileName, destPath)
            fileQueue.task_done()
            with self.lock:
                self.copyCount += 1
                percent = (self.copyCount * 100) / self.totalFiles
                print str(percent) + " percent copied."

    def threadWorkerCopy(self, fileNameList):
        for i in range(16):
            t = threading.Thread(target=self.CopyWorker)
            t.daemon = True
            t.start()
        for fileName in fileNameList:
            fileQueue.put(fileName)
        fileQueue.join()

ThreadedCopy()

Answer 3

在重新实现@Spencer 发布的代码时，我遇到了与帖子下方的 cmets 中提到的相同的错误（更具体地说：OSError: [Errno 24] Too many open files）。 我通过远离守护线程并改用concurrent.futures.ThreadPoolExecutor 解决了这个问题。这似乎以更好的方式处理要复制的文件的打开和关闭。通过这样做，除了现在看起来像这样的threadWorkerCopy(self, filename_list: List[str]) 方法之外，所有代码都保持不变：

    def threadWorkerCopy(self, filename_list: List[str]):
    """
    This function initializes the workers to enable the multi-threaded process. The workers are handles automatically with
    ThreadPoolExecutor. More infos about multi-threading can be found here: https://realpython.com/intro-to-python-threading/.
    A recurrent problem with the threading here was "OSError: [Errno 24] Too many open files". This was coming from the fact
    that deamon threads were not killed before the end of the script. Therefore, everything opened by them was never closed.

    Args:
        filename_list (List[str]): List containing the name of the files to copy.
    """
    with concurrent.futures.ThreadPoolExecutor(max_workers=cores) as executor:
        executor.submit(self.CopyWorker)

        for filename in filename_list:
            self.file_queue.put(filename)
        self.file_queue.join()  # program waits for this process to be done.

Answer 4

使用ThreadPool 怎么样？

import os
import glob
import shutil
from functools import partial
from multiprocessing.pool import ThreadPool

DST_DIR = '../path/to/new/dir'
SRC_DIR = '../path/to/files/to/copy'

# copy_to_mydir will copy any file you give it to DST_DIR
copy_to_mydir = partial(shutil.copy, dst=DST_DIR))

# list of files we want to copy
to_copy = glob.glob(os.path.join(SRC_DIR, '*'))

with ThreadPool(4) as p:
  p.map(copy_to_mydir, to_copy)

Answer 5

如果您只想将目录树从一个路径复制到另一个路径，这是我的解决方案，它比以前的解决方案更简单。它利用multiprocessing.pool.ThreadPool 并为shutil.copytree 使用自定义复制功能：

import shutil
from multiprocessing.pool import ThreadPool


class MultithreadedCopier:
    def __init__(self, max_threads):
        self.pool = ThreadPool(max_threads)

    def copy(self, source, dest):
        self.pool.apply_async(shutil.copy2, args=(source, dest))

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.pool.close()
        self.pool.join()


src_dir = "/path/to/src/dir"
dest_dir = "/path/to/dest/dir"


with MultithreadedCopier(max_threads=16) as copier:
    shutil.copytree(src_dir, dest_dir, copy_function=copier.copy)