How do I run code concurrently or in parallel in EK9?

← Concurrency · Ref: Q159

EK9 uses the async operation in stream pipelines for parallel processing. Instead of managing threads directly, you express parallelism as a pipeline stage that invokes each stream element concurrently.

STREAM PIPELINES WITH ASYNC

The stream must contain zero-arg function delegates. The async stage invokes each one in parallel on a virtual thread; the return value flows into the next stage:

  cat workers | async | map with toLine > stdout

async itself takes NO function parameter — it calls the stream element. There is no 'async with processor' form; the function to call IS the stream element.

THE ASYNC OPERATION

async is a stream pipeline operation like call, filter, or map. The stream type must be a zero-arg function delegate that returns a mainValue; that return value becomes the next stream's element type. async only differs from call in that elements are dispatched concurrently to virtual threads.

WHEN TO USE ASYNC

Use async for I/O-bound or computation-heavy operations where items can be processed independently:
- Fetching multiple URLs (build a List of url-fetcher functions, then `| async`)
- Processing multiple files (List of file-reader functions, then `| async`)
- Making parallel API calls (List of request functions, then `| async`)

TURNING DATA INTO ASYNC WORK

If you start with data rather than function delegates, use `map with` to project each item to a zero-arg function delegate, then `| async` to invoke them in parallel:

  cat urls | map with toFetcher | async | filter by isSuccess > stdout

Here toFetcher takes a URL and returns a zero-arg function whose body actually performs the fetch. The async stage then runs all those fetchers concurrently.

VIRTUAL THREADS

EK9 runs on Java 25 with virtual threads. The async operation leverages virtual threads for lightweight concurrency. You get parallelism without managing thread pools or executors.

THREAD-BOUNDARY IMPLICATIONS

Each `| async` invocation is a NEW thread context. If a worker captures a MutexLock that the calling pipeline is already holding, you will get E08253 (cross-thread same lock) because MutexLock reentrancy is per-thread. See Q1299 for the recovery pattern.

See Q89 for stream pipelines. See Q158 for MutexLock (which shows the canonical LockableAddressSet pattern). See Q160 for concurrency model. See Q1299 for the async + same-lock pitfall. See Q125 for head/tail.

Example

defines module qa.concurrency.parallel

  defines function

    doubleIt() as pure
      -> number as Integer
      <- rtn as Integer: number * 2

    isLarge() as pure
      -> number as Integer
      <- rtn <- Boolean()
      batchSize <- 50
      rtn: number > batchSize

  defines program

    ParallelProcessingDemo()
      stdout <- Stdout()

      // === STREAM PIPELINE ===

      items <- List() of Integer
      items += 10
      items += 20
      items += 30
      items += 40
      items += 50

      // Standard pipeline (sequential)
      cat items | map with doubleIt | filter by isLarge > stdout

      // For parallel: stream must contain zero-arg function delegates.
      //   cat workers | async | map with toLine > stdout
      // async takes NO parameter — the function IS the stream element.
      // To start from data: cat urls | map with toFetcher | async > stdout

      // See Q158 for the canonical async-safe shared-state
      // pattern (LockableAddressSet via MutexLock of AddressSet).

      stdout.println("EK9 uses async in stream pipelines for parallelism")

Common mistakes

E50060 — EK9 has no .stream() method chains. Use pipe syntax: 'cat items | map with fn | filter by fn > stdout'. EK9 expresses pipelines through operators, not method calls. See ek9 -h E50060 for details.

Incorrect:

items.stream().map(doubleIt).filter(isLarge).forEach(stdout)

Correct:

cat items | map with doubleIt | filter by isLarge > stdout

E50060 — EK9 List has no .add() method. Use the += operator. See ek9 -h E50060 for details.

Incorrect:

items.add(10)

Correct:

items += 10

Coming from another language?

Java: ExecutorService, CompletableFuture, or parallel streams. Python: asyncio, multiprocessing, or concurrent.futures. Rust: async/await with tokio, or rayon for data parallelism. Go: goroutines with channels. EK9: async in stream pipelines (cat funcs | async) where each stream element is a zero-arg function delegate that async invokes in parallel — leveraging Java 25 virtual threads for lightweight concurrency. No coloured-function problem; no thread pool wiring.

Keywords: stream, thread, migrate, concurrent, compile, concurrency, async, pipeline, parallel, virtual, process