并行运行函数

Question

我正在用系统时钟模拟CPU。我目前使用 asyncio 设置它的方式是

clock.run()

函数有一个无限循环，等待一段时间然后触发一个事件，然后再次等待并清除该事件，这就像时钟高低脉冲一样。

现在CPU也将无限运行，并安排函数在时钟下一次变高时运行，因此CPU执行将阻塞，直到时钟变高并且函数运行并返回。

问题是 asyncio 在单个主线程上运行（我认为），因此如果它运行时钟，它似乎也不能并行运行 cpu。

这是我当前的设置：

class Clock:
    def __init__(self, frequency: int):
        if not isinstance(frequency, int) or frequency <= 0:
            raise ValueError('Frequency must be a positive integer.')

        self.frequency = frequency
        self.period = 1 / frequency
        self.half_period = self.period / 2
        self._clock_pulse = asyncio.Event()

    async def run(self):
        while True:
            self._clock_pulse.set()
            print('high', self._clock_pulse.is_set())
            await asyncio.sleep(self.half_period)
            self._clock_pulse.clear()
            print('low', self._clock_pulse.is_set())
            await asyncio.sleep(self.half_period)

    async def schedule(self, func, *args):
        print('scheduled')
        await self._clock_pulse.wait()
        return await func(*args)


class CPU:
    def __init__(self, clock):
        self.clock = clock

    async def do_nothing(self, n):
        return n

    async def run(self):
        self.n = 0
        while True:
            value = await clock.schedule(self.do_nothing, self.n)
            print(value)
            self.n += 1


clock = Clock(1)
cpu = CPU(clock)


async def main():
    clock_task = asyncio.create_task(clock.run())
    cpu_task = asyncio.create_task(cpu.run())


asyncio.run(main())

因此，我期望

clock.run

循环与

cpu.run

并行连续运行。 也许我可以使用线程，但我对此了解不多？感谢您的帮助！

Answer 1

如果我了解您的情况：

1. 您有一个正在运行的时钟，周期性地在“高”和“低”脉冲之间交替发送脉冲。
2. 您有一个循环，您希望在其中安排并等待任务完成，并且该任务将在下一个“高”脉冲时开始。根据任务完成所需的时间，时钟可能在任务的连续调度之间生成许多高脉冲和低脉冲。我们所能保证的是任务始终以高脉冲开始运行。

您当前的代码

Clock.run

方法似乎无法区分高脉冲和低脉冲。我建议使用

asyncio.Event

实例来表示正在生成的高脉冲，而不是使用

asyncio.Condition

来显示已发生脉冲。

Clock.schedule

功能只需等待高脉冲条件发生即可。

请注意，与其定义具有

func

和 args 参数的 Clock.schedule 方法，不如传递协程参数更简单。另外，您的

main

功能需要一些修改（见下文）：

import asyncio

class Clock:
    def __init__(self, frequency: int):
        if not isinstance(frequency, int) or frequency <= 0:
            raise ValueError('Frequency must be a positive integer.')

        self.frequency = frequency
        self.period = 1 / frequency
        self.half_period = self.period / 2
        self._high_pulse_condition = asyncio.Condition()

    async def run(self):
        while True:
            async with self._high_pulse_condition:
                self._high_pulse_condition.notify_all()  # high pulse event
            await asyncio.sleep(self.period)

    async def schedule(self, coro):
        async with self._high_pulse_condition:
            await self._high_pulse_condition.wait()
        return await coro


class CPU:
    def __init__(self, clock):
        self.clock = clock

    async def do_nothing(self, n):
        return n

    async def run(self):
        import time

        n = 0
        while True:
            value = await self.clock.schedule(self.do_nothing(n))
            print(f'value = {value} at time = {time.time()}')
            n += 1

async def main():
    clock = Clock(1)
    cpu = CPU(clock)

    await asyncio.gather(cpu.run(), clock.run())

asyncio.run(main())

打印：

value = 0 at time = 1710281657.515421
value = 1 at time = 1710281658.5301206
value = 2 at time = 1710281659.53623
value = 3 at time = 1710281660.5377345
value = 4 at time = 1710281661.5463734
value = 5 at time = 1710281662.5613523
value = 6 at time = 1710281663.5721672
value = 7 at time = 1710281664.5855374
value = 8 at time = 1710281665.5871134
value = 9 at time = 1710281666.6020265
value = 10 at time = 1710281667.6114671
value = 11 at time = 1710281668.6124766
value = 12 at time = 1710281669.6271718
...