Python/Multiprocessing：进程似乎没有启动

Question

我有一个函数可以读取二进制文件并将每个字节转换为相应的字符序列。例如，0x05 变为 'AACC'，0x2A 变为 'AGGG' 等等......读取文件并转换字节的函数目前是线性函数，由于要转换的文件在 25kb 和 2Mb 之间，这可能需要相当长的时间一会儿。

因此，我正在尝试使用多处理来划分任务并希望提高速度。但是，我就是无法让它工作。下面是线性函数，虽然速度很慢；

def fileToRNAString(_file):

    if (_file and os.path.isfile(_file)):
        rnaSequences = []
        blockCount = 0
        blockSize = 2048
        printAndLog("!", "Converting %s into RNA string (%d bytes/block)" % (_file, blockSize))
        with open(_file, "rb") as hFile:
            buf = hFile.read(blockSize)
            while buf:
                decSequenceToRNA(blockCount, buf, rnaSequences)
                blockCount = blockCount + 1
                buf = hFile.read(blockSize)
    else:
        printAndLog("-", "Could not find the specified file. Please verify that the file exists:" + _file)
    return rnaSequences

注意：函数 'decSequenceToRNA' 读取缓冲区并将每个字节转换为所需的字符串。执行时，该函数返回一个包含块号和字符串的元组，例如(1, 'ACCGTAGATTA...') 最后，我有一个可用的这些元组的数组。

我已尝试将函数转换为使用 Python 的多处理；

def fileToRNAString(_file):
    rnaSequences = []
    if (_file and os.path.isfile(_file)):
        blockCount = 0
        blockSize = 2048
        printAndLog("!", "Converting %s into RNA string (%d bytes/block)" % (_file, blockSize))
        workers = []
        with open(_file, "rb") as hFile:
            buf = hFile.read(blockSize)
            while buf:
                p = Process(target=decSequenceToRNA, args=(blockCount, buf, rnaSequences))
                p.start()
                workers.append(p)
                blockCount = blockCount + 1
                buf = hFile.read(blockSize)
        for p in workers:
            p.join()
    else:
        printAndLog("-", "Could not find the specified file. Please verify that the file exists:" + _file)
    return rnaSequences

但是，似乎没有任何进程启动，因为当运行此函数时，会返回一个空数组。 'decSequenceToRNA' 中打印到控制台的任何消息都不会显示；

>>>fileToRNAString(testfile)
[!] Converting /root/src/amino56/M1H2.bin into RNA string (2048 bytes/block).

与这里的question 不同，我正在运行 Linux shiva 3.14-kali1-amd64 #1 SMP Debian 3.14.5-1kali1 (2014-06-07) x86_64 GNU/Linux 并使用PyCrust 在 Python 版本：2.7.3 上测试函数。我正在使用以下软件包：

import os
import re
import sys
import urllib2
import requests
import logging
import hashlib
import argparse
import tempfile
import shutil
import feedparser
from multiprocessing import Process

我想知道为什么我的代码不起作用，或者我是否在其他地方遗漏了一些东西以使流程正常工作。也对改进代码的建议持开放态度。下面是 'decSequenceToRNA' 供参考：

def decSequenceToRNA(_idxSeq, _byteSequence, _rnaSequences):
    rnaSequence = ''
    printAndLog("!", "Processing block %d (%d bytes)" % (_idxSeq, len(_byteSequence)))
    for b in _byteSequence:
        rnaSequence = rnaSequence + base10ToRNA(ord(b))
    printAndLog("+", "Block %d completed. RNA of %d nucleotides generated." % (_idxSeq, len(rnaSequence)))
    _rnaSequences.append((_idxSeq, rnaSequence))

Answer 1

decSequenceToRNA 在自己的进程中运行，这意味着它在主进程中获得自己的每个数据结构的单独副本。这意味着当您在decSequenceToRNA 中追加_rnaSequences 时，它对父进程中的rnaSequences 没有影响。这可以解释为什么返回一个空列表。

您有两种选择来解决这个问题。首先是创建一个list，可以使用multiprocessing.Manager在进程之间共享。例如：

import multiprocessing

def f(shared_list):
    shared_list.append(1)

if __name__ == "__main__":
    normal_list = []
    p = multiprocessing.Process(target=f, args=(normal_list,))
    p.start()
    p.join()
    print(normal_list)

    m = multiprocessing.Manager()
    shared_list = m.list()
    p = multiprocessing.Process(target=f, args=(shared_list,))
    p.start()
    p.join()
    print(shared_list)

输出：

[]   # Normal list didn't work, the appended '1' didn't make it to the main process
[1]  # multiprocessing.Manager() list works fine

将此应用于您的代码只需要替换

rnaSequences = []

有

m = multiprocessing.Manager()
rnaSequences = m.list()

或者，您可以（并且可能应该）使用multiprocessing.Pool，而不是为每个块创建单独的Process。我不确定hFile 有多大或您正在阅读的块有多大，但如果有超过multiprocessing.cpu_count() 块，您将通过为每个块生成进程来损害性能。使用Pool，您可以保持进程计数不变，并轻松创建rnaSequence 列表：

def decSequenceToRNA(_idxSeq, _byteSequence):
    rnaSequence = ''
    printAndLog("!", "Processing block %d (%d bytes)" % (_idxSeq, len(_byteSequence)))
    for b in _byteSequence:
        rnaSequence = rnaSequence + base10ToRNA(ord(b))
    printAndLog("+", "Block %d completed. RNA of %d nucleotides generated." % (_idxSeq, len(rnaSequence)))
    return _idxSeq, rnaSequence

def fileToRNAString(_file):
    rnaSequences = []
    if (_file and os.path.isfile(_file)):
        blockCount = 0
        blockSize = 2048
        printAndLog("!", "Converting %s into RNA string (%d bytes/block)" % (_file, blockSize))
        results = []
        p = multiprocessing.Pool()  # Creates a pool of cpu_count() processes
        with open(_file, "rb") as hFile:
            buf = hFile.read(blockSize)
            while buf:
                result = pool.apply_async(decSequenceToRNA, blockCount, buf)
                results.append(result)
                blockCount = blockCount + 1
                buf = hFile.read(blockSize)
        rnaSequences = [r.get() for r in results]
        pool.close()
        pool.join()
    else:
        printAndLog("-", "Could not find the specified file. Please verify that the file exists:" + _file)
    return rnaSequences

请注意，我们不再将rnaSequences 列表传递给孩子。取而代之的是，我们只是将我们本应附加的结果返回给父级（我们无法使用 Process），并在那里构建列表。

Answer 2

尝试这样写（参数列表末尾的逗号）

p = Process(target=decSequenceToRNA, args=(blockCount, buf, rnaSequences,))