多线程生成#
四个核心分布(random
、standard_normal
、standard_exponential
和 standard_gamma
)都允许使用 out
关键字参数填充现有数组.现有数组需要是连续且行为良好的(可写和已对齐).在正常情况下,使用常见的构造函数(如 numpy.empty
)创建的数组将满足这些要求.
这个例子使用了 Python 3 concurrent.futures
来使用多个线程填充一个数组.线程是长期存在的,因此重复调用不需要任何来自线程创建的额外开销.
生成的随机数是可重复的,因为相同的种子在给定线程数不变的情况下会产生相同的输出.
from numpy.random import default_rng, SeedSequence
import multiprocessing
import concurrent.futures
import numpy as np
class MultithreadedRNG:
def __init__(self, n, seed=None, threads=None):
if threads is None:
threads = multiprocessing.cpu_count()
self.threads = threads
seq = SeedSequence(seed)
self._random_generators = [default_rng(s)
for s in seq.spawn(threads)]
self.n = n
self.executor = concurrent.futures.ThreadPoolExecutor(threads)
self.values = np.empty(n)
self.step = np.ceil(n / threads).astype(np.int_)
def fill(self):
def _fill(random_state, out, first, last):
random_state.standard_normal(out=out[first:last])
futures = {}
for i in range(self.threads):
args = (_fill,
self._random_generators[i],
self.values,
i * self.step,
(i + 1) * self.step)
futures[self.executor.submit(*args)] = i
concurrent.futures.wait(futures)
def __del__(self):
self.executor.shutdown(False)
多线程随机数生成器可以用来填充数组.``values`` 属性显示填充前的零值和填充后的随机值.
In [2]: mrng = MultithreadedRNG(10000000, seed=12345)
...: print(mrng.values[-1])
Out[2]: 0.0
In [3]: mrng.fill()
...: print(mrng.values[-1])
Out[3]: 2.4545724517479104
使用多线程所需的时间可以与使用单线程生成所需的时间进行比较.
In [4]: print(mrng.threads)
...: %timeit mrng.fill()
Out[4]: 4
...: 32.8 ms ± 2.71 ms per loop (mean ± std. dev. of 7 runs, 10 loops each)
单线程调用直接使用 BitGenerator.
In [5]: values = np.empty(10000000)
...: rg = default_rng()
...: %timeit rg.standard_normal(out=values)
Out[5]: 99.6 ms ± 222 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)
收益是显著的,即使对于仅适度大的数组,扩展也是合理的.与不使用现有数组的调用相比,由于数组创建开销,收益甚至更大.
In [6]: rg = default_rng()
...: %timeit rg.standard_normal(10000000)
Out[6]: 125 ms ± 309 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)
请注意,如果用户没有设置 threads
,它将由 multiprocessing.cpu_count()
决定.
In [7]: # simulate the behavior for `threads=None`, if the machine had only one thread
...: mrng = MultithreadedRNG(10000000, seed=12345, threads=1)
...: print(mrng.values[-1])
Out[7]: 1.1800150052158556