时间戳

首页 > 解决方案 > 在 cython 中有线程本地数组以便我可以调整它们的大小？

multithreading - 在 cython 中有线程本地数组以便我可以调整它们的大小？

问题描述

我有一个区间树算法，我想为使用线程的许多查询并行运行。问题是每个线程都需要自己的数组，因为我无法提前知道会有多少次命中。

还有其他类似的问题，建议的解决方案总是有一个大小为 (K, t) 的数组，其中 K 是输出长度，t 是线程数。这对我不起作用，因为每个线程的 K 可能不同，并且每个线程可能需要调整数组的大小以适应它获得的所有结果。

伪代码：

for i in prange(len(starts)):

    qs, qe, qx = starts[i], ends[i], index[i]

    results = t.search(qs, qe)

    if len(results) + nfound < len(output):
        # add result to output
    else:
        # resize array
        # then add results

标签： multithreadingopenmpcython

解决方案

通常的模式是每个线程都有自己的容器，这是速度/复杂性和内存开销之间的权衡：

  1. 无需锁定访问此容器，因为只有一个线程访问它。
  2. 与“每个任务都有自己的容器（即每个i值）”相比，开销要少得多。

在并行部分之后，数据必须要么在后处理步骤中收集到最终容器中（这也可以并行发生），要么后续算法应该能够处理容器集合。

这是一个使用 c++-vector 的示例（它已经内置了内存管理和增加大小）：

%%cython -+ -c=/openmp --link-args=/openmp

from cython.parallel import prange, threadid
from libcpp.vector cimport vector
cimport openmp

def calc_in_parallel(N):    
    cdef int i,k,tid
    cdef int n = N
    cdef vector[vector[int]] vecs
    # every thread gets its own container
    vecs.resize(openmp.omp_get_max_threads())
    for i in prange(n, nogil=True):  
        tid = threadid()
        for k in range(i):
            # use container of the thread
            vecs[tid].push_back(k) # dummy for calculation

    return vecs

在许多情况下，使用omp_get_max_threads()线程数会高估实际线程数。在 中明确设置线程数可能更健壮prange，即

...
NUM_THREADS = 2
vecs.resize(NUM_THREADS)
for i in prange(n, nogil=True, num_threads = NUM_THREADS): 
...

可以使用纯 C 应用类似的方法，但在这种情况下将需要更多样板代码（内存管理）。

推荐阅读