Function SpscQueueThroughput

cpp/src/arrow/util/queue_benchmark.cc:35–80

static constexpr int64_t kSize = 100000;

void SpscQueueThroughput(benchmark::State& state) {
  SpscQueue<std::shared_ptr<Buffer>> queue(16);

  std::vector<std::shared_ptr<Buffer>> source;
  std::vector<std::shared_ptr<Buffer>> sink;
  source.reserve(kSize);
  sink.resize(kSize);
  const uint8_t data[1] = {0};
  for (int64_t i = 0; i < kSize; i++) {
    source.push_back(std::make_shared<Buffer>(data, 1));
  }

  for (auto _ : state) {
    std::thread producer([&] {
      auto itr = std::make_move_iterator(source.begin());
      auto end = std::make_move_iterator(source.end());
      while (itr != end) {
        while (!queue.Write(*itr)) {
        }
        itr++;
      }
    });

    std::thread consumer([&] {
      auto itr = sink.begin();
      auto end = sink.end();
      while (itr != end) {
        auto next = queue.FrontPtr();
        if (next != nullptr) {
          (*itr).swap(*next);
          queue.PopFront();
          itr++;
        }
      }
    });

    producer.join();
    consumer.join();
    std::swap(source, sink);
  }

  for (const auto& buf : source) {
    ARROW_CHECK(buf && buf->size() == 1);
  }
  state.SetItemsProcessed(state.iterations() * kSize);
}

BENCHMARK(SpscQueueThroughput)->UseRealTime();