1. Vấn đề nó giải quyết
Một producer và một consumer có thể cần channel bounded, low-latency mà không block bằng mutex. Ring buffer với index có owner riêng có thể cài concurrency contract hẹp này.
2. Kiến thức cần có
Compiler hỗ trợ C++17, được gọi với warning đầy đủ, cùng các bài trước theo thứ tự của khóa học.
Biết ring buffer, atomic, acquire/release publication, cache contention và tradeoff fixed capacity.
3. Ý tưởng cốt lõi
Chỉ producer ghi head index và chỉ consumer ghi tail. Release store publish slot đã hoàn tất hoặc removal đã xong; acquire load tương ứng ngăn một phía reuse slot trước khi work phía kia visible.
4. Cú pháp tối thiểu
buffer[head] = value;
head_.store(next, std::memory_order_release);
const auto head = head_.load(std::memory_order_acquire);5. Cách nó hoạt động
Producer push bốn integer cố định vào bounded ring trong khi một consumer pop đúng bốn.
Một slot để trống nhằm phân biệt full với empty, còn operation fail sẽ yield tới khi có progress.
Chương trình in
received: 10 20 30 40, tạo test oracle nhỏ để so sánh với dự đoán trước khi biên dịch.
6. Lỗi thường gặp
Thêm producer hoặc consumer thứ hai phá assumption về index ownership; wraparound, destruction và non-trivial element lifetime cũng cần design đầy đủ hơn.
Build thành công chưa đủ; phải kiểm tra lifetime, ordering và error path.
7. Khi nào nên dùng
Nên dùng kỹ thuật này khi topology đúng một producer một consumer, capacity bounded và latency measurement chứng minh lock-free complexity đáng giá.
Nên chọn cách C++11/14 đơn giản hơn nếu quy tắc C++17 không cải thiện safety, độ rõ hoặc hiệu năng đã đo trên các toolchain phải hỗ trợ.
8. Ví dụ đơn giản
Ví dụ dùng integer nên slot lifetime đơn giản. Thread join và fixed count tránh cần shutdown protocol riêng.
Mã mẫu hoàn chỉnh
Tệp mã nguồn
cpp17/46_lock_free_spsc_queue/main.cpp
#include <array>
#include <atomic>
#include <cstddef>
#include <iostream>
#include <thread>
template <class T, std::size_t Capacity>
class SpscQueue {
static_assert(Capacity > 1);
public:
bool push(T value) {
const auto head = head_.load(std::memory_order_relaxed);
const auto next = (head + 1) % Capacity;
if (next == tail_.load(std::memory_order_acquire)) return false;
buffer_[head] = value;
head_.store(next, std::memory_order_release);
return true;
}
bool pop(T& value) {
const auto tail = tail_.load(std::memory_order_relaxed);
if (tail == head_.load(std::memory_order_acquire)) return false;
value = buffer_[tail];
tail_.store((tail + 1) % Capacity, std::memory_order_release);
return true;
}
private:
std::array<T, Capacity> buffer_{};
std::atomic<std::size_t> head_{0};
std::atomic<std::size_t> tail_{0};
};
int main() {
SpscQueue<int, 5> queue;
const std::array<int, 4> sent{{10, 20, 30, 40}};
std::array<int, 4> received{};
std::thread producer([&] {
for (int value : sent) {
while (!queue.push(value)) std::this_thread::yield();
}
});
std::thread consumer([&] {
for (int& value : received) {
while (!queue.pop(value)) std::this_thread::yield();
}
});
producer.join();
consumer.join();
std::cout << "received:";
for (int value : received) std::cout << ' ' << value;
std::cout << '\n';
}
9. Điều cần nhớ
Lock-free nghĩa là system progress không cần lock, không tự động bảo đảm wait-freedom, fairness hay nhanh hơn.
Phải chọn rõ chế độ C++17; default của compiler mới hơn có thể che lỗi portability.
Warning và ví dụ xác định biến quy tắc ghi nhớ thành bằng chứng.
10. Câu hỏi tự kiểm tra
Dễ — Lock-free SPSC queue giải quyết vấn đề chính nào?
Trung bình — Vì sao ring bốn slot chỉ giữ tối đa ba queued value trong design này?
Khó — Happens-before edge nào bảo vệ việc đọc slot sau khi producer ghi?