@shutterstock/p-map-iterable
    Preparing search index...

    Class IterableMapper<Element, NewElement>

    Iterates over a source iterable / generator with specified concurrency, calling the mapper on each iterated item, and storing the mapper result in a queue of maxUnread size, before being iterated / read by the caller.

    Also exported as ConcurrentMapper, with the same constructor and instance types. Concurrency overlaps asynchronous callback work in the current JavaScript process. Results can arrive out of input order when concurrency is greater than one.

    • Enriching metadata or probing capabilities for a sync or async iterable of inputs
    • Prefetching items from an async I/O source
    • In the simple sequential (concurrency: 1) case, allows items to be prefetched async, preserving order, while caller processes an item
    • Can allow parallel prefetches for sources that allow for out of order reads (concurrency: 2+)
    • Prevents the producer from racing ahead of the consumer if maxUnread is reached

    The mapper should ideally handle all errors internally to enable error handling closest to where they occur. However, if errors do escape the mapper:

    When stopOnMapperError is true (default):

    • First error immediately stops processing
    • Error is thrown from the AsyncIterator's next() call

    When stopOnMapperError is false:

    • Processing continues despite errors
    • All errors are collected and thrown together
    • Errors are thrown as AggregateError after all items complete
    • Items are exposed to the mapper via an iterator or async iterator (this includes generator and async generator functions)
    • Consume results with for await while mapping proceeds
    • The mapper will not be invoked when maxUnread is reached, until items are consumed
    • The iterable will set done when the input has indicated done and all mapper promises have resolved
    const source = new SomeSource();
    const sourceIds = [1, 2,... 1000];
    const sink = new SomeSink();
    for (const sourceId of sourceIds) {
      const item = await source.read(sourceId);     // takes 300 ms of I/O wait, no CPU
      const outputItem = doSomeOperation(item);     // takes 20 ms of CPU
      await sink.write(outputItem);                 // takes 500 ms of I/O wait, no CPU
    }

    Each iteration takes 820ms total, but we waste time waiting for I/O. We could prefetch the next read (300ms) while processing (20ms) and writing (500ms), without changing the order of reads or writes.

    concurrency: 1 on the prefetcher preserves the order of the reads, and writes remain sequential and blocking.

    const source = new SomeSource();
    const sourceIds = [1, 2,... 1000];
    // Prefetches serially within maxUnread and releases results in sequential order
    const sourcePrefetcher = new IterableMapper(sourceIds,
      async (sourceId) => source.read(sourceId),
      { concurrency: 1, maxUnread: 10 }
    );
    const sink = new SomeSink();
    for await (const item of sourcePrefetcher) {    // may not block for fast sources
      const outputItem = doSomeOperation(item);     // takes 20 ms of CPU
      await sink.write(outputItem);                 // takes 500 ms of I/O wait, no CPU
    }

    This reduces iteration time to 520ms by overlapping reads with processing/writing.

    concurrency: 1 on the prefetcher preserves the order of the reads. concurrency: 1 on the flusher preserves the order of the writes, but allows the loop to iterate while last write is completing.

    const source = new SomeSource();
    const sourceIds = [1, 2,... 1000];
    const sourcePrefetcher = new IterableMapper(sourceIds,
      async (sourceId) => source.read(sourceId),
      { concurrency: 1, maxUnread: 10 }
    );
    const sink = new SomeSink();
    const flusher = new IterableQueueMapperSimple(
      async (outputItem) => sink.write(outputItem),
      { concurrency: 1 }
    );
    for await (const item of sourcePrefetcher) {    // may not block for fast sources
      const outputItem = doSomeOperation(item);     // takes 20 ms of CPU
      await flusher.enqueue(outputItem);            // will periodically block for portion of write time
    }
    // Close input permanently after the last enqueue and wait for all writes
    await flusher.onIdle();
    // Check for errors
    if (flusher.errors.length > 0) {
     // ...
    }

    This reduces iteration time to about max((max(readTime, writeTime) - cpuOpTime, cpuOpTime)) by overlapping reads and writes with the CPU processing step. In this contrived example, the loop time is reduced to 500ms - 20ms = 480ms. In cases where the CPU usage time is higher, the impact can be greater.

    For maximum throughput, allow out of order reads and writes with IterableQueueMapper (to iterate results with backpressure when too many unread items) or IterableQueueMapperSimple (to handle errors at end without custom iteration and applying backpressure to block further enqueues when concurrency items are in process):

    const source = new SomeSource();
    const sourceIds = [1, 2,... 1000];
    const sourcePrefetcher = new IterableMapper(sourceIds,
      async (sourceId) => source.read(sourceId),
      { concurrency: 10, maxUnread: 20 }
    );
    const sink = new SomeSink();
    const flusher = new IterableQueueMapperSimple(
      async (outputItem) => sink.write(outputItem),
      { concurrency: 10 }
    );
    for await (const item of sourcePrefetcher) {    // typically will not block
      const outputItem = doSomeOperation(item);     // takes 20 ms of CPU
      await flusher.enqueue(outputItem);            // typically will not block
    }
    // Close input permanently after the last enqueue and wait for all writes
    await flusher.onIdle();
    // Check for errors
    if (flusher.errors.length > 0) {
     // ...
    }

    This reduces iteration time to about 20ms by overlapping reads and writes with the CPU processing step. In this contrived (but common) example we would get a 41x improvement in throughput, removing 97.5% of the time to process each item and fully utilizing the CPU time available in the JS event loop.

    Type Parameters

    • Element
    • NewElement

    Implements

    Index
    constructor
    [asyncIterator] next

    • Used by the iterator returned from [Symbol.asyncIterator] Called every time an item is needed

      Returns Promise<IteratorResult<NewElement, any>>

      Iterator result

    Settings
    Member Visibility

    On This Page
    Typical Use CasesError HandlingUsageTypical Processing Loop without Prefetching with and Blocking Sequential WritesPrefetching with and Background Sequential Writes with Prefetching with and Out of Order Background Writes with
    Constructors
    Methods