How Bounded Fan-Out Concurrency Works in analyzeMany

The analyzeMany function implements bounded fan-out concurrency by spawning a fixed-size pool of async workers that atomically claim ticker indices from a shared counter, ensuring parallel analysis never exceeds a configurable ceiling while the research desk processes large ticker lists.

The analyzeMany function in the anthropics/cwc-workshops repository serves as the core driver for parallel stock analysis in the research desk application. When analyzing multiple tickers simultaneously, the function employs bounded fan-out concurrency to prevent resource exhaustion by limiting concurrent analyst sessions to a configurable maximum. This worker-pool pattern, implemented in research-desk/src/lib/analysis.ts, balances parallelism with system stability by capping the number of simultaneous API calls while efficiently distributing work across all available slots.

The Worker-Pool Architecture

Instead of spawning an unbounded promise for each ticker—which could overwhelm the Anthropic API or exhaust local resources—analyzeMany creates a fixed-size pool of async workers. Each worker repeatedly grabs the next unprocessed ticker from a shared atomic index (nextIndex), processes it via analyzeTicker, and continues until all tickers complete.

The concurrency limit is determined at runtime by evaluating options.concurrency against DEFAULT_CONCURRENCY from config.ts. The actual pool size becomes the smaller of the concurrency setting and the ticker count, preventing idle workers when processing small batches.

Step-by-Step Implementation Walkthrough

Step 1: Establish the Concurrency Ceiling

The function first calculates the maximum number of parallel workers, defaulting to DEFAULT_CONCURRENCY if no override is provided:

const concurrency = Math.max(1, options.concurrency ?? DEFAULT_CONCURRENCY);

As implemented in [research-desk/src/lib/analysis.ts lines 60-61](https://github.com/anthropics/cwc-workshops/blob/main/research-desk/src/lib/analysis.ts#L60-L61), this guarantees at least one worker while respecting user-defined limits.

Step 2: Prepare Analysis Records

The function initializes the result array, either reusing existing records or creating fresh AnalysisRecord objects:

const records = options.records ?? tickers.map((ticker) => newRecord(ticker));

This pattern, found at lines 62-63, supports both fresh analyses and resuming existing dispatch jobs.

Step 3: Create the Worker Pool

The bounded fan-out mechanism centers on a fixed-size array of async functions, each containing a while loop that atomically increments a shared index:

let nextIndex = 0;
const workers = Array.from({ length: Math.min(concurrency, records.length) }, async () => {
  while (nextIndex < records.length) {
    const record = records[nextIndex++];
    await analyzeTicker(client, cfg, record.ticker, options.focus ?? "", record);
    options.onProgress?.(record);
  }
});

This implementation at lines 66-74 creates exactly concurrency workers (or fewer if the ticker list is short). The nextIndex++ operation ensures each ticker is claimed by exactly one worker without requiring a separate queue structure.

Step 4: Execute All Workers

The function awaits completion of all workers via Promise.all:

await Promise.all(workers);

As shown at line 75, this resolves only when every worker has processed its assigned tickers, guaranteeing complete coverage of the input list.

Step 5: Return Populated Records

Finally, the function returns the fully populated records array containing status indicators, scorecards, and error states for each ticker (lines 76-77).

Key Implementation Characteristics

Bounded Concurrency
The pool size never exceeds concurrency, even when processing hundreds of tickers. This hard limit protects the Anthropic API rate limits and local memory resources.

Dynamic Work Distribution
Workers pull work dynamically using the shared nextIndex counter. Faster workers automatically claim more tickers, balancing load without complex scheduling logic.

Progress Feedback
The optional onProgress callback fires after each analyzeTicker completion, enabling real-time UI updates showing which tickers have finished processing.

Graceful Downsizing
When the ticker list is smaller than the concurrency setting, Math.min(concurrency, records.length) prevents creation of unnecessary idle workers.

Configuration and Limits

The concurrency ceiling defaults to DEFAULT_CONCURRENCY, defined in [research-desk/src/lib/config.ts](https://github.com/anthropics/cwc-workshops/blob/main/research-desk/src/lib/config.ts). Users can override this per-call via the concurrency option, useful when running on different hardware tiers or API quota levels.

Practical Usage Examples

Basic Analysis with Default Concurrency

import { analyzeMany, loadDeskConfigOrThrow } from "@/lib/analysis";

const client = getClient();                     // Anthropic SDK client
const cfg = loadDeskConfigOrThrow();            // Loads environment config
const tickers = ["AAPL", "MSFT", "NVDA"];

const records = await analyzeMany(client, cfg, tickers);
// Each record contains analysis results for one ticker

Custom Concurrency with Progress Tracking

await analyzeMany(client, cfg, tickers, {
  concurrency: 4,                 // Run up to 4 analyses in parallel
  focus: "risk & margin durability",
  onProgress: (rec) => {
    console.log(`${rec.ticker}: ${rec.status}`);
  },
});

Integration with the Dispatch Orchestrator

The research desk's orchestrator calls analyzeMany from [research-desk/src/lib/orchestrator.ts](https://github.com/anthropics/cwc-workshops/blob/main/research-desk/src/lib/orchestrator.ts) to fan out work across tickers before compiling results:

await analyzeMany(client, cfg, dispatch.tickers, {
  focus: dispatch.focus,
  concurrency: DEFAULT_CONCURRENCY,
  records: dispatch.records,
});
const resultPayload = compileDispatchResult(dispatch.records);

This pattern (around lines 76-80 in orchestrator.ts) demonstrates how bounded fan-out plugs into higher-level workflow orchestration while maintaining predictable resource usage.

Summary

• analyzeMany implements bounded fan-out concurrency through a fixed-size worker pool defined in analysis.ts
• Workers atomically claim ticker indices via the shared nextIndex counter, ensuring no duplicate processing without complex queue infrastructure
• The concurrency ceiling defaults to DEFAULT_CONCURRENCY but supports per-call overrides via the options parameter
• Dynamic work distribution allows faster workers to process more tickers automatically, optimizing total completion time
• Promise.all guarantees all tickers complete before returning the populated AnalysisRecord array
• The pattern integrates directly with the orchestrator.ts dispatch flow, supporting the research desk's batch analysis workflows

Frequently Asked Questions

What is bounded fan-out concurrency?

Bounded fan-out concurrency is a pattern that limits the number of simultaneous operations to a fixed maximum, preventing resource exhaustion while allowing parallel processing. In analyzeMany, this means running up to N analyst sessions concurrently (where N is configurable), regardless of how many tickers need analysis.

How does the shared counter prevent race conditions?

The nextIndex variable uses the JavaScript increment operator (nextIndex++), which is atomic for single-threaded async contexts. Because JavaScript is single-threaded and the await yields control between operations, no two workers can retrieve the same index simultaneously. This simple "pull" model eliminates race conditions without requiring locks or separate queue libraries.

What happens if I set concurrency higher than the number of tickers?

The code automatically adjusts via Math.min(concurrency, records.length), creating only as many workers as there are tickers. This prevents idle workers and unnecessary memory allocation when processing small batches, ensuring efficient resource usage regardless of the configured ceiling.

How does this pattern handle API rate limits?

By capping the number of concurrent analyzeTicker calls—which internally invoke the Anthropic API—analyzeMany naturally constrains the request rate. The fixed pool size prevents the thundering-herd problem that would occur if all tickers launched simultaneously, allowing the system to respect API quotas while maximizing throughput within those limits.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}