I/O 멀티플렉싱

fd_set read_fds;
FD_ZERO(&read_fds);
FD_SET(server_fd, &read_fds);                  /* 새 연결을 기다리는 리스닝 소켓 */
int max_fd = server_fd;
for (int i = 0; i < num_clients; i++) {        /* 매번 집합을 처음부터 다시 만든다 */
    FD_SET(client_fds[i], &read_fds);
    if (client_fds[i] > max_fd) max_fd = client_fds[i];
}
int ready = select(max_fd + 1, &read_fds, NULL, NULL, NULL);
if (ready < 0 && errno != EINTR) { perror("select"); exit(1); }
if (FD_ISSET(server_fd, &read_fds)) {
    int fd = accept(server_fd, NULL, NULL);
    if (fd >= FD_SETSIZE) close(fd);           /* fd_set에 넣을 수 없는 번호 */
    else if (fd >= 0) client_fds[num_clients++] = fd;
}
for (int i = 0; i < num_clients; i++) {
    if (FD_ISSET(client_fds[i], &read_fds)) {
        /* 이 연결에서 recv()를 불러도 멈추지 않는다 */
    }
}

struct pollfd fds[MAX_CLIENTS];
fds[0].fd = server_fd;
fds[0].events = POLLIN;                        /* 읽을 수 있게 되면 알려 달라 */
int ret = poll(fds, nfds, -1);                 /* -1이면 하나라도 준비될 때까지 기다린다 */
for (int i = 0; i < nfds; i++) {
    if (fds[i].revents & POLLIN) {
        /* 준비된 소켓 */
    }
}

int epfd = epoll_create1(0);
struct epoll_event ev = { .events = EPOLLIN, .data.fd = server_fd };
epoll_ctl(epfd, EPOLL_CTL_ADD, server_fd, &ev);     /* 한 번 등록하면 커널이 기억한다 */

struct epoll_event events[MAX_EVENTS];
for (;;) {
    int n = epoll_wait(epfd, events, MAX_EVENTS, -1);   /* 준비된 것만 돌아온다 */
    for (int i = 0; i < n; i++) {
        int fd = events[i].data.fd;
        /* fd가 리스닝 소켓이면 accept 후 EPOLL_CTL_ADD, 아니면 recv */
    }
}

import selectors
import socket
import struct
import threading

MAX_FRAME, MAX_QUEUE = 64 * 1024, 256 * 1024
SEND_LIMIT = 2                                         # send 한 번에 넘기는 최대 바이트(실험용)
sel = selectors.DefaultSelector()
closed = []

def close_client(client, reason):
    closed.append(reason)
    sel.unregister(client)                             # 닫기 전에 감시 목록에서 뺀다
    client.close()

def update_interest(client, state):
    events = 0 if state["eof"] else selectors.EVENT_READ
    if state["output"]:
        events |= selectors.EVENT_WRITE                # 보낼 것이 있을 때만 쓰기 준비를 기다린다
    if events:
        sel.modify(client, events, data=state)
    else:
        close_client(client, "EOF; output drained")

def queue_frames(state):
    incoming, outgoing = state["input"], state["output"]
    while len(incoming) >= 4:
        (length,) = struct.unpack("!I", incoming[:4])
        if length > MAX_FRAME:
            raise ValueError("frame too large")
        size = 4 + length
        if len(incoming) < size:
            return                                     # 덜 온 프레임은 다음 읽기까지 남겨 둔다
        incoming[4:size].decode("utf-8")               # 본문이 UTF-8인지 확인한다
        if len(outgoing) + size > MAX_QUEUE:
            raise ValueError("output queue limit")
        outgoing.extend(incoming[:size])               # 받은 프레임을 그대로 출력 큐에 넣는다
        del incoming[:size]

def service(client, state, mask):
    try:
        if mask & selectors.EVENT_READ and not state["eof"]:
            try:
                chunk = client.recv(4096)
            except BlockingIOError:
                chunk = None                           # 알림을 받았어도 읽을 것이 없을 수 있다
            if chunk:
                state["input"].extend(chunk)
                queue_frames(state)
            elif chunk == b"":
                state["eof"] = True                    # 상대가 보내기를 마쳤다
                if state["input"]:
                    raise ValueError("EOF inside frame")
        if mask & selectors.EVENT_WRITE and state["output"]:
            try:
                sent = client.send(state["output"][:SEND_LIMIT])
                del state["output"][:sent]             # 실제로 보낸 만큼만 큐에서 지운다
            except BlockingIOError:
                pass                                   # 남은 출력은 다음 쓰기 준비 때 보낸다
        update_interest(client, state)
    except (OSError, ValueError) as error:
        close_client(client, str(error))

def run():
    while True:
        for key, mask in sel.select():
            if key.data is None:                       # 리스닝 소켓
                try:
                    client, _ = key.fileobj.accept()
                except BlockingIOError:
                    continue
                client.setblocking(False)
                state = {"input": bytearray(), "output": bytearray(), "eof": False}
                sel.register(client, selectors.EVENT_READ, data=state)
            else:
                service(key.fileobj, key.data, mask)

server = socket.create_server(("127.0.0.1", 0))
server.setblocking(False)
sel.register(server, selectors.EVENT_READ, data=None)
threading.Thread(target=run, daemon=True).start()

messages = ["", "한글🙂"] + [f"m{i}" for i in range(40)]
expected = b"".join(struct.pack("!I", len(m.encode())) + m.encode() for m in messages)
with socket.create_connection(server.getsockname(), timeout=5) as client:
    for offset in range(0, len(expected), 3):
        client.sendall(expected[offset:offset + 3])    # 3바이트씩 잘게 나누어 쓴다
    client.shutdown(socket.SHUT_WR)
    received = bytearray()
    while part := client.recv(3):                      # 3바이트씩 읽는다
        received.extend(part)
print("같은 바이트를 돌려받았는가:", received == expected, f"({len(messages)}개 프레임, {len(received)}바이트)")
print("서버가 연결을 닫은 이유:", closed)

같은 바이트를 돌려받았는가: True (42개 프레임, 288바이트)
서버가 연결을 닫은 이유: ['EOF; output drained']