package saturn

You can search for identifiers within the package.

in-package search v0.2.0

On This Page

API
1. Consumer functions
2. Producer functions
Examples
1. Sequential example
2. Parallel example

saturn
- Saturn

Legend:
Library
Module
Module type
Parameter
Class
Class type

Optimized Michael-Scott lock-free multi-producer multi-consumer queue.

All functions are lockfree. It is the recommended starting point when needing FIFO structure. It is inspired by Simple, Fast, and Practical Non-Blocking and Blocking Concurrent Queue Algorithms.

If you need a length function, you can use the bounded queue Saturn.Bounded_queue instead with maximun capacity (default value). However, this adds a general overhead to the operation.

API

type 'a t

Represents a lock-free queue holding elements of type 'a.

val create : unit -> 'a t

create () returns a new queue, initially empty.

val of_list : 'a list -> 'a t

of_list list creates a new queue from a list.

🐌 This is a linear-time operation.

# open Saturn.Queue
# let t : int t = of_list [1;2;3;4]
val t : int t = <abstr>
# pop_opt t
- : int option = Some 1
# pop_opt t 
- : int option = Some 2

val is_empty : 'a t -> bool

is_empty q returns true if q is empty and false otherwise.

Consumer functions

exception Empty

Raised when pop_exn, peek_exn, or drop_exn is applied to an empty queue.

val peek_exn : 'a t -> 'a

peek_exn queue returns the first element of the queue without removing it.

raises Empty
if the queue is empty.

val peek_opt : 'a t -> 'a option

peek_opt queue returns Some of the first element of the queue without removing it, or None if the queue is empty.

val pop_exn : 'a t -> 'a

pop_exn queue removes and returns the first element of the queue.

raises Empty
if the queue is empty.

val pop_opt : 'a t -> 'a option

pop_opt q removes and returns the first element in queue q, or returns None if the queue is empty.

val drop_exn : 'a t -> unit

drop_exn queue removes the top element of the queue.

raises Empty
if the queue is empty.

Producer functions

val push : 'a t -> 'a -> unit

push q v adds the element v at the end of the queue q.

Examples

Sequential example

An example top-level session:

# open Saturn.Queue
# let t : int t = of_list [1;2;3]
val t : int t = <abstr>
# push t 42
- : unit = ()
# pop_exn t
- : int = 1
# peek_opt t
- : int option = Some 2
# drop_exn t
- : unit = ()
# pop_opt t
- : int option = Some 3
# pop_opt t
- : int option = Some 42
# pop_exn t
Exception: Saturn__Michael_scott_queue.Empty.

Parallel example

Note: The barrier is used in this example solely to make the results more interesting by increasing the likelihood of parallelism. Spawning a domain is a costly operation, especially compared to the relatively small amount of work being performed here. In practice, using a barrier in this manner is unnecessary.

# open Saturn.Queue
# let t : string t = create ()
val t : string t = <abstr>
# Random.self_init ()
- : unit = ()
# let barrier = Atomic.make 2
val barrier : int Atomic.t = <abstr>

# let work id =
    Atomic.decr barrier;
    while Atomic.get barrier <> 0 do
      Domain.cpu_relax ()
    done;
    for _ = 1 to 4 do
      Domain.cpu_relax ();
      if Random.bool () then push t id
      else
       match pop_opt t with
        | None -> Format.printf "Domain %s sees an empty queue.\n%!" id
        | Some v -> Format.printf "Domain %s pops values pushed by %s.\n%!" id v
    done
val work : string -> unit = <fun>

# let domainA = Domain.spawn (fun () -> work "A")
val domainA : unit Domain.t = <abstr>
# let domainB = Domain.spawn (fun () -> work "B")
Domain B pops values pushed by B.
Domain A pops values pushed by A.
Domain B pops values pushed by A.
Domain B pops values pushed by A.
val domainB : unit Domain.t = <abstr>

# Domain.join domainA; Domain.join domainB
- : unit = ()