[Concurrency Patterns] 10. 이벤트 아키텍처 II: Proactor와 Half-Sync/Half-Async

1
2
3

1. "읽을 데이터 있나?" 확인 (select/poll/epoll)
2. 있으면 read() 호출 (블로킹, 하지만 준비됨)
3. 데이터 처리

1
2
3

1. "이 소켓에서 읽어줘" 요청 (즉시 반환)
2. 나중에 OS가 "읽기 완료, 데이터 있음" 통지
3. 콜백에서 데이터 처리

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

// proactor_sim.cpp
// 빌드: g++ -std=c++20 -pthread -Wall -Wextra -O2 proactor_sim.cpp -o proactor_sim
#include <future>
#include <functional>
#include <iostream>
#include <thread>
#include <vector>
#include <unistd.h>

// std::async 기반으로 Proactor의 "요청 즉시 반환 + 완료 콜백" 의미를 모사한다.
// 실제 IOCP/io_uring은 커널이 I/O를 수행하지만, 호출자 인터페이스의 의미는 동일하다.
class SimulatedProactor {
public:
    using CompletionHandler = std::function<void(ssize_t bytesRead, std::string data)>;

    // 즉시 반환. 읽기는 별도 실행 흐름(std::async)에서 수행되고,
    // 끝나면 handler가 호출된다.
    void asyncRead(int fd, size_t maxBytes, CompletionHandler handler) {
        futures_.push_back(std::async(std::launch::async,
            [fd, maxBytes, handler] {
                std::string buf(maxBytes, '\0');
                ssize_t n = read(fd, buf.data(), maxBytes);
                if (n < 0) n = 0;
                buf.resize(static_cast<size_t>(n));
                handler(n, std::move(buf));  // 완료 통지
            }));
    }

    // 등록된 모든 비동기 작업의 완료를 기다린다.
    // 실제 Proactor의 run()은 이벤트 루프이지만, 데모에서는 join에 해당한다.
    void waitAll() {
        for (auto& f : futures_) f.wait();
        futures_.clear();
    }

private:
    std::vector<std::future<void>> futures_;
};

int main() {
    int fds[2];
    pipe(fds);
    write(fds[1], "async-io", 8);
    close(fds[1]);

    SimulatedProactor proactor;

    // (1) 요청: 즉시 반환된다 — read()는 아직 끝나지 않았을 수 있다.
    proactor.asyncRead(fds[0], 64, [](ssize_t bytes, std::string data) {
        // (3) 완료 통지: 별도 실행 흐름에서 호출된다.
        std::cout << "완료: " << bytes << " bytes, data=" << data << '\n';
    });

    std::cout << "요청 발행 직후 (읽기가 끝났는지 알 수 없음)\n";

    // (2) 완료 대기
    proactor.waitAll();
    close(fds[0]);
    return 0;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

┌─────────────────────────────────────────┐
│  Synchronous Service Layer (Workers)    │
│  - CPU-intensive processing             │
│  - Multiple threads, blocking I/O ok    │
└──────────────┬──────────────────────────┘
               │ (sync queue: condition_variable 기반)
┌──────────────▼──────────────────────────┐
│  Asynchronous Service Layer (Reactor)   │
│  - Event loop, single thread            │
│  - Non-blocking I/O (poll)              │
└─────────────────────────────────────────┘

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151

// half_sync_half_async.cpp
// 빌드: g++ -std=c++20 -pthread -Wall -Wextra -O2 half_sync_half_async.cpp -o hsha
#include <atomic>
#include <condition_variable>
#include <cstring>
#include <functional>
#include <iostream>
#include <map>
#include <mutex>
#include <poll.h>
#include <queue>
#include <thread>
#include <unistd.h>
#include <vector>

// --- 동기 계층: Bounded Queue (03~04장과 동일한 패턴) ---
template <typename T>
class BoundedQueue {
public:
    explicit BoundedQueue(size_t capacity) : capacity_(capacity) {}

    void push(T item) {
        std::unique_lock<std::mutex> lock(mu_);
        notFull_.wait(lock, [this] { return queue_.size() < capacity_ || done_; });
        if (done_) return;
        queue_.push(std::move(item));
        notEmpty_.notify_one();
    }

    bool pop(T& out) {
        std::unique_lock<std::mutex> lock(mu_);
        notEmpty_.wait(lock, [this] { return !queue_.empty() || done_; });
        if (queue_.empty()) return false;  // done_ && empty -> 종료
        out = std::move(queue_.front());
        queue_.pop();
        notFull_.notify_one();
        return true;
    }

    void shutdown() {
        std::lock_guard<std::mutex> lock(mu_);
        done_ = true;
        notEmpty_.notify_all();
        notFull_.notify_all();
    }

private:
    size_t capacity_;
    std::queue<T> queue_;
    std::mutex mu_;
    std::condition_variable notEmpty_, notFull_;
    bool done_ = false;
};

// --- 비동기 계층: 09장의 PollReactor ---
class PollReactor {
public:
    using EventHandler = std::function<void(int fd)>;

    void registerHandler(int fd, EventHandler handler) {
        handlers_[fd] = std::move(handler);
    }
    void unregisterHandler(int fd) { handlers_.erase(fd); }
    void stop() { running_ = false; }

    void run() {
        running_ = true;
        while (running_ && !handlers_.empty()) {
            std::vector<struct pollfd> pfds;
            pfds.reserve(handlers_.size());
            for (auto& [fd, handler] : handlers_) pfds.push_back({fd, POLLIN, 0});

            int n = poll(pfds.data(), pfds.size(), -1);
            if (n < 0) { if (errno == EINTR) continue; break; }

            for (auto& pfd : pfds) {
                if (pfd.revents & (POLLIN | POLLHUP | POLLERR)) {
                    auto it = handlers_.find(pfd.fd);
                    if (it != handlers_.end()) it->second(pfd.fd);
                }
            }
        }
    }

private:
    std::map<int, EventHandler> handlers_;
    bool running_ = false;
};

// --- Half-Sync/Half-Async 서버 ---
class HalfSyncHalfAsyncServer {
public:
    HalfSyncHalfAsyncServer(size_t numWorkers, size_t queueCapacity)
        : taskQueue_(queueCapacity) {
        for (size_t i = 0; i < numWorkers; ++i) {
            workers_.emplace_back([this] {
                std::function<void()> task;
                while (taskQueue_.pop(task)) {
                    task();  // CPU 바운드 처리: 워커 스레드, blocking I/O 허용
                }
            });
        }
    }

    ~HalfSyncHalfAsyncServer() {
        taskQueue_.shutdown();
        for (auto& w : workers_) w.join();
    }

    // 비동기 계층: 이벤트가 도착하면 짧게 읽고, 처리는 워커에 위임한다.
    void onDataReady(int fd) {
        char buf[256];
        ssize_t n = read(fd, buf, sizeof(buf));
        if (n <= 0) {
            reactor_.unregisterHandler(fd);
            close(fd);
            return;
        }
        std::string data(buf, static_cast<size_t>(n));
        taskQueue_.push([data] {
            // 동기 계층: CPU 바운드 처리 (파싱, 비즈니스 로직)
            std::cout << "[worker " << std::this_thread::get_id()
                      << "] processed: " << data << '\n';
        });
    }

    void registerFd(int fd) {
        reactor_.registerHandler(fd, [this](int f) { onDataReady(f); });
    }

    void run() { reactor_.run(); }  // 비동기 계층: 이벤트 루프 (단일 스레드)

private:
    PollReactor reactor_;
    BoundedQueue<std::function<void()>> taskQueue_;
    std::vector<std::thread> workers_;
};

int main() {
    HalfSyncHalfAsyncServer server(/*numWorkers=*/4, /*queueCapacity=*/16);

    int fds[2];
    pipe(fds);
    write(fds[1], "request-1", 9);
    close(fds[1]);

    server.registerFd(fds[0]);
    server.run();  // handlers_가 비면 종료 (데모용)
    close(fds[0]);
    return 0;
}