Repository F# setup

open System
open System.IO
open System.Threading
open System.Threading.Tasks
open Axial
open Axial.Layers
open Axial.Console
open Axial.FileSystem
open Axial.Hosting
open Axial.Hosting.Browser
open Axial.Hosting.Node
open Axial.PlatformService
open Axial.State
open Axial.Telemetry
open Axial.Telemetry.JavaScript

Batching and Parallelism

Strict batches and continuously replenished parallel work have different observable schedules and use separate operators.

Strict batches

Use FlowStream.chunkBySize with Flow.sequencePar when each complete batch must finish before the next starts:

Verified

> open Axial.PlatformService;;
> let completeAfter (label, milliseconds: int) =
-     Flow.sleep (System.TimeSpan.FromMilliseconds(float milliseconds)) |> Flow.map (fun () -> label)
- in
- (FlowStream.fromSeq [ ("A", 80); ("B", 10); ("C", 80); ("D", 10) ]
-  : FlowStream<ClockEnvironment, Never, string * int>)
- |> FlowStream.chunkBySize 2
- |> FlowStream.mapFlow (List.map completeAfter >> Flow.sequencePar)
- |> FlowStream.runCollect
- |> Flow.run (ClockEnvironment Clock.live);;

val it: Exit<string list list,Never> = Success [["A"; "B"]; ["C"; "D"]]

Axial

PlatformService

completeAfter: 'a * int -> Flow<'b,'c,'a>

label: 'a

milliseconds: int

int

An abbreviation for the CLI type . Basic Types

Axial.Flow

sleep: TimeSpan -> Flow<'env,'error,unit>

Suspends the flow for the specified duration, observing cancellation. The duration to sleep. A flow that completes after the specified delay, or is interrupted if cancelled first.

System

FromMilliseconds: float -> TimeSpan

Returns a that represents a specified number of milliseconds. A number of milliseconds. An object that represents . is less than or greater than . -or- is . -or- is . is equal to .

System.TimeSpan

Represents a time interval.

float: ^T -> float

Converts the argument to 64-bit float. This is a direct conversion for all primitive numeric types. For strings, the input is converted using Double.Parse() with InvariantCulture settings. Otherwise the operation requires an appropriate static conversion method on the input type. The input value. The converted float float 'A' // evaluates to 65.0 float 0xff // evaluates to 255.0 float -10 // evaluates to -10.0

(|>): 'T1 -> ('T1 -> 'U) -> 'U

Apply a function to a value, the value being on the left, the function on the right The argument. The function. The function result. let doubleIt x = x * 2 3 |> doubleIt // Evaluates to 6

map: ('value -> 'next) -> Flow<'env,'error,'value> -> Flow<'env,'error,'next>

Transforms the successful value of a flow. If the source fails, the is not executed. The original failure cause is preserved, including typed failures, interruption, and defects. Use map for pure value transformations after an effect has succeeded. A function of type 'value -> 'next to transform the successful value. The source flow of type to transform. A new with the transformed success value of type 'next. let flow = Flow.succeed 1 |> Flow.map (fun x -> x + 1)

Axial.FlowStreamModule

fromSeq: 'value seq -> FlowStream<'env,'error,'value>

Creates a stream from a synchronous sequence of values. The sequence of values to be emitted by the stream. A that yields each value from the sequence. FlowStream.fromSeq [1..10] |> FlowStream.runCollect |> Flow.run ()

Axial.FlowStream`3

Represents a cold stream of values that requires an environment, can fail with a typed error, and supports backpressure. The type of the environment dependency. The type of the failure value. The type of the success values in the stream.

Axial.ClockEnvironment

An environment containing only a clock, for timed flows with no other services.

Axial.Never

Represents an error channel that cannot occur.

string

An abbreviation for the CLI type . Basic Types

chunkBySize: int -> FlowStream<'a,'b,'c> -> FlowStream<'a,'b,'c list>

Groups consecutive values into non-empty lists of at most elements. The operator pulls and retains at most upstream values for each emitted list.

mapFlow: ('a -> Flow<'b,'c,'d>) -> FlowStream<'b,'c,'a> -> FlowStream<'b,'c,'d>

Transforms every value with a Flow effect. ids |> FlowStream.mapFlow load

Microsoft.FSharp.Collections.ListModule

Contains operations for working with values of type . Operations for collections such as lists, arrays, sets, maps and sequences. See also F# Collection Types in the F# Language Guide.

map: ('T -> 'U) -> 'T list -> 'U list

Builds a new collection whose elements are the results of applying the given function to each of the elements of the collection. The function to transform elements from the input list. The input list. The list of transformed elements. let inputs = [ "a"; "bbb"; "cc" ] inputs |> List.map (fun x -> x.Length) Evaluates to [ 1; 3; 2 ] This is an O(n) operation, where n is the length of the list.

(>>): ('T1 -> 'T2) -> ('T2 -> 'T3) -> 'T1 -> 'T3

Compose two functions, the function on the left being applied first The first function to apply. The second function to apply. The composition of the input functions. let addOne x = x + 1 let doubleIt x = x * 2 let addThenDouble = addOne >> doubleIt addThenDouble 3 // Evaluates to 8

sequencePar: Flow<'env,'error,'value> list -> Flow<'env,'error,'value list>

Runs all flows concurrently and returns their values in input order. An empty input succeeds immediately. If any flow fails, remaining flows are interrupted through the same structured parallel composition as zipPar. Scoped resources follow zipPar too: when every flow succeeds they move to the caller's scope; when any fails, all of them are released before sequencePar returns.

runCollect: FlowStream<'a,'b,'c> -> Flow<'a,'b,'c list>

Collects all emitted values into a list. stream |> FlowStream.runCollect

run: 'env -> Flow<'env,'error,'value> -> Exit<'value,'error>

Runs the workflow and blocks until the final exit is available. The environment used by the workflow. The workflow to run. The final workflow exit. let exit = workflow |> Flow.run environment

``.ctor``: IClock -> ClockEnvironment

Axial.PlatformService.Clock

Helpers for the clock service.

live: IClock

Creates a live clock backed by and a monotonic timer.

B and D complete first inside their respective batches, but Flow.sequencePar restores input order. The resulting nested list also exposes the batch barrier: C and D belong to the second result only after the first batch completes.

This pipeline:

pulls at most two source values for each batch in this example;
runs those two mappings concurrently;
returns results in input order;
does not begin the next batch until the previous batch has been emitted as a single list value; the next pull starts it, regardless of what the caller does with that list's elements;
interrupts sibling mappings when one fails.

Flow.sequencePar runs every Flow in the supplied list, so FlowStream.chunkBySize supplies the bound.

Continuously replenished work

Use FlowStream.mapFlowPar when a completed mapping should immediately open capacity for another input:

Verified

> open Axial.PlatformService;;
> let completeAfter (label, milliseconds: int) =
-     Flow.sleep (System.TimeSpan.FromMilliseconds(float milliseconds)) |> Flow.map (fun () -> label)
- in
- (FlowStream.fromSeq [ ("A", 150); ("B", 10); ("C", 10) ]
-  : FlowStream<ClockEnvironment, Never, string * int>)
- |> FlowStream.mapFlowPar (Parallelism.bounded 2) completeAfter
- |> FlowStream.runCollect
- |> Flow.run (ClockEnvironment Clock.live);;

val it: Exit<string list,Never> = Success ["B"; "C"; "A"]

Axial

PlatformService

completeAfter: 'a * int -> Flow<'b,'c,'a>

label: 'a

milliseconds: int

int

An abbreviation for the CLI type . Basic Types

Axial.Flow

sleep: TimeSpan -> Flow<'env,'error,unit>

Suspends the flow for the specified duration, observing cancellation. The duration to sleep. A flow that completes after the specified delay, or is interrupted if cancelled first.

System

FromMilliseconds: float -> TimeSpan

Returns a that represents a specified number of milliseconds. A number of milliseconds. An object that represents . is less than or greater than . -or- is . -or- is . is equal to .

System.TimeSpan

Represents a time interval.

float: ^T -> float

(|>): 'T1 -> ('T1 -> 'U) -> 'U

Apply a function to a value, the value being on the left, the function on the right The argument. The function. The function result. let doubleIt x = x * 2 3 |> doubleIt // Evaluates to 6

map: ('value -> 'next) -> Flow<'env,'error,'value> -> Flow<'env,'error,'next>

Axial.FlowStreamModule

fromSeq: 'value seq -> FlowStream<'env,'error,'value>

Axial.FlowStream`3

Axial.ClockEnvironment

An environment containing only a clock, for timed flows with no other services.

Axial.Never

Represents an error channel that cannot occur.

string

An abbreviation for the CLI type . Basic Types

mapFlowPar: Parallelism -> ('a -> Flow<'b,'c,'d>) -> FlowStream<'b,'c,'a> -> FlowStream<'b,'c,'d>

Maps values with a continuously replenished, bounded set of child fibers. Results are emitted in completion order. After each result is consumed, the next upstream value starts, so there are no strict batch barriers. At most the configured number of mappings are active or retained. The first failure observed stops the stream; the terminal consumer's child scope interrupts and awaits all remaining mappings before returning.

Axial.ParallelismModule

Creates bounds for parallel Flow and stream operators.

bounded: int -> Parallelism

Creates a positive concurrency bound. Thrown when is not positive.

runCollect: FlowStream<'a,'b,'c> -> Flow<'a,'b,'c list>

Collects all emitted values into a list. stream |> FlowStream.runCollect

run: 'env -> Flow<'env,'error,'value> -> Exit<'value,'error>

Runs the workflow and blocks until the final exit is available. The environment used by the workflow. The workflow to run. The final workflow exit. let exit = workflow |> Flow.run environment

``.ctor``: IClock -> ClockEnvironment

Axial.PlatformService.Clock

Helpers for the clock service.

live: IClock

Creates a live clock backed by and a monotonic timer.

With two slots, completion of B starts C without waiting for A. The returned list makes completion-order emission visible: B and C are emitted before the earlier but slower A.

At most two mappings are active or retained in this example. Results are emitted in completion order, so a slow earlier input does not block later results or failures. Consuming a result opens one slot and starts the next upstream mapping.

If consumption stops early, the terminal stream scope interrupts and awaits outstanding mapping fibers before returning. Parallel work remains bounded in both count and lifetime.

Choosing between batches and parallel mapping

Choose strict batches when order and batch barriers are part of the contract, for example checkpointing one page group before fetching the next. Choose FlowStream.mapFlowPar for worker-pool behavior where throughput matters and completion order is acceptable.

Batching and Parallelism

Strict batches

Continuously replenished work

Choosing between batches and parallel mapping

Overview

Docs

API Reference