ブロッキングを行う処理を並列処理する方法がわからない

Question

前提・実現したいこと

初心者です。間違いがあればご指摘ください。
ブロッキングを行う処理を並列処理する方法がわからない
ブロッキングを行う処理を並列処理で複数実行したいのですが、1行目しか実行されません。
2つとも実行させる方法を教えてほしいです。
エラーメッセージはないです。

該当のソースコード

threading.Thread(target1)
threading.Thread(target2)

試したこと

asyncio.start()を使用してもうまくいきませんでした。

###Pythonバージョン
Python 3.8

Comments

[dodox86]

ご提示のコードでは何をブロックしようとしているのか全く分かりませんが。

[AariyJP]

target1やtarget2がブロッキングを行う関数だと考えてくださればと思います。

[maisumakun]

「target1やtarget2」の処理内容はどのようなものでしょうか？

（「ブロッキング」と言っても、Python内の都合で起きるものなのか、I/Oのブロックなのかで全く話が異なります）

[AariyJP]

こんな感じの処理をしています。
def run(self, *args, **kwargs):
loop = self.loop

try:
loop.add_signal_handler(signal.SIGINT, lambda: loop.stop())
loop.add_signal_handler(signal.SIGTERM, lambda: loop.stop())
except NotImplementedError:
pass

async def runner():
try:
await self.start(*args, **kwargs)
finally:
if not self.is_closed():
await self.close()

def stop_loop_on_completion(f):
loop.stop()

future = asyncio.ensure_future(runner(), loop=loop)
future.add_done_callback(stop_loop_on_completion)
try:
loop.run_forever()
except KeyboardInterrupt:
log.info('Received signal to terminate bot and event loop.')
finally:
future.remove_done_callback(stop_loop_on_completion)
log.info('Cleaning up tasks.')
_cleanup_loop(loop)

if not future.cancelled():
try:
return future.result()
except KeyboardInterrupt:
# I am unsure why this gets raised here but suppress it anyway
return None

[teamikl]

質問に、問題の現象を再現できるコードを提示してください。
asyncio を複数のスレッドで実行したいということでしょうか？

気になる点: self.loop はスレッド毎に作成されてますか？
ブロッキング処理がasyncioのイベントループ( loop.run_forever() )の事だとすると、
スレッド毎にloopが必要になるはずです。

asyncio を使っているなら、やりたいことによっては
run_in_executor を使う方が適切な場合もあります。

Answer 1

記載頂いたコードではThreadの構築しかしていないように思えます。

こちらを参考にして実装してみてはいかがでしょうか。
https://qiita.com/tchnkmr/items/b05f321fa315bbce4f77

Comments

[AariyJP]

回答ありがとうございます。
指定された記事を参考にこのように書き直してみたところ、1行目のt1 = threading.Thread(target1)が実行された後、2行目のt2 = threading.Thread(target2)から実行されないのです。

t1 = threading.Thread(target1)
t2 = threading.Thread(target2)
t1.start()
t2.start()

[plasticgrammer]

「threading.Thread(target1) が実行された後」とのことですが、『実行された』とは何をもって確認しているのでしょうか。
また、threading.Thread(target1) の実行時例外は発生していないのでしょうか。

[AariyJP]

実行されたか否かはIDEのbreakpointで確認しています。該当のソースコードにbreakpointを設定して、1行目でしか一時停止しないので、2行目から実行されていないのかなと思っています。
特にエラーメッセージは出力されていません。

[plasticgrammer]

target1, target2 は引数なしの関数という認識ですが正しいでしょうか。
その場合に「target=」という形で指定しても同じ挙動になるのでしょうか。
t1 = threading.Thread(target=target1)

[AariyJP]

やはり2行目のt2が実行されません。
以下のように変更しました。
t1 = threading.Thread(target=target1(hoge))
t2 = threading.Thread(target=target2(hogehoge))
t1.start()
t2.start()

[plasticgrammer]

引数があるのですか？投稿されたソースから変更があったのでしょうか。

引数があるのでしたら、記述は以下のようになるかと思います。
t1 = threading.Thread(target=target1, args=(hoge,))

[AariyJP]

t1 = threading.Thread(target=target1(), args=hoge)
のようにしてみましたがエラーが発生してしまいました。
target1()←に引数が必要みたいで...

Traceback (most recent call last):
File "C:\Users(ユーザー名)\AppData\Roaming\JetBrains\IdeaIC2020.2\plugins\python-ce\helpers\pydev\pydevd.py", line 1448, in _exec
pydev_imports.execfile(file, globals, locals) # execute the script
File "C:\Users(ユーザー名)\AppData\Roaming\JetBrains\IdeaIC2020.2\plugins\python-ce\helpers\pydev\_pydev_imps\_pydev_execfile.py", line 18, in execfile
exec(compile(contents+"\n", file, 'exec'), glob, loc)
File "E:\Documents\IdeaProjects(プロジェクト名)\src\main\python\pyy\Initialize.py", line 96, in main
t1 = threading.Thread(target=target1(), args=hoge)
TypeError: target1() missing 1 required positional argument: 'hoge'
python-BaseException

[plasticgrammer]

解決したとのことですが、一応コメントしておきます。

> target1()←に引数が必要みたいで...
前回例示したコードの通り、()は不要です。
記載したコードそのままで試して頂けるとありがたかったです。

Answer 2

自己解決しました。
Javaを使ってPythonのプロセスを複数起動させてあげれば無事エラーもなく完成しました。
ご回答いただいた皆様、本当にありがとうございます！