并发任务确定性地生成伪随机数

Question

我有一个协程返回伪随机字节列表

import asyncio

import random

import os

async def random_bytes():

    random.seed(a=1234)

    results = []

    for _ in range(0, 128):

        await asyncio.sleep(int(os.urandom(1)[0])/4096)

        results.append(random.getrandbits(8))

    return results

loop = asyncio.get_event_loop()

task_1 = loop.create_task(random_bytes())

print(loop.run_until_complete(task_1))

正如预期的那样，这个列表在每次运行时总是相同的，即使每一代之间的时间间隔不同[基于来自 的一些外部熵os.urandom]，伪随机数生成器的种子值也相同。

现在，如果我同时运行其中的两个，则创建两个列表...

loop = asyncio.get_event_loop()

task_1 = loop.create_task(random_bytes())

task_2 = loop.create_task(random_bytes())

print(loop.run_until_complete(asyncio.gather(task_1, task_2)))

...列表总是不同的：任务基本上相互干扰，不再是确定性的。

如何生成两个并发的伪随机数任务，其中每个伪随机数列表都具有确定性，就好像没有其他任务在运行一样？

[我的用例：测试并发行为，使用大量伪随机数，以非确定性间隔生成，但希望伪随机数本身对于每个任务的每次测试运行都相同]

Answer 1

从random模块文档：该模块提供的函数实际上是 random.Random 类的隐藏实例的绑定方法。您可以实例化自己的 Random 实例以获取不共享状态的生成器。async def random_bytes():    generator = random.Random()    generator.seed(1234)    results = []    for _ in range(0, 128):        await asyncio.sleep(int(os.urandom(1)[0])/4096)        results.append(generator.getrandbits(8))    return results

Answer 2

您可以使用random.getstateand random.setstate，确保在设置状态、生成随机数和获取状态之间，您不会屈服于另一个任务。async def random_bytes():    random.seed(a=1234)    state = random.getstate()    results = []    for _ in range(0, 128):        await asyncio.sleep(int(os.urandom(1)[0])/4096)        random.setstate(state)        results.append(random.getrandbits(8))        state = random.getstate()    return results