thread_pool_smoke_test.cpp

// thread_pool_smoke_test.cpp
//
// Standalone smoke test for ThreadPool. Not a unit test framework - just
// three checks run in sequence, each printing PASS/FAIL. Good enough to
// catch the two classic bugs (predicate misuse, task-queue race) before
// wiring the pool into the proxy.
//
// Checks:
//   1. Concurrency: tasks that sleep actually overlap in wall-clock time
//      (proves multiple workers run simultaneously, not serialized).
//   2. Correctness: every submitted task actually executes, exactly once,
//      with no lost/duplicated work (this is what would have caught the
//      shadowed `task` bug - that bug throws std::bad_function_call,
//      which this test also implicitly guards against just by not
//      crashing).
//   3. Clean shutdown: destructor drains the queue and joins all threads
//      without hanging or crashing (this is what would have caught the
//      unlocked stop_ write - ThreadSanitizer catches that, not a plain
//      build, so build this with -fsanitize=thread separately).
//
// Build (plain):
//   g++ -std=c++17 -Wall -Wextra -pthread thread_pool_smoke_test.cpp -o
//   smoke_test
// Build (with ThreadSanitizer, catches the data-race class of bug):
//   g++ -std=c++17 -Wall -Wextra -pthread -fsanitize=thread
//       thread_pool_smoke_test.cpp -o smoke_test_tsan

#include "thread_pool.h"

#include <atomic>
#include <chrono>
#include <iostream>
#include <mutex>
#include <set>
#include <thread>
#include <vector>

namespace {

using Clock = std::chrono::steady_clock;

int g_checks_failed = 0;

void check(bool condition, const std::string &description) {
  if (condition) {
    std::cout << "  [PASS] " << description << "\n";
  } else {
    std::cout << "  [FAIL] " << description << "\n";
    ++g_checks_failed;
  }
}

// --- Check 1: tasks actually run concurrently, not serialized ---------
//
// Submit kNumTasks tasks that each sleep for kSleepMs. If the pool is
// truly concurrent with >= kNumTasks workers, total wall time should be
// close to kSleepMs (all overlap), not kNumTasks * kSleepMs (serialized).
// We assert total time is well under the serial bound as a sanity check
// - this is what a broken "cv_.wait(lock, some_bool)" would fail, since
// that fails to compile in the first place, but a pool that accidentally
// executes tasks *while holding the lock* would also serialize and fail
// this check.
void check_concurrency() {
  std::cout << "Check 1: concurrency (tasks should overlap in time)\n";

  constexpr int kNumTasks = 8;
  constexpr int kSleepMs = 200;

  ThreadPool pool(kNumTasks); // one worker per task -> full overlap expected

  auto start = Clock::now();
  std::atomic<int> completed{0};

  for (int i = 0; i < kNumTasks; ++i) {
    pool.submit([&completed, kSleepMs] {
      std::this_thread::sleep_for(std::chrono::milliseconds(kSleepMs));
      completed.fetch_add(1, std::memory_order_relaxed);
    });
  }

  // Pool destructor (end of scope) drains queue and joins all threads,
  // so by the time we get past this block, everything has finished.
  // We measure inside a nested scope so the timer stops right after
  // the tasks are known to be done, before destructor teardown noise.
  while (completed.load(std::memory_order_relaxed) < kNumTasks) {
    std::this_thread::sleep_for(std::chrono::milliseconds(5));
  }
  auto elapsed = Clock::now() - start;
  auto elapsed_ms =
      std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count();

  std::cout << "    " << kNumTasks << " tasks x " << kSleepMs
            << "ms each completed in " << elapsed_ms << "ms\n";

  // Serial execution would take kNumTasks * kSleepMs (1600ms here).
  // Concurrent execution with 8 workers should take roughly kSleepMs
  // (~200ms) plus scheduling slop. Use a generous cutoff to avoid
  // flakiness on a loaded CI box: anything under half the serial time
  // proves real overlap happened.
  check(elapsed_ms < (kNumTasks * kSleepMs) / 2,
        "wall time is well under serial bound -> tasks ran concurrently");
  check(completed.load() == kNumTasks, "all tasks completed");
}

// --- Check 2: every task runs exactly once, no lost/duplicated work ---
//
// Submit many small tasks that each record their own index. Afterwards,
// verify the set of recorded indices is exactly {0, ..., N-1} - this
// would have caught the shadowed `task` bug (empty std::function throws
// std::bad_function_call, so this test would abort/crash before ever
// reaching the assertions).
void check_correctness() {
  std::cout << "Check 2: correctness (every task runs exactly once)\n";

  constexpr int kNumTasks = 500;
  std::mutex results_mutex;
  std::vector<int> results;
  results.reserve(kNumTasks);

  {
    ThreadPool pool(4);
    for (int i = 0; i < kNumTasks; ++i) {
      pool.submit([&results, &results_mutex, i] {
        std::lock_guard<std::mutex> lock(results_mutex);
        results.push_back(i);
      });
    }
    // pool destructs here -> drains queue, joins workers, all tasks
    // guaranteed complete before we inspect `results`.
  }

  check(results.size() == static_cast<size_t>(kNumTasks),
        "exactly " + std::to_string(kNumTasks) + " tasks recorded (got " +
            std::to_string(results.size()) + ")");

  std::set<int> unique_results(results.begin(), results.end());
  check(unique_results.size() == static_cast<size_t>(kNumTasks),
        "no duplicate task executions");

  bool all_present = true;
  for (int i = 0; i < kNumTasks; ++i) {
    if (unique_results.find(i) == unique_results.end()) {
      all_present = false;
      break;
    }
  }
  check(all_present, "no task indices missing (nothing dropped on the floor)");
}

// --- Check 3: destructor drains remaining queue and shuts down clean --
//
// Submit a batch of tasks and immediately let the pool go out of scope
// (no artificial delay to "let workers catch up" first). The destructor
// contract is: stop accepting new work conceptually, but finish whatever
// is already queued, then join. If shutdown either hangs (lost wakeup)
// or drops queued-but-not-yet-started tasks, this check catches it.
void check_shutdown_drains_queue() {
  std::cout << "Check 3: shutdown drains queued tasks, doesn't hang\n";

  constexpr int kNumTasks = 50;
  std::atomic<int> completed{0};

  auto start = Clock::now();
  {
    ThreadPool pool(2); // few workers relative to task count, so a
                        // meaningful backlog is still queued when the
                        // destructor runs immediately after this loop.
    for (int i = 0; i < kNumTasks; ++i) {
      pool.submit(
          [&completed] { completed.fetch_add(1, std::memory_order_relaxed); });
    }
    // No sleep here - destructor fires right away, while tasks are
    // still likely queued/in-flight.
  }
  auto elapsed = Clock::now() - start;
  auto elapsed_ms =
      std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count();

  std::cout << "    shutdown + drain completed in " << elapsed_ms << "ms\n";

  check(elapsed_ms < 5000, "destructor returned promptly (didn't hang)");
  check(completed.load() == kNumTasks,
        "all " + std::to_string(kNumTasks) +
            " queued tasks executed before shutdown completed (none dropped)");
}

} // namespace

int main() {
  std::cout << "=== ThreadPool smoke test ===\n\n";

  check_concurrency();
  std::cout << "\n";
  check_correctness();
  std::cout << "\n";
  check_shutdown_drains_queue();

  std::cout << "\n=== "
            << (g_checks_failed == 0 ? "ALL CHECKS PASSED"
                                     : "SOME CHECKS FAILED")
            << " (" << g_checks_failed << " failure(s)) ===\n";

  return g_checks_failed == 0 ? 0 : 1;
}