【发布时间】：2020-02-02 03:48:16
【问题描述】：
我正在尝试模拟在 Python 中并行运行的不同任务，并且每个并行进程都以不同的频率运行（例如 200 Hz、100 Hz 和 50 Hz）。我使用来自this question 的代码创建了一个 Timer 类以“实时”运行这些进程，但是这些进程会随着时间的推移而去同步（即，三个 200 Hz 任务有时可以在两个 100 Hz 任务之间运行）。
为了同步我的进程，我在它们的共享内存中创建了滴答计数器。 200 Hz 过程的每次迭代都会增加一个计数器，然后等待计数器在达到 2 时将其重置为 0，而 100 Hz 过程的每次迭代都会等待该计数器达到 2 后再将其重置。 50 Hz 过程也是如此，但使用另一个计数器。我使用 while/pass 方法进行等待。
代码如下：
from multiprocessing import Process, Event, Value
import time
# Add Timer class for multiprocessing
class Timer(Process):
def __init__(self, interval, iteration, function, args=[], kwargs={}):
super(Timer, self).__init__()
self.interval = interval
self.iteration = iteration
self.iterationLeft = self.iteration
self.function = function
self.args = args
self.kwargs = kwargs
self.finished = Event()
def cancel(self):
"""Stop the timer if it hasn't finished yet"""
self.finished.set()
def run(self):
startTimeProcess = time.perf_counter()
while self.iterationLeft > 0:
startTimePeriod = time.perf_counter()
self.function(*self.args, **self.kwargs)
# print(self.interval-(time.clock() - startTimePeriod))
self.finished.wait(self.interval-(time.perf_counter() - startTimePeriod))
self.iterationLeft -= 1
print(f'Process finished in {round(time.perf_counter()-startTimeProcess, 5)} seconds')
def func0(id, freq, tick_p1):
# Wait for 2 runs of Process 1 (tick_p1)
while tick_p1.value < 2:
pass
tick_p1.value = 0 # Reset tick_p1
# Add fake computational time depending on the frequency of the process
print(f'id: {id} at {freq} Hz')
if freq == 400:
time.sleep(0.002)
elif freq == 200:
time.sleep(0.003)
elif freq == 100:
time.sleep(0.007)
elif freq == 50:
time.sleep(0.015)
def func1(id, freq, tick_p1, tick_p2):
# Wait for tick_p1 to have been reset by Process0
while tick_p1.value >= 2:
pass
# Wait for 2 runs of Process 2 (tick_p2)
while tick_p2.value < 2:
pass
tick_p2.value = 0 # Reset tick_p2
# Add fake computational time depending on the frequency of the process
print(f'id: {id} at {freq} Hz')
if freq == 400:
time.sleep(0.002)
elif freq == 200:
time.sleep(0.003)
elif freq == 100:
time.sleep(0.007)
elif freq == 50:
time.sleep(0.015)
# Increment tick_p1
tick_p1.value += 1
def func2(id, freq, tick_p2):
# Wait for tick_p2 to have been reset by Process1
while tick_p2.value >= 2:
pass
# Add fake computational time depending on the frequency of the process
print(f'id: {id} at {freq} Hz')
if freq == 400:
time.sleep(0.002)
elif freq == 200:
time.sleep(0.003)
elif freq == 100:
time.sleep(0.007)
elif freq == 50:
time.sleep(0.015)
# Increment tick_p2
tick_p2.value += 1
if __name__ == '__main__':
freqs = [50,100,200]
# freqs = [0.25,0.5,1]
Tf = 10
tick_p1 = Value('i', 1)
tick_p2 = Value('i', 1)
processes = []
p0 = Timer(interval=1/freqs[0], iteration=round(Tf*freqs[0]), function = func0, args=(0,freqs[0], tick_p1))
p1 = Timer(interval=1/freqs[1], iteration=round(Tf*freqs[1]), function = func1, args=(1,freqs[1], tick_p1, tick_p2))
p2 = Timer(interval=1/freqs[2], iteration=round(Tf*freqs[2]), function = func2, args=(2,freqs[2], tick_p2))
processes.append(p0)
processes.append(p1)
processes.append(p2)
start = time.perf_counter()
for process in processes:
process.start()
for process in processes:
process.join()
finish = time.perf_counter()
print(f'Finished in {round(finish-start, 5)} seconds')
如您所见，我在进程中添加了睡眠时间，以模拟计算时间。当我删除进程中的打印命令时，脚本需要 10.2 秒的运行时间来模拟 10 秒的“实时”计算（增加 2%，这是可以接受的）。
我的问题是，这是实现我想要做的最好的方式吗？有没有更好/更快的方法？
谢谢！
【问题讨论】：
标签： python timer multiprocessing synchronization