// Engineer Atlas: native locality experiment and bounded SPSC queue.
// Build: c++ -std=c++20 -O3 -pthread -Wall -Wextra latency_lab.cpp -o latency_lab
// Run: ./latency_lab
// Results describe this process and workload, not exchange or per-message latency.
#include <algorithm>
#include <array>
#include <atomic>
#include <chrono>
#include <cstdint>
#include <iostream>
#include <numeric>
#include <random>
#include <stdexcept>
#include <thread>
#include <vector>

template <std::size_t Capacity>
class SpscRing {
  static_assert(Capacity > 1 && (Capacity & (Capacity - 1)) == 0);
  // Padding is a teaching assumption: measure the target's cache-line size.
  alignas(64) std::atomic<std::size_t> head{0};
  alignas(64) std::atomic<std::size_t> tail{0};
  std::array<std::uint64_t, Capacity> slots{};
 public:
  // Exactly one producer owns head. One slot remains unused to distinguish full.
  bool push(std::uint64_t value) {
    const auto h = head.load(std::memory_order_relaxed);
    const auto next = (h + 1) & (Capacity - 1);
    if (next == tail.load(std::memory_order_acquire)) return false;
    slots[h] = value;
    head.store(next, std::memory_order_release);
    return true;
  }
  // Exactly one consumer owns tail. Acquire sees the producer's slot write.
  bool pop(std::uint64_t& value) {
    const auto t = tail.load(std::memory_order_relaxed);
    if (t == head.load(std::memory_order_acquire)) return false;
    value = slots[t];
    tail.store((t + 1) & (Capacity - 1), std::memory_order_release);
    return true;
  }
};

using Clock = std::chrono::steady_clock;
struct Node { std::uint64_t value; std::size_t next; };
std::uint64_t scan(const std::vector<Node>& nodes) {
  std::uint64_t sum = 0;
  for (const auto& node : nodes) sum += node.value;
  return sum;
}
std::uint64_t chase(const std::vector<Node>& nodes, std::size_t current) {
  std::uint64_t sum = 0;
  for (std::size_t i = 0; i < nodes.size(); ++i) {
    sum += nodes[current].value;
    current = nodes[current].next;
  }
  return sum;
}
void report(const char* name, std::vector<double> batches) {
  std::sort(batches.begin(), batches.end());
  // Nearest-rank quantiles of whole-batch elapsed times, including OS pauses.
  std::cout << name << ": batch_ms min=" << batches.front()
            << " median=" << batches[batches.size() / 2]
            << " p95=" << batches[(95 * batches.size() + 99) / 100 - 1]
            << " max=" << batches.back() << '\n';
}
int main() {
  constexpr std::size_t count = 1 << 20;
  constexpr int rounds = 21;
  constexpr std::uint64_t expected = count * (count - 1ULL) / 2;
  std::vector<Node> nodes(count);
  std::vector<std::size_t> order(count);
  std::iota(order.begin(), order.end(), 0);
  std::mt19937 random(42);
  std::shuffle(order.begin(), order.end(), random);
  for (std::size_t i = 0; i < count; ++i) nodes[order[i]] = {order[i], order[(i + 1) % count]};
  // Allocate and build before timing. Warm both paths; alternate their order.
  if (scan(nodes) != expected || chase(nodes, order[0]) != expected) throw std::runtime_error("bad setup");
  std::vector<double> contiguous, dependent;
  for (int round = 0; round < rounds; ++round) {
    for (int phase = 0; phase < 2; ++phase) {
      const bool follow = (round + phase) % 2;
      const auto start = Clock::now();
      const auto sum = follow ? chase(nodes, order[0]) : scan(nodes);
      const auto stop = Clock::now();
      if (sum != expected) throw std::runtime_error("checksum mismatch");
      (follow ? dependent : contiguous).push_back(std::chrono::duration<double, std::milli>(stop - start).count());
    }
  }
  std::cout << "C++=" << __cplusplus << " compiler=" << __VERSION__ << '\n'
            << "nodes=" << count << " bytes=" << nodes.size() * sizeof(Node)
            << " rounds=" << rounds << " seed=42 hardware_threads=" << std::thread::hardware_concurrency() << '\n';
  report("contiguous scan", contiguous);
  report("dependent shuffled traversal", dependent);
  // This test exercises wrap-around, full/empty backpressure and FIFO under load.
  SpscRing<1024> queue;
  std::atomic<bool> valid{true};
  const auto start = Clock::now();
  std::thread producer([&] { for (std::uint64_t i = 0; i < count; ++i) while (!queue.push(i)) std::this_thread::yield(); });
  std::thread consumer([&] {
    for (std::uint64_t i = 0; i < count; ++i) {
      std::uint64_t value;
      while (!queue.pop(value)) std::this_thread::yield();
      if (value != i) valid.store(false, std::memory_order_relaxed);
    }
  });
  producer.join(); consumer.join();
  std::uint64_t unused;
  if (!valid.load() || queue.pop(unused)) throw std::runtime_error("FIFO check failed");
  std::cout << "SPSC FIFO checks passed: " << count << " values, batch_ms="
            << std::chrono::duration<double, std::milli>(Clock::now() - start).count() << '\n';
}