24. IOCP(I/O Completion Port)와 워커 스레드 풀의 내부 동작

// 문제 24 — IOCP(I/O Completion Port)와 워커 스레드 풀의 내부 동작
//
// 실제 Windows IOCP API(CreateIoCompletionPort, GetQueuedCompletionStatus 등)를
// 그대로 쓰면 리눅스에서 컴파일이 안 되므로, 아래 코드는 IOCP의 핵심 동작을
// 이식 가능한 C++ 표준 라이브러리만으로 "시뮬레이션"한 것이다.
// 실제 IOCP를 다뤄본 적이 없어도, 아래 시뮬레이션 코드의 동작을 근거로
// 실제 Windows IOCP가 왜 이렇게 설계되었는지 설명할 수 있어야 한다.
//
// 아래 코드에서:
//   - CompletionPort::post()  는 실제 IOCP의 "완료 통지가 완료 큐에 쌓이는 것"에 대응한다.
//   - CompletionPort::getNext() 는 GetQueuedCompletionStatus() 호출에 대응한다.
//   - concurrencyHint 는 CreateIoCompletionPort()의 마지막 인자(NumberOfConcurrentThreads)에 대응한다.
//   - runningWorkers_ 카운터로 "현재 실제로 실행 중인(대기 상태가 아닌) 워커 수"를 시뮬레이션한다.
//
// (A) 워커 스레드 개수(NUM_WORKER_THREADS)와 concurrencyHint 값이 다르게 설정되어 있다.
// (B) OverlappedContext 안에 "다음 단계에서 다시 post 할 체이닝 콜백"이 들어있다.
//
// 과제:
// 1. 아래 코드를 컴파일 관점에서만 보고(직접 실행 결과를 몰라도), 이 프로그램의
//    워커 스레드 수가 8개인데 concurrencyHint가 2로 설정된 상황에서 실제 Windows IOCP라면
//    "동시에 실행 가능한 워커 스레드 수"가 어떻게 제한되는지 설명하라.
// 2. GetQueuedCompletionStatus (시뮬레이션에서는 getNext())가 큐가 비어 있을 때
//    워커 스레드를 어떻게 처리하는지, 그리고 완료 통지가 도착했을 때 "이미 실행 중인
//    스레드 수가 concurrencyHint 미만이면" 어떤 스레드가 깨어나는지 설명하라.
// 3. (B)의 체이닝 콜백처럼 하나의 OVERLAPPED 완료 처리 안에서 또 다른 비동기 I/O를
//    걸고 그 결과를 다시 완료 포트로 흘려보내는 패턴이 실제 게임 서버 네트워크 코드에서
//    왜 흔히 쓰이는지(recv 완료 → 다음 recv 재요청 등) 설명하라.
// 4. 워커 스레드 수를 concurrencyHint보다 훨씬 크게(cpu_count * 4 등) 잡는 이유와,
//    그럼에도 실제 "동시 실행" 스레드 수는 왜 concurrencyHint에 가깝게 유지되는지 설명하라.

#include <atomic>
#include <condition_variable>
#include <cstdio>
#include <deque>
#include <functional>
#include <mutex>
#include <thread>
#include <vector>

struct OverlappedContext {
    int connectionId;
    std::function<void(int)> onComplete; // 완료 시 실행할 콜백 (체이닝 가능)
};

class CompletionPort {
public:
    explicit CompletionPort(int concurrencyHint)
        : concurrencyHint_(concurrencyHint), runningWorkers_(0), stop_(false) {}

    // 실제 PostQueuedCompletionStatus() 에 대응
    void post(OverlappedContext ctx) {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            queue_.push_back(std::move(ctx));
        }
        cv_.notify_one();
    }

    // 실제 GetQueuedCompletionStatus() 에 대응
    bool getNext(OverlappedContext& out) {
        std::unique_lock<std::mutex> lock(mutex_);
        cv_.wait(lock, [this] { return !queue_.empty() || stop_; });
        if (stop_ && queue_.empty()) return false;
        out = std::move(queue_.front());
        queue_.pop_front();
        return true;
    }

    void shutdown() {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            stop_ = true;
        }
        cv_.notify_all();
    }

    int concurrencyHint() const { return concurrencyHint_; }
    std::atomic<int>& runningWorkers() { return runningWorkers_; }

private:
    int concurrencyHint_;
    std::atomic<int> runningWorkers_;
    bool stop_;
    std::mutex mutex_;
    std::condition_variable cv_;
    std::deque<OverlappedContext> queue_;
};

constexpr int NUM_WORKER_THREADS = 8; // (A) 워커 스레드 수
constexpr int CONCURRENCY_HINT = 2;   // (A) IOCP concurrency value

void workerLoop(CompletionPort& iocp, int workerIndex) {
    OverlappedContext ctx;
    while (iocp.getNext(ctx)) {
        iocp.runningWorkers().fetch_add(1);
        std::printf("[worker %d] conn=%d 처리 시작 (running=%d)\n",
                    workerIndex, ctx.connectionId, iocp.runningWorkers().load());

        if (ctx.onComplete) {
            ctx.onComplete(ctx.connectionId); // (B) 체이닝: 다음 recv를 다시 post
        }

        iocp.runningWorkers().fetch_sub(1);
    }
}

int main() {
    CompletionPort iocp(CONCURRENCY_HINT);

    std::vector<std::thread> workers;
    for (int i = 0; i < NUM_WORKER_THREADS; ++i) {
        workers.emplace_back(workerLoop, std::ref(iocp), i);
    }

    // 클라이언트 5개의 recv 완료를 흉내낸 초기 이벤트 post
    for (int connId = 0; connId < 5; ++connId) {
        iocp.post(OverlappedContext{
            connId,
            [&iocp](int cid) {
                // recv 완료 처리 후 "다음 recv"를 다시 완료 포트에 걸어주는 체이닝
                std::printf("  conn=%d 다음 recv 재요청\n", cid);
            }
        });
    }

    std::this_thread::sleep_for(std::chrono::milliseconds(200));
    iocp.shutdown();
    for (auto& w : workers) w.join();

    return 0;
}