Обратное давление: почему write() возвращает false и что такое drain

Пишущему потоку нужен предел. Проще всего отказывать: очередь полна — кусок не принят. Но к вызову write кусок уже создан и лежит в памяти у того, кто пишет. Отказ не освободил бы память, а переложил бы кусок в чужую очередь, и свою очередь пришлось бы заводить каждому, кто пишет в поток. Поэтому Node кусок берёт всегда, а возвращаемым значением говорит, стоит ли давать следующий: true — давай, false — очередь дошла до порога, подожди. Порог называется highWaterMark, «отметка высокой воды».

В миниатюре появились счётчик length — сколько символов ждёт записи — и порог в конструкторе. write прибавляет длину куска и отвечает, меньше ли очередь порога. Вычитает её не next, когда отдаёт кусок устройству, а колбэк done: кусок на записи ещё занимает память. У ответа есть пара. Сказав false, поток ставит пометку needDrain и, когда очередь опустеет, испускает событие drain: «можно продолжать». Порог в примере — 16 символов, чтобы переполнение было видно на четырёх кусках по восемь; у Node по умолчанию — 64 КиБ. Прогноз: что вернёт каждый write, примет ли поток третий и четвёртый кусок и когда придёт drain?

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

  constructor({ write, highWaterMark = 64 * 1024 }) {
    super();
    this.writeChunk = write;
    this.highWaterMark = highWaterMark;
  }

  write(chunk) {
    if (this.ending) {
      this.emit("error", new Error("write после end"));
      return false;
    }
    this.length += chunk.length;
    this.buffer.push(chunk);
    this.next();
    // ответ после попытки записи: кусок, записанный сразу, место уже освободил
    const ok = this.length < this.highWaterMark;
    if (!ok) this.needDrain = true;
    return ok;
  }

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

  next() {
    if (this.writing) return;
    if (this.buffer.length === 0) {
      if (this.needDrain && !this.ending) {
        this.needDrain = false;
        this.emit("drain");
      }
      if (this.ending && !this.finished) {
        this.finished = true;
        this.emit("finish");
      }
      return;
    }
    this.writing = true;
    const chunk = this.buffer.shift();
    this.writeChunk(chunk, () => {
      this.writing = false;
      // кусок держит память, пока устройство не позвало done
      this.length -= chunk.length;
      this.next();
    });
  }
}

const out = new Writable({
  highWaterMark: 16,
  write(chunk, done) {
    setTimeout(done, 5);
  },
});
out.on("drain", () => console.log("drain, в очереди:", out.length));

for (const chunk of ["кусок-1!", "кусок-2!", "кусок-3!", "кусок-4!"]) {
  const ok = out.write(chunk);
  console.log(`write ${chunk}`, ok, `${out.length}/${out.highWaterMark}`);
}

Первый write ответил true: в очереди 8 из 16, и это кусок, который уже пишется. Второй довёл очередь до порога и ответил false. Третий и четвёртый тоже ответили false, но очередь выросла до 32. false — не отказ, а просьба, и тот, кто её не слышит, копит очередь так же, как миниатюра прошлой статьи. Порог защищает память, только если пишущий слушает ответ.

drain пришёл один раз, при пустой очереди. Когда записался второй кусок, в очереди осталось 16, после третьего — 8, уже меньше порога, но события не было: next испускает drain, только когда очередь пуста. Приходи drain на первом шаге ниже порога, источник дописал бы один кусок, снова упёрся бы в порог и ждал следующего drain: «стоп» и «пуск» чередовались бы на каждом куске. Пустая очередь даёт источнику разгон на целый порог.

Ответ и событие вместе дают протокол для того, кто пишет: писать, пока write отвечает true, а получив false — остановиться и продолжить по drain. Функция writeMore в примере пишет шесть кусков. На false она выходит из цикла и подписывает себя на drain; счётчик n живёт снаружи, поэтому следующий вызов продолжит с того же места. Подписка — через once: с on каждый false добавлял бы ещё одного слушателя, и на длинном файле их стало бы больше десяти — то самое предупреждение об утечке из статьи про излучатель. Writable с порогом теперь в подложке. Прогноз: сколько раз придёт drain и в каком порядке встанут строки?

// Writable с порогом из прошлого примера — в подложке
const out = new Writable({
  highWaterMark: 16,
  write(chunk, done) {
    setTimeout(done, 5);
  },
});
out.on("drain", () => console.log("drain"));
out.on("finish", () => console.log("finish"));

let n = 0;
function writeMore() {
  while (n < 6) {
    n += 1;
    const ok = out.write(`кусок-${n}!`);
    console.log(`write #${n}`, ok, `${out.length}/${out.highWaterMark}`);
    if (!ok) {
      out.once("drain", writeMore);
      return;
    }
  }
  out.end();
}
writeMore();

Шесть кусков ушли тремя парами: true, false, пауза до drain — и цикл с того же места. Очередь ни разу не перешла порог, а finish пришёл после последнего drain: разбуженная им writeMore дошла до конца цикла и позвала end. Отсюда же правило, которое легко пропустить: drain обещан только тому, кому write ответил false. Поток, который ни разу не дошёл до порога, этого события не испускает.

Сверим оба примера с настоящим Writable. Прогон ниже повторяет их подряд на node:stream с теми же порогами и кусками: второй начинается, когда первый дождался drain. В остальном код отличается в трёх местах: writableLength и writableHighWaterMark вместо length и highWaterMark, кодировка между куском и done и опция decodeStrings: false. Без неё Node превратил бы строку в Buffer и считал порог в байтах, а у кириллицы на букву два байта. Прогноз: совпадёт ли вывод с миниатюрой строка в строку?

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

function slowWritable() {
  return new Writable({
    highWaterMark: 16,
    // строка остаётся строкой, и порог считается в символах, как в миниатюре
    decodeStrings: false,
    write(chunk, encoding, done) {
      setTimeout(done, 5);
    },
  });
}

// порог: четыре записи подряд
const first = slowWritable();
first.on("drain", () => console.log("drain, в очереди:", first.writableLength));
for (const chunk of ["кусок-1!", "кусок-2!", "кусок-3!", "кусок-4!"]) {
  const ok = first.write(chunk);
  console.log(`write ${chunk}`, ok, `${first.writableLength}/${first.writableHighWaterMark}`);
}
await once(first, "drain");

// писать до false, продолжать по drain
const out = slowWritable();
out.on("drain", () => console.log("drain"));
out.on("finish", () => console.log("finish"));

let n = 0;
function writeMore() {
  while (n < 6) {
    n += 1;
    const ok = out.write(`кусок-${n}!`);
    console.log(`write #${n}`, ok, `${out.writableLength}/${out.writableHighWaterMark}`);
    if (!ok) {
      out.once("drain", writeMore);
      return;
    }
  }
  out.end();
}
writeMore();

write кусок-1! true 8/16
write кусок-2! false 16/16
write кусок-3! false 24/16
write кусок-4! false 32/16
drain, в очереди: 0
write #1 true 8/16
write #2 false 16/16
drain
write #3 true 8/16
write #4 false 16/16
drain
write #5 true 8/16
write #6 false 16/16
drain
finish

Совпало строка в строку. У настоящего Writable то же правило ответа — очередь вместе с куском на записи меньше порога — и тот же момент drain: очередь пуста, а write перед этим ответил false. Пометку Node тоже показывает: свойство writableNeedDrain.

Теперь соединение из прошлой статьи. Слушатель data пишет кусок в сокет, и раз write научился отвечать, слушатель может остановить источник. Для этого у читающего потока есть pause — пара к resume из прошлой статьи: resume включает течение, pause выключает. В миниатюре pause — одна строка, flowing = false. К ней одна правка во flow: флаг проверяется в условии цикла на каждом куске, а не один раз перед циклом. Иначе pause, позванная из слушателя data, подействовала бы, только когда flow раздаст весь буфер. А чтобы было видно, сколько лежит в буфере, у него появился счётчик length, как у очереди записи.

Соединение становится протоколом из трёх строк: false — pause, drain — resume, end — end. Диск тот же, что в прошлой статье: 64 куска по 64 КиБ, отданные циклом. Порог у сокета по умолчанию, как в Node, — 64 КиБ, так что write отвечает false на каждом куске. Readable с pause и Writable с порогом лежат в подложке. Прогноз: сколько будет в очереди записи и где окажется файл, пока клиент его принимает?

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

source.on("data", (chunk) => {
  const ok = socket.write(chunk);
  maxQueue = Math.max(maxQueue, socket.length);
  if (!ok) source.pause();
});
socket.on("drain", () => source.resume());
source.on("end", () => socket.end());
socket.on("finish", () => {
  console.log(`finish, отправлено: ${sent} | очередь записи доходила до ${maxQueue / KiB} КиБ`);
});

for (let i = 1; i <= 64; i++) source.push("x".repeat(64 * KiB));
source.push(null);
console.log("диск отдал файл, в буфере источника:", source.length / KiB, "КиБ");

Очередь записи не выросла больше одного куска: pause сработала на первом же false. А файл никуда не делся — все 4096 КиБ лежат в буфере читающего потока. Гора переехала из одной очереди в другую. Причина в цикле диска: он кладёт куски через push, не спрашивая, нужны ли они. pause останавливает раздачу, но не того, кто наполняет буфер. Давление дошло от клиента до потока, но не до диска.

Настоящий читающий поток не ждёт, пока источник принесёт куски, а просит их сам. При создании ему передают функцию read, и поток зовёт её, когда в буфере есть место. read добывает следующий кусок и отдаёт его через push. У fs.createReadStream она так и устроена: читает с диска очередной кусок файла и зовёт push. push же отвечает по тому же правилу, что write: true — в буфере ещё есть место, false — буфер дошёл до своего порога. Места нет — поток не зовёт read, и диск не читается.

К счётчику length у буфера добавились порог highWaterMark, как у пишущего потока, и флаг reading: read уже позвали, а push ещё не пришёл, и звать её второй раз рано. Спрос проверяет последняя ветка flow: конца нет, чтение не идёт, в буфере есть место — позвать read. Позвать не на месте, а микрозадачей: push из read снова заходит во flow, и без отсрочки большой файл читался бы рекурсией глубиной в число кусков. Место проверяется и на паузе. Ниже Readable целиком, со всеми правками, и слушатель, который ставит поток на паузу после каждого куска. Прогноз: сколько кусков прочитает диск, пока поток на паузе, и сколько — после resume?

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

  constructor({ read, highWaterMark = 64 * 1024 } = {}) {
    super();
    this.readChunk = read;
    this.highWaterMark = highWaterMark;
  }

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

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

  pause() {
    this.flowing = false;
  }

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

  flow() {
    // флаг на каждом куске: pause() из слушателя data действует сразу
    while (this.flowing && this.buffer.length > 0) {
      const chunk = this.buffer.shift();
      this.length -= chunk.length;
      this.emit("data", chunk);
    }
    if (this.ended) {
      if (this.flowing && this.buffer.length === 0 && !this.endEmitted) {
        this.endEmitted = true;
        this.emit("end");
      }
      return;
    }
    if (this.readChunk && !this.reading && this.length < this.highWaterMark) {
      this.reading = true;
      // после текущего кода: push из read снова зовёт flow, и без отсрочки
      // чтение большого файла шло бы рекурсией глубиной в число кусков
      queueMicrotask(() => this.readChunk());
    }
  }
}

let made = 0;
const source = new Readable({
  highWaterMark: 16,
  // диск: следующий кусок читается, только когда поток его попросил
  read() {
    made += 1;
    this.push(`кусок-${made}!`);
  },
});

source.on("data", (chunk) => {
  console.log("data:", chunk);
  source.pause();
});

setTimeout(() => {
  console.log("прочитано кусков:", made, "| в буфере:", `${source.length}/${source.highWaterMark}`);
  source.resume();
  setTimeout(() => {
    console.log("после resume прочитано кусков:", made, "| в буфере:", `${source.length}/${source.highWaterMark}`);
  }, 10);
}, 10);

Поток на паузе замолк не сразу: он дочитал буфер до порога — два куска по восемь символов — и только тогда перестал звать read. Это запас: после resume следующий кусок уходит без ожидания диска. После resume ушёл один кусок, слушатель снова поставил паузу, и поток дочитал ровно один — до того же порога. Сколько бы ни лежало на диске, в буфере потока не больше порога.

Теперь вся цепочка. Диск отдаёт куски только через read, соединение — те же три строки, что в прошлом разделе. Readable из прошлого примера лежит в подложке. Пример считает, сколько кусков диск прочитал к моменту, когда клиенту ушёл первый, и сколько памяти заняли вместе буфер источника и очередь записи. Прогноз: сколько будет того и другого?

// Emitter, Readable с read и Writable с порогом — в подложке
const KiB = 1024;
let made = 0;
let madeAtFirstSend = 0;
let sent = 0;
let peak = 0;
const measure = () => {
  peak = Math.max(peak, source.length + socket.length);
};

const source = new Readable({
  // диск на 64 куска по 64 КиБ
  read() {
    if (made === 64) {
      this.push(null);
      return;
    }
    made += 1;
    this.push("x".repeat(64 * KiB));
    measure();
  },
});
const socket = new Writable({
  // медленный клиент: каждый кусок уходит по сети за 4 мс
  write(chunk, done) {
    measure();
    setTimeout(() => {
      sent += 1;
      if (sent === 1) madeAtFirstSend = made;
      done();
    }, 4);
  },
});

source.on("data", (chunk) => {
  if (!socket.write(chunk)) source.pause();
});
socket.on("drain", () => source.resume());
source.on("end", () => socket.end());
socket.on("finish", () => {
  console.log("к первой отправке диск прочитал кусков:", madeAtFirstSend);
  console.log(`отправлено: ${sent} | в памяти одновременно до ${peak / KiB} КиБ`);
});

К первой отправке диск прочитал два куска: один пишется в сокет, второй ждёт в буфере. Дальше каждый новый кусок читался, только когда предыдущий уходил клиенту, и в памяти одновременно было не больше 128 КиБ — по порогу на каждой стороне. Это число складывается из порогов, а не из размера файла: на видео в 512 МиБ из прошлой статьи оно было бы тем же. Давление прошло всю цепочку: клиент не успевает → write отвечает false → pause → буфер источника доходит до порога → read не зовут → диск не читается.

Обратное давление — не отдельный механизм, а цепочка ответов: каждое звено слушает соседа ниже по течению. Одно звено, которое не слушает, — и гора копится перед ним.

В Node три строки соединения уже написаны: source.pipe(socket) подписывается на data и end источника, на false зовёт pause, на drain — resume. Прогон ниже соединяет настоящий Readable с настоящим медленным Writable тремя способами: как в прошлой статье, data прямо в write; тремя строками из этой; и через pipe. Диск и клиент те же: 64 куска по 64 КиБ через read, запись по 4 мс. Опции encoding и decodeStrings: false оставляют куски строками, как в миниатюре. Прогноз: сколько кусков прочитает диск к первой отправке и сколько памяти займёт каждый способ?

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

const KiB = 1024;

// диск на 64 куска по 64 КиБ и медленный клиент; join соединяет их по-разному
function send(name, join) {
  let made = 0;
  let madeAtFirstSend = 0;
  let sent = 0;
  let peak = 0;
  const measure = () => {
    peak = Math.max(peak, source.readableLength + socket.writableLength);
  };

  const source = new Readable({
    encoding: "utf8",
    read() {
      if (made === 64) {
        this.push(null);
        return;
      }
      made += 1;
      this.push("x".repeat(64 * KiB));
      measure();
    },
  });
  const socket = new Writable({
    decodeStrings: false,
    write(chunk, encoding, done) {
      measure();
      setTimeout(() => {
        sent += 1;
        if (sent === 1) madeAtFirstSend = made;
        done();
      }, 4);
    },
  });

  join(source, socket);
  return new Promise((resolve) => {
    socket.on("finish", () => {
      console.log(`${name}: к первой отправке диск прочитал кусков: ${madeAtFirstSend}`);
      console.log(`${name}: отправлено: ${sent} | в памяти одновременно до ${peak / KiB} КиБ`);
      resolve();
    });
  });
}

await send("data → write", (source, socket) => {
  source.on("data", (chunk) => socket.write(chunk));
  source.on("end", () => socket.end());
});

await send("pause и drain", (source, socket) => {
  source.on("data", (chunk) => {
    if (!socket.write(chunk)) source.pause();
  });
  socket.on("drain", () => source.resume());
  source.on("end", () => socket.end());
});

await send("pipe", (source, socket) => source.pipe(socket));

data → write: к первой отправке диск прочитал кусков: 64
data → write: отправлено: 64 | в памяти одновременно до 4096 КиБ
pause и drain: к первой отправке диск прочитал кусков: 2
pause и drain: отправлено: 64 | в памяти одновременно до 128 КиБ
pipe: к первой отправке диск прочитал кусков: 2
pipe: отправлено: 64 | в памяти одновременно до 128 КиБ

Без ответа write диск прочитан целиком раньше, чем клиенту ушёл первый кусок, и все 4096 КиБ побывали в памяти разом. Три строки и pipe дали одно и то же — и ровно то, что миниатюра: два куска к первой отправке и 128 КиБ. pipe и есть эти три строки, плюс немного сверху: несколько приёмников у одного источника и отписка, когда приёмник закончил или закрылся. Ошибку приёмника pipe не переносит: сокет оборвался — источник не уничтожен и стоит на паузе с полным буфером. Соединение, которое убирает за собой, — в статье про pipeline.

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

Порог — компромисс между памятью и числом переключений. Чем он выше, тем реже пауза и drain, но тем больше памяти на соединение: тысяча одновременных скачиваний по два порога в 64 КиБ — около 125 МиБ в одних буферах потоков. Задаётся порог опцией highWaterMark при создании потока, у createReadStream тоже. Где этот порог стоит у настоящих потоков — файловых, сетевых и Web Streams браузера — покажет статья про настоящие потоки.