260610_0948_대표적 TCP 장애 유형(Tokio 기반 비동기 리시브를 곁들인)

[송신 스레드] ---- 데이터 전송 (TCP Packet) ----> [커널 수신 버퍼 (Rx Buffer)] │ (소켓 읽기 지연 시 적체) ▼ [수신 스레드] <--- 소켓 읽기 (recv / read) --------- [애플리케이션 영역]

역할 분리 (Separation of Concerns) ┌──────────────────────┐ ┌──────────────────────────┐ │ 리시브 스레드 (1개) │ push │ 처리 스레드 풀 (N개) │ │ - recv() 만 수행 │ ─────▶ │ - 파싱 / 역직렬화 │ │ - 가볍고 빠름 │ 큐 │ - 비즈니스 로직 / DB / IO │ │ - 절대 블로킹 금지 │ │ - CPU 코어별 병렬 처리 │ └──────────────────────┘ └──────────────────────────┘

use std::io::{ErrorKind, Read}; use std::net::TcpStream; /// 소켓에 들어온 데이터를 EAGAIN/WouldBlock 이 발생할 때까지 모두 읽어 /// 원시 바이트 블록 단위로 큐에 밀어 넣는다. (No 파싱) fn drain_socket<F>(stream: &mut TcpStream, mut on_block: F) -> std::io::Result<()> where F: FnMut(Vec<u8>), { // 반드시 논블로킹 모드여야 한다. stream.set_nonblocking(true)?; let mut buf = [0u8; 64 * 1024]; // 64KB 버스트 읽기 loop { match stream.read(&mut buf) { Ok(0) => break, // 상대가 연결 종료 (FIN) Ok(n) => { // 파싱하지 않고 원시 바이트만 복사해 큐로 전달 on_block(buf[..n].to_vec()); } // 커널 버퍼를 모두 비웠다 → 루프 종료, 다음 epoll 이벤트 대기 Err(e) if e.kind() == ErrorKind::WouldBlock => break, // 시그널에 의한 인터럽트 → 재시도 Err(e) if e.kind() == ErrorKind::Interrupted => continue, Err(e) => return Err(e), } } Ok(()) }

// Cargo.toml // [dependencies] // crossbeam-queue = "0.3" # 락프리 MPMC/SPSC 큐 use crossbeam_queue::ArrayQueue; use std::sync::Arc; use std::thread; fn main() { // 유한 크기(Bounded) 락프리 링 버퍼 — 3.3의 "방안 B"에 해당 let queue: Arc<ArrayQueue<Vec<u8>>> = Arc::new(ArrayQueue::new(1024)); // --- 단일 생산자 (리시브 스레드) --- let producer_q = Arc::clone(&queue); let producer = thread::spawn(move || { for i in 0..10_000u32 { let raw = i.to_be_bytes().to_vec(); // 원시 바이트 블록 // push 가 실패하면(큐가 가득 참) → 백프레셔 발동 지점 (3.3 참고) while producer_q.push(raw.clone()).is_err() { // 큐가 가득 참: 리시브를 잠시 멈춰 TCP 흐름 제어 유도 std::hint::spin_loop(); } } }); // --- 다중 소비자 (LWP 워커 풀) --- let mut workers = Vec::new(); for id in 0..4 { let consumer_q = Arc::clone(&queue); workers.push(thread::spawn(move || { loop { match consumer_q.pop() { Some(raw) => process_packet(id, &raw), // 무거운 로직 전담 None => std::hint::spin_loop(), // 큐가 비어 있음 } } })); } producer.join().unwrap(); // (실제 코드에서는 종료 신호 채널로 워커를 정리) let _ = workers; } /// 처리 스레드: 파싱 + 비즈니스 로직 (CPU 바운드 작업) fn process_packet(worker_id: usize, raw: &[u8]) { // 여기서 비로소 파싱/역직렬화/비즈니스 로직 수행 let _value = u32::from_be_bytes(raw.try_into().unwrap_or([0; 4])); // ... DB / 비즈니스 처리 ... let _ = worker_id; }

use std::cell::UnsafeCell; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; /// 고정 크기 락프리 링 버퍼 (SPSC). 용량은 2의 거듭제곱이어야 마스킹이 가능. pub struct RingBuffer<T> { buffer: Box<[UnsafeCell<Option<T>>]>, capacity: usize, mask: usize, head: AtomicUsize, // 생산자만 증가 (write pointer) tail: AtomicUsize, // 소비자만 증가 (read pointer) } unsafe impl<T: Send> Sync for RingBuffer<T> {} impl<T> RingBuffer<T> { pub fn new(capacity: usize) -> Arc<Self> { assert!(capacity.is_power_of_two(), "capacity must be a power of two"); let buffer = (0..capacity) .map(|_| UnsafeCell::new(None)) .collect::<Vec<_>>() .into_boxed_slice(); Arc::new(Self { buffer, capacity, mask: capacity - 1, head: AtomicUsize::new(0), tail: AtomicUsize::new(0), }) } /// 생산자: 큐가 가득 차면 Err(value) 반환 → 백프레셔 신호 pub fn push(&self, value: T) -> Result<(), T> { let head = self.head.load(Ordering::Relaxed); let tail = self.tail.load(Ordering::Acquire); if head.wrapping_sub(tail) >= self.capacity { return Err(value); // FULL → 리시브 멈춤(의도적 Zero Window) } let slot = &self.buffer[head & self.mask]; unsafe { *slot.get() = Some(value); } // Release: 위 데이터 쓰기가 head 공개보다 먼저 보이도록 보장 self.head.store(head.wrapping_add(1), Ordering::Release); Ok(()) } /// 소비자: 큐가 비면 None pub fn pop(&self) -> Option<T> { let tail = self.tail.load(Ordering::Relaxed); let head = self.head.load(Ordering::Acquire); if tail == head { return None; // EMPTY } let slot = &self.buffer[tail & self.mask]; let value = unsafe { (*slot.get()).take() }; self.tail.store(tail.wrapping_add(1), Ordering::Release); value } }

use crossbeam_queue::ArrayQueue; use std::sync::Arc; enum PushResult { Ok, Backpressure, // 큐가 가득 참 → 리시브를 멈춰 TCP Zero Window 유도 } fn try_enqueue(queue: &Arc<ArrayQueue<Vec<u8>>>, raw: Vec<u8>) -> PushResult { match queue.push(raw) { Ok(_) => PushResult::Ok, Err(_rejected) => { // 핵심: 여기서 절대 무한 재할당(unbounded)하지 않는다. // 리시브 루프를 잠시 멈추면 커널 Rx 버퍼가 차고, // OS가 송신 측에 Window=0 을 보내 흐름을 자연스럽게 제어한다. PushResult::Backpressure } } }

use crossbeam_queue::ArrayQueue; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; use std::thread; use std::time::Duration; struct Pool { queue: Arc<ArrayQueue<Vec<u8>>>, capacity: usize, active_workers: AtomicUsize, min_workers: usize, max_workers: usize, } impl Pool { /// 큐 점유율이 70% 이상이고 코어 여유가 있으면 워커를 추가 spawn fn scale(self: &Arc<Self>) { let occupancy = self.queue.len() as f64 / self.capacity as f64; let current = self.active_workers.load(Ordering::Relaxed); if occupancy > 0.70 && current < self.max_workers { self.active_workers.fetch_add(1, Ordering::Relaxed); let pool = Arc::clone(self); let worker_id = current + 1; thread::spawn(move || pool.worker_loop(worker_id, true /* shrinkable */)); println!("[SCALE-UP] worker {} spawned (occupancy={:.0}%)", worker_id, occupancy * 100.0); } } fn worker_loop(self: Arc<Self>, id: usize, shrinkable: bool) { let idle = AtomicBool::new(false); loop { match self.queue.pop() { Some(raw) => { idle.store(false, Ordering::Relaxed); // 파싱 + 비즈니스 로직 let _ = raw; } None => { // 자원 회수: 동적 생성된 워커는 큐가 비면 최소 인원만 남기고 종료 if shrinkable && self.active_workers.load(Ordering::Relaxed) > self.min_workers { self.active_workers.fetch_sub(1, Ordering::Relaxed); println!("[SCALE-DOWN] worker {} retired", id); return; } thread::sleep(Duration::from_millis(1)); } } } } }

// Cargo.toml // [dependencies] // tokio = { version = "1", features = ["full"] } use tokio::io::AsyncReadExt; use tokio::net::{TcpListener, TcpStream}; use tokio::sync::mpsc::{self, error::TrySendError}; #[tokio::main] async fn main() -> std::io::Result<()> { let listener = TcpListener::bind("0.0.0.0:9000").await?; // 유한(bounded) 채널 = 백프레셔 내장. 큐가 차면 송신이 대기/거절됨. let (tx, mut rx) = mpsc::channel::<Vec<u8>>(1024); // --- 처리(소비자) 태스크 풀: CPU 바운드는 spawn_blocking 으로 분리 --- tokio::spawn(async move { while let Some(raw) = rx.recv().await { tokio::task::spawn_blocking(move || { // 무거운 파싱 / 비즈니스 로직 let _ = raw; }); } }); // --- 리시브(생산자) 루프 --- loop { let (socket, _addr) = listener.accept().await?; let tx = tx.clone(); tokio::spawn(async move { receive_loop(socket, tx).await }); } } async fn receive_loop(mut socket: TcpStream, tx: mpsc::Sender<Vec<u8>>) { let mut buf = vec![0u8; 64 * 1024]; loop { match socket.read(&mut buf).await { Ok(0) => break, // 연결 종료 Ok(n) => { let raw = buf[..n].to_vec(); // No 파싱 → 원시 바이트만 // try_send: 큐가 가득 차면 더 이상 읽지 않음 → 커널 Rx 버퍼가 차고 // OS가 Zero Window 로 송신 측을 자연 제어 (의도적 백프레셔) match tx.try_send(raw) { Ok(_) => {} Err(TrySendError::Full(_)) => { // 큐 포화: 잠시 대기하여 흐름 제어 유도 tx.send(buf[..n].to_vec()).await.ok(); // 또는 await로 블록 } Err(TrySendError::Closed(_)) => break, } } Err(_) => break, } } }

# 현재 설정 확인 sysctl net.ipv4.tcp_rmem # 커널 설정 파일 수정 (/etc/sysctl.conf) # [최소크기 초기크기 최대크기] 순서 설정 (최대 16MB 또는 32MB 권장) net.ipv4.tcp_rmem = 4096 87380 16777216 net.core.rmem_max = 16777216 # 설정 반영 sysctl -p

int sock = socket(AF_INET, SOCK_STREAM, 0); int rcvbuf_size = 16 * 1024 * 1024; // 16MB // 소켓 수신 버퍼 크기 강제 고정 setsockopt(sock, SOL_SOCKET, SO_RCVBUF, &rcvbuf_size, sizeof(rcvbuf_size));

// Cargo.toml // [dependencies] // socket2 = "0.5" use socket2::{Domain, Protocol, Socket, Type}; use std::net::SocketAddr; fn build_tuned_listener(addr: SocketAddr) -> std::io::Result<std::net::TcpListener> { let socket = Socket::new(Domain::IPV4, Type::STREAM, Some(Protocol::TCP))?; // 수신 버퍼 16MB 로 상향 (커널 net.core.rmem_max 한도 내에서 적용됨) socket.set_recv_buffer_size(16 * 1024 * 1024)?; // 논블로킹 + 주소 재사용 socket.set_reuse_address(true)?; socket.set_nonblocking(true)?; socket.bind(&addr.into())?; socket.listen(1024)?; Ok(socket.into()) }

[ 외부 송신 단말 ] │ │ (TCP/IP 패킷 전송) ▼ [ OS Kernel Rx Buffer ] <--- sysctl로 최대 16MB 확보 │ │ (Epoll Edge-Triggered / Non-blocking Read) - 지연 요소 0% ▼ [ 리시브 스레드 (Producer) ] │ │ (No 파싱 / 원시 바이트 블록 그대로 복사) ▼ [ 공유 큐 (Lock-Free / Ring Buffer) ] <--- 유한 크기로 메모리 보호 │ ├───────────────────────┼───────────────────────┐ (Atomic CAS Pop) ▼ ▼ ▼ [ LWP 스레드 1 ] [ LWP 스레드 2 ] [ LWP 스레드 N ] (패킷 파싱 & 비즈니스) (패킷 파싱 & 비즈니스) (동적 확장 워커)