Effect.Concurrency 0.1.0-alpha.2

Effect.Concurrency

Structured concurrency primitives for Effect.

Fibers, concurrent data structures, streams, and combinators for parallel and racing effects.

Install

dotnet add package Effect.Concurrency --prerelease

Requires the Effect package.

Fibers

open Eff

let program =
    effect {
        let! fiber = EffectConcurrency.fork () longRunningEffect
        // do other work...
        let! result = Fiber.join fiber
        return result
    }

// Interrupt a fiber
do! Fiber.interrupt fiber

// Race multiple effects
let! fastest = EffectConcurrency.raceAll [ server1; server2 ]

// Timeout
let! result = myEffect |> EffectConcurrency.timeout (TimeSpan.FromSeconds 5.0)
// Ok (Some value) or Ok None

Ref — thread-safe mutable state

let program =
    effect {
        let! counter = Ref.make 0
        let inc = effect { for _ in 1..1000 do do! Ref.update ((+) 1) counter }
        let! _ = EffectConcurrency.zipPar inc inc
        return! Ref.get counter  // Ok 2000
    }

Deferred — one-shot promise

let program =
    effect {
        let! promise = Deferred.make<unit, AppError, string> ()
        let! fiber = EffectConcurrency.fork () (effect {
            do! Effect.sleep 100
            let! _ = Deferred.succeed "done" promise
            return ()
        })
        let! value = Deferred.await promise
        return value
    }

Queue — bounded async channel

Backed by System.Threading.Channels.Channel<T>.

let program =
    effect {
        let! q = Queue.bounded<unit, AppError, int> 5
        let producer = effect { for i in 1..10 do do! Queue.put i q }
        let consumer = effect { for _ in 1..10 do let! _ = Queue.take q in () }
        let! _ = EffectConcurrency.zipPar producer consumer
        return ()
    }

Stream — lazy async sequences

Push-based streams with backpressure, typed errors, and cancellation.

let result =
    Stream.range 1 100
    |> Stream.filter (fun x -> x % 2 = 0)
    |> Stream.map (fun x -> x * x)
    |> Stream.take 10
    |> Stream.runCollect
    |> Effect.run ()

// Merge concurrent streams
let merged = Stream.merge stream1 stream2

// Pipe to/from queues
do! myStream |> Stream.toQueue queue
let fromQ = Stream.fromQueue queue |> Stream.take 100

TestClock — deterministic time

Test time-dependent effects without real delays.

let clock = TestClock()

let program = effect {
    do! clock.Sleep 5000
    return "done"
}

let resultTask = Effect.run () program
clock.AdvanceByMillis 5000  // resolves instantly