Собираем читающий и пишущий поток

Файл в Node можно прочитать двумя способами. readFile из node:fs/promises отдаёт его целиком — одним Buffer, объектом Node для байтов. createReadStream из node:fs отдаёт тот же файл кусками: каждый кусок приходит событием data, а о конце сообщает событие end. Код с readFile короче, и на файле настроек разницы не видно. Она видна на большом файле.

Файл в примере — видео на 512 МиБ, которого нет: truncate растягивает пустой файл до нужной длины, не записывая байты. Читается такой файл как настоящий, а на дисках Linux и macOS места почти не занимает. Память под байты Node считает отдельно от кучи JavaScript, в поле arrayBuffers у process.memoryUsage(). Прогноз: на сколько вырастет эта память при каждом способе, сколько кусков отдаст поток и что readFile сделает с файлом на 3 ГиБ?

import { createReadStream } from "node:fs";
import { readFile, rm, truncate, writeFile } from "node:fs/promises";

const MiB = 1024 * 1024;
const file = "video.bin";
// truncate дописывает пустоту, а не байты: в Linux и macOS файл на 512 МиБ почти не занимает диска
await writeFile(file, "");
await truncate(file, 512 * MiB);

// байты Buffer лежат вне кучи V8, их считает отдельный счётчик
const buffers = () => process.memoryUsage().arrayBuffers;

let before = buffers();
const whole = await readFile(file);
console.log("readFile: кусков 1, длина", whole.length);
console.log("readFile: память под буферы выросла на весь файл:", buffers() - before >= 512 * MiB);

before = buffers();
let pieces = 0;
let biggest = 0;
let peak = 0;
await new Promise((resolve, reject) => {
  createReadStream(file)
    .on("data", (chunk) => {
      pieces += 1;
      biggest = Math.max(biggest, chunk.length);
      peak = Math.max(peak, buffers() - before);
    })
    .on("end", resolve)
    .on("error", reject);
});
console.log("поток: кусков", pieces, "самый большой", biggest);
console.log("поток: память выросла меньше чем на четверть файла:", peak < 128 * MiB);

await truncate(file, 3 * 1024 * MiB);
try {
  await readFile(file);
} catch (err) {
  console.log("readFile на 3 ГиБ:", err.message);
}
await rm(file);

readFile: кусков 1, длина 536870912
readFile: память под буферы выросла на весь файл: true
поток: кусков 8192 самый большой 65536
поток: память выросла меньше чем на четверть файла: true
readFile на 3 ГиБ: File size (3221225472) is greater than 2 GiB

readFile отдал файл одним куском в 536 870 912 байт, и память под буферы выросла на весь файл. Поток отдал те же байты 8192 кусками по 64 КиБ, и память выросла меньше чем на четверть файла. Поток куски не копит: кусок живёт, пока его обрабатывает слушатель, а дальше становится мусором, и рост памяти — это прочитанные куски, до которых сборщик мусора ещё не дошёл.

Последняя строка — предел: файл больше 2 ГиБ readFile не отдаёт вовсе. Но и до предела счёт идёт не на один файл. Сервер, который отдаёт видео через readFile, держит в памяти по файлу на каждого, кто скачивает: двадцать клиентов — 10 ГиБ. И каждый ждёт первый байт, пока файл не прочитан целиком, хотя первые 64 КиБ можно было отправить сразу. Поток меняет обе величины: память — на кусок, ожидание — до первого куска. Платить приходится кодом, который умеет работать с данными порциями. Такой код и соберём.

Поток в этой главе — stream, данные, которые приходят кусками. Главный поток и пул потоков из статьи про среду Node — это threads, исполнители кода. По-русски оба зовутся потоками, и путать их легко.

Поток на излучателе напрашивается сам: источник зовёт emit("data", кусок), потребитель подписывается через on. Излучатель берём целиком из последнего раздела прошлой статьи — с копией списка, once, броском на error и счётчиком слушателей. Дальше он лежит в подложке и в коде примеров не повторяется.

Обычный обработчик запроса сначала проверяет токен — идёт в базу — и только потом читает тело. Пока база отвечает, сеть успевает прислать тело. Прогноз: что напечатает пример и дойдёт ли тело до обработчика?

// Emitter из статьи про излучатель — в подложке
function checkToken() {
  // ответ базы приходит не сразу
  return new Promise((resolve) => setTimeout(resolve, 10));
}

async function handle(req) {
  await checkToken();
  req.on("data", (chunk) => console.log("кусок:", chunk));
  console.log("подписались на data");
}

const req = new Emitter();
handle(req);

// тело приходит по сети, пока обработчик ждёт базу
console.log("emit вернул", req.emit("data", '{"title":'));
console.log("emit вернул", req.emit("data", '"хлеб"}'));

Оба emit вернули false: слушателей не было, и излучатель, как ему и положено, ничего не запомнил. Обработчик подписался через десять миллисекунд, когда тело уже пролетело мимо, и будет ждать его вечно. На сервере это не редкий случай, а обычный порядок: между приходом байтов и подпиской почти всегда стоит await.

Лечение — буфер. Источник не испускает кусок, а кладёт его в массив методом push. Раздавать куски поток начинает, когда его попросили, и просьба в Node — сама подписка на data: on зовёт resume. resume ставит флаг flowing — поток течёт — и зовёт flow, который отдаёт слушателям накопленное по одному куску в порядке прихода. Пока поток течёт, push тоже зовёт flow, и новый кусок уходит сразу. Одна тонкость: resume включает течение не на месте, а микрозадачей через queueMicrotask. Прогноз: в каком порядке встанут четыре строки?

class Readable extends Emitter {
  buffer = [];
  flowing = false;

  push(chunk) {
    this.buffer.push(chunk);
    this.flow();
  }

  on(name, fn) {
    super.on(name, fn);
    if (name === "data") this.resume();
    return this;
  }

  resume() {
    queueMicrotask(() => {
      this.flowing = true;
      this.flow();
    });
  }

  flow() {
    if (!this.flowing) return;
    while (this.buffer.length > 0) this.emit("data", this.buffer.shift());
  }
}

function checkToken() {
  return new Promise((resolve) => setTimeout(resolve, 10));
}

async function handle(req) {
  await checkToken();
  req.on("data", (chunk) => console.log("кусок:", chunk));
  console.log("подписались на data");
}

const req = new Readable();
handle(req);
req.push('{"title":');
req.push('"хлеб"}');
console.log("в буфере кусков:", req.buffer.length);

Тело дошло. Пока обработчик ждал базу, два куска лежали в буфере, а когда он подписался, поток отдал их по порядку. Строка «подписались на data» стоит раньше кусков, хотя подписка случилась строкой выше: течение началось микрозадачей, после синхронного кода обработчика.

Зачем откладывать? Обработчик подписывается на несколько событий подряд: data, следом end и error. Раздай поток буфер прямо внутри первого on, короткое тело пролетело бы целиком — вместе с концом данных, о котором следующий раздел, — пока подписки на end ещё нет. Микрозадача ждёт, пока синхронный код допишет подписки. Node откладывает так же, только через process.nextTick — свою очередь, которую разбирает глава про цикл событий.

Буфер отдаёт кусок один раз: shift вынимает его из массива. Слушатель, подписанный позже, получит только новые куски. Отсюда правило, знакомое по Request: тело запроса вычерпывается один раз.

Как сказать, что кусков больше не будет? В Node источник зовёт push(null), а потребитель слушает end — там обработчик разбирает собранное тело и отвечает. Первое, что приходит в голову, — испустить end прямо в push(null). Readable из прошлого примера лежит в подложке, наивный конец дописан в наследнике. Короткое тело пришло одним пакетом: два куска и конец — подряд. Прогноз: что обработчик увидит в end?

// Readable с буфером из прошлого примера — в подложке
class NaiveEnd extends Readable {
  push(chunk) {
    if (chunk === null) this.emit("end");
    else super.push(chunk);
  }
}

const req = new NaiveEnd();
let body = "";
req.on("data", (chunk) => {
  body += chunk;
});
req.on("end", () => console.log(`end, тело: «${body}»`));

// короткое тело пришло целиком, одним пакетом
req.push('{"title":');
req.push('"хлеб"}');
req.push(null);
setTimeout(() => console.log(`через 10 мс тело: «${body}»`), 10);

Обработчик получил end с пустым телом, а через 10 мс тело оказалось целым. Куски ждали в буфере — течение начинается микрозадачей, — а push(null) испустил end сразу, раньше них. Сервер ответил бы «нужно поле title» на правильный запрос. Ошибка не в порядке вызовов, а в смысле события: end должен значить не «источник сказал, что всё», а «потребитель получил всё».

Поэтому push(null) только ставит пометку ended, а end испускает flow, когда буфер опустел. Флаг endEmitted не даёт испустить конец дважды: flow зовут и push, и resume. После пометки push кусков не принимает: опоздавший кусок — ошибка источника, и поток испускает error. Ниже Readable целиком, со всеми правками. Прогноз: какая строка напечатается первой и что станет с опоздавшим куском?

class Readable extends Emitter {
  buffer = [];
  flowing = false;
  ended = false;
  endEmitted = false;

  push(chunk) {
    if (this.ended) {
      this.emit("error", new Error("push после конца потока"));
      return;
    }
    if (chunk === null) this.ended = true;
    else this.buffer.push(chunk);
    this.flow();
  }

  on(name, fn) {
    super.on(name, fn);
    if (name === "data") this.resume();
    return this;
  }

  resume() {
    queueMicrotask(() => {
      this.flowing = true;
      this.flow();
    });
  }

  flow() {
    if (!this.flowing) return;
    while (this.buffer.length > 0) this.emit("data", this.buffer.shift());
    // конец — не сама пометка, а пометка плюс пустой буфер
    if (this.ended && !this.endEmitted) {
      this.endEmitted = true;
      this.emit("end");
    }
  }
}

const req = new Readable();
let body = "";
req.on("data", (chunk) => {
  body += chunk;
});
req.on("end", () => console.log(`end, тело: «${body}»`));

req.push('{"title":');
req.push('"хлеб"}');
req.push(null);

try {
  req.push("опоздавший кусок");
} catch (err) {
  console.log("вылетело:", err.message);
}

Тело целое. А первой напечаталась строка про опоздавший кусок: слушателя error нет, и излучатель бросил ошибку из emit — правило про error без слушателя из прошлой статьи. Этот push — синхронный код, а data и end пришли после него, из микрозадачи.

У нового смысла end есть обратная сторона: поток, который никто не читает, не кончается. Скажем, обработчик отклонил запрос по токену, тело ему не нужно, и на data он не подписался — только на end, чтобы после него закрыть запрос. Readable со всеми правками теперь в подложке. Прогноз: придёт ли end и что изменит resume?

// Emitter, Readable и Writable — в подложке
const req = new Readable();
let ended = false;
req.on("end", () => {
  ended = true;
});

req.push("тело, которое обработчику не нужно");
req.push(null);

setTimeout(() => {
  console.log("end пришёл:", ended, "| в буфере кусков:", req.buffer.length);
  req.resume();
  setTimeout(() => console.log("после resume end пришёл:", ended, "| в буфере кусков:", req.buffer.length), 10);
}, 10);

Без читателя кусок лежит в буфере, и end не приходит, сколько ни жди. resume без слушателя data пускает течение вхолостую: flow отдаёт куски в emit, которому некого звать, буфер пустеет, и конец приходит. В Node метод тот же, и req.resume() — штатный способ сказать «тело не нужно»: без него обработчик, который ждёт end, ждёт вечно.

Теперь в обратную сторону. Ответ сервера и файл на диске — пишущие потоки: HTTP-ответ можно отдавать кусками, ещё не зная, чем он кончится, и файл пишется так же. Само устройство — сокет или диск — работает асинхронно, и каждая запись занимает своё время. Запись файла Node и вовсе отправляет в пул, где несколько записей идут одновременно. Наивный пишущий поток отдаёт каждый кусок устройству сразу, а finish — событие «всё записано» — испускает в end. Три куска пишутся за 30, 5 и 15 мс. Прогноз: что будет в файле к finish и через 50 мс?

// Emitter — в подложке
// диск: каждая запись занимает своё время
const delays = { "[1]": 30, "[2]": 5, "[3]": 15 };
let file = "";
function disk(chunk, done) {
  setTimeout(() => {
    file += chunk;
    done();
  }, delays[chunk]);
}

class NaiveWritable extends Emitter {
  write(chunk) {
    disk(chunk, () => {});
  }

  end() {
    this.emit("finish");
  }
}

const out = new NaiveWritable();
out.on("finish", () => console.log(`finish, в файле: «${file}»`));
out.write("[1]");
out.write("[2]");
out.write("[3]");
out.end();
setTimeout(() => console.log(`через 50 мс в файле: «${file}»`), 50);

К finish файл пуст, а через 50 мс куски в нём стоят в порядке готовности, а не в порядке вызовов: второй записался быстрее первого и встал впереди. С настоящим файлом бывает то же. Документация fs.write прямо называет небезопасным повторный вызов на том же файле, пока не пришёл колбэк предыдущего, и для такого случая советует пишущий поток.

Пишущий поток чинит обе беды одним приёмом — очередью. write кладёт кусок в буфер, а next отдаёт устройству следующий, только когда предыдущий записан. Узнаёт поток об этом из колбэка: функция записи получает кусок и done и обязана позвать done, когда закончила. end ставит пометку ending, а finish приходит, когда пометка стоит, записи нет и очередь пуста. Функцию записи поток получает в конструкторе, так же как в Node. Прогноз: сколько кусков ждёт в очереди сразу после end и что будет в файле к finish?

class Writable extends Emitter {
  buffer = [];
  writing = false;
  ending = false;
  finished = false;

  constructor({ write }) {
    super();
    this.writeChunk = write;
  }

  write(chunk) {
    if (this.ending) {
      this.emit("error", new Error("write после end"));
      return;
    }
    this.buffer.push(chunk);
    this.next();
  }

  end() {
    this.ending = true;
    this.next();
  }

  next() {
    if (this.writing) return;
    if (this.buffer.length === 0) {
      if (this.ending && !this.finished) {
        this.finished = true;
        this.emit("finish");
      }
      return;
    }
    this.writing = true;
    this.writeChunk(this.buffer.shift(), () => {
      this.writing = false;
      this.next();
    });
  }
}

const delays = { "[1]": 30, "[2]": 5, "[3]": 15 };
let file = "";
const out = new Writable({
  write(chunk, done) {
    setTimeout(() => {
      file += chunk;
      done();
    }, delays[chunk]);
  },
});

out.on("finish", () => console.log(`finish, в файле: «${file}»`));
out.write("[1]");
out.write("[2]");
out.write("[3]");
out.end();
console.log("ждут в очереди:", out.buffer.length);

Первый кусок ушёл на запись сразу, внутри write, а два ждали в очереди. Записи шли по одной и в порядке вызовов, и finish пришёл после последнего done. Медленная первая запись задержала остальные, но не перемешала их. В этом и договор finish: всё, что передали в write, устройство приняло.

Очередь держится на колбэке, и отсюда самая тихая поломка пишущего потока. Функция записи ниже пропускает пустой кусок — писать его незачем — и выходит, не позвав done. Emitter, Readable и Writable со всеми правками лежат в подложке. Прогноз: что будет в файле, в очереди и с finish?

// Emitter, Readable и Writable — в подложке
let file = "";
const out = new Writable({
  write(chunk, done) {
    if (chunk === "") return; // пустой кусок писать незачем
    setTimeout(() => {
      file += chunk;
      done();
    }, 5);
  },
});

let finished = false;
out.on("finish", () => {
  finished = true;
});
out.write("[1]");
out.write("");
out.write("[2]");
out.end();

setTimeout(() => {
  console.log(`в файле: «${file}»`);
  console.log("ждут в очереди:", out.buffer.length, "| finish:", finished);
}, 50);

Записался только первый кусок. Пустой занял запись и не отпустил её: writing остался true, и «[2]» ждёт за записью, которая никогда не кончится. finish нет, ошибки нет, исключения нет. На сервере так выглядит ответ, который «иногда не доходит»: клиент ждёт до таймаута, а в журнале пусто. Правило простое: у функции записи каждая ветка кончается вызовом done, и та, где записывать нечего, тоже.

Осталось соединить. Отдать файл клиенту — значит связать читающий поток диска с пишущим потоком сокета: каждый data превращается в write, end — в end. Диск отдаёт куски быстро, а клиент на мобильном интернете принимает медленно. В примере 64 куска по 65 536 символов — как 64 КиБ в куске у createReadStream. Прогноз: сколько кусков ждёт в очереди записи, когда источник закончил?

// Emitter, Readable и Writable — в подложке
const source = new Readable(); // диск: отдаёт куски быстро
let written = 0;
let longestQueue = 0;
const socket = new Writable({
  // медленный клиент: каждый кусок уходит по сети за 4 мс
  write(chunk, done) {
    setTimeout(() => {
      written += 1;
      done();
    }, 4);
  },
});

source.on("data", (chunk) => {
  socket.write(chunk);
  longestQueue = Math.max(longestQueue, socket.buffer.length);
});
source.on("end", () => {
  console.log("источник закончил, ждут в очереди записи:", socket.buffer.length);
  socket.end();
});
socket.on("finish", () => console.log("finish, отправлено:", written, "| очередь доходила до", longestQueue));

for (let i = 1; i <= 64; i++) source.push("x".repeat(64 * 1024));
source.push(null);

В Node у write есть возвращаемое значение, которого в миниатюре нет: false значит «очередь полна, подожди». Как источник его слушает и зачем нужны pause и событие drain — следующая статья, про обратное давление. А ещё ручное соединение не видит ошибок: оборвись сокет, источник продолжит читать диск впустую. Соединение, которое убирает за собой, — в статье про pipeline.

Сверим правила с Node. В файле ниже шесть проверок на Readable и Writable из node:stream: тело, пришедшее раньше подписки, поток без читателя и resume, push после конца, три записи разной продолжительности, ветка без done и write после end.

Код проверок чуть длиннее миниатюры. Настоящему Readable передают функцию read: Node зовёт её, когда хочет данных, — здесь она пустая, куски кладёт сам пример. Опция encoding: "utf8" отдаёт куски строками, а decodeStrings: false не даёт Writable превращать строки в Buffer. У функции записи между куском и done стоит кодировка. once(поток, "end") из node:events — промис на первое событие. Прогноз: совпадёт ли вывод с примерами статьи?

import { once } from "node:events";
import { Readable, Writable } from "node:stream";

const wait = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

// 1: тело пришло раньше подписки
const req = new Readable({ read() {}, encoding: "utf8" });
req.push('{"title":');
req.push('"хлеб"}');
req.push(null);
let body = "";
req.on("data", (chunk) => {
  body += chunk;
});
req.on("end", () => console.log(`1. end, тело: «${body}»`));
console.log(`1. подписались, тело пока: «${body}»`);
await once(req, "end");

// 2: поток, который никто не читает
const unread = new Readable({ read() {} });
unread.push("тело, которое обработчику не нужно");
unread.push(null);
let ended = false;
unread.on("end", () => {
  ended = true;
});
await wait(20);
console.log("2. end без читателя пришёл:", ended);
unread.resume();
await wait(20);
console.log("2. после resume end пришёл:", ended);

// 3: push после конца
const late = new Readable({ read() {} });
late.on("error", (err) => console.log("3. error:", err.code));
late.push(null);
late.push("опоздавший кусок");
await wait(20);

// 4: три записи разной длины
const delays = { "[1]": 30, "[2]": 5, "[3]": 15 };
let file = "";
const out = new Writable({
  decodeStrings: false,
  write(chunk, encoding, done) {
    setTimeout(() => {
      file += chunk;
      done();
    }, delays[chunk]);
  },
});
out.write("[1]");
out.write("[2]");
out.write("[3]");
out.end();
await once(out, "finish");
console.log(`4. finish, в файле: «${file}»`);

// 5: ветка без done
let stuckFile = "";
const stuck = new Writable({
  decodeStrings: false,
  write(chunk, encoding, done) {
    if (chunk === "") return;
    setTimeout(() => {
      stuckFile += chunk;
      done();
    }, 5);
  },
});
let finished = false;
stuck.on("finish", () => {
  finished = true;
});
stuck.write("[1]");
stuck.write("");
stuck.write("[2]");
stuck.end();
await wait(50);
console.log(`5. в файле: «${stuckFile}» | finish пришёл: ${finished}`);

// 6: write после end
const closed = new Writable({
  write(chunk, encoding, done) {
    done();
  },
});
closed.on("error", (err) => console.log("6. error:", err.code));
closed.end();
closed.write("поздно");
await wait(20);

1. подписались, тело пока: «»
1. end, тело: «{"title":"хлеб"}»
2. end без читателя пришёл: false
2. после resume end пришёл: true
3. error: ERR_STREAM_PUSH_AFTER_EOF
4. finish, в файле: «[1][2][3]»
5. в файле: «[1]» | finish пришёл: false
6. error: ERR_STREAM_WRITE_AFTER_END

Совпало всё, что показывали примеры. Куски, пришедшие до подписки, дождались её, и строка про подписку стоит раньше тела. Поток без читателя не кончился, а resume его дочитал. Записи легли по порядку, а на ветке без done запись встала: в файле один «[1]», и finish не пришёл. push после конца и write после end — ошибки с кодами ERR_STREAM_PUSH_AFTER_EOF и ERR_STREAM_WRITE_AFTER_END. Только миниатюра испускает error прямо внутри вызова, а Node — после текущего кода, и без слушателя error исключение роняет не вызов, а весь процесс.

Чем ещё оригинал отличается от миниатюры — в таблице.

Потоки файлов и сокетов, Web Streams браузера и for await по потоку — в статье про настоящие потоки. А тело запроса, которое сегодня приходило двумя строками, в главе про HTTP придёт настоящими кусками сокета, и разбирать его придётся самим.