lock-free queue及micro-batching¶
优秀博客：https://moodycamel.com/blog/2014/a-fast-general-purpose-lock-free-queue-for-c++.htm
无锁操作：
1.LOCK 前缀指令确保单个操作的原子性
2.MESI 协议维护多核心数据一致性
3.内存屏障保障操作的可见性与有序性
简单spsc实现版本，高级多模式版本在简单实践处可见，因与初始版本兼容，所以没有实现阻塞获取消费位置的接口，只实现了防止生产者溢出覆盖旧数据的情况。
#pragma once
#include <immintrin.h>
#include <atomic>
#include <iostream>
#include <vector>
#include "macros.h"
namespace Common {
template <typename T>
class LFQueue final {
public:
explicit LFQueue(std::size_t num_elems)
: store_(round_up_to_power_of_2(num_elems), T()),
mask_(store_.size() - 1),
capacity_(store_.size()) {}
auto tryGetNextToWriteTo() noexcept -> T* {
auto current_write = next_write_index_.load(std::memory_order_relaxed);
auto current_read = next_read_index_.load(std::memory_order_acquire);
if (UNLIKELY((current_write + 1) & mask_ == current_read & mask_)) {
return nullptr;
}
return &store_[current_write & mask_];
}
auto getNextToWriteTo() noexcept -> T* {
while (true) {
auto slot = tryGetNextToWriteTo();
if (LIKELY(slot != nullptr)) {
return slot;
}
_mm_pause();
}
}
auto updateWriteIndex() noexcept {
auto current_write_index =
next_write_index_.load(std::memory_order_relaxed);
next_write_index_.store(current_write_index + 1, std::memory_order_release);
num_elements_.fetch_add(1, std::memory_order_release);
}
auto getNextToRead() const noexcept -> const T* {
auto current_read_index = next_read_index_.load(std::memory_order_relaxed);
auto current_element_count = num_elements_.load(std::memory_order_acquire);
if (LIKELY(current_element_count > 0)) {
std::size_t target_index = current_read_index & mask_;
return &store_[target_index];
} else {
return nullptr;
}
}
auto updateReadIndex() noexcept {
auto current_read_index = next_read_index_.load(std::memory_order_relaxed);
next_read_index_.store(current_read_index + 1, std::memory_order_release);
num_elements_.fetch_sub(1, std::memory_order_release);
}
auto size() const noexcept {
return num_elements_.load(std::memory_order_acquire);
}
// auto size() const
// {
// size_t h = next_write_index_.load(std::memory_order_acquire);
// size_t t = next_read_index_.load(std::memory_order_acquire);
// return (h + capacity_ - t) & mask_;
// }
auto is_full() const noexcept -> bool {
auto current_write = next_write_index_.load(std::memory_order_relaxed);
auto current_read = next_read_index_.load(std::memory_order_relaxed);
return (current_write + 1) & mask_ == current_read & mask_;
}
auto capacity() const noexcept -> std::size_t { return capacity_; }
LFQueue() = delete;
LFQueue(const LFQueue&) = delete;
LFQueue(const LFQueue&&) = delete;
LFQueue& operator=(const LFQueue&) = delete;
LFQueue& operator=(const LFQueue&&) = delete;
private:
static std::size_t round_up_to_power_of_2(std::size_t v) {
if (UNLIKELY(v == 0)) return 1;
--v;
v |= v >> 1;
v |= v >> 2;
v |= v >> 4;
v |= v >> 8;
v |= v >> 16;
v |= v >> 32;
++v;
return v;
}
std::vector<T> store_;
const std::size_t mask_;
const std::size_t capacity_;
alignas(64) std::atomic<std::size_t> next_write_index_{0};
alignas(64) std::atomic<std::size_t> next_read_index_{0};
alignas(64) std::atomic<std::size_t> num_elements_{0};
};
} // namespace Common
基于此的日志服务
#pragma once
#include <cstdio>
#include <fstream>
#include <string>
#include "lf_queue.h"
#include "macros.h"
#include "thread_utils.h"
#include "time_utils.h"
namespace Common {
/// Maximum size of the lock free queue of data to be logged.
constexpr size_t LOG_QUEUE_SIZE = 8 * 1024 * 1024;
/// Type of LogElement message.
enum class LogType : int8_t {
CHAR = 0,
INTEGER = 1,
LONG_INTEGER = 2,
LONG_LONG_INTEGER = 3,
UNSIGNED_INTEGER = 4,
UNSIGNED_LONG_INTEGER = 5,
UNSIGNED_LONG_LONG_INTEGER = 6,
FLOAT = 7,
DOUBLE = 8,
STRING = 9
};
/// Represents a single and primitive log entry.
struct LogElement {
LogType type_ = LogType::CHAR;
union {
char c;
int i;
long l;
long long ll;
unsigned u;
unsigned long ul;
unsigned long long ull;
float f;
double d;
char s[256];
} u_;
};
class Logger final {
public:
/// Consumes from the lock free queue of log entries and writes to the output
/// log file.
auto flushQueue() noexcept {
while (running_) {
for (auto next = queue_.getNextToRead(); queue_.size() && next;
next = queue_.getNextToRead()) {
switch (next->type_) {
case LogType::CHAR:
file_ << next->u_.c;
break;
case LogType::INTEGER:
file_ << next->u_.i;
break;
case LogType::LONG_INTEGER:
file_ << next->u_.l;
break;
case LogType::LONG_LONG_INTEGER:
file_ << next->u_.ll;
break;
case LogType::UNSIGNED_INTEGER:
file_ << next->u_.u;
break;
case LogType::UNSIGNED_LONG_INTEGER:
file_ << next->u_.ul;
break;
case LogType::UNSIGNED_LONG_LONG_INTEGER:
file_ << next->u_.ull;
break;
case LogType::FLOAT:
file_ << next->u_.f;
break;
case LogType::DOUBLE:
file_ << next->u_.d;
break;
case LogType::STRING:
file_ << next->u_.s;
break;
}
queue_.updateReadIndex();
}
file_.flush();
using namespace std::literals::chrono_literals;
std::this_thread::sleep_for(10ms);
}
}
explicit Logger(const std::string& file_name)
: file_name_(file_name), queue_(LOG_QUEUE_SIZE) {
file_.open(file_name);
ASSERT(file_.is_open(), "Could not open log file:" + file_name);
logger_thread_ = Common::createAndStartThread(
-1, "Common/Logger " + file_name_, [this]() { flushQueue(); });
ASSERT(logger_thread_ != nullptr, "Failed to start Logger thread.");
}
~Logger() {
std::string time_str;
std::cerr << Common::getCurrentTimeStr(&time_str)
<< " Flushing and closing Logger for " << file_name_ << std::endl;
while (queue_.size()) {
using namespace std::literals::chrono_literals;
std::this_thread::sleep_for(1s);
}
running_ = false;
logger_thread_->join();
file_.close();
std::cerr << Common::getCurrentTimeStr(&time_str) << " Logger for "
<< file_name_ << " exiting." << std::endl;
}
/// Overloaded methods to write different log entry types to the lock free
/// queue. Creates a LogElement of the correct type and writes it to the lock
/// free queue.
auto pushValue(const LogElement& log_element) noexcept {
*(queue_.getNextToWriteTo()) = log_element;
queue_.updateWriteIndex();
}
auto pushValue(const char value) noexcept {
pushValue(LogElement{LogType::CHAR, {.c = value}});
}
auto pushValue(const int value) noexcept {
pushValue(LogElement{LogType::INTEGER, {.i = value}});
}
auto pushValue(const long value) noexcept {
pushValue(LogElement{LogType::LONG_INTEGER, {.l = value}});
}
auto pushValue(const long long value) noexcept {
pushValue(LogElement{LogType::LONG_LONG_INTEGER, {.ll = value}});
}
auto pushValue(const unsigned value) noexcept {
pushValue(LogElement{LogType::UNSIGNED_INTEGER, {.u = value}});
}
auto pushValue(const unsigned long value) noexcept {
pushValue(LogElement{LogType::UNSIGNED_LONG_INTEGER, {.ul = value}});
}
auto pushValue(const unsigned long long value) noexcept {
pushValue(LogElement{LogType::UNSIGNED_LONG_LONG_INTEGER, {.ull = value}});
}
auto pushValue(const float value) noexcept {
pushValue(LogElement{LogType::FLOAT, {.f = value}});
}
auto pushValue(const double value) noexcept {
pushValue(LogElement{LogType::DOUBLE, {.d = value}});
}
auto pushValue(const char* value) noexcept {
LogElement l{LogType::STRING, {.s = {}}};
strncpy(l.u_.s, value, sizeof(l.u_.s) - 1);
pushValue(l);
}
auto pushValue(const std::string& value) noexcept {
pushValue(value.c_str());
}
/// Parse the format string, substitute % with the variable number of
/// arguments passed and write the string to the lock free queue.
template <typename T, typename... A>
auto log(const char* s, const T& value, A... args) noexcept {
while (*s) {
if (*s == '%') {
if (UNLIKELY(*(s + 1) == '%')) { // to allow %% -> % escape character.
++s;
} else {
pushValue(value); // substitute % with the value specified in the
// arguments.
log(s + 1, args...); // pop an argument and call self recursively.
return;
}
}
pushValue(*s++);
}
FATAL("extra arguments provided to log()");
}
/// Overload for case where no substitution in the string is necessary.
/// Note that this is overloading not specialization. gcc does not allow
/// inline specializations.
auto log(const char* s) noexcept {
while (*s) {
if (*s == '%') {
if (UNLIKELY(*(s + 1) == '%')) { // to allow %% -> % escape character.
++s;
} else {
FATAL("missing arguments to log()");
}
}
pushValue(*s++);
}
}
/// Deleted default, copy & move constructors and assignment-operators.
Logger() = delete;
Logger(const Logger&) = delete;
Logger(const Logger&&) = delete;
Logger& operator=(const Logger&) = delete;
Logger& operator=(const Logger&&) = delete;
private:
/// File to which the log entries will be written.
const std::string file_name_;
std::ofstream file_;
/// Lock free queue of log elements from main logging thread to background
/// formatting and disk writer thread.
Common::LFQueue<LogElement> queue_;
std::atomic<bool> running_ = {true};
/// Background logging thread.
std::thread* logger_thread_ = nullptr;
};
} // namespace Common
微批处理（Micro-Batching）的核心思想是：将多个消息攒成一批处理，减少单次处理的固定成本（消息处理和计算）。但批处理过大会增加延迟，过小则无法发挥批量优势。因此，这个处理器通过动态调整批大小，根据当前消息队列的积压量（backlog）自动选择最优策略：
- 队列积压少时，用小批量减少延迟；
- 队列积压多时，用大批量提高吞吐量。
以下代码需要QueueType有enqueue和dequeue接口
#include <atomic>
#include <chrono>
#include <functional>
#include <memory>
#include <thread>
#include <vector>
#ifdef _MSC_VER
#include <intrin.h>
#endif
template <typename Message, typename QueueType>
class MicroBatchProcessor {
private:
// Configuration constants determined at compile time for performance
static constexpr size_t MIN_BATCH_SIZE = 1; // Single message processing
static constexpr size_t NORMAL_BATCH_SIZE = 100; // Normal batching
static constexpr size_t MAX_BATCH_SIZE = 1000; // Aggressive batching
// Thresholds for determining queue state
static constexpr size_t LOW_BACKLOG_THRESHOLD = 10;
static constexpr size_t HIGH_BACKLOG_THRESHOLD = 100;
// Time thresholds in microseconds
static constexpr int LOW_BACKLOG_TIMEOUT_US = 100;
static constexpr int HIGH_BACKLOG_TIMEOUT_US = 1000;
std::atomic<size_t> message_queue_size{0};
std::atomic<bool> should_stop{false};
// Processing function provided by user
std::function<void(const std::vector<Message>&)> processor_func;
// Reference to external lock-free queue
QueueType& message_queue;
public:
explicit MicroBatchProcessor(std::function <
void(const std::vector<Message>& >) func,
QueueType& queue)
: processor_func(std::move(func)), message_queue(queue) {}
// Method to add messages to the processor
void add_message(const Message& msg) {
message_queue.enqueue(msg);
message_queue_size.fetch_add(1, std::memory_order_relaxed);
}
// Main processing loop - implements the dynamic micro-batching strategy
void run() {
std::vector<Message> batch_buffer;
batch_buffer.reserve(MAX_BATCH_SIZE);
while (!should_stop.load(std::memory_order_relaxed)) {
size_t current_backlog =
message_queue_size.load(std::memory_order_relaxed);
// Determine batch size based on backlog (dynamic adjustment)
size_t target_batch_size = determine_batch_size(current_backlog);
// Collect messages for batching
collect_messages(batch_buffer, target_batch_size);
// Process the batch if we have messages
if (!batch_buffer.empty()) {
processor_func(batch_buffer);
message_queue_size.fetch_sub(batch_buffer.size(),
std::memory_order_relaxed);
batch_buffer.clear();
}
// Small yield to prevent busy waiting
if (current_backlog < LOW_BACKLOG_THRESHOLD) {
// Use PAUSE instruction instead of sleep to reduce spin-wait penalties
#ifdef _MSC_VER
_mm_pause();
#else
__builtin_ia32_pause();
#endif
}
}
}
// Signal the processor to stop
void stop() { should_stop.store(true, std::memory_order_relaxed); }
private:
// Determine optimal batch size based on current queue backlog
size_t determine_batch_size(size_t backlog) const {
if (backlog <= LOW_BACKLOG_THRESHOLD) {
// No backlog - process immediately with minimal batch size
return MIN_BATCH_SIZE;
} else if (backlog <= HIGH_BACKLOG_THRESHOLD) {
// Small backlog - use normal batch size to improve throughput
return NORMAL_BATCH_SIZE;
} else {
// Large backlog - use aggressive batching to maximize throughput
// We're already behind, so minimize per-message overhead
return MAX_BATCH_SIZE;
}
}
// Collect messages into the batch buffer up to target size
void collect_messages(std::vector<Message>& buffer, size_t target_size) {
// Collect messages from the external lock-free queue
size_t collected = 0;
Message msg;
while (collected < target_size && message_queue.dequeue(msg)) {
buffer.push_back(msg);
collected++;
}
}
};