Python多线程中GIL相关死锁检测：PyGILSTATE_TryEnsure实现方案咨询

1. 模拟非阻塞获取GIL的逻辑 #

CPython公共API未提供直接的PyGILSTATE_TryEnsure，但可以通过以下步骤模拟非阻塞获取GIL的行为：

• 首先用PyThreadState_Get()检查当前线程是否已持有GIL：返回非NULL则说明已持有，无需额外操作。
• 若未持有GIL，通过CPython内部的GIL锁结构尝试非阻塞锁定。以下是兼容Python 3.8+的示例：

#include <Python.h>
#include "pythread.h"

int PyGILSTATE_TryEnsure(PyGILState_STATE *state) {
    PyThreadState *ts = PyThreadState_Get();
    if (ts != NULL) {
        *state = PyGILSTATE_UNLOCKED;
        return 0; // 已持有GIL，返回成功
    }

    // 获取内部GIL锁（Python 3.8+的内部结构）
    PyThread_type_lock gil_lock = ((PyInterpreterState*)PyInterpreterState_Get())->gil->lock;
    // 尝试非阻塞获取锁，超时设为0
    if (PyThread_acquire_lock_timed(gil_lock, 0, 0) == 0) {
        return -1; // 获取失败
    }

    // 获取成功，初始化线程状态
    ts = PyThreadState_New(PyInterpreterState_Get());
    PyEval_RestoreThread(ts);
    *state = PyGILSTATE_LOCKED;
    return 0;
}

注意：直接访问内部API存在兼容性风险，若需跨版本兼容，建议针对不同Python版本做适配或通过动态符号查找实现。

2. 跟踪线程持有的锁并检测死锁 #

当非阻塞获取GIL失败时，需要检测当前线程是否持有可能引发死锁的其他锁：

• 维护线程本地存储（TLS），记录当前线程持有的所有自定义锁（如C层面的互斥锁）。在获取锁时将其加入TLS列表，释放时移除。
• 当PyGILSTATE_TryEnsure返回失败时，遍历TLS中的锁列表，判断是否存在死锁风险：
  • 若当前线程持有任何锁，可直接判定存在死锁可能（因为持有GIL的线程大概率在等待这些锁），抛出Python异常。
  • 若需要更精准检测，可额外维护每个锁的等待线程列表，检查持有GIL的线程是否在等待当前线程持有的锁。

示例代码（C扩展中）：

#include <Python.h>
#include "pythread.h"
#include <stdlib.h>

// 线程本地存储：记录当前线程持有的锁
static Py_tss_t held_locks;

// 锁记录结构
typedef struct LockRecord {
    PyThread_type_lock lock;
    const char* lock_name;
    struct LockRecord* next;
} LockRecord;

// 初始化TLS
void init_lock_tracking() {
    Py_tss_create(&held_locks);
}

// 获取锁并记录到TLS
int acquire_tracked_lock(PyThread_type_lock lock, const char* name) {
    int acquired = PyThread_acquire_lock(lock, 1);
    if (acquired) {
        LockRecord* rec = malloc(sizeof(LockRecord));
        rec->lock = lock;
        rec->lock_name = name;
        rec->next = Py_tss_get(held_locks);
        Py_tss_set(held_locks, rec);
    }
    return acquired;
}

// 释放锁并从TLS移除
void release_tracked_lock(PyThread_type_lock lock) {
    LockRecord** head = (LockRecord**)&Py_tss_get(held_locks);
    LockRecord* prev = NULL;
    LockRecord* curr = *head;

    while (curr != NULL) {
        if (curr->lock == lock) {
            if (prev == NULL) {
                *head = curr->next;
            } else {
                prev->next = curr->next;
            }
            free(curr);
            break;
        }
        prev = curr;
        curr = curr->next;
    }
    PyThread_release_lock(lock);
}

// 尝试获取GIL并检测死锁
int try_ensure_gil_with_deadlock_check() {
    PyGILState_STATE state;
    int result = PyGILSTATE_TryEnsure(&state);
    if (result == -1) {
        // 检查是否持有其他锁
        LockRecord* curr = Py_tss_get(held_locks);
        if (curr != NULL) {
            // 抛出死锁异常
            PyErr_SetString(PyExc_DeadlockError, "Potential deadlock: holding locks while trying to acquire GIL");
            return -1;
        }
        // 无持有锁，只是暂时获取不到GIL，可返回重试
        return 1;
    }
    return 0;
}

3. Python层面的规避方案 #

如果是纯Python代码场景，避免GIL相关死锁的核心是：

• 不要在持有threading.Lock等同步原语时执行会长期占用GIL的操作（如计算密集型代码）。
• 使用threading.RLock避免同一线程重复获取锁导致的死锁。
• 对于需要调用C扩展的场景，确保C扩展在持有外部锁时不会尝试重新获取GIL（若已释放GIL，需先释放外部锁再重新获取GIL）。

内容的提问来源于stack exchange，提问作者Jan Wielemaker