// Multithreading: writer + cloned readers
// Textbook section: Volume II, chapter 11, "Multithreading",
//   sections "Sticky threads", "NOSTICKYTHREADS".
// What it demonstrates: a writer and parallel readers on one environment;
//   cloning a read-only transaction via txn_managed::clone() — all clones
//   see one snapshot, independent of the writer's commits.
// Build: g++ -std=c++17 -I<libmdbx-include> -I../common 14-multithreading.c++ -lmdbx -pthread
// Run: ./14-multithreading
// Expected output:
//   snapshot(reader0) = 10
//   snapshot(reader1) = 10
//   fresh(reader0) = 30
//   fresh(reader1) = 30
//   ok: readers see stable snapshot while writer commits
// On failure — message to stderr and exit 1.
//
// SPDX-License-Identifier: Apache-2.0

#include <atomic>
#include <iostream>
#include <string>
#include <thread>
#include <vector>

#include "common.h++"
#include <mdbx.h++>

int main() {
  try {
    const std::string path = std::string(example::tmpdir()) + "/mdbx-14-multithreading.mdbx";
    mdbx::env::remove(path);
    mdbx::env_managed env = example::env_open(path);

    // The initial 10 records.
    {
      auto txn = env.start_write();
      auto table = txn.open_map(nullptr);
      for (int i = 0; i < 10; ++i)
        txn.insert(table, mdbx::slice(std::to_string(i)), mdbx::slice("x"));
      txn.commit();
    }

    // The base snapshot for the clones.
    auto base = env.start_read();
    (void)base.open_map(nullptr);

    // The writer adds 20 more records in parallel with reading.
    std::atomic<bool> writer_done{false};
    std::thread writer([&] {
      for (int batch = 0; batch < 2; ++batch) {
        auto txn = env.start_write();
        auto table = txn.open_map(nullptr);
        for (int i = 0; i < 10; ++i) {
          const int k = 10 + batch * 10 + i;
          txn.insert(table, mdbx::slice(std::to_string(k)), mdbx::slice("w"));
        }
        txn.commit();
        std::this_thread::yield();
      }
      writer_done = true;
    });

    // Two readers: each works with its own clone of the base snapshot.
    std::atomic<uint64_t> snapshot_count[2];
    std::atomic<uint64_t> fresh_count[2];
    std::vector<std::thread> readers;
    for (int r = 0; r < 2; ++r) {
      readers.emplace_back([&, r] {
        // a snapshot clone for this thread
        auto clone = base.clone();
        auto table = clone.open_map(nullptr);
        snapshot_count[r] = clone.get_map_stat(table).ms_entries;
        clone.abort();

        while (!writer_done.load())
          std::this_thread::yield();

        // After the writer finishes, a fresh transaction sees all the data.
        auto fresh = env.start_read();
        auto ftable = fresh.open_map(nullptr);
        fresh_count[r] = fresh.get_map_stat(ftable).ms_entries;
        fresh.abort();
      });
    }
    for (auto &t : readers)
      t.join();
    writer.join();
    base.abort();

    std::cout << "snapshot(reader0) = " << snapshot_count[0].load() << "\n";
    std::cout << "snapshot(reader1) = " << snapshot_count[1].load() << "\n";
    std::cout << "fresh(reader0) = " << fresh_count[0].load() << "\n";
    std::cout << "fresh(reader1) = " << fresh_count[1].load() << "\n";

    if (snapshot_count[0] != 10 || snapshot_count[1] != 10 || fresh_count[0] != 30 || fresh_count[1] != 30) {
      std::cerr << "FAIL: unexpected counts\n";
      return EXIT_FAILURE;
    }

    std::cout << "ok: readers see stable snapshot while writer commits\n";
    return EXIT_SUCCESS;
  } catch (const std::exception &ex) {
    std::cerr << "error: " << ex.what() << "\n";
    return EXIT_FAILURE;
  }
}