現在実行中のcoroutineを得る方法

Question

自分で非同期libraryを書いているのですが、実行中のコルーチンが自身のコルーチンオブジェクトを得る簡単な方法はありますか？例えば以下のような感じです。

async def async_fn():
    自身のcoroutineを得る何らかの関数()

async def main():
    await my_async_library.複数のcoroutineの完了を待機(async_fn(), async_fn(), ...他にも色々なcoroutine)

my_async_library.run(main())

asyncioにおいてはcurrent_task()という関数を使って実行中のcoroutineを得られるようなのでちょっと実装codeを覗いてみたのですが、どうやらglobal変数を用いて実行中のtaskを自分で管理しているようです。そんな事をせずとも例えばinspect.get_current_coroutine()のような形で簡単に実行中のcoroutineを得る方法があれば教えてほしいです。

最終的にやりたい事

最終的にやりたいのは とあるGUI libraryのイベントのcallback関数からcoroutine.throw()を呼ぶことで、問題はcallback関数がそのcoroutineのinner関数として作られることです。だから実行中のcoroutineから自身のcoroutine objectを得て、それをcallback関数から見えるようにしたいです。

Answer 1

余談になりますが、フレームオブジェクトを利用しての実装。

コルーチンオブジェクトの捕捉はデコレータですが、
受け渡し方についての別アプローチです。

第一引数にコルーチンオブジェクトを直接渡す

"""
Update f_local via frame object
see more details https://www.python.org/dev/peps/pep-0558/

> The PyFrame_LocalsToFast() function will be changed to always emit RuntimeError, 
explaining that it is no longer a supported operation, 
and affected code should be updated to use PyFrame_GetLocals(frame), 
PyFrame_GetLocalsCopy(frame), or PyFrame_GetLocalsView(frame) instead.

"""

import asyncio
import inspect
import functools
import ctypes


def self_coro_instance(func):
    assert inspect.iscoroutinefunction(func)
    @functools.wraps(func)
    def _func(*args, **kw):
        # 第一引数は仮で適当な値を置きます
        coro = func(None, *args, **kw)
        assert inspect.iscoroutine(coro)

        # 第一引数の変数名は "coro" として、
        # フレームオブジェクト経由でコルーチン内のローカル変数 coro を置き換え
        frame = coro.cr_frame
        frame.f_locals.update({"coro": coro})

        # 通常は、f_locals は、書き換えても反映されませんが、
        # 変更を反映する内部API を呼び出し。 
        # (※ 但し、このAPIは廃止が検討されているようです) 
        ctypes.pythonapi.PyFrame_LocalsToFast(
            ctypes.py_object(frame), ctypes.c_int(1))
        
        return coro
    return _func


async def async_fn1():
    coro = async_fn2()
    print(coro)
    await coro


@self_coro_instance
async def async_fn2(coro):
    print(coro)
    await asyncio.sleep(1)


async def main():
    await asyncio.gather(async_fn1())


if __name__ == '__main__':
    asyncio.run(main())

欠点: 処理系依存してしまう上に、先の方法と比べても、恩恵は関数呼び出しの括弧が外れるだけなので、
このコード自体は、お勧めの方法という訳ではありません。

---

get_current_coroutine() の実装。
コルーチンオブジェクトをコルーチン内のローカル変数に保持します。

import asyncio
import inspect
import functools
import ctypes


def self_coro_instance(func):
    assert inspect.iscoroutinefunction(func)
    @functools.wraps(func)
    def _func(*args, **kw):
        coro = func(*args, **kw)
        assert inspect.iscoroutine(coro)

        frame = coro.cr_frame
        frame.f_locals.update({"__coro__": coro})
        ctypes.pythonapi.PyFrame_LocalsToFast(
            ctypes.py_object(frame), ctypes.c_int(1))
        
        return coro
    return _func


def get_current_coroutine():
    # 注意: デコレーターを多重に使う状況は想定してません。
    frame = inspect.currentframe().f_back
    return frame.f_locals.get("__coro__", None)


async def async_fn1():
    coro = async_fn2()
    print(coro)
    await coro


@self_coro_instance
async def async_fn2():
    # ローカル変数一覧を表示
    print(locals())

    # ローカル変数 __coro__ は存在しますが、
    # 実行時にデコレータでの代入の為、コンパイル時には存在しません。
    # print(__coro__) はバイトコードでは、グローバル変数参照となる為 NameError

    # locals() 経由で参照
    print(locals()["__coro__"])
    
    coro = get_current_coroutine()
    print(coro)

    await asyncio.sleep(1)


async def main():
    await asyncio.gather(async_fn1())


if __name__ == '__main__':
    asyncio.run(main())

---

実用に当たっては、PyFrame_LocalsToFast API の動向
(代替方法や、各バージョンでの状況)を調べる必要がありそうなので、
上記のコード自体は、ライブラリ向きではないデモンストレーションなのですが、

ここで紹介したかったのは、フレームオブジェクトの存在で、
inspect モジュール内部で良く扱われる実行時環境の情報です。
コルーチンの場合 coro.cr_frame.f_locals でコルーチン内のローカル変数にアクセス可能です。

これを応用して、例えば、inner関数で callback をという事でしたが、
コルーチン内のローカル変数を利用する為に inner 関数 にする必要があったのだとしたのなら、

別の方法としては、外部から callback を登録する方法にして、
必要なコルーチン内のローカル変数を coro.cr_frame.f_locals 経由で参照する、
等も考えられるのではないでしょうか。※ 但し処理系に依存する可能性は高い。

(実際の状況は把握してないので、この辺りは想像です）

Comments

[gottadiveintopy]

> 実際の状況は把握してないので、この辺りは想像です

実は本当にやりたいのはGUI libraryへ渡したcallback関数からasync generatorを進めたり閉じたりする事なんですが、ここで投稿するために問題をなんとか簡略化しようとした結果「現在実行中のcoroutineを得る方法」という全然意味の違う質問になってしまっています、ごめんなさい。

> 別の方法としては、外部から callback を登録する方法にして、必要なコルーチン内のローカル変数をcoro.cr_frame.f_locals 経由で参照する、等も考えられるのではないでしょうか。

ちょっとうっかりしていたのですが、coroutineへの操作(send(), close(), throw()等)は常に一番外側のcoroutine(awaitで待たれていないcoroutine)から行わないといけないのでしたね。なのでcallback関数から内側かもしれない(awaitされているかもしれない)coroutineを見えるようにした所で何の意味も無いことになります。だからcallback関数はcoroutineのinner関数にしておいて coro.cr_frame.f_locals のような物を触らずに済ませたいなと考えています。

そして一番外側のcoroutineをどうやって得るかなんですがasyncioであればasyncio.current_task().get_coro()とすれば簡単に得られます。問題は私が作っている非同期libraryでどうやって得るかなんですが、global変数を用いずとも意外と簡単だったことに今日気付いたのでこれは解決しました。どのように解決したかはPullRequestを作ってlinkを貼る予定です。

後は色々とまだ試せてないことがあるのでそれをした後にここで報告できたらなと思います。

処理系の実装依存のやり方なんですが、何とか理解できました。最終手段としてこういったやり方があるのは頭に入れておこうと思います。 m(_ _)m

[teamikl]

処理系依存は、スタンドアローンなアプリであれば有りかもしれませんが、
ライブラリであれば避けたいところですよね。

> 実は本当にやりたいのはGUI libraryへ渡したcallback関数からasync generatorを進めたり閉じたりする事なんですが、

なるほど。状況が少し解りました、納得です。

内部から coroutine.throw() 呼び出しはエラーになるはずなので、
raise を使えばよいのでは…とか、関数定義が内部にあるだけで
呼び出しは callback からなので実行コンテキストは異なるのかな？等、考えましたが
実際に、そのコルーチンがどのように扱われるか、というところ迄は未考慮でした。

＞後は色々とまだ試せてないことがあるのでそれをした後にここで報告できたらなと思います。

GUI + 非同期ということで、個人的にも興味がある所です。
是非お願いします。

[gottadiveintopy]

作ってるのはこんな感じの非同期libraryです。
https://youtu.be/8XP1KgRd3jI
これはあくまでtkinter用で今つまづいているkivy用の物とは異なりますが。

> 呼び出しは callback からなので実行コンテキストは異なるのかな？

ですね。callback関数は完全に同期codeから呼ばれるので。

[teamikl]

私もジェネレータで似たようなことをしたことがありました。
コルーチンは扱ってませんが、同じくtkinterで

https://repl.it/@MiKLTea/tkGenCountdown#main.py

入力イベントの待機等は、自分はタイマーでキューを監視する方法を取ってましたが、
タイマーを止めて、GUIイベントのcallbackを起点に再開という方法もできそうですね。
ここのアイデアは参考になりました。ありがとうございます。

inner関数にする理由と、一番外側のコルーチンオブジェクトが必要というのも、状況を把握出来ました。

[gottadiveintopy]

ごめんなさい「coroutineに対する操作は一番外側...」っていうのは私の誤解でした。多分内側のcoroutineからでも一番外側の物からでも可能だと思います。(これも100%の自信は無いですが)

[gottadiveintopy]

詳しい情況を載せられずエスパー回答に近い事を求めてしまったので、情況と自分が調べて分かった事をまとめました。

https://qiita.com/gotta_dive_into_python/private/ea3fb0477a1b3110df69

少し長いので面倒くさければ読まなくとも構いません m(_ _)m

[teamikl]

まとめ有難うございます。やはり実際のコードを元にすると解りやすい。
個人的にも興味のある所だったので、とても参考になりました。

Answer 2

「現在実行中のcoroutineを得る方法」

素直にデコレーターで実装してみました。
第一引数に、「実行中のコルーチンを得る関数」を渡します。

import asyncio
import inspect
import functools

def Y(func):
    assert inspect.iscoroutinefunction(func)
    @functools.wraps(func)
    def _func(*args, **kw):
        coro = None
        coro = func((lambda: coro), *args, **kw)
        assert inspect.iscoroutine(coro)
        return coro
    return _func

async def async_fn1():
    coro = async_fn2()
    print(coro)
    await coro

@Y
async def async_fn2(getCoro):
    coro = getCoro()
    print(coro)
    await asyncio.sleep(1)

async def main():
    await asyncio.gather(async_fn1())


if __name__ == '__main__':
    asyncio.run(main())

デメリット: デコレータを使うと型チェックやannotation 周りがすっきりしない。
定義と実際の呼び出しに齟齬が出てしまいます。

---

inspect 等の実行時情報を参照するアプローチ

そんな事をせずとも例えばinspect.get_current_coroutine()のような形で
簡単に実行中のcoroutineを得る方法があれば教えてほしいです。

inspect関連の多くは処理系依存になる為、以下はCPythonを前提とします。

inspect.currentframe で参照できるスタックフレームの情報ですが、
この実行時情報にコルーチンの情報は含まれません。

正確には、C言語側での構造体 にはあるかもしれないが、
Pythonからアクセス可能な情報　PyFrameObjectとしてはないようです。

より具体的には、f_gen がコルーチン内部のジェネレータ・オブジェクト(※ ここは確認してないので自信なし)を持っているようですが frame_memberlist には f_gen が無い。

一応、状況によっては使える事もあります、

• ローカル変数にはアクセスできるので、変数名を予め規約として決めておけば、

変数名決め打ちでアクセスする等は可能です。しかし、汎用的な解決策には出来ません。

• ctypes を使い、アドレス範囲を特定して探す方法も考えられますが、簡単な方法ではありません。

　CPython処理系の知識が要求され、デバッグが困難になります。
内部仕様の変更にも弱いので、お勧めできる方法ではありません。

個人的な意見ですが、
外部へ提供されてない内部情報は、理由があって隠蔽しているはずなので、
「実行中のコルーチンを得る」には独自に管理する方が無難だと思います。

---

最終的にやりたいのは とあるGUI libraryのイベントのcallback関数からcoroutine.throw()を呼ぶことで、

別の方法を検討した方が脈はありそうです。

想定している状況に合うかは解らないのですが、
coroutine.throw() は外部からコルーチン内で例外を発生させる方法なので、
内部からなら通常の例外(raise ～)が使えるはずです。

Comments

[gottadiveintopy]

かなり深く調べてくれて本当にありがとうございます。私は処理系の実装には明るくないので勉強になります。

デコレーターを使ったやり方ですが、欲しいのはまさにこれでした！(これに気付かなかったのが恥ずかしい)。というのも私としてはasyncioやtrioがやっているような本格的なtaskの管理を行いたくないからです。本格的な事をすればするほどasyncioやtrioに近づいていき、この自作libraryの存在意味が無くなるので。

> 別の方法を検討した方が脈はありそうです。

確かにGUI libraryのcallback関数側から直接例外を起こさない方法もありそうですね。asyncioでいうならcallback関数側からはasyncio.Event.set()を呼ぶだけにしておいて、それを受けてcoroutine側の方で色々動くようにしておけばcallback関数側はcoroutineを知らずに済みそうです。なので上のデコレーターのやり方よりもこっちをまず試そうと思います。ただ投稿の題はあくまでcoroutineを得る方法なので、best answerとさせて頂きます。

ともかく投稿する前にもっと熟慮が必要でした (汗)