Skuld.Effects.Parallel (skuld v0.23.0)

View Source

Parallel effect - simple fork-join concurrency with built-in boundaries.

Provides high-level parallel operations where each call is self-contained with automatic task management. Unlike Async, there are no handles to track or boundaries to manage - each operation spawns tasks, awaits results, and cleans up automatically.

Basic Usage

use Skuld.Syntax
alias Skuld.Comp
alias Skuld.Effects.{Parallel, Throw}

# Run multiple computations in parallel, get all results
comp do
  Parallel.all([
    comp do expensive_work_1() end,
    comp do expensive_work_2() end,
    comp do expensive_work_3() end
  ])
end
|> Parallel.with_handler()
|> Throw.with_handler()
|> Comp.run!()
#=> [:result1, :result2, :result3]

Operations

  • all/1 - Run all computations, return all results (fails if any fail)
  • race/1 - Run all computations, return first to complete (cancels others)
  • map/2 - Map a function over items in parallel

Error Handling

Task failures are caught and returned. For all/1 and map/2, the first failure stops waiting and returns the error. For race/1, task failures are ignored unless all tasks fail.

Handlers

BEAM Process Constraints

Same as Async - child tasks get a snapshot of env at fork time. Changes don't propagate back. Use AtomicState for shared mutable state.

Summary

Functions

__handle__(comp, atom)

Install Parallel handler via catch clause syntax.

map(items, fun)

Map a function over items in parallel.

with_handler(comp)

Install a production Parallel handler using Task.Supervisor.

with_sequential_handler(comp)

Install a sequential handler for testing.

Functions

__handle__(comp, atom)

Install Parallel handler via catch clause syntax.

Config selects handler type:

catch
  Parallel -> nil           # production handler (Task.Supervisor)
  Parallel -> :sequential   # test handler (runs tasks sequentially)

map(items, fun)

@spec map([term()], (term() -> Skuld.Comp.Types.computation())) ::
  Skuld.Comp.Types.computation()

Map a function over items in parallel.

The function receives each item and should return a computation. Results are returned in the same order as the input items. If any computation fails, returns {:error, {:task_failed, reason}}.

Example

Parallel.map([1, 2, 3], fn id ->
  comp do
    fetch_user(id)
  end
end)
#=> [%User{id: 1}, %User{id: 2}, %User{id: 3}]

with_handler(comp)

@spec with_handler(Skuld.Comp.Types.computation()) :: Skuld.Comp.Types.computation()

Install a production Parallel handler using Task.Supervisor.

Creates a Task.Supervisor that manages all parallel tasks. The supervisor is stopped when the handler scope exits.

Example

comp do
  Parallel.all([
    comp do :a end,
    comp do :b end
  ])
end
|> Parallel.with_handler()
|> Throw.with_handler()
|> Comp.run!()

with_sequential_handler(comp)

@spec with_sequential_handler(Skuld.Comp.Types.computation()) ::
  Skuld.Comp.Types.computation()

Install a sequential handler for testing.

Runs parallel computations sequentially for deterministic behavior. Useful for testing logic without concurrency complexity.

Example

comp do
  Parallel.all([
    comp do :a end,
    comp do :b end
  ])
end
|> Parallel.with_sequential_handler()
|> Throw.with_handler()
|> Comp.run!()
#=> [:a, :b]