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

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实例，以获得不共享状态的生成器。

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.getstate和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