1. Vấn đề nó giải quyết
Một luồng tạo dữ liệu, một luồng khác lấy dữ liệu theo thứ tự FIFO. Hàng đợi vòng SPSC (single producer, single consumer) có thể làm việc này với vùng nhớ cố định và hai chỉ số atomic; phải đồng bộ cả lúc công bố phần tử lẫn lúc cho phép dùng lại ô nhớ.
2. Kiến thức cần có
Biết
std::atomic, memory order acquire/release và quan hệ happens-before: thứ tự bảo đảm một thao tác hoàn tất trước khi thao tác khác được phép truy cập dữ liệu liên quan.Biết vòng đời đối tượng (lifetime), move constructor và destructor.
Hiểu rõ chỉ một luồng được đẩy và chỉ một luồng được lấy; hai phía có thể chạy đồng thời.
3. Ý tưởng cốt lõi
Để dành một ô trống trong mảng N ô (N > 1): head == tail nghĩa là rỗng; next(head) == tail nghĩa là đầy. Dung lượng sử dụng là N - 1.
Chỉ phía ghi thay head, chỉ phía đọc thay tail. Phía ghi dựng phần tử rồi release-store head; phía đọc acquire-load head trước khi lấy phần tử. Phía đọc hủy phần tử rồi release-store tail; phía ghi acquire-load tail trước khi tái sử dụng ô. Hai chiều đồng bộ bảo vệ dữ liệu và vòng đời; xem quy tắc acquire/release.
4. Cú pháp tối thiểu
Đoạn công bố phần tử bên trong try_push:
slots_[current].emplace(std::move(value));
head_.store(following, std::memory_order_release);Nó chỉ chạy sau khi acquire-load tail_ xác nhận còn chỗ. slots_ là mảng std::optional<T>; một ô trống chưa chứa đối tượng T đang sống.
5. Cách nó hoạt động
Phía ghi đọc chỉ số của mình bằng
relaxed, tính chỉ số kế tiếp và acquire-load chỉ số phía đọc. Nếu đầy, trảfalse; nếu còn chỗ, dựng phần tử trước khi công bốheadmới.Phía đọc đọc
tailbằngrelaxed, acquire-loadhead. Nếu rỗng, trảstd::nullopt; nếu có dữ liệu, chuyển phần tử sang kết quả,reset()ô rồi công bốtailmới.Chỉ số luôn nằm trong
[0, N). Mã mẫu dùngindex + 1 == N ? 0 : index + 1; không dựa vào tràn số có dấu hay giả địnhNlà lũy thừa của hai.Mã mẫu yêu cầu
Tcó move constructor và destructor không ném lỗi. Tham sốtry_push(T value)được tạo trước khi vào thân hàm: truyền lvalue có thể sao chép và ném lỗi; truyền rvalue có thể đã làm đổi nguồn ngay cả khi hàng đợi trảfalse.Chỉ hủy hàng đợi sau khi hai phía đã ngừng truy cập. Các
optionalcòn chứa phần tử sẽ hủy chúng.
6. Lỗi thường gặp
Công bố
headtrước khi dựng xongT, hoặc công bốtailtrước khi dùng xong/hủy phần tử.Đổi mọi truy cập atomic thành
relaxed; tính nguyên tử của chỉ số không tự công bố dữ liệu trong ô.Thêm phía ghi/đọc thứ hai mà giữ thuật toán cũ.
Gọi cả thao tác là lock-free chỉ vì không thấy mutex. Atomic chỉ số phải lock-free trên nền tảng đích; thao tác của
Tcũng không được khóa/chặn.noexceptkhông chứng minh điều đó.
7. Khi nào nên dùng
Dùng thiết kế này khi số phía ghi/đọc cố định và có chính sách rõ cho trạng thái đầy: thử lại, bỏ dữ liệu hoặc báo ngược áp lực xử lý. Nó không phù hợp nguyên trạng cho MPSC/MPMC. Đo độ trễ và tranh chấp cache trước khi thêm tối ưu như tách hai chỉ số sang cache line khác nhau.
8. Ví dụ đơn giản
Mã mẫu C++20 dùng SpscRing<int, 4>: có bốn ô vật lý nhưng chỉ chứa tối đa ba số. Nó đẩy 40, 2, rồi lấy ra theo đúng thứ tự đó.
Ví dụ này chỉ minh họa hoạt động tuần tự, chưa chứng minh an toàn đa luồng. Khi kiểm chứng thiết kế, cần thử rỗng/đầy, dung lượng nhỏ nhất, nhiều vòng quay chỉ số, và hai luồng truyền dãy số để phát hiện mất/trùng/đảo thứ tự. Chạy sanitizer phù hợp để tìm data race hoặc lỗi bộ nhớ; kết quả sạch không thay thế lập luận đồng bộ.
Mã mẫu hoàn chỉnh
Tệp mã nguồn
dailycppinterview/254_bounded-lock-free-spsc-ring-buffer/main.cpp
// Real-World C++ Interviews Q254: How would you design and validate a bounded lock-free SPSC
// ring buffer, including full/empty states, object lifetime, wraparound, and acquire-release
// ordering?
// Key: With exactly one producer and one consumer, let the producer own head and the consumer
// own tail. One simple representation reserves a slot: head equal to tail means empty, while
// advancing head to tail means full, so physical size N stores at most N-1 elements. The
// producer constructs a free slot before a release-store to head; the consumer acquire-loads
// head before reading that object, destroys it, then release-stores tail so the producer's
// acquire-load can safely reuse the slot. Index arithmetic must wrap without overflow
// assumptions, and T's construction, move, and destruction contracts must be handled. Test
// capacity boundaries, many wraparounds, long stress runs, and sanitizer-supported misuse; also
// verify that the index atomics are lock-free on the target. The design is not MPSC or MPMC.
#include <array>
#include <atomic>
#include <cstddef>
#include <iostream>
#include <optional>
#include <type_traits>
#include <utility>
template<class T, std::size_t N>
requires (N > 1 && std::is_nothrow_move_constructible_v<T> &&
std::is_nothrow_destructible_v<T>)
class SpscRing {
public:
static constexpr std::size_t capacity = N - 1;
static constexpr bool always_lock_free =
std::atomic<std::size_t>::is_always_lock_free;
bool try_push(T value) noexcept {
const std::size_t current = head_.load(std::memory_order_relaxed);
const std::size_t following = next(current);
if (following == tail_.load(std::memory_order_acquire)) return false;
slots_[current].emplace(std::move(value));
head_.store(following, std::memory_order_release);
return true;
}
std::optional<T> try_pop() noexcept {
const std::size_t current = tail_.load(std::memory_order_relaxed);
if (current == head_.load(std::memory_order_acquire)) return std::nullopt;
std::optional<T> result{std::in_place, std::move(*slots_[current])};
slots_[current].reset();
tail_.store(next(current), std::memory_order_release);
return result;
}
private:
static constexpr std::size_t next(std::size_t index) noexcept {
return index + 1 == N ? 0 : index + 1;
}
std::array<std::optional<T>, N> slots_{};
std::atomic<std::size_t> head_{};
std::atomic<std::size_t> tail_{};
};
int main() {
SpscRing<int, 4> queue;
queue.try_push(40);
queue.try_push(2);
while (const auto value = queue.try_pop()) {
std::cout << *value << std::endl;
}
}
9. Điều cần nhớ
always_lock_free trong mã mẫu cho biết thuộc tính của atomic chỉ số; nó không cưỡng chế nền tảng phải hỗ trợ. Đánh giá thiết kế cần cả lập luận happens-before, quy ước của T và kết quả thử trên nền tảng đích.
10. Câu hỏi tự kiểm tra
Đề đầy đủ: Bạn sẽ thiết kế và kiểm chứng bounded lock-free SPSC ring buffer như thế nào, gồm trạng thái full/empty, object lifetime, wraparound và acquire-release ordering?
Với N = 4, vì sao chỉ chứa ba phần tử? Cặp release/acquire trên tail bảo vệ việc gì? Nếu move constructor của T không ném lỗi nhưng lấy mutex, có thể gọi toàn bộ try_push là lock-free không?