Skip to content
编程语言书库

第14章 线程安全

本章讲解 pybind11 中的线程安全问题，包括 GIL 与线程的关系、异步编程模型以及线程本地存储。

14.1 多线程基础概念

Section titled “14.1 多线程基础概念”

Python 的多线程模型基于操作系统线程，但受 GIL 限制。

#include <pybind11/pybind11.h>
#include <thread>
#include <atomic>
#include <vector>


namespace py = pybind11;


// 全局计数器（需要原子操作）
std::atomic<int> global_counter(0);


// 工作函数
void worker_function(int iterations) {
    for (int i = 0; i < iterations; ++i) {
        global_counter.fetch_add(1, std::memory_order_relaxed);
    }
}


// 启动多线程处理
py::dict parallel_counter(size_t num_threads, int iterations_per_thread) {
    std::vector<std::thread> threads;


    for (size_t t = 0; t < num_threads; ++t) {
        threads.emplace_back(worker_function, iterations_per_thread);
    }


    for (auto& t : threads) {
        t.join();
    }


    py::gil_scoped_acquire gil;
    py::dict result;
    result["counter"] = global_counter.load();
    result["expected"] = num_threads * iterations_per_thread;
    return result;
}


PYBIND11_MODULE(threading_module, m) {
    m.def("parallel_counter", &parallel_counter);
}
>>> import threading_module
>>> threading_module.parallel_counter(4, 10000)
{'counter': 40000, 'expected': 40000}

14.2 GIL 与线程

Section titled “14.2 GIL 与线程”

GIL（全局解释器锁）意味着同一时刻只有一个线程执行 Python 字节码。这是 Python 的设计选择，用于简化内存管理。

#include <pybind11/pybind11.h>
#include <thread>
#include <chrono>
#include <vector>


namespace py = pybind11;


// CPU 密集型计算（不需要 Python 对象）
double cpu_bound_computation(double value) {
    double result = value;
    for (int i = 0; i < 1000000; ++i) {
        result = std::sqrt(result * result + 1.0);
    }
    return result;
}


// 受 GIL 限制的操作（持有 Python 对象）
py::object process_with_gil(const std::vector<double>& data) {
    py::gil_scoped_acquire gil;  // 获取 GIL
    py::list results;


    for (double val : data) {
        results.append(cpu_bound_computation(val));  // 每次循环都持有 GIL
    }


    return results;
}


// 释放 GIL 的计算（只做纯 C++ 计算）
py::object parallel_process_without_gil(const std::vector<double>& data) {
    size_t num_threads = std::thread::hardware_concurrency();
    std::vector<double> results(data.size(), 0.0);
    std::vector<std::thread> threads;


    size_t chunk_size = data.size() / num_threads;


    for (size_t t = 0; t < num_threads; ++t) {
        size_t start = t * chunk_size;
        size_t end = (t == num_threads - 1) ? data.size() : (t + 1) * chunk_size;


        threads.emplace_back([&, start, end]() {
            py::gil_scoped_acquire gil;  // 每个线程需要获取 GIL
            for (size_t i = start; i < end; ++i) {
                results[i] = cpu_bound_computation(data[i]);
            }
        });
    }


    for (auto& t : threads) {
        t.join();
    }


    py::gil_scoped_acquire gil;
    return py::cast(results);
}


PYBIND11_MODULE(gil_threads_module, m) {
    m.def("process_with_gil", &process_with_gil);
    m.def("parallel_process", &parallel_process_without_gil);
}

关键洞察：GIL 确保同一时刻只有一个线程执行 Python C API。释放 GIL 后，C++ 代码可以并行执行，但无法访问任何 Python 对象。设计算法时，应在持有 GIL 的时间段内只做必要的 Python 操作。

14.3 异步编程模型

Section titled “14.3 异步编程模型”

Python 的异步编程模型（asyncio）与 C++ 的异步机制可以结合使用。

#include <pybind11/pybind11.h>
#include <future>
#include <thread>


namespace py = pybind11;


// 异步执行 CPU 密集型任务
py::object async_compute(py::object callback, double input) {
    // 在新线程中执行，避免阻塞 Python 主线程
    std::async(std::launch::async, [callback, input]() {
        // 执行耗时计算
        double result = 0.0;
        for (int i = 0; i < 1000000; ++i) {
            result += std::sqrt(input * input + i);
        }


        // 获取 GIL 后调用回调
        py::gil_scoped_acquire gil;
        callback(result);


        return result;
    });


    return py::none();  // 立即返回，不阻塞
}


// 使用 Python 的 asyncio 兼容接口
py::object awaitable_compute(double input) {
    struct Awaitable {
        double input;
        std::future<double> future;


        Awaitable(double inp) : input(inp) {
            future = std::async(std::launch::async, [inp]() {
                double result = 0.0;
                for (int i = 0; i < 1000000; ++i) {
                    result += std::sqrt(inp * inp + i);
                }
                return result;
            });
        }
    };


    // 返回一个可等待对象
    return py::cast(std::make_shared<Awaitable>(input));
}
>>> import asyncio
>>> import gil_threads_module as m


>>> # 异步回调模式
>>> def callback(result):
...     print(f"Computation completed: {result}")


>>> m.async_compute(callback, 42.0)
>>> # 回调会在后台线程完成时被调用


>>> # 使用 asyncio
>>> async def main():
...     awaitable = m.awaitable_compute(42.0)
...     # 需要额外实现 __await__ 协议
...     pass


>>> asyncio.run(main())

14.4 并行计算最佳实践

Section titled “14.4 并行计算最佳实践”

实现高性能并行计算需要正确管理 GIL 和线程同步。

#include <pybind11/pybind11.h>
#include <thread>
#include <vector>
#include <numeric>
#include <algorithm>


namespace py = pybind11;


// 线程安全的累加器
class ThreadSafeAccumulator {
public:
    void add(double value) {
        // 使用原子操作或锁保护
        std::lock_guard<std::mutex> lock(mutex_);
        sum_ += value;
        count_++;
    }


    double get_sum() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return sum_;
    }


    size_t get_count() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return count_;
    }


private:
    mutable std::mutex mutex_;
    double sum_ = 0.0;
    size_t count_ = 0;
};


// 生产者-消费者模式
class Pipeline {
public:
    void produce(const std::vector<double>& items) {
        py::gil_scoped_acquire gil;
        for (const auto& item : items) {
            queue_.push_back(item);
        }
        cv_.notify_all();
    }


    py::list consume(size_t max_items) {
        py::gil_scoped_acquire gil;
        py::list results;


        std::unique_lock<std::mutex> lock(mutex_);
        cv_.wait_for(lock, std::chrono::milliseconds(100), [this] {
            return !queue_.empty();
        });


        size_t count = std::min(max_items, queue_.size());
        for (size_t i = 0; i < count; ++i) {
            results.append(queue_.front());
            queue_.erase(queue_.begin());
        }


        return results;
    }


private:
    mutable std::mutex mutex_;
    std::condition_variable cv_;
    std::vector<double> queue_;
};


PYBIND11_MODULE(parallel_module, m) {
    py::class_<ThreadSafeAccumulator>(m, "ThreadSafeAccumulator")
        .def(py::init<>())
        .def("add", &ThreadSafeAccumulator::add)
        .def("get_sum", &ThreadSafeAccumulator::get_sum)
        .def("get_count", &ThreadSafeAccumulator::get_count);


    py::class_<Pipeline>(m, "Pipeline")
        .def(py::init<>())
        .def("produce", &Pipeline::produce)
        .def("consume", &Pipeline::consume);
}

并行计算最佳实践：

  1. 识别 GIL 临界区：将 Python 对象访问限制在最小范围
  2. 批量处理：减少 GIL Acquire/Release 开销
  3. 使用线程池：避免频繁创建/销毁线程
  4. 避免全局状态：使用线程局部存储替代全局变量

14.5 线程本地存储（thread_local）

Section titled “14.5 线程本地存储（thread_local）”

thread_local 关键字为每个线程提供独立的变量副本。

#include <pybind11/pybind11.h>
#include <thread>
#include <vector>


namespace py = pybind11;


// 线程局部计数器
thread_local int thread_local_counter = 0;


// 线程局部缓冲区
thread_local std::vector<double> thread_local_buffer;


void increment_and_store(double value) {
    thread_local_counter++;
    thread_local_buffer.push_back(value);
}


py::dict get_thread_local_state() {
    py::gil_scoped_acquire gil;
    py::dict state;
    state["counter"] = thread_local_counter;
    state["buffer_size"] = thread_local_buffer.size();
    state["buffer_sum"] = std::accumulate(thread_local_buffer.begin(),
                                          thread_local_buffer.end(), 0.0);
    return state;
}


// 使用 thread_local 缓存
class ThreadLocalCache {
public:
    void set(const std::string& key, py::object value) {
        py::gil_scoped_acquire gil;
        cache().insert({key, value});
    }


    py::object get(const std::string& key) {
        py::gil_scoped_acquire gil;
        auto& local_cache = cache();
        auto it = local_cache.find(key);
        if (it != local_cache.end()) {
            return it->second;
        }
        return py::none();
    }


private:
    // 每个线程有自己的 unordered_map
    static thread_local std::unordered_map<std::string, py::object>& cache() {
        static thread_local std::unordered_map<std::string, py::object> instance;
        return instance;
    }
};


PYBIND11_MODULE(tls_module, m) {
    m.def("increment_and_store", &increment_and_store);
    m.def("get_thread_local_state", &get_thread_local_state);


    py::class_<ThreadLocalCache>(m, "ThreadLocalCache")
        .def(py::init<>())
        .def("set", &ThreadLocalCache::set)
        .def("get", &ThreadLocalCache::get);
}
>>> import tls_module as m
>>> def worker():
...     m.increment_and_store(42.0)
...     m.increment_and_store(58.0)
...     print(m.get_thread_local_state())


>>> import threading
>>> threads = [threading.Thread(target=worker) for _ in range(3)]
>>> for t in threads:
...     t.start()
>>> for t in threads:
...     t.join()
{'counter': 2, 'buffer_size': 2, 'buffer_sum': 100.0}
{'counter': 2, 'buffer_size': 2, 'buffer_sum': 100.0}
{'counter': 2, 'buffer_size': 2, 'buffer_sum': 100.0}

关键洞察：thread_local 确保每个线程有独立的变量副本，避免线程间数据竞争。对于需要在线程中缓存 Python 对象的场景，线程局部存储是理想的解决方案。每个线程的 ThreadLocalCache 实例有自己的 unordered_map，不会产生竞争。

线程安全 checklist：

  • 访问 Python 对象前确保获取 GIL
  • 释放 GIL 后不访问任何 py::object
  • 使用 thread_local 避免全局可变状态
  • 对于共享数据，使用适当的同步原语（std::mutex、std::atomic）
  • 考虑使用线程池避免频繁创建线程的开销