はじめに
この記事の動機は、Windows上でPythonを使ってある定周期実行をするアプリのパフォーマンステストをしないといけないことがあったのですが、私は普段は組み込みソフトウェアを触っていて、ものを動かすアプリを設計するときはITRON準拠のリアルタイムOSっぽく設計したいなぁという思いがありました。そこで、Pythonで周期ハンドラっぽいのを作ってみよう！と思い立ったのが動機です。
もちろんWindowsというシステムで動かしている以上リアルタイム性を確保できないのは承知です。
（ふと思ったがWSL上のRT Linuxで動かしたらWindows上でもリアルタイム性をある程度確保できるのでは？次回以降のネタにしよう…）
コード
仕様：指定周期で2つのプロセスを同期実行可能なコード
- mainから3つのmultiprocessingを生成（Threadでも可能）
- プロセスA、プロセスB、周期ハンドラ
- 周期ハンドラは生成時に周期を指定。その周期ごとにイベントオブジェクトを使って他プロセスにイベントを発行
- プロセスA、プロセスBは開始後無限ループに入り、周期ハンドラからのイベントを待つ。イベントセットを受けて何かを実行し、再度イベント待ちに入る
- 各プロセスはMainから終了のイベントセットを受けて、無限ループを抜け終了（Thread用に作っていたのでこうなっているが、Processなら.terminate()が使えるのでそれがいいか？）
###############################################################
# periodic handler test
###############################################################
import time
import multiprocessing
import datetime
def a(flag, end):
print("[a]: I'm ready")
while True:
if end.is_set():
break
# wait set_flag from time monitoring
flag.wait()
# do something
print(f' [a]: {datetime.datetime.now()}')
def b(flag, end):
print("[b]: I'm ready")
while True:
if end.is_set():
break
# wait set_flag from time monitoring
flag.wait()
# do something
print(f' [b]: {datetime.datetime.now()}')
def periodic_handler(flag, end, period):
t0 = t1 = 0
while True:
if end.is_set():
break
t0 = time.perf_counter()
while t1 - t0 < period:
t1 = time.perf_counter()
flag.set() # start other task
print(f'[periodic_handler]: {datetime.datetime.now()}')
flag.clear()
if __name__ == '__main__':
# period setting
period = 1.0 #s
# event object for theads
flag = multiprocessing.Event() # to share timing of every period
end = multiprocessing.Event() # to kill threads
# make process in advance
p1 = multiprocessing.Process(target=a, args=(flag,end,))
p2 = multiprocessing.Process(target=b, args=(flag,end,))
p3 = multiprocessing.Process(target=periodic_handler, args=(flag,end,period,))
# start three theads
p1.start()
p2.start()
p3.start()
# finish all threads after 5 sec
time.sleep(5)
end.set() # stop all theads
ちょこっと解説（メモ）
ほとんど言う事はないのですが、
そもそものコンセプトは、周期実行をするために各プロセス（プロセスA、プロセスB）内でタイマーを見てそれぞれが周期実行するのではなく、タイマーを監視する周期ハンドラからイベントを受けて各プロセスが動くことで一つのタイマーに同期して複数のプロセスが動くというもの。
なのでこのperiodic_handlerの周期を図る仕組みが大変重要になってくる。そこでWindows上のPythonから使える最も精度の高いタイマー（と風の噂で聞きました…）のtime.perf_counter()とwhileを使って周期タイミングをはかっています。もっといいのがあるよ！という賢い方がいらっしゃいましたら教えて下さい…
問題になりうるのがこの周期ハンドラの最小時間分解能がwhileループ一周期の時間になるということ。そしてその時間がどれくらいかわかっていないこと…そして周期タイミングを計るところでオーバーヘッドがあること・・
パフォーマンス計測
※冒頭でいったアプリの計測ではなく、本疑似周期ハンドラがどれくらいの精度かを見るためのテストです。
実行結果ログ
[b]: I'm ready
[a]: I'm ready
[periodic_handler]: 2023-03-23 23:06:20.900484
[b]: 2023-03-23 23:06:20.900484
[a]: 2023-03-23 23:06:20.900484
[periodic_handler]: 2023-03-23 23:06:21.900679
[b]: 2023-03-23 23:06:21.900679
[a]: 2023-03-23 23:06:21.900679
[periodic_handler]: 2023-03-23 23:06:22.902062
[a]: 2023-03-23 23:06:22.902129
[b]: 2023-03-23 23:06:22.902129
[periodic_handler]: 2023-03-23 23:06:23.902266
[a]: 2023-03-23 23:06:23.902266
[b]: 2023-03-23 23:06:23.902266
この結果を信用するなら、４回中３回はミリ秒以下の精度で実行できていて、１回はミリ秒以上の誤差がある。たった４回の試行で発生するならだいぶダメな印象。このdatetime.datetime.now()の時間表示も何も信頼していないけど、Windowsのリアルタイム性も信頼していないからまあこんなもんだろうなぁという感じ…
一応実際どれくらいの周期でイベントを立てられているかをtime.perf_counter()を使って見てみたが、毎回1.0003s±50usほどになっている。イベントオブジェクトのset, clear等の時間がオーバーヘッドとして毎回300us上乗せされている感じか。やっぱりタイマーからの割り込みがないとこんな感じなのか…？
今後もこれ関係の調査などは進めていきたいと思います。