Demand & Buffering

A library can have only two active jobs and still allocate a million queued promises. Bound the work that is started and inspect how inputs are obtained. A synchronous iterator lets the consumer pull the next item when capacity exists instead of first spreading the source into an array.

This is a form of demand control: the consumer asks for more work at its own pace. It is not a complete streaming system. A result array retained until the end still grows with total input, and a source that eagerly allocated everything has already used that memory.

For early exit, close an unfinished iterator. A small take helper can express a bounded view of a source; it must not pull an extra item merely to discover that the quota has been reached. Avoid a for/of implementation that fetches the next item before checking a limit inside its body. Check demand before next().

Predict, run, and explain

Write the output and trace the important state changes before checking it. Then run the program. Each editor is independent. Async examples end with main() so this editor waits for their work.

let reads = 0;

function* source() {
  for (let n = 0; n < 5; n++) {
    reads++;
    yield n;
  }
}

const iterator = source();
const pair = [iterator.next().value, iterator.next().value];

console.log(pair.join(","));
console.log(reads);
iterator.return();

0,1
2

Read iterator.return?.() as “call return if that method is present.” This is the method-call form of the optional chaining you already used for properties. Some iterators have no return method. For a generator, calling return enters its finally block and closes unfinished iteration.

Count pulls at the source boundary

A limit on active mapper calls does not automatically limit queued promises or input reads. Count next calls at the source to see whether the consumer asks for values it has no capacity to process. Demand checks should precede the next pull.

A zero quota is especially revealing: the consumer should not need one value merely to discover that it wants none. Early closure and normal exhaustion are distinct states; close an unfinished owned iterator without inventing an extra read to detect its end.

let reads = 0;

function* source() {
  for (let n = 0; n < 4; n++) {
    reads++;
    yield n;
  }
}

const cursor = source();

console.log(reads);
cursor.next();
cursor.next();
console.log(reads);
cursor.return();

0
2

Creating the generator performs no pulls. The two explicit next calls are the only demand in this example. Instrument the boundary rather than guessing from output count.

let reads = 0;

function* source() {
  reads++;
  yield 1;
}

const cursor = source();
const quota = 0;

for (let n = 0; n < quota; n++) {
  cursor.next();
}

cursor.return();
console.log(reads);

0

The source body was never entered. Closing an unstarted generator also does not run its body’s finally block; resources must be acquired inside the lifetime you can actually observe.

Try the rule on new inputs

Test quotas zero, one, exactly the source length and larger than the source. Keep separate counters for pulls and mapper calls. Explain why returning a collected result array still needs memory proportional to output size.

Your turn

// Check the quota before iterator.next; use finally for unfinished-source cleanup.

function* take(values, count) {
  if (!Number.isInteger(count) || count < 0) {
    throw new RangeError("Invalid count");
  }

  const iterator = values[Symbol.iterator]();
  let exhausted = false;

  try {
    for (let n = 0; n < count; n++) {
      const step = iterator.next();

      if (step.done) {
        exhausted = true;
        return;
      }

      yield step.value;
    }
  } finally {
    if (!exhausted) {
      iterator.return?.();
    }
  }
}

let reads = 0;

function* source() {
  for (let n = 0; n < 5; n++) {
    reads++;
    yield n;
  }
}

console.log([...take(source(), 2)].join(","));
console.log(reads);

0,1
2

Change it and test it

Test count zero, an empty source and a quota larger than the source. Show cleanup with a finally message. Explain why collecting every output from a bounded pool still needs O(n) memory; active concurrency and total memory are different quantities.