Concurrency Control in Streams

Concurrency Control in Streams

In this lesson, you will learn how to limit the number of simultaneous asynchronous operations within a stream to ensure your application remains stable and efficient under heavy load.

When we use Transform streams to process data—like resizing images or calling external APIs—it is easy to accidentally trigger too many tasks at once. If you receive 1,000 items and fire off 1,000 asynchronous network requests simultaneously, you risk exhausting your memory, hitting API rate limits, or crashing the process.

The Bottleneck Problem

Standard stream pipelines often pull data as fast as the consumer can take it. If your processing logic is asynchronous, the stream might push new chunks into your logic before the previous ones have finished. To control this, we need a mechanism to signal that we are "full" and cannot accept more work until current tasks complete.

Implementing a Semaphore

The most common way to handle concurrency control is by maintaining an internal counter of "active tasks." We only call the stream’s callback() function when we have finished processing a chunk, which signals to the stream that we are ready for the next one.

Consider this Transform stream pattern:

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

let activeTasks = 0;
const MAX_CONCURRENT = 5;

const limitedTransform = new Transform({
  async transform(chunk, encoding, callback) {
    if (activeTasks >= MAX_CONCURRENT) {
      // We are at capacity; wait until a task finishes
      await waitForCapacity(); 
    }

    activeTasks++;
    
    // Process the task
    performExpensiveTask(chunk)
      .finally(() => {
        activeTasks--;
        callback(); // Signal that we are ready for the next chunk
      });
  }
});

By manually tracking activeTasks and delaying the callback(), we exert explicit backpressure. The stream will stop pulling data from the readable source until our active task count drops below the threshold. This creates a "sliding window" of execution where only a set number of operations progress at any given moment. This simple gatekeeper prevents resource exhaustion by forcing the pipeline to respect your application's physical limits.

Key Takeaway: Concurrency control prevents resource exhaustion by using a counter and delaying the stream's callback to ensure no more than a defined number of tasks process data at once.