Python中的多线程文件传输？

Question

我手头有一项特殊的小任务，但我不知道如何最好地实施解决方案。

我有三个工作站通过带宽为 40gbps 的 InfiniBand 连接到运行 Ubuntu 20.04 LTS 的 NAS。此 NAS 配备 2TB NVMe SSD 作为写入缓存，以及 7 个 RAID0 单元作为主存储。

这些工作站会将原始数据输出到此 NAS 以供以后使用，每台机器每天都会输出大约 6TB 的数据文件，每个文件的大小在 100 到 300 GB 之间。为了防止网络过于拥挤，我让他们先将数据输出到 NVMe 缓存，然后我计划从那里分发数据文件 - 每个 RAID0 单元同时分配一个文件以最大化磁盘 IO。例如，file1 到 array0，file2 到 array1，file3 到 array2，等等。

现在我正在 NAS 端编写一个脚本（最好是 systemd 服务，但我可以使用 nohup）来监控缓存，并将文件发送到这些 RAID 阵列。

这是我想出的，感谢this post，它非常接近我的目标。

import queue, threading, os, time
import shutil

bfr_drive = '/home/test_folder' # cache
ext = ".dat" # data file extension
array = 0 # simluated array as t0-t6
fileList = [] # list of files to be moved from cache to storage
destPath = '/home/test_folder/t'
fileQueue = queue.Queue()


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

    def __init__(self):
        for file_name in os.listdir(bfr_drive):
            if file_name.endswith(ext):
                fileList.append(os.path.join(bfr_drive, file_name))
                fileList.sort()

        self.totalFiles = len(fileList)

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


    def CopyWorker(self):
        global array
        while True:
            fileName = fileQueue.get()
            shutil.copy(fileName, destPath+str(array))
            array += 1
            if array > 6:
                array = 0
            fileQueue.task_done()

            with self.lock:
                self.copyCount += 1
                
                percent = (self.copyCount * 100) / self.totalFiles
                
                print (str(percent) + " percent copied.")

    def threadWorkerCopy(self, fileNameList):
        # global array
        for i in range(4):
            t = threading.Thread(target=self.CopyWorker)
            t.daemon = True
            t.start()
            # array += 1
            # if array > 6:
                # array = 0
            print ("current array is:" + str(array)) # output prints array0 for 4 times, did not iterate
            
          
        for fileName in fileNameList:
            fileQueue.put(fileName)
        fileQueue.join()

ThreadedCopy()

现在，Python 脚本可以成功分发文件，但只能在 for i in range(4) 中的数字之后。例如，如果我将它设置为 4，那么工作人员将对前 4 个文件使用相同的路径（array0），然后他们才会开始遍历数组到 1、2、3 等。

有人能指出我如何分发这些文件吗？我认为我正朝着正确的方向前进，但是，我无法理解为什么这些工人一开始就被困在同一个目录中。

编辑：我有早期版本的代码，其中路径迭代处于生成过程threadWorkerCopy。我现在把它移到了实际的工作函数CopyWorker。问题仍然存在。

Answer 1

问题是您不会在工作线程中生成 array 的新值，而只会在 threadWorkerCopy 中创建线程时生成。
结果将取决于系统上的实际时间。每个工作线程在读取值时都会使用array 的值。这可能与threadWorkerCopy 增加值同时发生或之后，因此您可能会获得不同目录中的文件或同一目录中的所有文件。

要为每个复制进程获取一个新编号，array 中的编号必须在工作线程中递增。在这种情况下，您必须防止两个或多个线程同时访问array。你可以用另一个锁来实现它。

为了测试，我将目录列表替换为示例文件名的硬编码列表，并将复制替换为打印值。

import queue, threading, os, time
import shutil

bfr_drive = '/home/test_folder' # cache
ext = ".dat" # data file extension
array = 0 # simluated array as t0-t6
fileList = [] # list of files to be moved from cache to storage
destPath = '/home/test_folder/t'
fileQueue = queue.Queue()


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

    def __init__(self):
        # directory listing replaced with hard-coded list for testing
        for file_name in [ 'foo.dat', 'bar.dat', 'baz.dat', 'a.dat', 'b.dat', 'c.dat', 'd.dat', 'e.dat', 'f.dat', 'g.dat' ] :
        #for file_name in os.listdir(bfr_drive):
            if file_name.endswith(ext):
                fileList.append(os.path.join(bfr_drive, file_name))
                fileList.sort()

        self.totalFiles = len(fileList)

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


    def CopyWorker(self):
        global array
        while True:
            fileName = fileQueue.get()

            with self.lockArray:
                myArray = array
                array += 1
                if array > 6:
                    array = 0

            # actual copying replaced with output for testing
            print('copying', fileName, destPath+str(myArray))
            #shutil.copy(fileName, destPath+str(myArray))

            with self.lock:
                self.copyCount += 1

                percent = (self.copyCount * 100) / self.totalFiles

                print (str(percent) + " percent copied.")

            # moved to end because otherwise main thread may terminate before the workers
            fileQueue.task_done()

    def threadWorkerCopy(self, fileNameList):
        for i in range(4):
            t = threading.Thread(target=self.CopyWorker)
            t.daemon = True
            t.start()

        for fileName in fileNameList:
            fileQueue.put(fileName)
        fileQueue.join()

ThreadedCopy()

这会打印出类似这样的内容（可能会在不同的运行之间发生变化）：

10 files to copy.
copying /home/test_folder\a.dat /home/test_folder/t0
10.0 percent copied.
copying /home/test_folder\baz.dat /home/test_folder/t3
20.0 percent copied.
copying /home/test_folder\b.dat /home/test_folder/t1
copying /home/test_folder\c.dat /home/test_folder/t4
copying /home/test_folder\bar.dat /home/test_folder/t2
copying /home/test_folder\d.dat /home/test_folder/t5
30.0 percent copied.
copying /home/test_folder\e.dat /home/test_folder/t6
40.0 percent copied.
copying /home/test_folder\f.dat /home/test_folder/t0
50.0 percent copied.
copying /home/test_folder\foo.dat /home/test_folder/t1
60.0 percent copied.
copying /home/test_folder\g.dat /home/test_folder/t2
70.0 percent copied.
80.0 percent copied.
90.0 percent copied.
100.0 percent copied.

注意事项：

我将fileQueue.task_done() 行移到CopyWorker 的末尾。否则我不会得到所有百分比输出行，有时还会出现错误消息

Fatal Python error: could not acquire lock for <_io.BufferedWriter name='<stdout>'> at interpreter shutdown, possibly due to daemon threads

也许你应该在主线程结束之前等待所有工作线程结束。

我没有检查代码中是否还有其他错误。

---

编辑问题中的代码更改后：

修改后的代码仍然存在工作线程在fileQueue.task_done()之后仍然会做一些输出的问题，这样主线程可能会在worker之前结束。

修改后的代码包含工作线程访问array时的竞态条件，因此可能出现意外行为。