Node.js ReadableStream backpressure 처리 방법

Introduction

Node.js 스트림은 대용량 데이터를 효율적으로 처리하지만, 읽기 속도와 쓰기 속도의 차이로 인해 backpressure 문제가 발생할 수 있습니다. 이번 포스트에서는 backpressure의 원인과 해결 방법을 살펴보겠습니다.

Environment

node --version
# v20.11.0

# 스트림 모니터링을 위한 모듈
npm list stream-metrics
# stream-metrics@1.0.1

Problem

대용량 파일을 처리할 때 메모리 사용량이 급격히 증가하는 문제가 발생했습니다:

const fs = require('fs');

// 문제 있는 코드
function processLargeFile(inputPath, outputPath) {
    const readStream = fs.createReadStream(inputPath);
    const writeStream = fs.createWriteStream(outputPath);

    readStream.on('data', (chunk) => {
        // backpressure 무시하고 무조건 쓰기
        writeStream.write(chunk);
    });

    readStream.on('end', () => {
        writeStream.end();
    });
}

// 메모리 사용량 급증
// 1GB 파일 처리 시 메모리 사용량: 2GB+

모니터링 결과

# 메모리 사용량 추이
Time    RSS         Heap Used
0s      50MB        30MB
30s     500MB       450MB
60s     1.2GB       1.1GB
90s     2.5GB       2.3GB

Analysis

Backpressure 발생 원인

ReadableStream ──速度快→ WritableStream
      ↑                      ↑
  데이터 생성 속도      데이터 쓰기 속도 (느림)
      │                      │
      └── Backpressure! ─────┘

스트림 동작 원리

Solution

1. pipe() 메서드 사용

const fs = require('fs');
const { pipeline } = require('stream');

// pipe() 사용 (backpressure 자동 처리)
function processLargeFile(inputPath, outputPath) {
    const readStream = fs.createReadStream(inputPath);
    const writeStream = fs.createWriteStream(outputPath);

    readStream.pipe(writeStream);

    writeStream.on('finish', () => {
        console.log('File processing completed');
    });

    writeStream.on('error', (err) => {
        console.error('Write error:', err);
    });

    readStream.on('error', (err) => {
        console.error('Read error:', err);
    });
}

2. pipeline() 사용 (권장)

const fs = require('fs');
const { pipeline, Transform } = require('stream');
const { promisify } = require('util');

const pipelineAsync = promisify(pipeline);

// Transform 스트림 정의
class UpperCaseTransform extends Transform {
    _transform(chunk, encoding, callback) {
        const uppercased = chunk.toString().toUpperCase();
        this.push(uppercased);
        callback();
    }
}

// pipeline 사용 (에러 처리 및 cleanup 자동)
async function processFile(inputPath, outputPath) {
    const readStream = fs.createReadStream(inputPath);
    const transformStream = new UpperCaseTransform();
    const writeStream = fs.createWriteStream(outputPath);

    try {
        await pipelineAsync(
            readStream,
            transformStream,
            writeStream
        );
        console.log('Pipeline completed successfully');
    } catch (err) {
        console.error('Pipeline failed:', err);
    }
}

3. 수동 backpressure 처리

const fs = require('fs');

class BackpressureHandler {
    constructor(readStream, writeStream) {
        this.readStream = readStream;
        this.writeStream = writeStream;
        this.paused = false;
    }

    handle() {
        this.readStream.on('data', (chunk) => {
            // write 스트림이 아직 쓸 수 있는지 확인
            const canContinue = this.writeStream.write(chunk);

            if (!canContinue) {
                // write 스트림이 가득 찼으므로 읽기 일시정지
                this.readStream.pause();
                this.paused = true;

                console.log('Backpressure detected, pausing read stream');

                // write 스트림이 비워질 때까지 대기
                this.writeStream.once('drain', () => {
                    console.log('Drain event, resuming read stream');
                    this.readStream.resume();
                    this.paused = false;
                });
            }
        });

        this.readStream.on('end', () => {
            this.writeStream.end();
        });

        this.readStream.on('error', (err) => {
            console.error('Read stream error:', err);
            this.writeStream.destroy(err);
        });

        this.writeStream.on('error', (err) => {
            console.error('Write stream error:', err);
            this.readStream.destroy(err);
        });
    }
}

// 사용 예시
const readStream = fs.createReadStream('large-file.txt');
const writeStream = fs.createWriteStream('output.txt');

const handler = new BackpressureHandler(readStream, writeStream);
handler.handle();

4. 고급 Transform 스트림

const { Transform } = require('stream');

class BufferedTransform extends Transform {
    constructor(options) {
        super(options);
        this.buffer = [];
        this.bufferSize = options.bufferSize || 1024 * 1024; // 1MB
        this.currentBufferSize = 0;
    }

    _transform(chunk, encoding, callback) {
        this.buffer.push(chunk);
        this.currentBufferSize += chunk.length;

        if (this.currentBufferSize >= this.bufferSize) {
            this._flushBuffer();
        }

        callback();
    }

    _flush(callback) {
        this._flushBuffer();
        callback();
    }

    _flushBuffer() {
        if (this.buffer.length > 0) {
            const combined = Buffer.concat(this.buffer);
            this.push(combined);
            this.buffer = [];
            this.currentBufferSize = 0;
        }
    }
}

// 사용 예시
const fs = require('fs');
const { pipeline } = require('stream');
const { promisify } = require('util');

const pipelineAsync = promisify(pipeline);

async function processWithBuffer(inputPath, outputPath) {
    await pipelineAsync(
        fs.createReadStream(inputPath),
        new BufferedTransform({ bufferSize: 2 * 1024 * 1024 }), // 2MB 버퍼
        fs.createWriteStream(outputPath)
    );
}

5. 스트림 모니터링

const { PassThrough } = require('stream');

class MonitoringStream extends PassThrough {
    constructor(options) {
        super(options);
        this.bytesRead = 0;
        this.bytesWritten = 0;
        this.startTime = Date.now();
    }

    _write(chunk, encoding, callback) {
        this.bytesWritten += chunk.length;
        this._logProgress();
        super._write(chunk, encoding, callback);
    }

    _logProgress() {
        const elapsed = (Date.now() - this.startTime) / 1000;
        const rate = (this.bytesWritten / 1024 / 1024 / elapsed).toFixed(2);
        process.stdout.write(`\rProcessed: ${(this.bytesWritten / 1024 / 1024).toFixed(2)} MB | Rate: ${rate} MB/s`);
    }
}

// 사용 예시
const monitoringStream = new MonitoringStream();

pipeline(
    fs.createReadStream('large-file.txt'),
    monitoringStream,
    fs.createWriteStream('output.txt'),
    (err) => {
        if (err) {
            console.error('Pipeline failed:', err);
        } else {
            console.log('\nPipeline completed');
        }
    }
);

Lessons Learned

1. pipe() 사용: 스트림 연결 시 항상 pipe() 또는 pipeline()을 사용하세요
2. backpressure 인식: 스트림의 backpressure를 인식하고 적절히 처리하세요
3. 에러 처리: 모든 스트림에 에러 핸들러를 연결하세요
4. 메모리 모니터링: 대용량 데이터 처리 시 메모리 사용량을 모니터링하세요
5. bufferSize 조정: Transform 스트림의 버퍼 크기를 데이터에 맞게 조정하세요

This blog does not accept any external sponsorships, affiliate marketing, or ad revenue.