首頁
編程語言
C語言|JAVA編程
Python編程
網頁編程
ASP編程|PHP編程
JSP編程
數據庫
MYSQL數據庫|SqlServer數據庫
Oracle數據庫|DB2數據庫
您现在的位置: 程式碼之家 >> 數據庫知識 >  >> MySQL 資料庫 >> MySQL 綜合教學

Python 執行模型底層探秘:自定義字節碼 DEBUG_OP、字節碼注入、堆棧/幀追蹤實現全記錄

     編輯:MySQL 綜合教學

動機

想在 **字節碼級別** 追蹤 Python 函數執行:每條 opcode 執行前/後的堆棧狀態、幀信息、異常拋出瞬間。sys.settrace 粒度太粗(行級),決定動手擴展 CPython 3.5,實現一個 DEBUG_OP opcode + 運行時注入框架。

第一步:定義新 Opcode DEBUG_OP

修改 Include/opcode.h:

#define DEBUG_OP 0   // 佔用原 NOP=0,或在末尾新增
#define POP_TOP 1
#define ROT_TWO 2
...

修改 Python/opcode_targets.h 註冊跳轉表:

static void *opcode_targets[256] = {
    &&TARGET_DEBUG_OP,
    &&TARGET_POP_TOP,
    ...
};

第二步:實現 TARGET(DEBUG_OP)(Python/ceval.c)

TARGET(DEBUG_OP) {
    PyObject *target = NULL, *value = NULL, *res = NULL, **sp = NULL, *tmp;
    int i;
    _Py_IDENTIFIER(op_target);                    // 固定回調函數名
    target = _PyDict_GetItemId(f->f_globals, &PyId_op_target);
    if (target == NULL && _PyErr_OCCURRED()) {
        if (!PyErr_ExceptionMatches(PyExc_KeyError)) goto error;
        PyErr_Clear(); DISPATCH();                // 無回調則靜默跳過
    }
    value = PyList_New(0);                        // 收集堆棧
    for (i = 1; i <= STACK_LEVEL(); i++) {
        tmp = PEEK(i);
        if (tmp == NULL) tmp = Py_None;
        PyList_Append(value, tmp);
    }
    Py_INCREF(target);
    PUSH(target); PUSH(value); Py_INCREF(f); PUSH(f);  // 參數:回調、堆棧列表、幀對象
    sp = stack_pointer;
    res = call_function(&sp, 2);                  // 調用 op_target(stack, frame)
    stack_pointer = sp;
    if (res == NULL) goto error;
    Py_DECREF(res);
    DISPATCH();
}

編譯:./configure && make 生成自定義 python 二進制。

第三步:運行時字節碼注入

核心難點:code.co_code 不可變、**分支指令偏移需同步調整**。

可變 CodeObject 包裝器

class MutableCodeObject:
    attrs = ("co_argcount","co_kwonlyargcount","co_nlocals","co_stacksize",
             "co_flags","co_code","co_consts","co_names","co_varnames",
             "co_filename","co_name","co_firstlineno","co_lnotab",
             "co_freevars","co_cellvars")
    def __init__(self, code):
        self.code = code
        for a in self.attrs:
            v = getattr(code, a)
            setattr(self, a, list(v) if isinstance(v, tuple) else v)
    def get_code(self):
        args = [tuple(getattr(self,a)) if isinstance(getattr(self,a),list) else getattr(self,a) for a in self.attrs]
        return self.code.__class__(*args)

注入邏輯:處理絕對/相對跳轉偏移

JUMP_REL = {'FOR_ITER','JUMP_FORWARD','SETUP_LOOP','SETUP_WITH','SETUP_EXCEPT','SETUP_FINALLY'}
JUMP_ABS = {'POP_JUMP_IF_TRUE','POP_JUMP_IF_FALSE','JUMP_ABSOLUTE'}

def insert_debug_op(code, offset):
    """在指定字節偏移插入 DEBUG_OP(0x00),同步修正後續跳轉目標"""
    new_bytes = bytearray()
    inserted = False
    for instr in dis.Bytecode(code):
        if instr.offset == offset:
            new_bytes.append(0x00)        # DEBUG_OP
            inserted = True
        # 相對跳轉:目標在插入點之後 → arg+1
        if instr.opname in JUMP_REL and not inserted:
            if offset < instr.offset + 3 + instr.arg:
                new_bytes.extend(struct.pack(" 插入點 → arg+1
        if instr.opname in JUMP_ABS:
            if instr.arg > offset:
                new_bytes.extend(struct.pack("

全函數注入

def add_debug_op_everywhere(code):
    offsets = [i.offset for i in dis.Bytecode(code)]
    for n, off in enumerate(sorted(offsets)):
        code = insert_debug_op(code, off + n)  # 每插一個,後續偏移+1
    # 需同步增加 co_stacksize(DEBUG_OP 壓入 3 個參數)
    m = MutableCodeObject(code)
    m.co_stacksize += 3
    return m.get_code()

第四步:Python 級追蹤器封裝

def op_target(stack, frame, exc=None):
    if op_target.callback:
        op_target.callback(stack, frame, exc)

class Trace:
    def __init__(self, func):
        self.func = func
    def __call__(self, *a, **kw):
        self._instrument(self.func)
        op_target.callback = self.callback
        try: return self.func(*a, **kw)
        finally: op_target.callback = None
    def _instrument(self, f):
        if not hasattr(f, "op_debug") and hasattr(f, "__code__"):
            f.__code__ = add_debug_op_everywhere(f.__code__)
            f.__globals__["op_target"] = op_target
            f.op_debug = True
    def callback(self, stack, frame, exc): pass  # 子類實現

# 示例:自動追蹤被調用函數
def do_auto_follow(self, stack, frame):
    next_instr = FrameAnalyser(frame).next_instr()
    if "CALL" in next_instr.opname:
        f_idx = (next_instr.arg & 0xFF) + 2 * (next_instr.arg >> 8)
        called = stack[f_idx]
        if not hasattr(called, "op_debug"):
            self._instrument(called)

效果演示

GitHub Repo(原文末尾鏈接)。

  1. 上一篇文章:
  2. 下一篇文章:
Copyright © 程式碼之家 All Rights Reserved