pyqtgraphでの更新処理とマルチプロセス

Question

外部からデータを取り込み、それを解析・表示するプログラムをpyqtgraphを用いて開発しています。
QThreadでデータを取り込み、シグナル・スロットでメインスレッドにて描画を更新するようにしています。
1つのaxでは、取り込んだデータをそのまま表示しています。この処理は処理時間も短いため特にラグなく描画できます。
その他のaxでは数秒ごとにたまったデータを解析し、その結果を表示するようにしています。この解析処理に300msほどかかるため、この処理をするシグナルが出るたびに全体の描画がその時間分止まってしまいます。
取り込んだデータをそのまま表示する部分に影響を出さずに数秒ごとの更新を行う方法はあるでしょうか。

試したこと：
解析結果はndarrayとして帰ってくるため、concurrent.futures.ProcessPoolExecutorとmultiprocessing.shared_memoryを利用して裏で計算をさせてその間に前回の解析結果を表示すれば可能かと思ったのですが、これらの実装がうまくいきません。
ご教授いただければ幸いです。

Comments

[teamikl]

＞裏で計算をさせてその間に前回の解析結果を表示すれば可能かと思ったのですが、これらの実装がうまくいきません。

別プロセスで処理できているなら、
これで「GUIが止まる問題」については解決するはずですが、
どの様にうまくいかないのでしょうか？

Answer 1

その他のaxでは数秒ごとにたまったデータを解析し、その結果を表示するようにしています。この解析処理に300msほどかかるため、この処理をするシグナルが出るたびに全体の描画がその時間分止まってしまいます。

この 300ms かかる解析処理はサブスレッドで処理されているのでしょうか？

メインスレッドで処理されている場合、その間GUIのイベントループが止まります。
スレッドのクラスを使いコードを書いたが、
シグナル＆スロットの呼び出しにより、メインで処理されてしまっている
等も考えられます。

例えば、QThread の使い方には2通りあり、誤った使い方をしていると、
スレッド間とのシグナル＆スロットが意図通りになってない可能性があります。

ログ等を取ってサブスレッド側で実行されてるのを確認してください。

もし、メインスレッドで実行されている場合は、
「データの読み取り」のスレッドとは別に
「データの解析」をするスレッドやプロセスを作ってください。

取り込んだデータをそのまま表示する部分に影響を出さずに数秒ごとの更新を行う方法はあるでしょうか。

問題が起こる原因:
GUIのイベントループに処理が戻っていないと、
描画やイベント等の処理がその間停止になります。

具体駅な情報が提示されていない為、何処かは解りませんが。
解決手段としては、

時間のかかっているデータの解析部分を別スレッドもしくは別プロセスで行い、
メインスレッドでは結果の表示のみを行ってください。

スロットとして記述する関数は、基本直ぐに処理が終わる様にして、
処理が滞ることなくイベントループに戻る様にします。

もし、イベントループ内で時間の掛かるループ処理をする場合は、
ループ内で定期的に processEvents を呼ぶ事で GUI を止めない様にはできますが、
これは応急処置的な扱いなので、お勧めはしません。
(イベントループがネストするので、
時間の掛かるイベントが複数ある場合は破綻する事があります)

---

裏で計算をさせてその間に前回の解析結果を表示すれば可能かと思ったのですが、これらの実装がうまくいきません。

に関しては、問題を再現できる最小限のコードを提示してください。
何処で躓いているかわからないので、要点のみ。

別プロセスとのデータのやり取りは pickle で永続化出来る形式が必要です。
shared_memory利用とのことなので大丈夫だとは思いますが、
参考: 異なるプロセスから参照可能な共有メモリ ndarray の例があります。

Python の concurrent モジュールを使う場合、
サブプロセスからのシグナル・スロットによる通知はできないので、
キュー等で処理の終了をメインのプロセスへ通知する手段が必要ですが、
その辺りはどうなってますか？

また、データの生存期間にも注意が必要です。
データがローカル関数ということは有りませんか？

---

参考

• 複数のプロセスからの単一ファイルへのログ記録

　logging はスレッドセーフですが、マルチプロセスの場合は構成が少し複雑、
Queue を使い同期を取ります。

以下は簡易的な設定方法（マルチスレッド安全ではありませんが必要十分）
関数が実行されているプロセスとスレッドを確認してください。

import logging
from concurrent.futures import ProcessPoolExecutor

# マルチプロセスの場合、別プロセス側でも設定が必要の為、
# 簡易的にはグローバルで設定
logging.basicConfig(
    level=logging.DEBUG,
    format="[%(process)s %(threadName)-18s %(levelname)-6s %(funcName)-12s] %(message)s",
)


def worker():
    import time
    logging.debug("start")
    time.sleep(5)
    logging.debug("end")

def done(future):
    logging.debug("done")

def main():
    logging.debug("start")
    with ProcessPoolExecutor() as e:
        future = e.submit(worker)
        future.add_done_callback(done)
    logging.debug("end")

if __name__ == "__main__":
    main()

※ ここでは、デバッグ用途で、ログで関数が実行されてるプロセスとスレッドを確認するのが目的なので、
多少の表示の乱れは気にしないこととします。
（ログをファイルに保存する場合には、プロセス間での同期が必要です）

Comments

[mrhonen]

ご回答ありがとうございます。確認が遅くなってしまい申し訳御座いません。
ご指摘の通り別スレッドで行うはずだった解析処理がメインスレッドで行われていた部分と、プロセス間での通知がうまく実装されていなかったことが問題でした。
その部分を修正し、GUIはスムーズに更新されるようになりましたが時間が進むにつで非常に重くなってしまいました。指摘いただいた生存期間の問題が考えられるので、検討してみたいと思います。
ありがとうございました。