python - 尝试使用多线程并行化 python 算法并避免 GIL 限制

Question

我正在使用 Biopython 在 Python 中实现一个算法 。我在 FASTA 中存储了多个比对(等长序列集)文件。每个比对包含 500 到 30000 个序列，每个序列的长度约为 17000 个元素。每个序列都存储为 Bio.SeqRecord.SeqRecord对象(查看 SeqRecord object's API documentation 了解更多信息)，它不仅包含序列，还包含有关它的一些信息。我使用 Bio.AlignIO.read() 从磁盘读取它(查看 AlignIO module's API documentation 了解更多信息)返回 MultipleSeqAlignment对象:

seqs = AlignIO.read(seqs_filename, 'fasta')
len_seqs = seqs.get_alignment_length()
stats = {'-': [0.0] * len_seqs, 'A': [0.0] * len_seqs,
         'G': [0.0] * len_seqs, 'C': [0.0] * len_seqs,
         'T': [0.0] * len_seqs}

为了清楚起见，我包括了这个草图:

因为我想并行化对齐分析，所以我使用 threading module 为每个可用的 cpu 分配了一个片段。 (有关我稍后做出此决定的原因的更多详细信息):

num_cpus = cpu_count()
num_columns = ceil(len_seqs / float(num_cpus))
start_column = 0
threads = []
for cpu in range(0, num_cpus):
    section = (start_column, start_column + num_columns)
    threads.append(CI_Thread(seqs_type, seqs, section, stats))
    threads[cpu].start()

    start_column += num_columns

stats 变量是一个共享变量，我在其中存储了一些有关分析的信息(如您在第一个代码片段中所见)。因为每个 cpu 都会修改这个共享变量 的不同位置，所以我认为不需要访问控制，也不需要任何同步原语。我使用线程而不是进程的主要原因是我想避免复制整个 MultipleSeqAlignment每个 cpu 的对象。我做了一些研究，发现 some计算器 posts关于这个。

我还阅读了一些关于“可怕的”Python 全局解释器锁 (GIL) 的信息(我在堆栈交换中的 stackoverflow 和 programmers 找到了很好的信息)但我仍然不能 100% 确定我的算法是否受其影响。据我所知，我一次将一个序列加载到内存中，因此在每次迭代时都进行 IO。这就是为什么我认为使用线程是个好主意，如 this stackoverflow post 中所述我之前已经提过。分析的基本结构(每个线程正在执行)看起来像这样:

for seq in seqs:
    num_column = start_column
    for column in seq.seq[start_column:end_column].upper():
        # [...]
            try:
                stats[elem][num_column] = get_some_info(column)
            except TypeError:
                raise TypeError(('"stats" argument should be a '
                                 'dict of lists of int'))

我使用 timeit module 做了一些性能测试和 the time command使用参数 -f "%e %M"不仅可以检查经过的实时时间(以秒为单位)，还可以检查进程在其生命周期内的最大驻留集大小(以千字节为单位)。似乎使用线程的执行时间是顺序执行的执行时间除以线程数。对于最大驻留集大小，我仍然无法找到模式。

如果您对性能或清晰度有任何其他建议，我将不胜感激。

提前致谢。

Answer 1

当您想对某些数据运行可能是 CPU 密集型算法时，线程将无济于事。尝试查看 multiprocessing 模块，我在处理一个在 100 MB 扫描图像中执行特殊 OCR 的项目时取得了巨大成功。

考虑这个来解决共享内存问题:

from multiprocessing import Pool

def f(x):
    return x*x

if __name__ == '__main__':
    pool = Pool(processes=4)              # start 4 worker processes
    result = pool.apply_async(f, [10])    # evaluate "f(10)" asynchronously
    print result.get(timeout=1)           # prints "100" unless your computer is *very* slow
    print pool.map(f, range(10))          # prints "[0, 1, 4,..., 81]"

看看 pool.imap() 和 pool.apply_async()。

希望对您有所帮助。