tdlight/tdutils/td/utils/MpmcQueue.h

450 lines
12 KiB
C
Raw Normal View History

//
// Copyright Aliaksei Levin (levlam@telegram.org), Arseny Smirnov (arseny30@gmail.com) 2014-2018
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
#pragma once
// MPMC queue
// Simple semaphore protected implementation
// To close queue, one should send as much sentinel elements as there are readers.
// Once there are no readers and writers, one may easily destroy queue
#include "td/utils/format.h"
#include "td/utils/HazardPointers.h"
#include "td/utils/logging.h"
#include "td/utils/port/thread.h"
#include "td/utils/ScopeGuard.h"
#include <array>
#include <atomic>
namespace td {
namespace detail {
struct MpmcStat {
void alloc_ok(size_t thread_id) {
s(thread_id).alloc_ok_cnt++;
}
void alloc_error(size_t thread_id) {
s(thread_id).alloc_error_cnt++;
}
void push_loop_error(size_t thread_id) {
s(thread_id).push_loop_error_cnt++;
}
void push_loop_ok(size_t thread_id) {
s(thread_id).push_loop_ok_cnt++;
}
void dump() {
int alloc_ok_cnt = 0;
int alloc_error_cnt = 0;
int push_loop_error_cnt = 0;
int push_loop_ok_cnt = 0;
for (auto &d : arr) {
alloc_ok_cnt += d.alloc_ok_cnt;
alloc_error_cnt += d.alloc_error_cnt;
push_loop_error_cnt += d.push_loop_error_cnt;
push_loop_ok_cnt += d.push_loop_ok_cnt;
}
LOG(ERROR) << tag("alloc_ok_cnt", alloc_ok_cnt) << tag("alloc_error_cnt", alloc_error_cnt)
<< tag("push_loop_error_cnt", push_loop_error_cnt) << tag("push_loop_ok_cnt", push_loop_ok_cnt);
}
private:
struct ThreadStat {
int alloc_ok_cnt{0};
int alloc_error_cnt{0};
int push_loop_ok_cnt{0};
int push_loop_error_cnt{0};
char pad[TD_CONCURRENCY_PAD - sizeof(int) * 4];
};
std::array<ThreadStat, 1024> arr;
ThreadStat &s(size_t thread_id) {
return arr[thread_id];
}
};
} // namespace detail
//detail::MpmcStat stat_;
template <class T>
class OneValue {
public:
bool set_value(T &value) {
value_ = std::move(value);
int state = Empty;
if (state_.compare_exchange_strong(state, Value, std::memory_order_acq_rel)) {
return true;
}
value = std::move(value_);
return false;
}
bool get_value(T &value) {
auto old_state = state_.exchange(Taken, std::memory_order_acq_rel);
if (old_state == Value) {
value = std::move(value_);
return true;
}
return false;
}
void reset() {
state_ = Empty;
value_ = T();
}
private:
enum Type : int { Empty = 0, Taken, Value };
std::atomic<int> state_{Empty};
T value_{};
};
template <class T>
class OneValue<T *> {
public:
bool set_value(T *value) {
T *was = nullptr;
return state_.compare_exchange_strong(was, value, std::memory_order_acq_rel);
}
bool get_value(T *&value) {
value = state_.exchange(Taken(), std::memory_order_acq_rel);
return value != nullptr;
}
void reset() {
state_ = nullptr;
}
OneValue() {
}
private:
std::atomic<T *> state_{nullptr};
T *Taken() {
static T xxx;
return &xxx;
}
};
template <class T>
class MpmcQueueBlock {
public:
explicit MpmcQueueBlock(size_t size) : nodes_(size) {
}
enum class PopStatus { Ok, Empty, Closed };
//blocking pop
//returns Ok or Closed
PopStatus pop(T &value) {
while (true) {
auto read_pos = read_pos_.fetch_add(1, std::memory_order_relaxed);
if (read_pos >= nodes_.size()) {
return PopStatus::Closed;
}
//TODO blocking get_value
if (nodes_[static_cast<size_t>(read_pos)].one_value.get_value(value)) {
return PopStatus::Ok;
}
}
}
//nonblocking pop
//returns Ok, Empty or Closed
PopStatus try_pop(T &value) {
while (true) {
auto read_pos = read_pos_.fetch_add(1, std::memory_order_relaxed);
if (read_pos >= nodes_.size()) {
return PopStatus::Closed;
}
if (nodes_[static_cast<size_t>(read_pos)].one_value.get_value(value)) {
return PopStatus::Ok;
}
auto write_pos = write_pos_.load(std::memory_order_relaxed);
if (write_pos <= read_pos + 1) {
return PopStatus::Empty;
}
}
}
enum class PushStatus { Ok, Closed };
PushStatus push(T &value) {
while (true) {
auto write_pos = write_pos_.fetch_add(1, std::memory_order_relaxed);
if (write_pos >= nodes_.size()) {
return PushStatus::Closed;
}
if (nodes_[static_cast<size_t>(write_pos)].one_value.set_value(value)) {
//stat_.push_loop_ok(0);
return PushStatus::Ok;
}
//stat_.push_loop_error(0);
}
}
private:
struct Node {
OneValue<T> one_value;
};
std::atomic<uint64> write_pos_{0};
char pad[TD_CONCURRENCY_PAD - sizeof(std::atomic<uint64>)];
std::atomic<uint64> read_pos_{0};
char pad2[TD_CONCURRENCY_PAD - sizeof(std::atomic<uint64>)];
std::vector<Node> nodes_;
char pad3[TD_CONCURRENCY_PAD - sizeof(std::vector<Node>)];
};
template <class T>
class MpmcQueueOld {
public:
explicit MpmcQueueOld(size_t threads_n) : MpmcQueueOld(1024, threads_n) {
}
static std::string get_description() {
return "Mpmc queue (fetch and add array queue)";
}
MpmcQueueOld(size_t block_size, size_t threads_n) : block_size_{block_size}, hazard_pointers_{threads_n} {
auto node = make_unique<Node>(block_size_);
write_pos_ = node.get();
read_pos_ = node.get();
node.release();
}
MpmcQueueOld(const MpmcQueueOld &other) = delete;
MpmcQueueOld &operator=(const MpmcQueueOld &other) = delete;
MpmcQueueOld(MpmcQueueOld &&other) = delete;
MpmcQueueOld &operator=(MpmcQueueOld &&other) = delete;
~MpmcQueueOld() {
auto *ptr = read_pos_.load(std::memory_order_relaxed);
while (ptr) {
auto *to_delete = ptr;
ptr = ptr->next_.load(std::memory_order_relaxed);
delete to_delete;
}
//stat_.dump();
//stat_ = MpmcStat();
}
size_t hazard_pointers_to_delele_size_unsafe() const {
return hazard_pointers_.to_delete_size_unsafe();
}
void gc(size_t thread_id) {
hazard_pointers_.retire(thread_id);
}
using PushStatus = typename MpmcQueueBlock<T>::PushStatus;
using PopStatus = typename MpmcQueueBlock<T>::PopStatus;
void push(T value, size_t thread_id) {
auto hazard_ptr_holder = hazard_pointers_.get_holder(thread_id, 0);
while (true) {
auto node = hazard_ptr_holder.protect(write_pos_);
auto status = node->block.push(value);
switch (status) {
case PushStatus::Ok:
return;
case PushStatus::Closed: {
auto next = node->next_.load(std::memory_order_acquire);
if (next == nullptr) {
auto new_node = new Node(block_size_);
new_node->block.push(value);
if (node->next_.compare_exchange_strong(next, new_node, std::memory_order_acq_rel)) {
//stat_.alloc_ok(thread_id);
write_pos_.compare_exchange_strong(node, new_node, std::memory_order_acq_rel);
return;
} else {
//stat_.alloc_error(thread_id);
new_node->block.pop(value);
//CHECK(status == PopStatus::Ok);
delete new_node;
}
}
//CHECK(next != nullptr);
write_pos_.compare_exchange_strong(node, next, std::memory_order_acq_rel);
break;
}
}
}
}
bool try_pop(T &value, size_t thread_id) {
auto hazard_ptr_holder = hazard_pointers_.get_holder(thread_id, 0);
while (true) {
auto node = hazard_ptr_holder.protect(read_pos_);
auto status = node->block.try_pop(value);
switch (status) {
case PopStatus::Ok:
return true;
case PopStatus::Empty:
return false;
case PopStatus::Closed: {
auto next = node->next_.load(std::memory_order_acquire);
if (!next) {
return false;
}
if (read_pos_.compare_exchange_strong(node, next, std::memory_order_acq_rel)) {
hazard_ptr_holder.clear();
hazard_pointers_.retire(thread_id, node);
}
break;
}
}
}
}
T pop(size_t thread_id) {
T value;
while (true) {
if (try_pop(value, thread_id)) {
return value;
}
td::this_thread::yield();
}
}
private:
struct Node {
explicit Node(size_t block_size) : block{block_size} {
}
std::atomic<Node *> next_{nullptr};
char pad[TD_CONCURRENCY_PAD - sizeof(std::atomic<Node *>)];
MpmcQueueBlock<T> block;
//Got pad in MpmcQueueBlock
};
std::atomic<Node *> write_pos_;
char pad[TD_CONCURRENCY_PAD - sizeof(std::atomic<Node *>)];
std::atomic<Node *> read_pos_;
char pad2[TD_CONCURRENCY_PAD - sizeof(std::atomic<Node *>)];
size_t block_size_;
HazardPointers<Node, 1> hazard_pointers_;
//Got pad in HazardPointers
};
template <class T>
class MpmcQueue {
public:
explicit MpmcQueue(size_t threads_n) : MpmcQueue(1024, threads_n) {
}
static std::string get_description() {
return "NEW Mpmc queue (fetch and add array queue)";
}
MpmcQueue(size_t block_size, size_t threads_n) : hazard_pointers_{threads_n} {
auto node = make_unique<Node>();
write_pos_ = node.get();
read_pos_ = node.get();
node.release();
}
MpmcQueue(const MpmcQueue &other) = delete;
MpmcQueue &operator=(const MpmcQueue &other) = delete;
MpmcQueue(MpmcQueue &&other) = delete;
MpmcQueue &operator=(MpmcQueue &&other) = delete;
~MpmcQueue() {
auto *ptr = read_pos_.load(std::memory_order_relaxed);
while (ptr) {
auto *to_delete = ptr;
ptr = ptr->next.load(std::memory_order_relaxed);
delete to_delete;
}
}
size_t hazard_pointers_to_delele_size_unsafe() const {
return hazard_pointers_.to_delete_size_unsafe();
}
void gc(size_t thread_id) {
hazard_pointers_.retire(thread_id);
}
void push(T value, size_t thread_id) {
SCOPE_EXIT {
hazard_pointers_.clear(thread_id, 0);
};
while (true) {
auto node = hazard_pointers_.protect(thread_id, 0, write_pos_);
auto &block = node->block;
auto pos = block.write_pos++;
if (pos >= block.data.size()) {
auto next = node->next.load();
if (next == nullptr) {
auto new_node = new Node{};
new_node->block.write_pos++;
new_node->block.data[0].set_value(value);
Node *null = nullptr;
if (node->next.compare_exchange_strong(null, new_node)) {
write_pos_.compare_exchange_strong(node, new_node);
return;
} else {
new_node->block.data[0].get_value(value);
delete new_node;
}
} else {
write_pos_.compare_exchange_strong(node, next);
}
} else {
if (block.data[static_cast<size_t>(pos)].set_value(value)) {
return;
}
}
}
}
bool try_pop(T &value, size_t thread_id) {
SCOPE_EXIT {
hazard_pointers_.clear(thread_id, 0);
};
while (true) {
auto node = hazard_pointers_.protect(thread_id, 0, read_pos_);
auto &block = node->block;
if (block.write_pos <= block.read_pos && node->next.load(std::memory_order_relaxed) == nullptr) {
return false;
}
auto pos = block.read_pos++;
if (pos >= block.data.size()) {
auto next = node->next.load();
if (!next) {
return false;
}
if (read_pos_.compare_exchange_strong(node, next)) {
hazard_pointers_.clear(thread_id, 0);
hazard_pointers_.retire(thread_id, node);
}
} else {
if (block.data[static_cast<size_t>(pos)].get_value(value)) {
return true;
}
}
}
}
T pop(size_t thread_id) {
T value;
while (true) {
if (try_pop(value, thread_id)) {
return value;
}
td::this_thread::yield();
}
}
private:
struct Block {
std::atomic<uint64> write_pos{0};
char pad[TD_CONCURRENCY_PAD - sizeof(std::atomic<uint64>)];
std::atomic<uint64> read_pos{0};
char pad2[TD_CONCURRENCY_PAD - sizeof(std::atomic<uint64>)];
std::array<OneValue<T>, 1024> data;
char pad3[TD_CONCURRENCY_PAD];
};
struct Node {
Node() = default;
Block block;
std::atomic<Node *> next{nullptr};
char pad[TD_CONCURRENCY_PAD - sizeof(std::atomic<Node *>)];
//Got pad in MpmcQueueBlock
};
std::atomic<Node *> write_pos_;
char pad[TD_CONCURRENCY_PAD - sizeof(std::atomic<Node *>)];
std::atomic<Node *> read_pos_;
char pad2[TD_CONCURRENCY_PAD - sizeof(std::atomic<Node *>)];
HazardPointers<Node, 1> hazard_pointers_;
//Got pad in HazardPointers
};
} // namespace td