package moonpool

  1. Overview
  2. Docs
package moonpool
Legend:
Page
Library
Module
Module type
Parameter
Class
Class type
Source

Module Moonpool.Ws_poolSource

Work-stealing thread pool.

A pool of threads with a worker-stealing scheduler. The pool contains a fixed number of worker threads that wait for work items to come, process these, and loop.

This is good for CPU-intensive tasks that feature a lot of small tasks. Note that tasks will not always be processed in the order they are scheduled, so this is not great for workloads where the latency of individual tasks matter (for that see Fifo_pool).

This implements Runner.t since 0.3.

If a pool is no longer needed, shutdown can be used to signal all threads in it to stop (after they finish their work), and wait for them to stop.

The threads are distributed across a fixed domain pool (whose size is determined by Domain.recommended_domain_count. See create for more details.

include module type of Runner
Sourcetype fiber = Picos.Fiber.t
Sourcetype task = unit -> unit
Sourcetype t

A runner.

If a runner is no longer needed, shutdown can be used to signal all worker threads in it to stop (after they finish their work), and wait for them to stop.

The threads are distributed across a fixed domain pool (whose size is determined by Domain.recommended_domain_count on OCaml 5, and simple the single runtime on OCaml 4).

Sourceval size : t -> int

Number of threads/workers.

Sourceval num_tasks : t -> int

Current number of tasks. This is at best a snapshot, useful for metrics and debugging.

Sourceval shutdown : t -> unit

Shutdown the runner and wait for it to terminate. Idempotent.

Sourceval shutdown_without_waiting : t -> unit

Shutdown the pool, and do not wait for it to terminate. Idempotent.

Sourceexception Shutdown
Sourceval run_async : ?fiber:fiber -> t -> task -> unit

run_async pool f schedules f for later execution on the runner in one of the threads. f() will run on one of the runner's worker threads/domains.

  • parameter fiber

    if provided, run the task with this initial fiber data

  • raises Shutdown

    if the runner was shut down before run_async was called.

Sourceval run_wait_block : ?fiber:fiber -> t -> (unit -> 'a) -> 'a

run_wait_block pool f schedules f for later execution on the pool, like run_async. It then blocks the current thread until f() is done executing, and returns its result. If f() raises an exception, then run_wait_block pool f will raise it as well.

NOTE be careful with deadlocks (see notes in Fut.wait_block about the required discipline to avoid deadlocks).

  • raises Shutdown

    if the runner was already shut down

Sourceval dummy : t

Runner that fails when scheduling tasks on it. Calling run_async on it will raise Failure.

  • since 0.6

Implementing runners

Sourcemodule For_runner_implementors : sig ... end

This module is specifically intended for users who implement their own runners. Regular users of Moonpool should not need to look at it.

Sourceval get_current_runner : unit -> t option

Access the current runner. This returns Some r if the call happens on a thread that belongs in a runner.

  • since 0.5
Sourceval get_current_fiber : unit -> fiber option

get_current_storage runner gets the local storage for the currently running task.

Sourcetype ('a, 'b) create_args = ?on_init_thread:(dom_id:int -> t_id:int -> unit -> unit) -> ?on_exit_thread:(dom_id:int -> t_id:int -> unit -> unit) -> ?on_exn:(exn -> Printexc.raw_backtrace -> unit) -> ?num_threads:int -> ?name:string -> 'a

Arguments used in create. See create for explanations.

Sourceval create : (unit -> t, _) create_args

create () makes a new thread pool.

  • parameter on_init_thread

    called at the beginning of each new thread in the pool.

  • parameter num_threads

    size of the pool, ie. number of worker threads. It will be at least 1 internally, so 0 or negative values make no sense. The default is Domain.recommended_domain_count(), ie one worker thread per CPU core.

Note that specifying num_threads=n means that the degree of parallelism is at most n. This behavior is different than the one of Domainslib, see https://github.com/c-cube/moonpool/issues/41 for context.

If you want to use all cores, use Domain.recommended_domain_count().

  • parameter on_exit_thread

    called at the end of each thread in the pool

  • parameter name

    a name for this thread pool, used if tracing is enabled (since 0.6)

Sourceval with_ : (unit -> (t -> 'a) -> 'a, _) create_args

with_ () f calls f pool, where pool is obtained via create. When f pool returns or fails, pool is shutdown and its resources are released.

Most parameters are the same as in create.

  • since 0.3