Файлы, процессы, Node-стримы

В 05-async уже разобрана общая модель стримов (push vs pull, backpressure, async iterators). В этом уроке смотрим, как теория ложится на конкретные API Node: fs, child_process, Buffer. Цель, понять, какой инструмент подходит для какой задачи и где у каждого острые углы.

1. fs: sync, callback, promises

Историческая эволюция API:

import fs from 'node:fs';
import { promises as fsp } from 'node:fs';
import { readFile } from 'node:fs/promises';

// 1. синхронный
const data1 = fs.readFileSync('./config.json', 'utf8');

// 2. callback (старый стиль)
fs.readFile('./config.json', 'utf8', (err, data2) => {
  if (err) throw err;
  console.log(data2);
});

// 3. promise (современный стиль)
const data3 = await readFile('./config.json', 'utf8');

Все три делают одно и то же. Разница в стоимости и в применимости.

Sync. Блокирует event loop до возврата. Допустим только в момент старта процесса (читать конфиг до server.listen), в CLI-скриптах, миграциях. В HTTP-обработчике, никогда: одно readFileSync на 200 KB заморозит весь сервер на десятки миллисекунд.

Callback. Старый API. Не блокирует, но цепочка вложенных колбэков читается тяжело. В новом коде не используем.

Promise. Дефолт для современного кода. Импорт через node:fs/promises. Тоже неблокирующий, но композируется с await, try/catch, Promise.all.

Под капотом промис-API всё равно дёргает callback-API через util.promisify, разница только в обёртке.

2. Когда читать целиком, когда стримом

// читаем целиком: ок для маленького файла
const config = JSON.parse(await readFile('./config.json', 'utf8'));

// стримом: для большого, чтобы не съесть память
import { createReadStream } from 'node:fs';
import { pipeline } from 'node:stream/promises';

await pipeline(
  createReadStream('./events.jsonl'),
  parseLines(),
  countByType(),
);

Граница на глаз: до 1 MB читай через readFile, выше, через стрим. Если файл может вырасти со временем (логи, событийные записи), сразу стрим, иначе через год получишь падение по OOM.

readFile под капотом всё равно читает чанками через тред-пул libuv, просто в конце склеивает в один Buffer. Стрим не делает финальной склейки, поэтому пик памяти у него ниже.

3. Buffer и TypedArray

Buffer, специфичный для Node класс над сырой памятью. Наследник Uint8Array, поэтому всё, что работает с TypedArray, работает с Buffer.

// четыре способа создать Buffer
const a = Buffer.alloc(8);                  // нули
const b = Buffer.allocUnsafe(8);            // быстрее, но память НЕ занулена
const c = Buffer.from('hello', 'utf8');     // из строки
const d = Buffer.from([0xde, 0xad, 0xbe, 0xef]); // из массива байтов

// конвертация
b.toString('utf8');     // в строку
b.toString('hex');      // 'deadbeef'
b.toString('base64');   // base64

allocUnsafe быстрее, потому что не зануляет: память берётся из внутреннего пула. Но в ней могут оказаться следы прежних данных. Использовать только если ты сразу запишешь в буфер всё нужное (например, заполнишь read()-ом из сокета). Если есть хоть малейший шанс прочитать незаписанный байт, бери alloc.

Buffer vs ArrayBuffer. ArrayBuffer, web-стандартный сырой блок памяти. TypedArray (Int8Array, Float64Array) даёт типизированный взгляд на ArrayBuffer. DataView, для смешанных типов и явного endianness. Buffer наследует Uint8Array, поэтому одновременно view над ArrayBuffer.

const buf = Buffer.alloc(8);
const view = new DataView(buf.buffer, buf.byteOffset, buf.byteLength);

view.setUint32(0, 0xdeadbeef, /* littleEndian */ false);
console.log(buf); // <Buffer de ad be ef 00 00 00 00>

Когда писать бинарный протокол (TLV, WAT, custom RPC), лучше работать через DataView: явное указание endianness не зависит от платформы.

4. Node-стримы: pipeline, не pipe

Общая теория стримов в 05-async:streams. Здесь, акценты, специфичные для Node.

fs.createReadStream и fs.createWriteStream базовые источники и приёмники I/O. Под капотом они используют libuv-thread-pool для файловых операций (HDD/SSD), не сокетную модель.

Базовое правило для прода: всегда pipeline, никогда pipe.

import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';

await pipeline(
  createReadStream('input.log'),
  createGzip(),
  createWriteStream('input.log.gz'),
);

Что pipeline даёт сверх pipe:

• Один промис на всю цепочку, удобно await.
• Корректное закрытие при ошибке, .destroy(err) на всех звеньях.
• Поддержка { signal } для отмены через AbortController.
• Async generator как звено, прямо в pipeline.

import { pipeline } from 'node:stream/promises';

async function* uppercase(source: AsyncIterable<Buffer>) {
  for await (const chunk of source) {
    yield Buffer.from(chunk.toString().toUpperCase());
  }
}

await pipeline(
  createReadStream('input.txt'),
  uppercase,
  createWriteStream('upper.txt'),
);

Это самый чистый способ собирать пайплайны в Node 18+. Минимум boilerplate, ошибки сами поднимаются вверх.

producer

flowing

buffer

0/8

consumer

0

0 produced0 consumed0 paused0 tps

Подними producer выше consumer, буфер заполняется. Когда длина достигает hwm, producer переходит в paused: это и есть backpressure. Опустится буфер ниже половины hwm, producer возвращается в flowing. Это упрощённая модель: в Node hwm, это порог для возврата false у write(), в Web Streams похожая логика реализована через возврат промиса из pull.

5. Backpressure руками (когда pipeline не подходит)

Если нужно ручное управление, помни про write():

import { once } from 'node:events';

async function writeAll(writable: NodeJS.WritableStream, chunks: AsyncIterable<Buffer>) {
  for await (const chunk of chunks) {
    if (writable.write(chunk) === false) {
      await once(writable, 'drain');
    }
  }
  writable.end();
}

writable.write() возвращает false, когда внутренний буфер достиг highWaterMark. Это сигнал “дай мне разгрести”, а не запрет писать. Если игнорировать, буфер растёт без ограничения. На больших объёмах, OOM.