Width-Limited Queue

#include <atomic>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <mutex>
#include <optional>
#include <queue>
#include <thread>
#include <vector>

class JobQueue {
public:
    using Job = std::function<void()>;

    JobQueue(size_t max_width) : running_(true) {
        // Create a thread pool with "max_width" number of worker threads.
        // Cap the maximum number of threads to the hardware limit if it is
        // available, otherwise use a hardcoded limit of 128 to avoid resource
        // exhaustion.
        size_t threads = std::thread::hardware_concurrency();
        if (threads == 0) {
            threads = 128;
        }

        for (size_t i = 0; i < std::min(max_width, threads); ++i) {
            thread_pool_.emplace_back([this] { thread_worker(); });
        }
    }

    ~JobQueue() {
        // Flip running flag so that waiting threads will terminate.
        running_ = false;

        // Wake up any waiting threads.
        cv_.notify_all();

        // Join all threads.
        for (std::thread& t : thread_pool_) {
            t.join();
        }
    }

    void add_job(const Job& job) {
        {
            std::unique_lock<std::mutex> lock(mutex_);
            pending_jobs_.push(job);
        }
        // Wake up a worker thread to start running job if there is one
        // available.
        cv_.notify_one();
    }

private:
    void thread_worker() {
        while (running_) {
            // Check if there is a job waiting in the queue.
            auto next_job = [this]() -> std::optional<Job> {
                std::unique_lock<std::mutex> lock(mutex_);
                cv_.wait(lock, [this] { return !pending_jobs_.empty() || !running_; });

                if (!running_ && pending_jobs_.empty()) {
                    return std::nullopt;
                }

                // Although wait handles this, double check safely
                if (pending_jobs_.empty()) {
                    return std::nullopt;
                }

                Job next_job = pending_jobs_.front();
                pending_jobs_.pop();
                return next_job;
            }();

            if (!next_job.has_value()) {
                return;
            }

            // Run the next job if available.
            (*next_job)();
        }
    }

    std::atomic<bool> running_;
    std::vector<std::thread> thread_pool_;
    std::queue<Job> pending_jobs_;
    std::mutex mutex_;
    std::condition_variable cv_;
};

// ==========================================
// Test Harness (from video footer)
// ==========================================

void test(const char* name, size_t width) {
    using namespace std::chrono_literals;

    JobQueue q(width);

    std::atomic<int> max_running_jobs = 0;
    std::atomic<int> running_jobs = 0;

    // Declare the function that will run as the job on the work queue.
    const auto job_func = [&]() {
        ++running_jobs;

        // Sleep for a few milliseconds so that multiple jobs have the
        // opportunity to run concurrently.
        std::this_thread::sleep_for(1ms);

        // Update the maximum number of running jobs based on how many are
        // currently running.
        int old_max_jobs = max_running_jobs;
        while (old_max_jobs < running_jobs &&
               !max_running_jobs.compare_exchange_weak(old_max_jobs, running_jobs)) {
            // max_running_jobs updated in another thread. Try again.
        }

        --running_jobs;
    };

    // Initialize job queue with specified width.
    // Schedule a number of jobs to run on the work queue.
    // Ensure we schedule enough jobs to saturate the width.
    int jobs_to_schedule = std::min((size_t)100, width * 10);

    for (int i = 0; i < jobs_to_schedule; ++i) {
        q.add_job(job_func);
    }

    // Wait for all the jobs to finish running.
    while (running_jobs > 0) {
        std::this_thread::sleep_for(1ms);
    }

    // Check conditions.
    if (max_running_jobs <= width) {
         std::cout << "PASS: " << name << " (Max: " << max_running_jobs << ", Limit: " << width << ")" << std::endl;
    } else {
         std::cerr << "FAIL: " << name << " (Max: " << max_running_jobs << ", Limit: " << width << ")" << std::endl;
    }
}

int main() {
    test("Serial Queue", 1);
    test("Parallel Queue", 4);
    return 0;
}