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

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)))


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

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

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

最佳答案

random module docs


  该模块提供的功能实际上是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

关于python - 确定性并发任务生成伪随机数,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/53242191/

10-12 20:52