#include <fcntl.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
#include <algorithm>
#include <array>
#include <atomic>
#include <chrono>
#include <cmath>
#include <cstring>
#include <thread>
#include <vector>
#include "macros.h"
// Basic types and constants
namespace Common {
using timestamp_t = uint64_t;
using checksum_t = uint32_t;
using order_id_t = int;
using quantity_t = int;
using price_t = double;
constexpr uint32_t BOOK_MAGIC_NUMBER = 0x424F4F4B; // "BOOK"
constexpr uint16_t BOOK_VERSION = 1;
constexpr size_t LOG_BUFFER_SIZE = 1048576; // 1M entries
constexpr size_t SNAPSHOT_BUFFER_SIZE = 1024;
// Order structure
struct Order {
order_id_t id;
price_t price;
quantity_t quantity;
bool is_bid;
timestamp_t timestamp;
checksum_t checksum;
// Constructors
Order()
: id(0),
price(0.0),
quantity(0),
is_bid(true),
timestamp(0),
checksum(0) {}
Order(order_id_t id, price_t price, quantity_t quantity, bool is_bid,
timestamp_t ts)
: id(id),
price(price),
quantity(quantity),
is_bid(is_bid),
timestamp(ts) {
checksum = calculate_checksum();
}
// Calculate checksum for data integrity
checksum_t calculate_checksum() const {
checksum_t sum = 0;
sum += static_cast<checksum_t>(id);
sum += static_cast<checksum_t>(*reinterpret_cast<const uint64_t*>(&price));
sum += static_cast<checksum_t>(quantity);
sum += static_cast<checksum_t>(is_bid ? 1 : 0);
sum += static_cast<checksum_t>(timestamp);
return sum;
}
// Verify checksum
bool verify_checksum() const { return checksum == calculate_checksum(); }
};
// Level data for market data queries
struct LevelData {
price_t price;
quantity_t total_quantity;
LevelData() : price(0.0), total_quantity(0) {}
LevelData(price_t p, quantity_t q) : price(p), total_quantity(q) {}
};
// Order log structure
struct OrderLog {
enum Action { ADD, UPDATE, DELETE, SNAPSHOT } action;
order_id_t order_id;
price_t price;
quantity_t quantity;
bool is_bid;
timestamp_t timestamp;
int price_level_idx;
checksum_t checksum;
OrderLog() = default;
OrderLog(Action a, const Order& o, int idx)
: action(a),
order_id(o.id),
price(o.price),
quantity(o.quantity),
is_bid(o.is_bid),
timestamp(o.timestamp),
price_level_idx(idx) {
calculate_checksum();
}
void calculate_checksum() {
checksum = 0;
checksum += static_cast<checksum_t>(action);
checksum += static_cast<checksum_t>(order_id);
checksum +=
static_cast<checksum_t>(*reinterpret_cast<const uint64_t*>(&price));
checksum += static_cast<checksum_t>(quantity);
checksum += static_cast<checksum_t>(is_bid ? 1 : 0);
checksum += static_cast<checksum_t>(timestamp);
checksum += static_cast<checksum_t>(price_level_idx);
}
bool verify_checksum() const {
checksum_t sum = 0;
sum += static_cast<checksum_t>(action);
sum += static_cast<checksum_t>(order_id);
sum += static_cast<checksum_t>(*reinterpret_cast<const uint64_t*>(&price));
sum += static_cast<checksum_t>(quantity);
sum += static_cast<checksum_t>(is_bid ? 1 : 0);
sum += static_cast<checksum_t>(timestamp);
sum += static_cast<checksum_t>(price_level_idx);
return sum == checksum;
}
};
// Price level structure
template <typename OrderContainer>
struct PriceLevel {
OrderContainer orders;
price_t price;
quantity_t total_quantity;
checksum_t checksum;
PriceLevel() : price(0.0), total_quantity(0), checksum(0) {}
void reset() {
orders.clear();
price = 0.0;
total_quantity = 0;
checksum = 0;
}
void add_order(const Order& order) {
orders.push_back(order);
total_quantity += order.quantity;
price = order.price;
calculate_checksum();
}
bool update_order(const Order& new_order) {
for (auto& order : orders) {
if (order.id == new_order.id) {
total_quantity += (new_order.quantity - order.quantity);
order.quantity = new_order.quantity;
order.timestamp = new_order.timestamp;
order.checksum = new_order.calculate_checksum();
calculate_checksum();
return true;
}
}
return false;
}
bool delete_order(order_id_t order_id) {
auto it =
std::remove_if(orders.begin(), orders.end(),
[order_id](const Order& o) { return o.id == order_id; });
if (it == orders.end()) return false;
total_quantity -= it->quantity;
orders.erase(it, orders.end());
calculate_checksum();
return true;
}
void calculate_checksum() {
checksum = 0;
checksum +=
static_cast<checksum_t>(*reinterpret_cast<const uint64_t*>(&price));
checksum += static_cast<checksum_t>(total_quantity);
checksum += static_cast<checksum_t>(orders.size());
size_t count = 0;
for (const auto& order : orders) {
checksum += order.checksum;
if (++count >= 5) break;
}
}
bool verify_checksum() const {
checksum_t sum = 0;
sum += static_cast<checksum_t>(*reinterpret_cast<const uint64_t*>(&price));
sum += static_cast<checksum_t>(total_quantity);
sum += static_cast<checksum_t>(orders.size());
size_t count = 0;
for (const auto& order : orders) {
sum += order.checksum;
if (++count >= 5) break;
}
return sum == checksum;
}
};
// Price level snapshot
struct PriceLevelSnapshot {
price_t price;
quantity_t total_quantity;
int order_count;
checksum_t checksum;
PriceLevelSnapshot()
: price(0.0), total_quantity(0), order_count(0), checksum(0) {}
template <typename OrderContainer>
PriceLevelSnapshot(const PriceLevel<OrderContainer>& level)
: price(level.price),
total_quantity(level.total_quantity),
order_count(static_cast<int>(level.orders.size())) {
calculate_checksum();
}
void calculate_checksum() {
checksum = 0;
checksum +=
static_cast<checksum_t>(*reinterpret_cast<const uint64_t*>(&price));
checksum += static_cast<checksum_t>(total_quantity);
checksum += static_cast<checksum_t>(order_count);
}
bool verify_checksum() const {
checksum_t sum = 0;
sum += static_cast<checksum_t>(*reinterpret_cast<const uint64_t*>(&price));
sum += static_cast<checksum_t>(total_quantity);
sum += static_cast<checksum_t>(order_count);
return sum == checksum;
}
};
// Book snapshot header
struct BookSnapshotHeader {
uint32_t magic_number;
uint16_t version;
timestamp_t timestamp;
int ptr_bid_ini_idx;
int ptr_bid_end_idx;
int ptr_offer_ini_idx;
int ptr_offer_end_idx;
size_t bid_level_count;
size_t offer_level_count;
size_t total_order_count;
size_t data_size;
checksum_t checksum;
BookSnapshotHeader()
: magic_number(BOOK_MAGIC_NUMBER),
version(BOOK_VERSION),
timestamp(0),
ptr_bid_ini_idx(-1),
ptr_bid_end_idx(-1),
ptr_offer_ini_idx(-1),
ptr_offer_end_idx(-1),
bid_level_count(0),
offer_level_count(0),
total_order_count(0),
data_size(0),
checksum(0) {}
template <typename Container>
BookSnapshotHeader(const struct BookSnapshot<Container>& snapshot)
: magic_number(BOOK_MAGIC_NUMBER),
version(BOOK_VERSION),
timestamp(snapshot.timestamp),
ptr_bid_ini_idx(snapshot.ptr_bid_ini_idx),
ptr_bid_end_idx(snapshot.ptr_bid_end_idx),
ptr_offer_ini_idx(snapshot.ptr_offer_ini_idx),
ptr_offer_end_idx(snapshot.ptr_offer_end_idx),
bid_level_count(snapshot.bids.size()),
offer_level_count(snapshot.offers.size()),
total_order_count(snapshot.total_order_count) {
data_size = sizeof(BookSnapshotHeader) +
(bid_level_count * sizeof(PriceLevelSnapshot)) +
(offer_level_count * sizeof(PriceLevelSnapshot)) +
(total_order_count * sizeof(OrderLog));
calculate_checksum();
}
void calculate_checksum() {
checksum = 0;
checksum += magic_number;
checksum += static_cast<checksum_t>(version);
checksum += static_cast<checksum_t>(timestamp);
checksum += static_cast<checksum_t>(ptr_bid_ini_idx);
checksum += static_cast<checksum_t>(ptr_bid_end_idx);
checksum += static_cast<checksum_t>(ptr_offer_ini_idx);
checksum += static_cast<checksum_t>(ptr_offer_end_idx);
checksum += static_cast<checksum_t>(bid_level_count);
checksum += static_cast<checksum_t>(offer_level_count);
checksum += static_cast<checksum_t>(total_order_count);
checksum += static_cast<checksum_t>(data_size);
}
bool verify() const {
if (magic_number != BOOK_MAGIC_NUMBER) return false;
if (version != BOOK_VERSION) return false;
checksum_t sum = 0;
sum += magic_number;
sum += static_cast<checksum_t>(version);
sum += static_cast<checksum_t>(timestamp);
sum += static_cast<checksum_t>(ptr_bid_ini_idx);
sum += static_cast<checksum_t>(ptr_bid_end_idx);
sum += static_cast<checksum_t>(ptr_offer_ini_idx);
sum += static_cast<checksum_t>(ptr_offer_end_idx);
sum += static_cast<checksum_t>(bid_level_count);
sum += static_cast<checksum_t>(offer_level_count);
sum += static_cast<checksum_t>(total_order_count);
sum += static_cast<checksum_t>(data_size);
return sum == checksum;
}
};
// Book snapshot
template <typename Container>
struct BookSnapshot {
timestamp_t timestamp;
std::vector<PriceLevelSnapshot> bids;
std::vector<PriceLevelSnapshot> offers;
size_t total_order_count;
int ptr_bid_ini_idx;
int ptr_bid_end_idx;
int ptr_offer_ini_idx;
int ptr_offer_end_idx;
BookSnapshot()
: timestamp(0),
total_order_count(0),
ptr_bid_ini_idx(-1),
ptr_bid_end_idx(-1),
ptr_offer_ini_idx(-1),
ptr_offer_end_idx(-1) {}
template <typename T>
BookSnapshot(timestamp_t ts, const class LimitOrderBook<T>& book);
};
// Limit order book base class
template <typename Container>
class LimitOrderBook {
protected:
Container bids;
Container offers;
int precision;
int depth;
double step_value;
std::atomic<PriceLevel<typename Container::value_type::OrderContainer>*>
ptr_bid_ini;
std::atomic<PriceLevel<typename Container::value_type::OrderContainer>*>
ptr_bid_end;
std::atomic<PriceLevel<typename Container::value_type::OrderContainer>*>
ptr_offer_ini;
std::atomic<PriceLevel<typename Container::value_type::OrderContainer>*>
ptr_offer_end;
int price_to_index(price_t price, bool is_bid, bool update_pointers = true) {
using LevelType =
PriceLevel<typename Container::value_type::OrderContainer>;
Container& target = is_bid ? bids : offers;
std::atomic<LevelType*>& ptr_ini = is_bid ? ptr_bid_ini : ptr_offer_ini;
std::atomic<LevelType*>& ptr_end = is_bid ? ptr_bid_end : ptr_offer_end;
LevelType* current_ini = ptr_ini.load(std::memory_order_acquire);
if (current_ini == nullptr) {
if (update_pointers) {
LevelType* new_ini = &target[0];
new_ini->price = price;
ptr_ini.store(new_ini, std::memory_order_release);
ptr_end.store(new_ini, std::memory_order_release);
}
return 0;
}
LevelType* current_end = ptr_end.load(std::memory_order_acquire);
if ((is_bid && price >= current_ini->price &&
price <= current_end->price) ||
(!is_bid && price >= current_ini->price &&
price <= current_end->price)) {
int qty_steps =
static_cast<int>((price - current_ini->price) * step_value);
int index = (current_ini - &target[0]) + qty_steps;
return (index >= 0 && index < depth) ? index : -1;
}
double range = std::abs(std::max(current_end->price, price) -
std::min(current_ini->price, price));
if (range * step_value < depth) {
int qty_steps = static_cast<int>(
std::round((price - current_end->price) * step_value));
int cycles = std::abs(qty_steps) / depth;
if ((is_bid && price > current_end->price && cycles >= 1) ||
(!is_bid && price > current_end->price && cycles >= 1)) {
if (update_pointers) {
LevelType* new_ini = &target[0];
new_ini->reset();
new_ini->price = price;
while (!ptr_ini.compare_exchange_weak(current_ini, new_ini,
std::memory_order_release,
std::memory_order_acquire)) {
}
ptr_end.store(new_ini, std::memory_order_release);
}
return 0;
}
int current_ini_idx = current_ini - &target[0];
int current_end_idx = current_end - &target[0];
int new_ini_idx =
((qty_steps >= 0 ? current_ini_idx : current_end_idx) + qty_steps) %
depth;
int new_end_idx =
((qty_steps >= 0 ? current_end_idx : current_ini_idx) + qty_steps) %
depth;
new_ini_idx = (new_ini_idx + depth) % depth;
new_end_idx = (new_end_idx + depth) % depth;
if (update_pointers) {
if (price < current_ini->price) {
LevelType* expected = current_ini;
LevelType* desired = &target[new_ini_idx];
ptr_ini.compare_exchange_strong(expected, desired,
std::memory_order_release,
std::memory_order_acquire);
}
if (price > current_end->price) {
LevelType* expected = current_end;
LevelType* desired = &target[new_end_idx];
desired->price = price;
ptr_end.compare_exchange_strong(expected, desired,
std::memory_order_release,
std::memory_order_acquire);
}
}
return price < current_ini->price ? new_ini_idx : new_end_idx;
}
return -1;
}
public:
LimitOrderBook(int precision, int depth)
: precision(precision), depth(depth) {
step_value = std::pow(10, precision);
bids.resize(depth);
offers.resize(depth);
ptr_bid_ini.store(nullptr, std::memory_order_relaxed);
ptr_offer_ini.store(nullptr, std::memory_order_relaxed);
ptr_bid_end.store(nullptr, std::memory_order_relaxed);
ptr_offer_end.store(nullptr, std::memory_order_relaxed);
}
LimitOrderBook(const LimitOrderBook& other)
: precision(other.precision),
depth(other.depth),
step_value(other.step_value),
bids(other.bids),
offers(other.offers) {
ptr_bid_ini.store(other.ptr_bid_ini.load()
? &bids[other.ptr_bid_ini.load() - &other.bids[0]]
: nullptr,
std::memory_order_relaxed);
ptr_bid_end.store(other.ptr_bid_end.load()
? &bids[other.ptr_bid_end.load() - &other.bids[0]]
: nullptr,
std::memory_order_relaxed);
ptr_offer_ini.store(
other.ptr_offer_ini.load()
? &offers[other.ptr_offer_ini.load() - &other.offers[0]]
: nullptr,
std::memory_order_relaxed);
ptr_offer_end.store(
other.ptr_offer_end.load()
? &offers[other.ptr_offer_end.load() - &other.offers[0]]
: nullptr,
std::memory_order_relaxed);
}
LimitOrderBook& operator=(const LimitOrderBook& other) {
if (this != &other) {
precision = other.precision;
depth = other.depth;
step_value = other.step_value;
bids = other.bids;
offers = other.offers;
ptr_bid_ini.store(other.ptr_bid_ini.load()
? &bids[other.ptr_bid_ini.load() - &other.bids[0]]
: nullptr,
std::memory_order_relaxed);
ptr_bid_end.store(other.ptr_bid_end.load()
? &bids[other.ptr_bid_end.load() - &other.bids[0]]
: nullptr,
std::memory_order_relaxed);
ptr_offer_ini.store(
other.ptr_offer_ini.load()
? &offers[other.ptr_offer_ini.load() - &other.offers[0]]
: nullptr,
std::memory_order_relaxed);
ptr_offer_end.store(
other.ptr_offer_end.load()
? &offers[other.ptr_offer_end.load() - &other.offers[0]]
: nullptr,
std::memory_order_relaxed);
}
return *this;
}
~LimitOrderBook() = default;
void add_order(const Order& order, bool is_bid) {
int index = price_to_index(order.price, is_bid);
if (index < 0 || index >= depth) return;
Container& target = is_bid ? bids : offers;
target[index].add_order(order);
}
void update_order(const Order& new_order, bool is_bid) {
int index = price_to_index(new_order.price, is_bid, false);
if (index < 0 || index >= depth) return;
Container& target = is_bid ? bids : offers;
target[index].update_order(new_order);
}
void delete_order(order_id_t order_id, price_t price, bool is_bid) {
int index = price_to_index(price, is_bid, false);
if (index < 0 || index >= depth) return;
Container& target = is_bid ? bids : offers;
target[index].delete_order(order_id);
}
LevelData get_best_bid() const {
PriceLevel<typename Container::value_type::OrderContainer>* end =
ptr_bid_end.load(std::memory_order_acquire);
if (end == nullptr) return LevelData();
return LevelData(end->price, end->total_quantity);
}
LevelData get_lowest_bid() const {
PriceLevel<typename Container::value_type::OrderContainer>* ini =
ptr_bid_ini.load(std::memory_order_acquire);
if (ini == nullptr) return LevelData();
return LevelData(ini->price, ini->total_quantity);
}
LevelData get_best_offer() const {
PriceLevel<typename Container::value_type::OrderContainer>* ini =
ptr_offer_ini.load(std::memory_order_acquire);
if (ini == nullptr) return LevelData();
return LevelData(ini->price, ini->total_quantity);
}
LevelData get_highest_offer() const {
PriceLevel<typename Container::value_type::OrderContainer>* end =
ptr_offer_end.load(std::memory_order_acquire);
if (end == nullptr) return LevelData();
return LevelData(end->price, end->total_quantity);
}
int get_depth() const { return depth; }
int get_precision() const { return precision; }
const Container& get_bids() const { return bids; }
const Container& get_offers() const { return offers; }
void print_bids() const {
for (int i = 0; i < depth; ++i) {
const auto& level = bids[i];
if (level.total_quantity <= 0) continue;
}
auto* ini = ptr_bid_ini.load(std::memory_order_acquire);
auto* end = ptr_bid_end.load(std::memory_order_acquire);
if (LIKELY(ini != nullptr && end != nullptr)) {
}
}
void print_offers() const {
for (int i = 0; i < depth; ++i) {
const auto& level = offers[i];
if (level.total_quantity <= 0) continue;
}
auto* ini = ptr_offer_ini.load(std::memory_order_acquire);
auto* end = ptr_offer_end.load(std::memory_order_acquire);
if (LIKELY(ini != nullptr && end != nullptr)) {
}
}
template <typename C>
friend struct BookSnapshot;
template <typename C>
friend class HistoricalLimitOrderBook;
};
template <typename Container>
template <typename T>
BookSnapshot<Container>::BookSnapshot(timestamp_t ts,
const LimitOrderBook<T>& book)
: timestamp(ts),
ptr_bid_ini_idx(book.ptr_bid_ini ? book.ptr_bid_ini - &book.bids[0] : -1),
ptr_bid_end_idx(book.ptr_bid_end ? book.ptr_bid_end - &book.bids[0] : -1),
ptr_offer_ini_idx(
book.ptr_offer_ini ? book.ptr_offer_ini - &book.offers[0] : -1),
ptr_offer_end_idx(
book.ptr_offer_end ? book.ptr_offer_end - &book.offers[0] : -1),
total_order_count(0) {
bids.reserve(book.bids.size());
for (const auto& level : book.bids) {
if (level.total_quantity > 0) {
bids.emplace_back(level);
total_order_count += level.orders.size();
}
}
offers.reserve(book.offers.size());
for (const auto& level : book.offers) {
if (level.total_quantity > 0) {
offers.emplace_back(level);
total_order_count += level.orders.size();
}
}
}
// Lock-free memory mapped file manager
class LockFreeMmapFileManager {
private:
int fd_ = -1;
void* mapped_addr_ = MAP_FAILED;
size_t file_size_;
std::string file_path_;
std::atomic<size_t> write_pos_ = {0};
std::atomic<bool> is_valid_ = {false};
void initialize_mmap() {
fd_ = open(file_path_.c_str(), O_CREAT | O_RDWR, S_IRUSR | S_IWUSR);
if (UNLIKELY(fd_ == -1)) {
is_valid_.store(false, std::memory_order_release);
return;
}
struct stat file_stat;
if (fstat(fd_, &file_stat) == 0) {
if (UNLIKELY(static_cast<size_t>(file_stat.st_size) != file_size_)) {
if (UNLIKELY(ftruncate(fd_, file_size_) == -1)) {
close(fd_);
is_valid_.store(false, std::memory_order_release);
return;
}
}
} else {
if (UNLIKELY(ftruncate(fd_, file_size_) == -1)) {
close(fd_);
is_valid_.store(false, std::memory_order_release);
return;
}
}
mapped_addr_ =
mmap(nullptr, file_size_, PROT_READ | PROT_WRITE, MAP_SHARED, fd_, 0);
if (UNLIKELY(mapped_addr_ == MAP_FAILED)) {
close(fd_);
is_valid_.store(false, std::memory_order_release);
return;
}
is_valid_.store(true, std::memory_order_release);
}
public:
LockFreeMmapFileManager(const std::string& path, size_t initial_size)
: file_path_(path), file_size_(initial_size) {
initialize_mmap();
}
~LockFreeMmapFileManager() {
if (is_valid_.load(std::memory_order_acquire)) {
if (mapped_addr_ != MAP_FAILED) {
msync(mapped_addr_, file_size_, MS_SYNC);
munmap(mapped_addr_, file_size_);
}
if (fd_ != -1) {
close(fd_);
}
is_valid_.store(false, std::memory_order_release);
}
}
bool write_data(const void* data, size_t size, size_t& written_pos) {
if (UNLIKELY(!is_valid_.load(std::memory_order_acquire))) {
written_pos = 0;
return false;
}
size_t current_pos = write_pos_.fetch_add(size, std::memory_order_relaxed);
if (UNLIKELY(current_pos + size > file_size_)) {
write_pos_.fetch_sub(size, std::memory_order_relaxed);
written_pos = 0;
return false;
}
memcpy(static_cast<char*>(mapped_addr_) + current_pos, data, size);
std::atomic_thread_fence(std::memory_order_release);
written_pos = current_pos;
return true;
}
bool read_data(void* dest, size_t offset, size_t size) const {
if (UNLIKELY(!is_valid_.load(std::memory_order_acquire) ||
offset + size > file_size_)) {
return false;
}
std::atomic_thread_fence(std::memory_order_acquire);
memcpy(dest, static_cast<char*>(mapped_addr_) + offset, size);
return true;
}
size_t get_write_position() const {
return write_pos_.load(std::memory_order_acquire);
}
bool set_write_position(size_t pos) {
if (UNLIKELY(!is_valid_.load(std::memory_order_acquire) ||
pos > file_size_)) {
return false;
}
write_pos_.store(pos, std::memory_order_release);
return true;
}
size_t get_file_size() const { return file_size_; }
bool sync() {
if (UNLIKELY(!is_valid_.load(std::memory_order_acquire))) return false;
return msync(mapped_addr_, file_size_, MS_SYNC) == 0;
}
void reset() { write_pos_.store(0, std::memory_order_release); }
const void* get_mapped_address() const { return mapped_addr_; }
const std::string& get_file_path() const { return file_path_; }
};
// Historical limit order book with persistence
template <typename Container>
class PersistentHistoricalOrderBook : public LimitOrderBook<Container> {
private:
using Base = LimitOrderBook<Container>;
std::array<OrderLog, LOG_BUFFER_SIZE> log_buffer;
std::atomic<size_t> log_write_pos{0};
std::atomic<size_t> log_read_pos{0};
std::array<BookSnapshot<Container>, SNAPSHOT_BUFFER_SIZE> snapshot_buffer;
std::atomic<size_t> snapshot_write_pos{0};
std::atomic<size_t> last_snapshot_idx{0};
timestamp_t snapshot_interval;
std::atomic<timestamp_t> last_snapshot_time{0};
LockFreeMmapFileManager mmap_manager_;
std::thread persistence_thread_;
std::atomic<bool> running_ = {true};
std::atomic<timestamp_t> last_persisted_timestamp_ = {0};
timestamp_t persistence_interval_;
std::atomic<bool> trigger_flag_ = {false};
struct SnapshotLocation {
timestamp_t timestamp;
size_t file_offset;
size_t size;
};
std::array<SnapshotLocation, SNAPSHOT_BUFFER_SIZE> snapshot_locations_;
std::atomic<size_t> snapshot_location_write_pos{0};
std::atomic<size_t> last_persisted_log_pos_ = {0};
static timestamp_t get_timestamp() {
auto now = std::chrono::high_resolution_clock::now();
return std::chrono::duration_cast<std::chrono::nanoseconds>(
now.time_since_epoch())
.count();
}
void add_log(const OrderLog& log) {
size_t pos = log_write_pos.fetch_add(1, std::memory_order_relaxed);
log_buffer[pos % LOG_BUFFER_SIZE] = log;
std::atomic_thread_fence(std::memory_order_release);
}
void create_snapshot() {
timestamp_t now = get_timestamp();
size_t pos = snapshot_write_pos.fetch_add(1, std::memory_order_relaxed);
size_t idx = pos % SNAPSHOT_BUFFER_SIZE;
snapshot_buffer[idx].~BookSnapshot<Container>();
new (&snapshot_buffer[idx]) BookSnapshot<Container>(now, *this);
std::atomic_thread_fence(std::memory_order_release);
last_snapshot_time.store(now, std::memory_order_relaxed);
last_snapshot_idx.store(pos, std::memory_order_relaxed);
}
void create_snapshot_if_needed() {
timestamp_t now = get_timestamp();
timestamp_t last_time = last_snapshot_time.load(std::memory_order_acquire);
if (now - last_time >= snapshot_interval) {
if (last_snapshot_time.compare_exchange_weak(last_time, now,
std::memory_order_release,
std::memory_order_acquire)) {
create_snapshot();
}
}
}
void persistence_worker() {
while (running_.load(std::memory_order_acquire)) {
timestamp_t start = get_timestamp();
while (get_timestamp() - start < persistence_interval_ &&
!trigger_flag_.load(std::memory_order_acquire) &&
running_.load(std::memory_order_acquire)) {
std::this_thread::sleep_for(std::chrono::nanoseconds(100));
}
if (!running_.load(std::memory_order_acquire)) break;
persist_data();
trigger_flag_.store(false, std::memory_order_release);
}
}
void persist_data() {
this->create_snapshot();
size_t last_snapshot_idx =
this->last_snapshot_idx.load(std::memory_order_acquire);
const BookSnapshot<Container>& latest_snapshot =
this->snapshot_buffer[last_snapshot_idx % SNAPSHOT_BUFFER_SIZE];
size_t snapshot_offset = 0;
size_t snapshot_size = 0;
if (serialize_and_write_snapshot(latest_snapshot, snapshot_size,
snapshot_offset)) {
size_t loc_pos =
snapshot_location_write_pos.fetch_add(1, std::memory_order_relaxed);
size_t loc_idx = loc_pos % SNAPSHOT_BUFFER_SIZE;
snapshot_locations_[loc_idx].timestamp = latest_snapshot.timestamp;
snapshot_locations_[loc_idx].file_offset = snapshot_offset;
snapshot_locations_[loc_idx].size = snapshot_size;
} else {
return;
}
size_t start_pos = last_persisted_log_pos_.load(std::memory_order_acquire);
size_t end_pos = this->log_write_pos.load(std::memory_order_acquire);
if (start_pos < end_pos) {
size_t count = 0;
for (size_t pos = start_pos; pos < end_pos; ++pos) {
std::atomic_thread_fence(std::memory_order_acquire);
const OrderLog& log = this->log_buffer[pos % LOG_BUFFER_SIZE];
if (UNLIKELY(!log.verify_checksum())) {
continue;
}
size_t written_pos;
if (!mmap_manager_.write_data(&log, sizeof(OrderLog), written_pos)) {
break;
}
count++;
}
if (count > 0) {
last_persisted_log_pos_.store(start_pos + count,
std::memory_order_release);
last_persisted_timestamp_.store(get_timestamp(),
std::memory_order_release);
}
}
mmap_manager_.sync();
}
bool serialize_and_write_snapshot(const BookSnapshot<Container>& snapshot,
size_t& out_size, size_t& out_offset) {
BookSnapshotHeader header(snapshot);
out_size = header.data_size;
size_t header_offset;
if (!mmap_manager_.write_data(&header, sizeof(BookSnapshotHeader),
header_offset)) {
return false;
}
out_offset = header_offset;
for (const auto& level : snapshot.bids) {
if (UNLIKELY(!level.verify_checksum())) {
continue;
}
size_t written_pos;
if (!mmap_manager_.write_data(&level, sizeof(PriceLevelSnapshot),
written_pos)) {
return false;
}
}
for (const auto& level : snapshot.offers) {
if (UNLIKELY(!level.verify_checksum())) {
continue;
}
size_t written_pos;
if (!mmap_manager_.write_data(&level, sizeof(PriceLevelSnapshot),
written_pos)) {
return false;
}
}
return true;
}
public:
PersistentHistoricalOrderBook(int precision, int depth,
const std::string& mmap_file_path,
size_t mmap_file_size,
timestamp_t snapshot_interval_ns = 1000000,
timestamp_t persistence_interval_ns = 10000000)
: Base(precision, depth, snapshot_interval_ns),
mmap_manager_(mmap_file_path, mmap_file_size),
persistence_interval_(persistence_interval_ns) {
persistence_thread_ =
std::thread(&PersistentHistoricalOrderBook::persistence_worker, this);
}
~PersistentHistoricalOrderBook() {
running_.store(false, std::memory_order_release);
trigger_flag_.store(true, std::memory_order_release);
if (persistence_thread_.joinable()) {
persistence_thread_.join();
}
}
void add_order(const Order& order, bool is_bid) {
timestamp_t ts = get_timestamp();
Order timed_order = order;
timed_order.timestamp = ts;
timed_order.checksum = timed_order.calculate_checksum();
int index = this->price_to_index(timed_order.price, is_bid);
if (index >= 0 && index < this->depth) {
Container& target = is_bid ? this->bids : this->offers;
target[index].add_order(timed_order);
add_log(OrderLog(OrderLog::ADD, timed_order, index));
}
create_snapshot_if_needed();
}
void update_order(const Order& new_order, bool is_bid) {
timestamp_t ts = get_timestamp();
Order timed_order = new_order;
timed_order.timestamp = ts;
timed_order.checksum = timed_order.calculate_checksum();
int index = this->price_to_index(timed_order.price, is_bid, false);
if (index >= 0 && index < this->depth) {
Container& target = is_bid ? this->bids : this->offers;
if (target[index].update_order(timed_order)) {
add_log(OrderLog(OrderLog::UPDATE, timed_order, index));
}
}
create_snapshot_if_needed();
}
void delete_order(order_id_t order_id, price_t price, bool is_bid) {
timestamp_t ts = get_timestamp();
Order dummy_order(order_id, price, 0, is_bid, ts);
dummy_order.checksum = dummy_order.calculate_checksum();
int index = this->price_to_index(price, is_bid, false);
if (index >= 0 && index < this->depth) {
Container& target = is_bid ? this->bids : this->offers;
if (target[index].delete_order(order_id)) {
add_log(OrderLog(OrderLog::DELETE, dummy_order, index));
}
}
create_snapshot_if_needed();
}
void get_persistence_stats(timestamp_t& last_persisted_ts, size_t& file_pos,
size_t& file_size) const {
last_persisted_ts =
last_persisted_timestamp_.load(std::memory_order_acquire);
file_pos = mmap_manager_.get_write_position();
file_size = mmap_manager_.get_file_size();
}
void trigger_persistence() {
trigger_flag_.store(true, std::memory_order_release);
}
};
using BasicLimitOrderBook =
LimitOrderBook<std::vector<PriceLevel<std::vector<Order>>>>;
using LFLimitOrderBook =
LimitOrderBook<std::vector<PriceLevel<std::vector<Order>>>>;
using BasicPersistentOrderBook =
PersistentHistoricalOrderBook<std::vector<PriceLevel<std::vector<Order>>>>;
using LFPersistentOrderBook =
PersistentHistoricalOrderBook<std::vector<PriceLevel<std::vector<Order>>>>;
} // namespace Common