发布于 由admin

CPython的GIL与C扩展线程：如何通过Py_BEGIN_ALLOW_THREADS实现IO密集型任务的释放

CPython的GIL与C扩展线程：如何通过Py_BEGIN_ALLOW_THREADS实现IO密集型任务的释放

大家好，今天我们来深入探讨CPython的全局解释器锁（GIL）以及如何利用C扩展线程，特别是Py_BEGIN_ALLOW_THREADS宏，来释放GIL，从而改善IO密集型任务的性能。

1. 全局解释器锁（GIL）的概念和影响

CPython解释器使用全局解释器锁（GIL）来保证同一时刻只有一个线程可以执行Python字节码。这简化了CPython的内存管理和线程安全，但也带来了一个显著的缺点：在多线程环境中，即使在多核处理器上，CPython程序也无法真正地并行执行CPU密集型的任务。这是因为GIL的存在使得多个线程无法同时持有解释器的控制权。

GIL的存在主要解决了两个问题：

  • 内存管理： CPython的垃圾回收机制依赖于引用计数。多个线程同时修改对象的引用计数可能导致数据竞争，从而导致内存泄漏或程序崩溃。GIL通过串行化对Python对象的访问，避免了这些问题。

  • C扩展兼容性： 许多现有的C扩展并不是线程安全的。GIL的存在保证了这些扩展在多线程环境中也能安全地运行，而无需进行大量的修改。

虽然GIL简化了CPython的内部实现并提高了C扩展的兼容性，但它也限制了Python程序在多核处理器上的并行能力。特别是对于CPU密集型的任务，多线程Python程序的性能提升通常很小，甚至可能不如单线程程序。

2. GIL对IO密集型任务的影响

虽然GIL限制了CPU密集型任务的并行性，但对于IO密集型任务，GIL的影响相对较小。这是因为线程在等待IO操作完成时（例如，等待网络响应或读取文件），可以主动释放GIL，允许其他线程执行。当IO操作完成后，线程会重新尝试获取GIL，然后继续执行。

这种释放GIL的机制使得IO密集型任务可以并发执行，从而提高程序的整体性能。多个线程可以同时等待不同的IO操作完成，而不会因为GIL的限制而互相阻塞。

3. 使用C扩展释放GIL：Py_BEGIN_ALLOW_THREADS和Py_END_ALLOW_THREADS

C扩展提供了一种机制，允许C代码主动释放GIL，从而允许其他Python线程执行。这个机制主要通过Py_BEGIN_ALLOW_THREADS和Py_END_ALLOW_THREADS宏来实现。

  • Py_BEGIN_ALLOW_THREADS： 这个宏会释放GIL，允许其他Python线程执行。它通常在C代码开始执行耗时的IO操作或CPU密集型操作之前调用。

  • Py_END_ALLOW_THREADS： 这个宏会尝试重新获取GIL。它通常在C代码完成IO操作或CPU密集型操作之后调用。如果GIL已经被其他线程持有，当前线程会阻塞，直到GIL可用。

使用这两个宏的典型模式如下：

// C代码
#include <Python.h>
#include <stdio.h>

static PyObject* my_io_function(PyObject* self, PyObject* args) {
    // ... 准备工作 ...

    Py_BEGIN_ALLOW_THREADS
    // 执行耗时的IO操作，例如网络请求或文件读取
    printf("执行IO操作...n");
    sleep(5); // 模拟IO等待
    printf("IO操作完成.n");
    Py_END_ALLOW_THREADS

    // ... 处理结果 ...

    Py_RETURN_NONE;
}

static PyMethodDef MyModuleMethods[] = {
    {"my_io_function",  my_io_function, METH_VARARGS, "执行一个IO密集型任务"},
    {NULL, NULL, 0, NULL}        /* Sentinel */
};

static struct PyModuleDef mymodule = {
    PyModuleDef_HEAD_INIT,
    "mymodule",   /* name of module */
    NULL,          /* module documentation, may be NULL */
    -1,          /* size of per-interpreter state of the module,
                     or -1 if the module keeps state in global variables. */
    MyModuleMethods
};

PyMODINIT_FUNC
PyInit_mymodule(void)
{
    return PyModule_Create(&mymodule);
}

在这个例子中，my_io_function是一个C函数，它会被Python调用。在执行耗时的IO操作之前，Py_BEGIN_ALLOW_THREADS被调用，释放GIL。在IO操作完成后，Py_END_ALLOW_THREADS被调用，重新获取GIL。

4. 一个更完整的例子：多线程下载文件

让我们看一个更完整的例子，演示如何使用C扩展和Py_BEGIN_ALLOW_THREADS来并行下载多个文件。

首先，我们需要一个C扩展模块，它包含一个函数，用于下载单个文件。这个函数会释放GIL，以便其他线程可以同时下载其他文件。

// download_module.c
#include <Python.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#ifdef _WIN32
#include <winsock2.h>
#include <ws2tcpip.h>
#pragma comment(lib, "ws2_32.lib")
#else
#include <sys/socket.h>
#include <netdb.h>
#include <unistd.h>
#endif

#define BUFFER_SIZE 1024

// Function to download a file from a URL
static PyObject* download_file(PyObject* self, PyObject* args) {
    const char* url;
    const char* filename;

    if (!PyArg_ParseTuple(args, "ss", &url, &filename)) {
        return NULL;
    }

    printf("Downloading %s to %s...n", url, filename);

    #ifdef _WIN32
    WSADATA wsaData;
    if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0) {
        PyErr_SetString(PyExc_RuntimeError, "WSAStartup failed");
        return NULL;
    }
    #endif

    Py_BEGIN_ALLOW_THREADS

    int sockfd;
    struct addrinfo hints, *servinfo, *p;
    int rv;
    char buffer[BUFFER_SIZE];
    size_t bytes_received;
    FILE* fp;

    memset(&hints, 0, sizeof hints);
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;

    // Extract hostname from URL (very basic, needs more robust parsing)
    char hostname[256];
    strncpy(hostname, url, sizeof(hostname) - 1);
    hostname[sizeof(hostname) - 1] = ''; // Ensure null termination

    char *path = strchr(hostname, '/');
    if (path != NULL) {
        *path = ''; // Null terminate hostname at the first slash
        path++;        // Move pointer to the path part
    } else {
        path = "/"; // Default path if no slash found
    }

    if ((rv = getaddrinfo(hostname, "80", &hints, &servinfo)) != 0) {
        fprintf(stderr, "getaddrinfo: %sn", gai_strerror(rv));
        Py_BLOCK_THREADS
        #ifdef _WIN32
        WSACleanup();
        #endif
        Py_RETURN_NONE;
    }

    // Loop through all the results and connect to the first we can
    for (p = servinfo; p != NULL; p = p->ai_next) {
        if ((sockfd = socket(p->ai_family, p->ai_socktype, p->ai_protocol)) == -1) {
            perror("socket");
            continue;
        }

        if (connect(sockfd, p->ai_addr, p->ai_addrlen) == -1) {
            #ifdef _WIN32
            closesocket(sockfd);
            #else
            close(sockfd);
            #endif
            perror("connect");
            continue;
        }

        break; // If we get here, we must have connected successfully
    }

    if (p == NULL) {
        fprintf(stderr, "failed to connectn");
        freeaddrinfo(servinfo); // All done with this structure
        Py_BLOCK_THREADS
        #ifdef _WIN32
        WSACleanup();
        #endif
        Py_RETURN_NONE;
    }

    freeaddrinfo(servinfo); // All done with this structure

    // Send HTTP GET request
    char request[512];
    snprintf(request, sizeof(request), "GET /%s HTTP/1.1rnHost: %srnConnection: closernrn", path, hostname);
    if (send(sockfd, request, strlen(request), 0) == -1) {
        perror("send");
        #ifdef _WIN32
        closesocket(sockfd);
        #else
        close(sockfd);
        #endif
        Py_BLOCK_THREADS
        #ifdef _WIN32
        WSACleanup();
        #endif
        Py_RETURN_NONE;
    }

    // Open file for writing
    fp = fopen(filename, "wb");
    if (fp == NULL) {
        perror("fopen");
        #ifdef _WIN32
        closesocket(sockfd);
        #else
        close(sockfd);
        #endif
        Py_BLOCK_THREADS
        #ifdef _WIN32
        WSACleanup();
        #endif
        Py_RETURN_NONE;
    }

    // Receive and write data to file
    while ((bytes_received = recv(sockfd, buffer, BUFFER_SIZE, 0)) > 0) {
        fwrite(buffer, 1, bytes_received, fp);
    }

    if (bytes_received == -1) {
        perror("recv");
    }

    fclose(fp);
    #ifdef _WIN32
    closesocket(sockfd);
    #else
    close(sockfd);
    #endif

    Py_END_ALLOW_THREADS

    #ifdef _WIN32
    WSACleanup();
    #endif

    printf("Downloaded %s to %s successfully.n", url, filename);

    Py_RETURN_NONE;
}

static PyMethodDef DownloadModuleMethods[] = {
    {"download_file", download_file, METH_VARARGS, "Downloads a file from a URL."},
    {NULL, NULL, 0, NULL}        /* Sentinel */
};

static struct PyModuleDef downloadmodule = {
    PyModuleDef_HEAD_INIT,
    "download_module",   /* name of module */
    NULL,          /* module documentation, may be NULL */
    -1,          /* size of per-interpreter state of the module,
                     or -1 if the module keeps state in global variables. */
    DownloadModuleMethods
};

PyMODINIT_FUNC
PyInit_download_module(void)
{
    return PyModule_Create(&downloadmodule);
}

注意，在网络操作开始前使用了Py_BEGIN_ALLOW_THREADS释放了GIL，并在操作完成后使用Py_END_ALLOW_THREADS重新获取GIL。 同时注意，windows系统socket编程需要初始化WSA，并在程序结束时Cleanup。 还有一些错误处理，如果出现错误，记得在释放GIL之前使用Py_BLOCK_THREADS，防止其他线程干扰错误处理。

然后，我们需要一个Python脚本，它使用concurrent.futures模块来创建多个线程，并使用C扩展模块来下载文件。

# main.py
import concurrent.futures
import download_module
import time

urls = [
    "http://example.com/file1.txt",
    "http://example.com/file2.txt",
    "http://example.com/file3.txt",
]

filenames = [
    "file1.txt",
    "file2.txt",
    "file3.txt",
]

def download(url, filename):
    download_module.download_file(url, filename)

if __name__ == "__main__":
    start_time = time.time()

    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        for url, filename in zip(urls, filenames):
            executor.submit(download, url, filename)

    end_time = time.time()
    print(f"Total time: {end_time - start_time:.2f} seconds")

这个脚本创建了一个线程池，并使用线程池来并行下载多个文件。由于download_file函数会释放GIL，因此多个线程可以同时等待网络响应，从而提高程序的整体性能。

编译C扩展:

将 download_module.c 保存到文件中。创建一个 setup.py 文件：

# setup.py
from distutils.core import setup, Extension

download_module = Extension('download_module',
                            sources=['download_module.c'])

setup (name = 'DownloadModule',
       version = '1.0',
       description = 'This is a demo package',
       ext_modules = [download_module])

使用以下命令编译并安装扩展：

python setup.py build_ext --inplace

这个命令会在当前目录生成一个 download_module.so (或者 Windows 下的 download_module.pyd) 文件。

5. 何时使用Py_BEGIN_ALLOW_THREADS和Py_END_ALLOW_THREADS

  • IO密集型操作： 当C代码执行耗时的IO操作时，例如网络请求、文件读取、数据库查询等，应该释放GIL。

  • CPU密集型操作： 当C代码执行耗时的CPU密集型操作时，例如图像处理、科学计算、加密解密等，也应该释放GIL。

  • 长时间运行的操作： 只要C代码的执行时间比较长，并且不需要访问Python对象，就可以释放GIL。

6. 注意事项

  • 线程安全： 在释放GIL之前，必须确保C代码是线程安全的。这意味着C代码不能访问共享的Python对象，或者必须使用适当的锁来保护对共享对象的访问。

  • 异常处理： 如果C代码在释放GIL之后发生异常，必须小心处理。由于GIL已经被释放，其他线程可能会修改Python对象，导致数据不一致或程序崩溃。可以使用PyErr_SetString和PyErr_Clear等函数来处理异常。 并在释放GIL之前使用Py_BLOCK_THREADS防止其他线程干扰异常处理。

  • 死锁： 如果多个线程互相等待对方释放GIL，可能会导致死锁。应该避免这种情况发生。

  • 性能： 释放和获取GIL会带来一定的性能开销。应该仔细权衡释放GIL带来的性能提升和性能开销。

  • Windows Socket: 在Windows下使用socket，请注意调用WSAStartup和WSACleanup来初始化和清理socket库。

7. 表格对比：GIL释放与不释放的影响

特性 不释放GIL 释放GIL
并行性 无法并行执行CPU密集型任务 允许其他Python线程执行，提高IO密集型任务的并发性
线程安全 C代码可以安全地访问Python对象 需要确保C代码是线程安全的，避免数据竞争
性能 适用于CPU密集型任务，避免GIL的开销 适用于IO密集型任务，提高程序的整体性能
复杂性 简单，不需要考虑线程安全问题 复杂，需要考虑线程安全问题、异常处理和死锁
适用场景 单线程程序，CPU密集型任务，不需要访问共享对象 多线程程序，IO密集型任务，需要并发执行
示例代码 直接访问Python对象 Py_BEGIN_ALLOW_THREADS和Py_END_ALLOW_THREADS宏
资源竞争 低 高，需要使用锁来保护共享资源
异常处理 简单，可以直接使用Python的异常处理机制 复杂，需要小心处理在释放GIL之后发生的异常，避免数据不一致或程序崩溃。 在释放GIL之前使用Py_BLOCK_THREADS防止其他线程干扰异常处理。

8. 总结

通过Py_BEGIN_ALLOW_THREADS和Py_END_ALLOW_THREADS宏，C扩展可以主动释放GIL，从而允许其他Python线程执行，提高IO密集型任务的并发性。使用这种机制需要小心处理线程安全问题、异常处理和死锁，并仔细权衡释放GIL带来的性能提升和性能开销。正确地使用C扩展线程可以显著提高Python程序的性能，特别是在处理IO密集型任务时。

更多IT精英技术系列讲座，到智猿学院

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注