sched -- 실시간 I/O 스케줄러 (리액터) #

#include <std/sched/reactor.h>
#include <std/sched/io_nb.h>   // non-blocking socket/file helpers meant to pair with it

#define SCHED_READ    1
#define SCHED_WRITE   2
#define SCHED_SIGNAL  3   // 이 필터에서는 'fd'가 실제로는 시그널 번호다

struct Reactor {
    int kq;   // 내부 OS 이벤트 큐 fd (kqueue fd, epoll fd; Win32 백엔드에서는 미사용)

    // 백엔드별 내부 관리용 — 필요 없는 백엔드에서는 비워진 채 사용되지
    // 않는다(각각이 왜 존재하는지는 reactor.h 참고):
    struct Vec sigfds;    // reactor_epoll.sc의 signum -> signalfd 테이블
    struct Vec watched;   // reactor_win32.sc의 등록된 (fd, filter) 테이블

    static struct Reactor new();                             // OS 큐는 아직 없음(kq = -1); 이어서 init()을 호출합니다
    int  init();                                             // 백엔드를 열고/초기화; 성공 0, 실패 -1
    void add(int fd, int filter);                             // 영구적인, 레벨 트리거 관심 등록
    void remove(int fd, int filter);
    int  poll(&TaskScheduler sched, long long timeout_ms);
        // timeout_ms: 0 = 대기 중인 것이 없으면 즉시 반환, 음수 = 무기한 블록.
        // 준비된 (fd, filter)마다 sched->unblock_fd()를 호출한다. 처리된
        // OS 이벤트 수를 반환한다(여러 태스크가 같은 fd를 기다린다면
        // 언블록된 태스크 수와 다를 수 있다).

    // poll()처럼 기다리지만, 준비된 (fd, filter)를 TaskScheduler 대신
    // on_ready(ctx, fd, filter)에 넘긴다: 다른 스케줄러(std/async의
    // IoDriver)가 리액터를 구동하는 방법. poll()은 sched->unblock_fd()를
    // 호출하는 콜백을 넘긴 이것이다. 각 백엔드는 이것만 구현한다.
    generic<T> int poll_events(fn void(?&T, int, int) on_ready, ?&T ctx, long long timeout_ms);
    void close_();   // OS 큐(epoll에서는 연 signalfd들, Windows에서는 WSAStartup)를 닫고
                     // 등록 테이블을 해제한다. 멱등: 닫힌 reactor는 Reactor::new()와 같아서
                     // init()으로 다시 열 수 있고, 다시 닫거나 drop해도 아무 일도 없다
    void drop();     // close_() — Reactor는 Drop 타입이므로 지역 변수는 모든 경로에서 drop한다
}

void reactor_run(&TaskScheduler sched, &Reactor r);

#include <std/sched/reactor.h>
#include <std/sched/io_nb.h>
#include <std/io.h>

struct Conn { int fd; };

static struct TaskScheduler g_sched;   // tasks reach the scheduler through a static

static int echo_conn(?&Conn c, int resume_point) {
    match (c) {
        case null: return 0;
        case some(conn): {
            if (resume_point == 0) {
                g_sched.await_fd(conn.fd, SCHED_READ);   // park until readable
                return 1;
            }
            // ... read/write conn.fd here ...
        }
    }
    return 0;   // done
}

int main() {
    g_sched = std::TaskScheduler::new();
    struct Reactor reactor = std::Reactor::new();
    reactor.init();

    int listenfd = std::tcp_listen_nb(8080);
    reactor.add(listenfd, SCHED_READ);

    // ... an accept task spawns echo_conn per connection:
    //     g_sched.spawn(echo_conn, conn_ref);

    std::reactor_run(&g_sched, &reactor);
    reactor.drop();
    return 0;
}

#define SCHED_AF_INET     2   // 모든 백엔드에서 동일한 값 (BSD 소켓에서
#define SCHED_SOCK_STREAM 1   // 비롯된 우연으로, 이후의 모든 API가 — Winsock을
#define SCHED_O_RDONLY 0      // 포함해 — 의도적으로 유지)
#define SCHED_O_WRONLY 1
#define SCHED_O_RDWR   2

unsigned short sched_htons(unsigned short host16);
unsigned int   sched_htonl(unsigned int host32);
unsigned int   sched_ipv4(unsigned char a, unsigned char b, unsigned char c, unsigned char d);

int fd_set_nonblocking(int fd);                         // 열린 fd에 대해 논블로킹 모드 활성화
int fd_set_blocking(int fd);                            // 블로킹 모드로 복귀 (연결 진행 중에도 안전)
int fd_open_nb(const &char path, int flags, int mode);   // 지원되는 곳에서 논블로킹 모드로 open()

int tcp_listen_nb(unsigned short port);                  // socket+bind+listen, 논블로킹; 리스닝 fd 또는 -1
int tcp_accept_nb(int listenfd);
    // 논블로킹 accept(); 아직 대기 중인 것이 없으면 -1/EAGAIN — 호출자는
    // await_fd(listenfd, SCHED_READ) 후 재시도해야 한다.
int tcp_connect_nb(unsigned int addr_network_order, unsigned short port);
    // 논블로킹 connect(); 즉시 EINPROGRESS와 함께 반환 — 호출자는
    // await_fd(fd, SCHED_WRITE)를 사용하고 쓰기 가능 상태를 "연결 완료"로
    // 취급해야 한다(성공/실패를 구분하려면 SO_ERROR를 확인).
int tcp_wait_connected(int fd);
    // tcp_connect_nb 핸드셰이크가 끝날 때까지 블록: 연결되면 0, 실패하면 -1
    // (동기 호출자용; Reactor 태스크는 대신 SCHED_WRITE를 await한다).
int tcp_connect_timeout(unsigned int addr_network_order, unsigned short port,
                        int timeout_ms);
    // 최대 timeout_ms 동안 기다리며 연결(<= 0: 추가 제한 없음); 블로킹 모드로
    // 돌린 연결된 소켓을 반환하거나 -1(소켓은 닫힘, 시간 초과면 ETIMEDOUT /
    // WSAETIMEDOUT, 아니면 연결 오류 자체).
int sock_set_io_timeout(int fd, int timeout_ms);
    // SO_RCVTIMEO + SO_SNDTIMEO: timeout_ms 동안 진전이 없는 읽기/쓰기 한 번이
    // 블록 대신 실패한다; <= 0이면 해제. 성공 시 0.
int sock_would_block();
    // 마지막 논블로킹 소켓 호출이 블록될 상황이었기 때문에만 실패했으면 1
    // (errno EAGAIN/EWOULDBLOCK, Windows는 WSAEWOULDBLOCK). 실패한 호출
    // 직후에 호출한다.