spmc共享内存无锁队列应用

来源1.https://github.com/MengRao/SPMC_Queue

来源2.https://zhuanlan.zhihu.com/p/15326347602

#pragma once
#include <atomic>

template <class T, uint32_t CNT>
class SPMCQueue {
 public:
  static_assert(CNT && !(CNT & (CNT - 1)), "CNT must be a power of 2");
  struct Reader {
    operator bool() const { return q; }
    T* read() {
      auto& blk = q->blks[next_idx % CNT];
      uint32_t new_idx =
          ((std::atomic<uint32_t>*)&blk.idx)->load(std::memory_order_acquire);
      if (int(new_idx - next_idx) < 0) return nullptr;
      next_idx = new_idx + 1;
      return &blk.data;
    }

    T* readLast() {
      T* ret = nullptr;
      while (T* cur = read()) ret = cur;
      return ret;
    }

    SPMCQueue<T, CNT>* q = nullptr;
    uint32_t next_idx;
  };

  Reader getReader() {
    Reader reader;
    reader.q = this;
    reader.next_idx = write_idx + 1;
    return reader;
  }

  template <typename Writer>
  void write(Writer writer) {
    auto& blk = blks[++write_idx % CNT];
    writer(blk.data);
    ((std::atomic<uint32_t>*)&blk.idx)
        ->store(write_idx, std::memory_order_release);
  }

 private:
  friend class Reader;
  struct alignas(64) Block {
    uint32_t idx = 0;
    T data;
  } blks[CNT];

  alignas(128) uint32_t write_idx = 0;
};

template <class Q>
Q* shmmap(const char* filename) {
  int fd = shm_open(filename, O_CREAT | O_RDWR, 0666);
  if (fd == -1) {
    std::cerr << "shm_open failed: " << strerror(errno) << std::endl;
    return nullptr;
  }
  if (ftruncate(fd, sizeof(Q))) {
    std::cerr << "ftruncate failed: " << strerror(errno) << std::endl;
    close(fd);
    return nullptr;
  }
  Q* ret = (Q*)mmap(0, sizeof(Q), PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
  close(fd);
  if (ret == MAP_FAILED) {
    std::cerr << "mmap failed: " << strerror(errno) << std::endl;
    return nullptr;
  }
  return ret;
}

// Level2 逐笔委托行情
struct Lev2Order {
  char SecurityID[31];  // 股票代码
  int TradeTime;        // 交易所时间
  char TickType;        // 报单类型
  char Side;            // 报单方向
  double Price;         // 价格
  long long Volume;     // 数量(股)
  int MainSeq;          // 主通道号
  long long SubSeq;     // 次通道号
  long long No;         // 报单号
};

// Level2 逐笔成交行情
struct Lev2Trans {
  char SecurityID[31];  // 股票代码
  int TradeTime;        // 交易所时间
  char TickType;        // 报单类型
  double Price;         // 价格
  long long Volume;     // 数量(股)
  int MainSeq;          // 主通道号
  long long SubSeq;     // 次通道号
  long long BuyNo;      // 买方报单号
  long long SellNo;     // 卖方报单号
};

const int CAPACITY = 524288;
using TradeQueue = SPMCQueue<Lev2Trans, CAPACITY>;  // 逐笔成交
using OrderQueue = SPMCQueue<Lev2Order, CAPACITY>;  // 逐笔委托

q_trade_sz = shmmap<TradeQueue>("market_data_trade_sz");
q_order_sz = shmmap<OrderQueue>("market_data_order_sz");
q_trade_sh = shmmap<TradeQueue>("market_data_trade_sh");
q_order_sh = shmmap<OrderQueue>("market_data_order_sh");

/// 上海逐笔数据通知
virtual void OnRtnNGTSTick(
    TORALEV2API::CTORATstpLev2NGTSTickField* pOrderDetail) {
  if (pOrderDetail->TickType == 'T') {
    // 逐笔成交
    q_trade_sh->write([=, this](Lev2Trans& msg) {
      strcpy(msg.SecurityID, pOrderDetail->SecurityID);
      msg.TradeTime = pOrderDetail->TickTime * 10;
      msg.TickType = pOrderDetail->TickType;
      msg.Price = pOrderDetail->Price;
      msg.Volume = pOrderDetail->Volume;
      msg.MainSeq = pOrderDetail->MainSeq;
      msg.SubSeq = pOrderDetail->SubSeq;
      msg.BuyNo = pOrderDetail->BuyNo;
      msg.SellNo = pOrderDetail->SellNo;
    });
  }
}

// 上海股票逐笔成交
void handle_sh_trans() {
  TradeQueue* q_trade_sh = shmmap<TradeQueue>("market_data_trade_sh");
  auto reader = q_trade_sh->getReader();
  mqtt::async_client mqtt_client_trans_sh(DFLT_ADDRESS, CLIENT_ID_TRANS_SH,
                                          maxBufferedMessages);
  mqtt_client_trans_sh.connect(connopts)->wait();

  while (1) {
    Lev2Trans* pOrderDetail = reader.read();
    if (pOrderDetail) {
      std::string SecurityID(pOrderDetail->SecurityID);

      std::string data = fmt::format(
          "{:s},{:s},{:d},{:d},{:c},{:.3f},{:d},{:d},{:d},{:d},{:d}",
          SecurityID, today, pOrderDetail->TradeTime,
          duration_cast<milliseconds>(system_clock::now().time_since_epoch())
              .count(),
          pOrderDetail->TickType, pOrderDetail->Price, pOrderDetail->Volume,
          pOrderDetail->MainSeq, pOrderDetail->SubSeq, pOrderDetail->BuyNo,
          pOrderDetail->SellNo);

      if (subscribe_trans_sh.find(SecurityID) != subscribe_trans_sh.end()) {
        mqtt::message_ptr pubmsg =
            mqtt::make_message("level2/trans/" + SecurityID, data);
        mqtt_client_trans_sh.publish(pubmsg);
      }
    }
  }

  delete q_trade_sh;
  mqtt_client_trans_sh.disconnect()->wait();
}

// 上海股票逐笔
void handle_sh() {
  TradeQueue* q_trade_sh = shmmap<TradeQueue>("market_data_trade_sh");
  auto reader_trans = q_trade_sh->getReader();

  std::string base_dir = "/data/";
  base_dir = base_dir + get_datetime(NULL, true);
  std::filesystem::create_directory(base_dir);
  std::string out_path = base_dir + "/" + "trans_sh.csv";
  FILE* fp_trans = fopen(out_path.c_str(), "a");

  while (1) {
    Lev2Trans* pTransDetail = reader_trans.read();
    if (pTransDetail) {
      std::string data = fmt::format(
          "{:s},{:s},{:d},{:d},{:c},{:.3f},{:d},{:d},{:d},{:d},{:d}",
          pTransDetail->SecurityID, today, pTransDetail->TradeTime,
          duration_cast<milliseconds>(system_clock::now().time_since_epoch())
              .count(),
          pTransDetail->TickType, pTransDetail->Price, pTransDetail->Volume,
          pTransDetail->MainSeq, pTransDetail->SubSeq, pTransDetail->BuyNo,
          pTransDetail->SellNo);

      fprintf(fp_trans, "%s\n", data.c_str());
    }

    fclose(fp_trans);
  }

main:

  std::thread thread_sh_trans(handle_sh_trans);
  thread_sh_trans.join();

  std::thread thread_sh_trans(handle_sh);
  thread_sh_trans.join();

  // 共享内存的权限问题，在容器运行时需要添加--ipc=host --pid=host
  // 标志，然后就可以利用原始行情做多粒度的Level2行情分析