TouchDesigner での並行処理　~ メインスレッドに乗れない私たち ~

TouchDesignerでマルチスレッドしたいモジュールとかがあったので、そのときのスタディーを書いてみました

今回は Win & Mac 両方で試して、いけました。

TouchDesigner 2021.12380
Python 3.7.2 (TD Version)

今回の検証などに使用したリポはこちらです

目次

• 結論
• 非同期とは

• これまでのやり方

  • 刻み toe
  • 別 python script
  • 非同期系モジュール駆使
  • Asyncio & trio
  • threading & multiprocessing & concurrent.futures
  • やり方

結論

別スレッドを threading あるいは concurrent.futures の ThreadpoolExecutor を利用して立てて、処理を別スレッド内での coroutine あるいは Future を使い、TouchDesigner とのやりとりを全て Queue を使用する。
別スレッドの処理が終わっているのかを Queue の内容で毎フレーム確認する、あるいは event で管理する。

その軸となる TextDAT はこちら

import asyncio
import queue
import threading



@asyncio.coroutine
async def workerFunction(fromMainQ, toMainQ):
    value = fromMainQ.get()
    toMainQ.put(value + 'bbb')
    return


def otherThread(fromMainQ, toMainQ):
    print('starting different thread')
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(workerFunction(fromMainQ, toMainQ))
    loop.close()


fromMainQ = queue.Queue()
toMainQ = queue.Queue()

me.parent().store('fromMainQ', fromMainQ)
me.parent().store('toMainQ', toMainQ)

me.parent().storeStartupValue('fromMainQ', None)
me.parent().storeStartupValue('toMainQ', None)

# our input Queue is their output queue
# so they will receive from our toMainQ and we get their reply from  fromMainQ
backgroundThread = threading.Thread(target=otherThread,
                                    args=(toMainQ, fromMainQ))

# this will start the background non blocking thread
backgroundThread.start()

非同期とは

今回の記事の概念を全て説明するのは大変なので、参考にする記事をいくつか紹介し、その中の内容を引用する。

並行処理は瞬間を切り取ったときには 1 つの処理をしているのですが、ある一定の時間でみると処理を切り替えながら複数の処理をこなしているものを指します。
Python では今から説明するマルチスレッド（ThreadPoolExecutor）とイベントループ（asyncio）がこれに当たります。

方法論

これまでのやり方としては、

• multithreading
• 刻み toe
• 別 python script などなど存在します。

刻み toe

刻みtoe は一つの巨大のTDではなく、処理を小刻みにして、別々のTDにすることを指します

公式でも言われてますが、そもそもTD自体が一つのpython プロセスなので、一番早く multiprocessingを保証する方法です

実際にはTouchDesignerコミュニティーでもそのように活用している人も多くみられます。

一つの TD 自体が一つの Python Process だとして考えられたら、分岐して、NDI IN 、 Touch In、 OSC などでつなげることは簡単かつ簡単な選択肢でもある。

欠点としては、

• 遅延問題
• 複数の toe になるために、デバッグするためのコストが増える
• 死活問題が増える など考えられます。

別 python script

Subprocess モジュールを利用して、外部の python script を叩くような作戦のことを指します。
こちらも OSC などで始まり、終わりなどを受け取ることができ、データについても OSC で送ったり、ファイル自体を書き込みによって実現することができます。

欠点としては刻み toe と同じく、

• 遅延問題
• 複数の toe になるために、デバッグするためのコストが増える
• 死活問題が増える
• プロトコル決め など考えられます

これらの死活問題を回避するためのプロセスもあったので、著者は記事を別に書くかもしれないです

非同期モジュール駆使

パッケージについて

最後の戦略としては、multithreading / concurrent.futures 周りの非同期処理をしてくれそうなモジュールを利用することです。
今回では

• threading
• concurrent.futures
• asyncio
• trio などのモジュールを使った方法について説明をします。

ざっくりしたモジュールの説明であるが、
multithreading と concurrent.futures はほぼ一緒で、concurrent.futures は python 3.6 から標準化され、 multithreading をより簡単に使うために作られたようなものである。

asyncio と trio もほぼ一緒で、 trio は asyncio をより簡単に使うためのパッケージです。

concurrent.futures についてはこちらの記事がわかりやすかったので、引用すると、、

Python には他に threading と multiprocessing というモジュールがありますが、これらが１つのスレッド・プロセスを扱うのに対して、concurrent.futures モジュールは複数のスレッド・プロセスを扱うことを目的としています。

concurrent.futuresモジュールには抽象クラスとしてExecutorクラスがあり、実装クラスとして2つのクラスが提供されています。
並列タスクを実行するにはこの２つのうちどちらかを使用します。

ThreadPoolExecutor
スレッドを使って並列タスクを実行します。
ネットワークアクセスなどCPUに負荷がかからない処理の並列実行に適しています。

ProcessPoolExecutor
プロセスを使って並列タスクを実行します。
CPUに負荷がかかる計算処理などの並列実行に適しています。

今回は ThreadPoolExecutor しか注目しません。なぜなら、ProcessPoolExecutor で何か TD 内で起こすと、必ず処理落ち太郎が出現します
こちらの例は example toe の 中に入っています。実行してみてください。
このような画面が出てくると思います。🥺

説明としては、TD 自体が Python Process なので、そこからまたさらに別の process pool を立ち上げること自体が難しいとのこと。forum でもこちらについては言及されています。

これまでの非同期チャレンジ

先人の知識なしではできなかったので、先人の挑戦について紹介をする。

@genkitoyama さんは Web server を立ち上げて、別スレッドを発火したタイミングでタイマーも発火させ、終わったタイミングを作るような戦略をとりました。
また、いくつかの multithread についての属性を上げてくれました。

欠点と対策
threading モジュールもとても便利なのですが、欠点がいくつかあります。
別スレッドでの処理がいつ終了したかを検知できない
別スレッドの処理中にTDのオペレータを参照できない

他の事例として Matthew Ragan パイセンの example 集がある
この中ではいくつかの multithreading と queue を用いたやり方を紹介しており、終わったタイミングなどを獲得する方法を Table DAT などを用いて紹介されている。

最後の最後に Matthew さんは forum の 2013 年のものを取り上げている。

2013 年、、、こりゃもしや昔の ofx を作り直すみたいな作業がいるのかと思いきや、動きます〜！わーい 🕺

応用した例は github を参照していただいた方が早いですが、ざっくり説明すると Queue を用いて、TD と別スレッドとの伝言ゲームをするような感じです

Queue には numpy array であったり、普通の int とか string なども渡せます
そこでさらに asyncio や trio などを加えることによって、いつ重い処理が終わったのかなどを検知して、スレッド自体を殺すこともでき、TDにはネイティブにないサーバーなどを立てることができます（自前のflaskとかdjango とか!）

それでは良い multithreaded TouchDesigner ライフを！