🔍
pythonコード実行の監視-カバレッジやコールグラフの自作
はじめに
Python 3.12よりsys.monitoringが追加されました。
これにより、コードオブジェクトに対して、開始イベントやRETURNイベントを監視することが可能になります。
この機能を応用することでカバレッジ計測や、コールグラフが自作できます。
本記事では、このライブラリの使いかたの調査と実験を行います。
また、コードオブジェクトを使用するので、バイトコードに対して最低限の知識を有していることを前提とします。
簡単なサンプル
特定の関数の呼び出しと、その戻り値のイベントを監視するためのサンプルコードを以下に紹介します。
サンプルコード
from sys import monitoring
from types import CodeType
import dis
def func_x(n):
a = func_y(n)
return a * 2
def func_y(n):
return 5
def show_code(code):
for ins in dis.Bytecode(code):
line_no = " "
if ins.line_number:
line_no = f"{ins.line_number:3}"
print(f" {line_no} {ins.offset:3}: {ins.baseopname} {ins.argval}")
print("func_xのバイトコード-----")
show_code(func_x)
print("func_yのバイトコード-----")
show_code(func_y)
events = monitoring.events
tool_id = monitoring.DEBUGGER_ID
# tool_id を使用する 前に呼び出す必要があります。tool_idは0 から 5 までの範囲でなければなりません
monitoring.use_tool_id(monitoring.DEBUGGER_ID, "SimpleMonitor")
def on_py_start(code: CodeType, offset: int):
# Python 関数の開始時のイベント(呼び出し直後に発生し、呼び出し先のフレームはスタック上にあります)
print(f"[on_py_start] {code.co_name} offset:{offset}")
def on_py_return(code: CodeType, offset: int, retval: object):
# Python 関数からの戻り (戻りの直前に発生し、呼び出し先のフレームはスタック上にあります)
print(f"[on_return] {code.co_name} -> {retval!r}({type(retval)}) offset:{offset}")
def on_call(caller_code: CodeType, offset:int, callee_obj: object, arg0: object):
# Python コード内の呼び出し (呼び出しの前にイベントが発生します)
# arg0には１つめの引数が格納される。（他の引数は観測できない）
print(f"[on_call] {caller_code.co_name} -> {callee_obj!r} offset:{offset} arg0:{arg0!r}")
#
monitoring.register_callback(tool_id, events.PY_START, on_py_start)
monitoring.register_callback(tool_id, events.PY_RETURN, on_py_return)
monitoring.register_callback(tool_id, events.CALL, on_call)
target_events = (
events.PY_START |
events.PY_RETURN |
events.CALL
)
# 特定のCodeTypeに対してのイベントのみ取得する
print('=====set_local_events 後のfunc_x')
monitoring.set_local_events(tool_id, func_x.__code__, target_events)
func_x(100)
# イベント登録を解除
monitoring.register_callback(tool_id, events.PY_START, None)
monitoring.register_callback(tool_id, events.PY_RETURN, None)
monitoring.register_callback(tool_id, events.CALL, None)
monitoring.free_tool_id(tool_id)
出力例：
func_xのバイトコード-----
5 0: RESUME 0
6 2: LOAD_GLOBAL func_y
6 12: LOAD_FAST n
6 14: CALL 1
6 22: STORE_FAST a
7 24: LOAD_FAST a
7 26: LOAD_CONST 2
7 28: BINARY_OP 5
7 32: RETURN_VALUE None
func_yのバイトコード-----
9 0: RESUME 0
10 2: RETURN_CONST 5
=====set_local_events 後のfunc_x
[on_py_start] func_x offset:0
[on_call] func_x -> <function func_y at 0x105523740> offset:14 arg0:100
[on_return] func_x -> 10(<class 'int'>) offset:32
シンプルなサンプルコードの解説
特定の関数の呼び出しなどのイベントを行う手順は以下の通りです。
- use_tool_idを使用してイベントIDの利用を宣言
- register_callbackを使用して、イベントと、そのコールバック関数を紐づけ
- set_local_eventsまたはset_eventsを使用してイベントの監視を有効にする
- 監視対象の関数の実行
- 監視の終了処理
use_tool_idを使用してイベントIDの利用を宣言
特定の関数の呼び出しなどのイベントはツールIDごとに監視できます。ツールIDは事前に以下のIDが割り当てられており、ツールごとの連携が可能になっています。
sys.monitoring.DEBUGGER_ID = 0
sys.monitoring.COVERAGE_ID = 1
sys.monitoring.PROFILER_ID = 2
sys.monitoring.OPTIMIZER_ID = 5
あらゆるイベントを監視する前にまずは、use_tool_idを使用してイベントIDの利用を宣言する必要があります。
register_callbackを使用して、イベントと、そのコールバック関数を紐づけ
次にregister_callbackを使用して、イベントと、そのコールバック関数を紐づけます。
今回は以下のイベントを監視します。
コールバック関数がどのような引数であるかは, Callback function arguments を参照してください。
PY_START
Python関数の開始時のイベントです。関数が開始した直後にイベントが発生します。
以下のようなコールバック関数を記述します。
def on_py_start(code: CodeType, offset: int):
# Python 関数の開始時のイベント(呼び出し直後に発生し、呼び出し先のフレームはスタック上にあります)
print(f"[on_py_start] {code.co_name} offset:{offset}")
このコールバック関数が呼び出された直後のコールスタックは以下のようになります。
on_py_start ← いま実行中（最上位）
func_x ← これから実行を開始（RESUME直前）
caller ← func_xを呼び出した関数
codeは開始された関数のCodeTypeとなっています。
offsetは、対象のコードのバイトコードのオフセットがどのタイミングでイベントが発火したかを記載しています。
今回は以下のようにoffset:0となっています。
[on_py_start] func_x offset:0
これはfunc_xのバイトコードの5行目、オフセット0である RESUME 0の位置であることが確認できます。
PY_RETURN
Python 関数からの戻る直前に発生します。
以下のようなコールバック関数を記述します。
def on_py_return(code: CodeType, offset: int, retval: object):
# Python 関数からの戻り (戻りの直前に発生し、呼び出し先のフレームはスタック上にあります)
print(f"[on_return] {code.co_name} -> {retval!r}({type(retval)}) offset:{offset}")
codeはreturnを行っている関数のCodeTypeとなっています。
offsetは、対象のコードのバイトコードのオフセットがどのタイミングでイベントが発火したかを記載しています
今回は以下のようにoffset:32となっています。
[on_return] func_x -> 10(<class 'int'>) offset:32
これはfunc_xのバイトコードの7行目、オフセット32である RETURN_VALUEの位置であることが確認できます。
CALL
バイトコード中のCALLが呼び出される直前にイベントが発生します。
以下のようなコールバック関数を記述します。
def on_call(caller_code: CodeType, offset:int, callee_obj: object, arg0: object):
# Python コード内の呼び出し (呼び出しの前にイベントが発生します)
# arg0には１つめの引数が格納される。（他の引数は観測できない）
print(f"[on_call] {caller_code.co_name} -> {callee_obj!r} offset:{offset} arg0:{arg0!r}")
caller_codeは呼び出し元の関数の行っている関数のCodeTypeとなっています。
offsetはCALLが実行されたオフセット値です。
今回は以下のように出力されています。
[on_call] func_x -> <function func_y at 0x105523740> offset:14 arg0:100
func_xのバイトコードのoffset:14はCALL 1になっていることが確認できます。
callee_objはこれから呼ばれる対象の関数やメソッドです。
arg0は最初の引数です。複数ある場合は１つしか観測できません。引数がない場合はsys.monitoring.MISSINGとなります。
set_local_eventsまたはset_eventsを使用してイベントの監視を有効にする
set_local_eventsは指定の関数で発生するイベントを監視できます。
target_events = (
events.PY_START |
events.PY_RETURN |
events.CALL
)
monitoring.set_local_events(tool_id, func_x.__code__, target_events)
今回のケースではfunc_xのみを監視しているため、以下のような監視結果となります。
[on_py_start] func_x offset:0
[on_call] func_x -> <function func_y at 0x105523740> offset:14 arg0:100
[on_return] func_x -> 10(<class 'int'>) offset:32
func_xのイベントのみしか監視できていないのが確認できます。
もし、set_eventsを使用した場合は全ての関数が対象となります。
monitoring.set_events(tool_id, target_events)
func_x(100)
この場合は以下のような出力となります。
=====set_local_events 後のfunc_x
[on_call] <module> -> <function func_x at 0x10e465ee0> offset:604 arg0:100
[on_py_start] func_x offset:0
[on_call] func_x -> <function func_y at 0x10e51f740> offset:14 arg0:100
[on_py_start] func_y offset:0
[on_return] func_y -> 5(<class 'int'>) offset:2
[on_return] func_x -> 10(<class 'int'>) offset:32
[on_call] <module> -> <built-in function register_callback> offset:664 arg0:0
全体を監視しているため、register_callbackでイベントを解除するような挙動やprintも監視対象となりますので、コールバック関数側での適切なフィルタリングが必要になります。
監視の終了処理
python3.13以前の場合は以下のように、登録を解除してからfree_tool_idを実行する必要があります。
# イベント登録を解除
monitoring.register_callback(tool_id, events.PY_START, None)
monitoring.register_callback(tool_id, events.PY_RETURN, None)
monitoring.register_callback(tool_id, events.CALL, None)
monitoring.free_tool_id(tool_id)
python3.14以降ではfree_tool_idの挙動が変わっているため、register_callbackでイベントの登録解除を行う必要はありません。
たとえば以下のように登録解除漏れの状態で再度、監視対象の関数を動かしたとします。
monitoring.register_callback(tool_id, events.PY_START, None)
# monitoring.register_callback(tool_id, events.PY_RETURN, None)
# monitoring.register_callback(tool_id, events.CALL, None)
monitoring.free_tool_id(tool_id)
print('===== free_tool_id実行後のfunc_x')
func_x(100)
python3.13では以下のようにコールバック関数が動きます。
===== free_tool_id実行後のfunc_x
[on_call] func_x -> <function func_y at 0x109523740> offset:14 arg0:100
[on_return] func_x -> 10(<class 'int'>) offset:32
しかしpython3.14ではfree_tool_idの内部でclear_tool_idを実行してイベントの登録の解除が行われているため、以下のようにコールバック関数は動作しなくなります。
===== free_tool_id実行後のfunc_x
複雑なサンプル
コールグラフ＋ラインカバレッジの作成例
以下のサンプルでは以下の監視を行います。
- 関数の呼び出し先をみてコールグラフの作成
- 関数中のラインカバレッジを収集する
- 関数中で発生した戻り値を収集する
- 関数中で発生したyieldの値を収集する
- 関数の呼び出し時の引数を収集する
- 関数中で発生した例外を収集する
コールグラフ作成用クラス
コールグラフ作成用クラス
import threading
import sys
import asyncio
import os
from inspect import ArgInfo, getargvalues
from sys import monitoring
from types import CodeType, FunctionType, BuiltinFunctionType, MethodType
from typing import Optional
from dataclasses import dataclass, field
events = monitoring.events
def _unwrap_func(obj):
if isinstance(obj, classmethod) or isinstance(obj, staticmethod):
return obj.__func__
if isinstance(obj, (FunctionType, BuiltinFunctionType)):
return obj
if isinstance(obj, MethodType):
return obj.__func__
return None
@dataclass
class CallNode:
code: CodeType
call_count: int
calls: set['CallNode'] = field(default_factory=set)
exceptions: list[BaseException] = field(default_factory=list)
arg_infos: list[ArgInfo] = field(default_factory=list)
cover_lines: set[int] = field(default_factory=set)
returns: list[object] = field(default_factory=list)
yields: list[object] = field(default_factory=list)
def __hash__(self) -> int:
return hash(self.code)
class NestFunctionMonitor:
def __init__(self, root_folder: str):
self.tool_id = monitoring.DEBUGGER_ID
monitoring.use_tool_id(monitoring.DEBUGGER_ID, "NestFunctionMonitor")
self.local_events = (
events.PY_RETURN |
events.PY_YIELD |
events.LINE
)
self.global_events = (
events.PY_START |
events.RAISE
)
self.lock = threading.Lock()
self.ignore_codes = self._collect_own_codes()
self.dict_nodes: dict[CodeType, CallNode] = {}
self.max_sample = 10
self.abs_root_folder = os.path.abspath(root_folder)
def _collect_own_codes(self):
codes = set()
# クラスに定義されたあらゆる関数を拾う（継承分も含めたければ __mro__ を回す）
for cls in type(self).__mro__: # 必要に応じて break で上位を切る
if cls is object:
break
for _, obj in cls.__dict__.items():
f = _unwrap_func(obj)
if f and getattr(f, "__code__", None):
codes.add(f.__code__)
# インスタンスに動的に付いたもの（必要なら）
for v in self.__dict__.values():
f = _unwrap_func(v)
if f and getattr(f, "__code__", None):
codes.add(f.__code__)
return codes
def start(self):
monitoring.register_callback(self.tool_id, events.PY_START, self.on_py_start)
monitoring.register_callback(self.tool_id, events.RAISE, self.on_raise)
monitoring.register_callback(self.tool_id, events.PY_RETURN, self.on_py_return)
monitoring.register_callback(self.tool_id, events.PY_YIELD, self.on_py_yield)
monitoring.register_callback(self.tool_id, events.LINE, self.on_line)
monitoring.set_events(self.tool_id, self.global_events)
def stop(self):
monitoring.register_callback(self.tool_id, events.PY_START, None)
monitoring.register_callback(self.tool_id, events.RAISE, None)
monitoring.register_callback(self.tool_id, events.PY_RETURN, None)
monitoring.register_callback(self.tool_id, events.PY_YIELD, None)
monitoring.register_callback(self.tool_id, events.LINE, None)
monitoring.free_tool_id(self.tool_id)
def __enter__(self):
self.start()
return self
def __exit__(self, exc_type, exc, tb):
self.stop()
def on_py_start(self, code: CodeType, instruction_offset: int):
if code in self.ignore_codes:
return monitoring.DISABLE
if not code.co_filename.startswith(self.abs_root_folder):
return monitoring.DISABLE
tid = threading.get_ident()
cur = sys._current_frames()[tid]
print(f"[on_py_start] {tid} {code.co_qualname} {code.co_filename} instruction_offset:{instruction_offset}")
with self.lock:
node = self.dict_nodes.get(code)
if not node:
node = CallNode(code, 0, set())
self.dict_nodes[code] = node
monitoring.set_local_events(self.tool_id, code, self.local_events)
node.call_count += 1
start_watch = False
while cur:
if start_watch and self.dict_nodes.get(cur.f_code):
self.dict_nodes.get(cur.f_code).calls.add(node)
if len(node.arg_infos) <= self.max_sample:
args_info = getargvalues(cur)
node.arg_infos.append(args_info)
break
elif code is cur.f_code:
start_watch = True
cur = cur.f_back
def on_raise(self, code: CodeType, instruction_offset: int, exception: BaseException):
# 反復終了を知らせるための StopIterationに起因するケースを除く例外が送出されたときに発生する
print(f'[on_raise] {threading.get_ident()} {code.co_name} {instruction_offset} {exception}')
with self.lock:
node = self.dict_nodes.get(code)
if node and len(node.exceptions) <= self.max_sample:
node.exceptions.append(exception)
def on_py_return(self, code, offset, retval):
print(f"[on_return] {code.co_name} -> {retval!r}({type(retval)}) offset:{offset}")
node = self.dict_nodes.get(code)
if node and len(node.returns) <= self.max_sample:
node.returns.append(retval)
def on_py_yield(self, code, offset, retval):
print(f"[on_py_yield] {code.co_name} -> {retval!r}({type(retval)}) offset:{offset}")
node = self.dict_nodes.get(code)
if node and len(node.yields) <= self.max_sample:
node.yields.append(retval)
def on_line(self, code: CodeType, line_number: int):
print(f'[on_line] {code.co_name} line_no:{line_number}')
node = self.dict_nodes.get(code)
if node:
node.cover_lines.add(line_number)
def dump(self):
self.stop()
roots = []
for _, node1 in self.dict_nodes.items():
has_parent = False
for node2 in self.dict_nodes.values():
if node1 is node2:
continue
if node1 in node2.calls:
has_parent = True
break
if not has_parent:
roots.append(node1)
for node in roots:
self.dump_node(node)
def dump_node(self, node: CallNode, route_path: Optional[list[CallNode]] = None):
depth = 0
if route_path:
depth = len(route_path)
else:
route_path = []
route_path.append(node)
print(f"{' '*depth}{node.code.co_qualname} (実行数:{node.call_count})")
print(f"{' '*(depth+2)} 引数のサンプル: {node.arg_infos}")
print(f"{' '*(depth+2)} 例外のサンプル: {node.exceptions}")
print(f"{' '*(depth+2)} 戻り値サンプル: {node.returns}")
print(f"{' '*(depth+2)} yieldsサンプル: {node.yields}")
print(f"{' '*(depth+2)} 実行済み行: {node.cover_lines}")
for child in node.calls:
if child in route_path:
# 循環コール
continue
self.dump_node(child, route_path)
解説
監視方法としてはグローバルのPY_START, RAISEイベントを監視します。
グローバルのイベントを監視する場合は、以下の条件でフィルタリングしています。
- 指定したフォルダ以下のcodeであること
- NestFunctionMonitorのメソッド以外とする
その後、PY_STARTのイベント中で、その関数のPY_RETURN、PY_YIELD、LINEイベントを監視しています。
マルチスレッド環境を考慮して、コールバック関数から操作される共有オブジェクトの操作時にはロックをかけて実施しています。
通常関数の呼び出し例
例外が発生しない場合の関数の呼び出し結果は以下のようになります。
def test_1(s:int, n: int)->int:
try:
return test_2(s) * n
except ValueError:
return -1
def test_2(s)->int:
return test_3(s)
def test_3(s)->int:
if s % 2 == 0:
return s * 2
else:
raise ValueError("例外")
with NestFunctionMonitor("./src") as mon:
test_1(2, 1)
以下の出力結果では、各関数の呼び出しツリーの関係と、引数、戻り値が観測できています。
test_1 (実行数:1)
引数のサンプル: []
例外のサンプル: []
戻り値サンプル: [4]
yieldsサンプル: []
実行済み行: {185, 186}
test_2 (実行数:1)
引数のサンプル: [ArgInfo(args=['s', 'n'], varargs=None, keywords=None, locals={'s': 2, 'n': 1})]
例外のサンプル: []
戻り値サンプル: [4]
yieldsサンプル: []
実行済み行: {191}
test_3 (実行数:1)
引数のサンプル: [ArgInfo(args=['s'], varargs=None, keywords=None, locals={'s': 2})]
例外のサンプル: []
戻り値サンプル: [4]
yieldsサンプル: []
実行済み行: {194, 195}
もし、例外が発生するような関数を観測した場合についてはtest_1, test_2, test_3で例外の発生が確認でき、例外のハンドリングをしていないtest_2, test3ではRETURNが発生していないことが観測できます。
with NestFunctionMonitor("./src") as mon:
test_1(1, 1)
mon.dump()
test_1 (実行数:1)
引数のサンプル: []
例外のサンプル: [ValueError('例外')]
戻り値サンプル: [-1]
yieldsサンプル: []
実行済み行: {185, 186, 187, 188}
test_2 (実行数:1)
引数のサンプル: [ArgInfo(args=['s', 'n'], varargs=None, keywords=None, locals={'s': 1, 'n': 1})]
例外のサンプル: [ValueError('例外')]
戻り値サンプル: []
yieldsサンプル: []
実行済み行: {191}
test_3 (実行数:1)
引数のサンプル: [ArgInfo(args=['s'], varargs=None, keywords=None, locals={'s': 1})]
例外のサンプル: [ValueError('例外')]
戻り値サンプル: []
yieldsサンプル: []
実行済み行: {194, 197}
ジェネレータの使用例
ジェネレータを使用した関数を観測した例を確認します。
def yield_sample(n, d):
for i in range(n):
yield i * d
def yield_from_sample():
yield from yield_sample(2, 2)
yield from yield_sample(10, 3)
with NestFunctionMonitor("./src") as mon:
for n in yield_from_sample():
print(n)
mon.dump()
この例ではyield_sample, yield_from_sampleでyieldが発生していることが確認できます。
yield_from_sample (実行数:1)
引数のサンプル: []
例外のサンプル: []
戻り値サンプル: [None]
yieldsサンプル: [0, 2, 0, 3, 6, 9, 12, 15, 18, 21, 24]
実行済み行: {204, 205}
yield_sample (実行数:2)
引数のサンプル: [ArgInfo(args=[], varargs=None, keywords=None, locals={}), ArgInfo(args=[], varargs=None, keywords=None, locals={})]
例外のサンプル: []
戻り値サンプル: [None, None]
yieldsサンプル: [0, 2, 0, 3, 6, 9, 12, 15, 18, 21, 24]
実行済み行: {200, 201}
非同期関数の使用例
非同期関数を使用した例を確認します。
async def async_test(name: str, n: float):
print("start...", name)
await asyncio.sleep(n)
print("end...", name)
return f"{name}:{n}"
async def async_main():
res1 = await async_test("1番目", 0.02)
res2 = await async_test("2番目", 0.01)
res3 = await async_test("3番目", 0.01)
print(res1)
print(res2)
print(res3)
with NestFunctionMonitor("./src") as mon:
with asyncio.Runner(debug=True) as runner:
runner.run(async_main())
mon.dump()
async_testの戻り値が確認できるのとともにyieldでFutureオブジェクトが実行されていることが確認できます。
async_main (実行数:1)
引数のサンプル: []
例外のサンプル: []
戻り値サンプル: [None]
yieldsサンプル: [<Future finished result=None created at /Users/user/.pyenv/versions/3.14.0/lib/python3.14/asyncio/base_events.py:459>, <Future finished result=None created at /Users/user/.pyenv/versions/3.14.0/lib/python3.14/asyncio/base_events.py:459>, <Future finished result=None created at /Users/user/.pyenv/versions/3.14.0/lib/python3.14/asyncio/base_events.py:459>]
実行済み行: {224, 225, 226, 227, 228, 223}
async_test (実行数:3)
引数のサンプル: [ArgInfo(args=[], varargs=None, keywords=None, locals={'res1': '1番目:0.02', 'res2': '2番目:0.01', 'res3': '3番目:0.01'}), ArgInfo(args=[], varargs=None, keywords=None, locals={'res1': '1番目:0.02', 'res2': '2番目:0.01', 'res3': '3番目:0.01'}), ArgInfo(args=[], varargs=None, keywords=None, locals={'res1': '1番目:0.02', 'res2': '2番目:0.01', 'res3': '3番目:0.01'})]
例外のサンプル: []
戻り値サンプル: ['1番目:0.02', '2番目:0.01', '3番目:0.01']
yieldsサンプル: [<Future finished result=None created at /Users/user/.pyenv/versions/3.14.0/lib/python3.14/asyncio/base_events.py:459>, <Future finished result=None created at /Users/user/.pyenv/versions/3.14.0/lib/python3.14/asyncio/base_events.py:459>, <Future finished result=None created at /Users/user/.pyenv/versions/3.14.0/lib/python3.14/asyncio/base_events.py:459>]
実行済み行: {216, 217, 218, 219}
バイトコードのカバレッジの作成例
以下のサンプルではバイトコードのカバレッジ収集の作成を行います。
- バイトコードレベルでのカバレッジを収集します。
- 命令の実行順序を保持します。
- これはスレッドごとに実行順序を管理します。
バイトコードのカバレッジの作成用クラス
import dis
from types import CodeType
from sys import monitoring
import threading
from concurrent.futures import ThreadPoolExecutor
events = monitoring.events
class ByteCodeMonitor:
def __init__(self, code: CodeType):
self.target_code = code
self.tool_id = monitoring.DEBUGGER_ID
self.watch_events = events.INSTRUCTION
self.lock = threading.Lock()
self.thread_steps = {}
monitoring.use_tool_id(monitoring.DEBUGGER_ID, "ByteCodeMonitor")
def start(self):
monitoring.register_callback(self.tool_id, events.INSTRUCTION, self.on_instruction)
monitoring.set_local_events(self.tool_id, self.target_code, self.watch_events)
def stop(self):
monitoring.register_callback(self.tool_id, events.INSTRUCTION, None)
monitoring.free_tool_id(self.tool_id)
def __enter__(self):
self.start()
return self
def __exit__(self, exc_type, exc, tb):
self.stop()
def on_instruction(self, code: CodeType, instruction_offset: int):
# VM の命令が、これから実行されようとしている
tid = threading.get_ident()
with self.lock:
if tid not in self.thread_steps:
self.thread_steps[tid] = []
self.thread_steps[tid].append(instruction_offset)
print(f'[on_instruction] ({tid}):({len(self.thread_steps[tid])}) {code.co_name} instruction_offset: {instruction_offset}')
def show_dump(self):
for tid, steps in self.thread_steps.items():
print(f'{tid}=============')
for ins in dis.Bytecode(self.target_code):
indexes = [f"{i + 1:3}" for i, step in enumerate(steps) if step == ins.offset]
line_no = " "
if ins.line_number:
line_no = f"{ins.line_number:3}"
print(f" {line_no} {ins.offset:3}: {ins.baseopname.ljust(36)} {str(ins.argval).ljust(16)} ... {indexes[:10]}")
この例ではINSTRUCTIONイベントを監視します。これはVM の命令が、これから実行されようとしている場合に発生します。
この発生順序をスレッドごとに監視しています。
バイトコードのカバレッジを観測してみます。
def sum_even_double(nums: list[int]) -> int:
total = 0
for n in nums:
if n % 2 == 0:
total += n * 2
return total
with ByteCodeMonitor(sum_even_double.__code__) as mon:
sum_even_double([1, 2, 5, 6, 2])
mon.show_dump()
出力結果：
この例は左から行番号, オフセット, 命令 , 命令の引数, 最後に実行順です。
ループがある場合は同じ命令を複数回実行します。
140704398269376=============
53 0: RESUME 0 ... []
54 2: LOAD_SMALL_INT 0 ... [' 1']
54 4: STORE_FAST total ... [' 2']
55 6: LOAD_FAST_BORROW nums ... [' 3']
55 8: GET_ITER None ... [' 4']
55 10: FOR_ITER 82 ... [' 5', ' 15', ' 29', ' 39', ' 53', ' 67']
55 14: STORE_FAST n ... [' 6', ' 16', ' 30', ' 40', ' 54']
56 16: LOAD_FAST_BORROW n ... [' 7', ' 17', ' 31', ' 41', ' 55']
56 18: LOAD_SMALL_INT 2 ... [' 8', ' 18', ' 32', ' 42', ' 56']
56 20: BINARY_OP 6 ... [' 9', ' 19', ' 33', ' 43', ' 57']
56 32: LOAD_SMALL_INT 0 ... [' 10', ' 20', ' 34', ' 44', ' 58']
56 34: COMPARE_OP == ... [' 11', ' 21', ' 35', ' 45', ' 59']
56 38: POP_JUMP_IF_TRUE 48 ... [' 12', ' 22', ' 36', ' 46', ' 60']
56 42: NOT_TAKEN None ... [' 13', ' 37']
56 44: JUMP_BACKWARD 10 ... [' 14', ' 38']
57 48: LOAD_FAST_BORROW_LOAD_FAST_BORROW ('total', 'n') ... [' 23', ' 47', ' 61']
57 50: LOAD_SMALL_INT 2 ... [' 24', ' 48', ' 62']
57 52: BINARY_OP 5 ... [' 25', ' 49', ' 63']
57 64: BINARY_OP 13 ... [' 26', ' 50', ' 64']
57 76: STORE_FAST total ... [' 27', ' 51', ' 65']
57 78: JUMP_BACKWARD 10 ... [' 28', ' 52', ' 66']
55 82: END_FOR None ... []
55 84: POP_ITER None ... [' 68']
58 86: LOAD_FAST_BORROW total ... [' 69']
58 88: RETURN_VALUE None ... [' 70']
その他注意点
- BRANCHイベントはPython3.14ではBRANCH_LEFT/BRANCH_RIGHTイベントになりました。
- ローカルのイベントの種類には制限があります。対象外のイベントを指定するとset_local_eventsはエラーとなります。
- コールバック関数でsys.monitoring.DISABLEを返すと、そのイベントは無効になります。
- C_RAISE、C_RETURNイベントを監視したい場合はかならずCALLイベントと合わせて監視する必要があります。
まとめ
今回はsys.monitoringの使用例を説明しました。
この機能を利用することで、コードカバレッジの収集や、コールグラフを自分で作成することが可能となります。
Discussion