쌓이는 바이트를 직접 잰다

목표

빠른 생산자와 느린 소비자 사이에서 메모리가 어떻게 늘어나는지를 직접 재고, write() 의 반환값과 drain 이벤트가 무엇을 막아 주는지 확인합니다.

왜 중요한가

"메모리 누수" 로 신고되는 것의 상당수는 새는 것이 아니라 줄이 서 있는 것입니다. 초당 10MB 를 만들어 내는 쪽과 초당 2MB 를 받아 가는 쪽을 그냥 이어 두면, 남는 8MB 는 어딘가에 쌓여야 하고 그 어딘가가 이 프로세스의 힙입니다.

Node 의 스트림은 이것을 막을 손잡이를 하나 줍니다 — write() 의 반환값입니다. 그런데 그 값은 그냥 두면 아무 일도 하지 않습니다. 읽고 멈춰 주는 것은 쓰는 쪽의 몫입니다. 그 한 줄이 빠지면 코드는 멀쩡히 돌고, 시험도 통과하고, 운영에 나가서 큰 파일 하나에 프로세스가 죽습니다.

그리고 배압을 지키면 느려질 것 같지만, 재 보면 그렇지 않습니다. 소비자가 받는 속도는 어차피 같기 때문입니다. 잃는 것은 거의 없고 막는 것은 큽니다.

단계

/root/work/backpressure/flow.mjs 에 makeSink(options) 를 만들고, 이 판의 기본 highWaterMark 를 재어 report.json 의 hwm 에 적습니다.
fillUntilFalse(stream, chunk) 로 버퍼가 차는 지점을 셉니다.
writeAll(stream, chunks) 로 배압을 지키며 씁니다.
floodNoWait(stream, chunks) 로 반환값을 무시하고 부어 runs.flood 에 적습니다.
같은 양을 writeAll 로 보내 runs.paced 에 적고 backpressure 에 비율을 적습니다.
pipeThrough(readable, writable) 로 같은 일을 하고 runs.pipeline 에 적습니다.
estimateQueueBytes 와 secondsUntil 로 쌓이는 속도를 계산합니다.

참고

makeSink({hwm, delayMs}) 는 받은 양을 receivedBytes·receivedChunks 에 세는 Writable 입니다. delayMs 가 0 이면 setImmediate 로, 0 보다 크면 그만큼 늦게 콜백을 부릅니다.
writeAll 과 floodNoWait 은 둘 다 {peakBuffered} 를 돌려줍니다. stream.writableLength 가 그 순간 쌓여 있는 바이트입니다.
4번과 5번은 같은 양으로 재야 비교가 됩니다. 조각을 배열로 미리 만들지 말고 보낼 때마다 새로 만드세요 — 미리 만들면 재기도 전에 이미 다 쓴 상태가 되고, 같은 버퍼를 다시 쓰면 참조만 쌓여 메모리가 늘지 않습니다.
rssDeltaMB 는 process.memoryUsage().rss 의 차이를 MB 로 적은 것입니다.
흔한 실수: readable.pipe(writable) 로 끝내는 것. 배압은 지켜 주지만 오류를 올려 주지 않아서, 소비자가 죽어도 생산자는 계속 읽습니다.

이 판의 기본값을 직접 잰다

/root/work/backpressure/flow.mjs 에 makeSink({hwm, delayMs}) 를 export 하고, /root/work/backpressure/report.json 에 node 와 hwm.writableDefault·hwm.objectMode·hwm.readableDefault 를 재어서 적으세요.

new Writable({write(c, e, cb) { cb(); }}).writableHighWaterMark 를 찍어 보세요. 기억하고 있는 숫자와 다를 수 있습니다.

makeSink 는 받은 바이트와 조각 수를 receivedBytes·receivedChunks 에 세어 두고, delayMs 가 0 보다 크면 그만큼 늦게 콜백을 부릅니다.

write() 가 false 를 내는 지점

fillUntilFalse(stream, chunk) 를 export 하세요. false 가 나올 때까지 write() 를 부르고, false 를 낸 그 호출까지 포함한 횟수를 돌려줍니다.

write() 는 "이 조각을 받아 넣은 뒤 쌓인 양이 한도보다 적은가" 를 돌려줍니다. 한도와 같아지는 순간부터 false 입니다.

그래서 64KB 한도에 64KB 를 한 번 쓰면 첫 호출이 이미 false 입니다. 이 숫자가 손에 잡혀야 다음 단계의 대기 규칙이 자연스러워집니다.

멈추고, drain 을 기다린다

writeAll(stream, chunks) 를 export 하세요. 순서를 지켜 모두 쓰되 write() 가 false 면 drain 을 기다렸다가 이어 쓰고, {peakBuffered} 를 돌려줍니다.

await new Promise(r => stream.once("drain", r)) 한 줄이 배압의 전부입니다.

on 이 아니라 once 를 쓰세요. 매번 on 으로 붙이면 리스너가 쌓여서 경고가 뜨고, 결국 그것도 메모리입니다.

반환값을 무시하면 어디까지 쌓이는가

floodNoWait(stream, chunks) 를 export 하고, 64KB 조각 2000개를 반환값을 무시한 채 부어 runs.flood 에 chunks·chunkBytes·hwm·peakBuffered·bufferedMB·rssDeltaMB·wallMs·receivedChunks 를 적으세요.

peakBuffered 는 stream.writableLength 의 최댓값입니다. 부은 만큼 그대로 쌓입니다 — 한도는 아무것도 막지 못합니다.

조각을 미리 배열로 만들면 재기도 전에 메모리를 이미 다 쓴 상태가 됩니다. 제너레이터로 보낼 때마다 만드세요. rssDeltaMB 가 거의 0 이라면 그 함정에 빠진 것입니다.

같은 양을 배압을 지켜 보낸다

같은 조각 수와 크기를 writeAll 로 보내 runs.paced 에 같은 항목을 적고, backpressure 에 bufferRatio(flood/paced 의 peakBuffered, 첫째 자리), rssRatio(첫째 자리), timeRatio(paced/flood 의 wallMs, 둘째 자리)를 적으세요.

쌓인 양은 수천 배 차이가 나는데 timeRatio 는 1 근처입니다.

소비자가 받아 가는 속도는 어느 쪽이든 같기 때문입니다. 기다리는 것이 느린 것이 아닙니다 — 기다리지 않으면 그 차이가 메모리에 쌓일 뿐입니다.

실패했을 때 누가 치우는가

pipeThrough(readable, writable) 를 export 하세요. stream/promises 의 pipeline 을 써서, 다 흘려보내면 완료되고 어느 쪽이 실패하면 그 오류로 거부되어야 합니다. 같은 양을 흘려보내 runs.pipeline 에 chunks·chunkBytes·hwm·peakBuffered·wallMs·receivedChunks 를 적으세요.

pipe() 도 배압은 지켜 줍니다. 그런데 소비자가 죽으면 아무 일도 일어나지 않아서, 생산자는 계속 읽고 오류는 아무도 안 듣는 곳에서 터집니다.

pipeline 은 한쪽이 실패하면 나머지를 정리하고 오류를 부른 쪽까지 올려 줍니다. 채점기는 일부러 실패하는 소비자를 물려 그것을 확인합니다.

몇 초 뒤에 터지는가

estimateQueueBytes({producerBps, consumerBps, seconds}) 와 secondsUntil({producerBps, consumerBps, limitBytes}) 를 export 하세요. 소비가 생산보다 빠르거나 같으면 각각 0 과 Infinity 입니다. 숫자가 아닌 값이 들어오면 예외를 던집니다.

쌓이는 속도는 max(0, 생산 - 소비) 이고, 한도까지 남은 시간은 한도 / 그 속도 입니다. 산수가 전부입니다.

이 두 줄이 있으면 "메모리가 좀 올라가는데요" 대신 "지금 속도면 4분 뒤에 한도에 닿습니다" 라고 말할 수 있습니다. 용량 계획은 여기서 시작합니다.