如何从主线程正确释放 Python C API GIL

Question

我正在尝试将 Python 嵌入到 C++ 多线程程序中。 我所做的是从 Python C API 调用两个统计函数来对我收集的一些数据执行Two Sample Kolmogorov-Smirnov Test 和Two Sample Anderson-Darling Test。所以我只是将 Python 嵌入到我的代码中，而不是扩展它或使用我自己的 Python 函数。

我最近发现，为了运行使用 Python C API 的多线程程序，您需要正确处理全局解释器锁 (GIL)，并且当您使用 Python C API 函数时，您需要获取 GIL 和然后在使用完 API 函数后释放它。

我仍然不明白的是如何从主线程中正确释放 GIL，以便让其他人执行 Python 代码。

我试过这个（选项 1）：

 int main(int argc, const char * argv[]) {

    int n = 4;
    std::thread threads[n];

    Py_Initialize();
    PyEval_InitThreads();
    PyEval_SaveThread();
    for (int i = 0; i < n; i++) {
        threads[i] = std::thread(exec, i);
    }
    for (int i = 0; i < n; i++) {
        threads[i].join();
    }
    Py_Finalize();
    return 0;
}

但是当我调用Py_Finalize()时它给了我一个segmentation fault。

所以我尝试了这个（选项 2）：

int main(int argc, const char * argv[]) {

    int n = 4;
    std::thread threads[n];

    Py_Initialize();
    PyEval_InitThreads();
    PyThreadState * Py_UNBLOCK_THREADS
    for (int i = 0; i < n; i++) {
        threads[i] = std::thread(exec, i);
    }
    for (int i = 0; i < n; i++) {
        threads[i].join();
    }
    Py_BLOCK_THREADS    
    Py_Finalize();
    return 0;
}

还有这个（选项 3）：

int main(int argc, const char * argv[]) {

    int n = 4;
    std::thread threads[n];

    Py_Initialize();
    PyEval_InitThreads();
    Py_BEGIN_ALLOW_THREADS
    for (int i = 0; i < n; i++) {
        threads[i] = std::thread(exec, i);
    }
    for (int i = 0; i < n; i++) {
        threads[i].join();
    }
    Py_END_ALLOW_THREADS
    Py_Finalize();
    return 0;
}

使用这最后两个选项，代码运行但以以下错误结束：

Exception ignored in: <module 'threading' from '/usr/local/opt/python3/Frameworks/Python.framework/Versions/3.6/lib/python3.6/threading.py'> Traceback (most recent call last): File "/usr/local/opt/python3/Frameworks/Python.framework/Versions/3.6/lib/python3.6/threading.py", line 1289, in _shutdown assert tlock.locked() AssertionError:

编辑： 生成的线程执行的代码是这样的：

double limited_rand(double lower_bound, double upper_bound) {
    return lower_bound + (rand() / (RAND_MAX / (upper_bound-lower_bound) ) );
}


double exec_1(std::vector<int> &left_sample, std::vector<int> &right_sample) {
    PyGILState_STATE gstate = PyGILState_Ensure(); // Acquiring GIL for thread-safe usage Python C API

    PyObject* scipy_stats_module = PyImport_ImportModule("scipy.stats"); // importing "scipy.stats" module

    import_array();
    npy_intp left_nparray_shape[] = {(npy_intp)left_sample.size()}; // Size of left nparray's first dimension
    PyObject* left_sample_nparray = PyArray_SimpleNewFromData(1, left_nparray_shape, NPY_INT, &left_sample[0]); // Creating numpy array with 1 dimension, taking "dim" as a dummy, elements are integers, and the data is taken from "sample1" as a int* pointer
    npy_intp right_nparray_shape[] = {(npy_intp)right_sample.size()}; // Size of right nparray's first dimension
    PyObject* right_sample_nparray = PyArray_SimpleNewFromData(1, right_nparray_shape, NPY_INT, &right_sample[0]);

    PyObject* ks_2samp = PyObject_GetAttrString(scipy_stats_module, "ks_2samp");
    Py_DecRef(scipy_stats_module);

    PyObject* ks_2samp_return_val = PyObject_CallFunctionObjArgs(ks_2samp, left_sample_nparray, right_sample_nparray, NULL);
    Py_DecRef(ks_2samp);
    Py_DecRef(right_sample_nparray);
    Py_DecRef(left_sample_nparray);

    double p_value = PyFloat_AsDouble(PyTuple_GetItem(ks_2samp_return_val, 1));
    Py_DecRef(ks_2samp_return_val);

    PyGILState_Release(gstate); // Releasing GIL
    return p_value;
}


void initialize_c_2d_int_array(int*& c_array, unsigned long row_length_c_array, std::vector<int> &row1, std::vector<int> &row2) {
    for (unsigned int i = 0; i < row_length_c_array; i++) {
        c_array[i] = row1[i];
        c_array[row_length_c_array + i] = row2[i];
    }
}
double exec_2(std::vector<int> &left_sample, std::vector<int> &right_sample){
    PyGILState_STATE gstate = PyGILState_Ensure(); // Acquiring GIL for thread-safe usage Python C API

    PyObject* scipy_stats_module = PyImport_ImportModule("scipy.stats"); // importing "scipy.stats" module
                                                                         //            import_array();
    unsigned long n_cols = std::min(left_sample.size(), right_sample.size());
    int* both_samples = (int*) (malloc(2 * n_cols * sizeof(int)));
    initialize_c_2d_int_array(both_samples, n_cols, left_sample, right_sample);
    npy_intp dim3[] = {2, (npy_intp) n_cols};
    PyObject* both_samples_nparray = PyArray_SimpleNewFromData(2, dim3, NPY_INT, both_samples);

    PyObject* anderson_ksamp = PyObject_GetAttrString(scipy_stats_module, "anderson_ksamp");
    Py_DecRef(scipy_stats_module);

    PyObject* anderson_2samp_return_val = PyObject_CallFunctionObjArgs(anderson_ksamp, both_samples_nparray, NULL);
    Py_DecRef(anderson_ksamp);
    Py_DecRef(both_samples_nparray);
    free(both_samples);

    double p_value = PyFloat_AsDouble(PyTuple_GetItem(anderson_2samp_return_val, 2));
    Py_DecRef(anderson_2samp_return_val);

    PyGILState_Release(gstate); // Releasing GIL

    return p_value;
}


void exec(int thread_id) {
    std::vector<int> left_sample;
    std::vector<int> right_sample;

    int n = 50;
    for (int j = 0; j < n; j++) {

        int size = 100;
        for (int i = 0; i < size; i++) {
            left_sample.push_back(limited_rand(0, 100));
            right_sample.push_back(limited_rand(0, 100));
        }

        exec_1(left_sample, right_sample);
        exec_2(left_sample, right_sample);
    }
}

我使用 Python C API 的函数只有 exec_1 和 exec_2，而 exec 只是重复调用新的随机数据。这是我能想到的最简单的代码，它模仿了我的真实代码的行为。在使用 Python API 时，我还省略了所有类型的错误检查，以提高可读性。

没有任何其他选择，我将运行我的代码，如 option 2 或 option 3 并忘记错误，但我真的很想了解发生了什么。你能帮帮我吗？

附：我在使用 Xcode 8.3.3 的 macOS 10.12.5 系统下运行 Python 3.6.1。如果您需要更多详细信息，请告诉我。

Answer 1

选项1：

我认为这是给你一个分段错误，因为你调用了 PyEval_SaveThread() （它释放了 gil，返回一个保存的线程状态，并将当前线程状态设置为 NULL）。

Py_Finalize 将尝试释放与解释器相关的所有内存，我猜这包括主线程状态。因此，您可以通过以下方式捕获此状态：

 PyEval_InitThreads(); //initialize and aquire the GIL
 //release the GIL, store thread state, set the current thread state to NULL
 PyThreadState *mainThreadState = PyEval_SaveThread();

 *main code segment*

 //re-aquire the GIL (re-initialize the current thread state)
 PyEval_RestoreThread(mainThreadState); 

 Py_Finalize();
 return 0;

或者您可以在调用 PyEval_InitThreads() 之后立即调用 PyEval_ReleaseLock()，因为看起来主代码段没有使用任何嵌入式 python。我有一个类似的问题，这似乎解决了它。

注意：其他线程仍需要在必要时获取/释放 GIL