ShowDocs
搜索文章Ctrl K

外观

← 返回C++列表
开发实践

C++ 完整 SPSC 实践：固定载荷、回绕与双线程测试 · SPSC完整项目的头文件入口测试怎样共享源码 ​

返回分篇目录

本系列以前置值语义、RAII、基本线程/函数接口为基础。示例未执行；本轮仅完成静态文本与模板核对，不将预期输出称为实际运行结果。

专题入口：C++ SPSC 模型：有界环、角色与槽位生存期 · C++ SPSC 内存序：发布消息与归还槽位 · C++ 完整 SPSC 实践：固定载荷、回绕与双线程测试 · C++ SPSC 排错：数据竞争、角色越界与退出协议

SPSC完整项目的头文件入口测试怎样共享源码 ​

SPSC完整项目的头文件入口测试怎样共享源码：include/p2/spsc.hpp 是唯一算法实现，app 和 tests 都包含它；CMake 以 INTERFACE 目标传播 C++20 与头目录，再由 Threads::Threads 提供平台线程用法要求。没有复制一个“测试专用队列”来替代真实代码。

以下同源包含该工程全部文件，README 与 TEST-SPEC 也使用固定文件。阅读时先用 N=2 演算满空，再跟随两个方向的发布链，最后看压力测试的收拢与超时。测试通过的预期不是已取得的运行证据，更不等于证明所有允许执行都无数据竞争。

.gitignore（同源文件，示例未执行）
text
/out/
CMakeLists.txt（同源文件，示例未执行）
text
cmake_minimum_required(VERSION 3.28)
project(P2Spsc VERSION 1.0.0 LANGUAGES CXX)
find_package(Threads REQUIRED)
include(CTest)
add_library(p2_spsc INTERFACE)
target_compile_features(p2_spsc INTERFACE cxx_std_20)
target_include_directories(p2_spsc INTERFACE ${CMAKE_CURRENT_SOURCE_DIR}/include)
add_executable(p2-spsc app/main.cpp)
target_link_libraries(p2-spsc PRIVATE p2_spsc Threads::Threads)
if(BUILD_TESTING)
  add_executable(p2-spsc-tests tests/spsc_test.cpp)
  target_link_libraries(p2-spsc-tests PRIVATE p2_spsc Threads::Threads)
  add_test(NAME spsc.contract COMMAND p2-spsc-tests)
  set_tests_properties(spsc.contract PROPERTIES TIMEOUT 30)
endif()
README.md（同源文件，示例未执行）
md
# 固定值 SPSC 有界环

示例未执行。完整实现和测试仅依赖 C++20 标准库，教学版本见 VERSIONS.md。仅支持一个生产者调用 try_push、一个消费者调用 try_pop；同角色并发调用不合法。启动前构造，停用并 join 后析构；不提供并发 reset、size、迭代器或引用借用。

SpscRing<N> 实际存储 N 个已构造 uint64_t，可用容量 N-1；N>=2。空条件 read==write，满条件 next(write)==read。try_push 失败无写入；try_pop 失败不改变输出。失败后可重试，跨线程快照不是全局精确计数。失败重试由调用方决定退避、限时或丢弃策略，不内置忙等。

## 读者命令（本轮不执行）

先准备支持 C++20 jthread 的编译器、标准库、CMake 3.28 与 Ninja。在 examples/native/p2-spsc 目录，PowerShell、Bash、Zsh 逐行执行相同配置命令：

```text
cmake -S . -B out/debug -G Ninja -DCMAKE_BUILD_TYPE=Debug -DBUILD_TESTING=ON
cmake --build out/debug
ctest --test-dir out/debug --output-on-failure
```

Windows 原生需在已初始化的开发者 PowerShell 执行，演示入口 `./out/debug/p2-spsc.exe`；Linux/macOS 为 `./out/debug/p2-spsc`。预期 received=10000 checksum=49995000，退出0；错误顺序、数量或期限超出返回1。没有把任何预期输出作为实际运行结果。

## 所有权与退出

生产线程独占写索引，消费线程独占读索引；槽位不是原子对象，跨线程读写靠索引发布来建立 happens-before。生产者 release 发布已写值，消费者 acquire 读取可见写索引；消费者 release 归还已读槽位，生产者 acquire 后才可覆盖。双方对自己索引使用 relaxed。移除任一方向的发布都需要重新证明，不能以 x86 上一次成功代替证明。

全部 uint64_t 元素在队列构造时已经存活，读写仅做赋值，不使用 placement-new，也不支持任意会抛异常的用户类型。类不可复制/移动。队列不拥有线程；应用以 jthread 局部作用域在队列析构前 join。stop 原子只是终止标记，relaxed 已足够，因为它不发布 sum/received；主线程在 join 后读取这些非原子结果。

## 测试与限制

TEST-SPEC.md 区分确定性边界/回绕测试和两线程顺序压力；它们已经编写但没有执行。索引以条件分支回到零，不让计数器无限累加，也不要求容量为2的幂。is_lock_free 只报告当前索引原子的实现属性，不证明整个应用无锁，更不保证公平调度或固定延迟。没有吞吐、false sharing 或缓存行大小的性能结论。

不要并发析构，不要让两个生产者共享该对象，不要用读取输出引用保留槽位地址。若需多生产者、多消费者或非平凡对象，应重新设计所有权和提交协议；不能仅把索引改 fetch_add 就复用此证明。
TEST-SPEC.md（同源文件，示例未执行）
md
# 测试规格（未执行）

测试入口 tests/spsc_test.cpp 不依赖 assert，Release 不会跳过契约。

| 场景 | 输入/动作 | 预期 |
| --- | --- | --- |
| 空队列 | SpscRing<2> 初态，out=99，try_pop | false，out保持99 |
| 最小容量 | push7、push8、pop、pop | true、false、true且7、false |
| 回绕 | SpscRing<4> 每轮填3、拒绝第4、排空，共1000轮 | 无重复丢失，3000值按序，索引多次过零 |
| 非2次幂槽数 | SpscRing<17> | 不依赖位掩码 |
| 两线程压力 | 独立 producer/consumer，发送0..99999 | 收到100000值，逐项等于期望值，最后为空 |
| 故障界限 | 达到5秒工作期限 | stop=true，两线程退出并join，返回1 |

成功预期标准输出 spsc: boundaries, wrap and ordered stress passed，退出0。deadline 只是避免常见逻辑卡住的测试保护，不是吞吐指标；极慢机器或调度停顿可能失败，需要区分环境超时与顺序错误。CTest 外层30秒用于更严重停滞，测试程序本身不能保证 OS 会按时调度。

线程创建抛 std::system_error 时由 main 的处理器捕获并返回1；向已存在的处理器展开栈时，已构造的 std::jthread 析构并 join，生产循环还有独立期限，避免第二线程创建失败后永久等待。若没有匹配的处理器，标准不保证 terminate 前展开栈，不能只凭 jthread 类型宣称异常路径已收拢。没有以异常文本作为稳定断言。

可选后续验证：在支持平台下用独立构建目录配置 ThreadSanitizer（编译/链接都需启用），运行同一测试；没有竞态报告也不能证明内存模型正确性。不得通过增加随机 sleep 修复同步错误。本轮没有运行普通测试、压力测试、TSan 或基准。
VERSIONS.md（同源文件，示例未执行）
md
# 版本与前提

示例未执行。2026-10-17 核对。教学基准：C++20，固定参照 WG21 N4861 的 [atomics.order] 和 [intro.races]；CMake 3.28。N4861 是 C++20 时期工作草案，不等于现行最新标准；这里仅使用 release/acquire、线程收拢与 uint64_t 的常规契约。

std::uint64_t 需由实现提供精确64位无符号类型；std::size_t 原子是否无锁由实现决定。Windows 预期 MSVC + 标准库，Linux 预期 GCC/Clang + 配套标准库，macOS 预期 AppleClang + 系统 SDK；均需 C++20 std::jthread 支持。实际版本、CPU/SDK、线程与原子运行库均待记录，没有三平台运行结果。

[WG21 N4861](https://www.open-std.org/jtc1/sc22/wg21/docs/papers/2020/n4861.pdf)、[N4861 atomics.order HTML](https://timsong-cpp.github.io/cppwp/n4861/atomics.order)、[N4861 intro.races HTML](https://timsong-cpp.github.io/cppwp/n4861/intro.races)、[CMake Threads](https://cmake.org/cmake/help/v3.28/module/FindThreads.html)。HTML 是固定草案的可读镜像，权威出处是 WG21 草案。
app/main.cpp（同源文件，示例未执行）
cpp
#include <p2/spsc.hpp>
#include <atomic>
#include <chrono>
#include <cstdint>
#include <exception>
#include <iostream>
#include <thread>

int main() try {
    constexpr std::uint64_t count = 10000;
    p2::SpscRing<64> queue;
    std::atomic<bool> stop{false};
    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
    std::uint64_t received = 0, sum = 0;
    bool ordered = true;
    {
        std::jthread producer([&] {
            for (std::uint64_t value = 0; value < count;) {
                if (stop.load(std::memory_order_relaxed)) return;
                if (queue.try_push(value)) ++value;
                else std::this_thread::yield();
                if (std::chrono::steady_clock::now() >= deadline) {
                    stop.store(true, std::memory_order_relaxed);
                    return;
                }
            }
        });
        while (received < count && !stop.load(std::memory_order_relaxed)) {
            std::uint64_t value = 0;
            if (queue.try_pop(value)) {
                ordered = ordered && value == received;
                sum += value;
                ++received;
            } else std::this_thread::yield();
            if (std::chrono::steady_clock::now() >= deadline) {
                stop.store(true, std::memory_order_relaxed);
            }
        }
        // jthread joins producer before queue and captured locals can die.
    }
    if (stop.load(std::memory_order_relaxed) || !ordered || received != count || sum != 49995000) {
        std::cerr << "transfer failed or deadline exceeded\n";
        return 1;
    }
    std::cout << "received=10000 checksum=49995000\n";
    return 0;
} catch (const std::exception& error) {
    std::cerr << "transfer failed: " << error.what() << '\n';
    return 1;
}
include/p2/spsc.hpp（同源文件，示例未执行）
cpp
#pragma once
#include <array>
#include <atomic>
#include <cstddef>
#include <cstdint>

namespace p2 {
// Exactly one producer and one consumer. Stop/join both before destruction.
// Slots is physical storage; one slot is always reserved to distinguish full.
template<std::size_t Slots>
class SpscRing {
    static_assert(Slots >= 2);
    std::array<std::uint64_t, Slots> values_{};
    std::atomic<std::size_t> write_{0};
    std::atomic<std::size_t> read_{0};
    static constexpr std::size_t next(std::size_t index) noexcept {
        return index == Slots - 1 ? 0 : index + 1;
    }
public:
    static constexpr std::size_t capacity = Slots - 1;
    SpscRing() noexcept = default;
    SpscRing(const SpscRing&) = delete;
    SpscRing& operator=(const SpscRing&) = delete;
    bool try_push(std::uint64_t value) noexcept {
        const auto write = write_.load(std::memory_order_relaxed);
        const auto after = next(write);
        if (after == read_.load(std::memory_order_acquire)) return false;
        values_[write] = value;
        write_.store(after, std::memory_order_release);
        return true;
    }
    bool try_pop(std::uint64_t& value) noexcept {
        const auto read = read_.load(std::memory_order_relaxed);
        if (read == write_.load(std::memory_order_acquire)) return false;
        value = values_[read];
        read_.store(next(read), std::memory_order_release);
        return true;
    }
    bool indices_lock_free() const noexcept {
        return write_.is_lock_free() && read_.is_lock_free();
    }
};
}
tests/spsc_test.cpp（同源文件，示例未执行）
cpp
#include <p2/spsc.hpp>
#include <atomic>
#include <chrono>
#include <cstdint>
#include <exception>
#include <iostream>
#include <thread>

bool boundaries() {
    p2::SpscRing<2> one;
    std::uint64_t out = 99;
    if (one.try_pop(out) || out != 99) return false;
    if (!one.try_push(7) || one.try_push(8)) return false;
    if (!one.try_pop(out) || out != 7 || one.try_pop(out)) return false;
    p2::SpscRing<4> ring;
    for (std::uint64_t round = 0; round != 1000; ++round) {
        for (std::uint64_t i = 0; i != 3; ++i)
            if (!ring.try_push(round * 3 + i)) return false;
        if (ring.try_push(999)) return false;
        for (std::uint64_t i = 0; i != 3; ++i)
            if (!ring.try_pop(out) || out != round * 3 + i) return false;
        if (ring.try_pop(out)) return false;
    }
    return true;
}

bool stress() {
    constexpr std::uint64_t count = 100000;
    p2::SpscRing<17> ring;
    std::atomic<bool> stop{false};
    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
    std::uint64_t received = 0;
    bool ordered = true;
    {
        std::jthread producer([&] {
            for (std::uint64_t value = 0; value != count;) {
                if (stop.load(std::memory_order_relaxed)) return;
                if (ring.try_push(value)) ++value;
                else std::this_thread::yield();
                if (std::chrono::steady_clock::now() >= deadline) {
                    stop.store(true, std::memory_order_relaxed);
                    return;
                }
            }
        });
        std::jthread consumer([&] {
            while (received != count && !stop.load(std::memory_order_relaxed)) {
                std::uint64_t value = 0;
                if (ring.try_pop(value)) {
                    if (value != received) ordered = false;
                    ++received;
                } else std::this_thread::yield();
                if (std::chrono::steady_clock::now() >= deadline) {
                    stop.store(true, std::memory_order_relaxed);
                }
            }
        });
        // Both jthreads join here; only then does main read consumer's locals.
    }
    std::uint64_t remaining = 0;
    return !stop.load(std::memory_order_relaxed) && ordered &&
           received == count && !ring.try_pop(remaining);
}

int main() try {
    if (!boundaries() || !stress()) {
        std::cerr << "spsc contract failed or deadline exceeded\n";
        return 1;
    }
    std::cout << "spsc: boundaries, wrap and ordered stress passed\n";
    return 0;
} catch (const std::exception& error) {
    std::cerr << "spsc contract failed: " << error.what() << '\n';
    return 1;
}

继续阅读