Module Hegel.Stateful

Introduction

A stateful test applies a random sequence of rules to a state. A rule is a step function that takes the test case and the state, draws whatever data it needs, and updates the state in place. An invariant is a property that must hold after each step.

With the ppx_hegel_test PPX, a state machine is a module written as module%hegel_state_machine M = struct … end. Mark rules with [@@rule] and invariants with [@@invariant]. See Rule and Invariant for valid options. The PPX generates the run function for the state machine. If the module defines sexp_of_state, run uses it to print the state after each step.

Without the PPX, create rules with Rule.create and the invariants with Invariant.create, put them in a module of type State_machine, and pass that to run.

Every invariant is checked on the initial and final states. Between steps, invariants are sampled unless marked always_check.

Examples in this documentation assume open Hegel.

Example: an integer stack.

module%hegel_state_machine Stack = struct
  type state = int list ref

  let sexp_of_state stack = sexp_of_list sexp_of_int !stack

  let push tc stack =
    let n = draw tc (integers ~min_value:0 ~max_value:100 ()) in
    stack := n :: !stack
  [@@rule]
  ;;

  let pop tc stack =
    match !stack with
    | [] -> assume tc false
    | _ :: rest -> stack := rest
  [@@rule]
  ;;

  let short tc stack =
    note tc (Printf.sprintf "%d elements" (List.length !stack));
    assert (List.length !stack < 10)
  [@@invariant { always_check = true }]
  ;;
end

let%hegel_test integer_stack tc = Stack.run tc ~init:(ref [])

Concurrent stateful testing

run_concurrent with max_concurrency > 1 runs rules on several workers concurrently. libhegel draws the number of workers in [min_concurrency, max_concurrency]. Rules belong to a group, and only rules in the same group may run concurrently. Rules without a group all belong to the <anonymous> group.

A round is a step of a concurrent test. Each round selects one rule group, and each worker runs a sequence of rules from that group. Invariants run after all workers finish the round.

How the workers run is a Concurrency.t. The default implementation is Concurrency.threads, which puts workers on systhreads. See Concurrency for more details.

A concurrent rule takes a context after the test case. If using Concurrency.threads or Concurrency.domains it is (). A capability wrapping a scheduler passes the scheduler's per-task handle. This can be used by the system under test or the rule to spawn its own tasks.

With the ppx_hegel_test PPX, a concurrent state machine is a module written as module%hegel_concurrent_state_machine M = struct … end. Mark rules with [@@rule] and invariants with [@@invariant]. See Concurrent_rule and Invariant for valid options. The PPX generates the run function for the state machine. If the module defines sexp_of_state, run uses it to print the state after each step.

The example store below has a bug. The store locks individual reads and writes, but releases the lock between reading a counter and writing its incremented value. Two workers can therefore overwrite each other's updates.

module Store = struct
  type t =
    { lock : Mutex.t
    ; values : (int, int) Hashtbl.t
    }

  let create () = { lock = Mutex.create (); values = Hashtbl.create 4 }

  let get store key =
    Mutex.protect store.lock (fun () -> Hashtbl.find_opt store.values key)
  ;;

  let put store key value =
    Mutex.protect store.lock (fun () -> Hashtbl.replace store.values key value)
  ;;

  let put_if_absent store key =
    Mutex.protect store.lock (fun () ->
      if Hashtbl.mem store.values key
      then false
      else (
        Hashtbl.add store.values key 0;
        true))
  ;;

  let increment store key =
    let value = Option.value (get store key) ~default:0 in
    put store key (value + 1)
  ;;

  let snapshot store = Mutex.protect store.lock (fun () -> Hashtbl.copy store.values)
end

module%hegel_concurrent_state_machine Key_value_store = struct
  type state =
    { store : Store.t
    ; keys : int Stateful.Concurrent_pool.t
    ; increments : int Atomic.t
    }

  let register tc () state =
    let key = draw tc (integers ~min_value:0 ~max_value:3 ()) in
    if Store.put_if_absent state.store key then Stateful.Concurrent_pool.add state.keys tc key
  [@@rule { group = "operations" }]
  ;;

  let increment tc () state =
    let key = draw_silent tc (Stateful.Concurrent_pool.values_reusable state.keys) in
    Store.increment state.store key;
    Atomic.incr state.increments
  [@@rule { group = "operations" }]
  ;;

  let read tc () state =
    let key = draw_silent tc (Stateful.Concurrent_pool.values_reusable state.keys) in
    match Store.get state.store key with
    | Some value -> note tc (Printf.sprintf "read %d -> %d" key value)
    | None -> note tc (Printf.sprintf "key %d is absent" key)
  [@@rule { group = "operations" }]
  ;;

  let snapshot tc () state =
    let count = Hashtbl.length (Store.snapshot state.store) in
    note tc (Printf.sprintf "snapshot holds %d keys" count)
  [@@rule { group = "snapshot" }]
  ;;

  let no_lost_updates _tc state =
    let stored =
      Hashtbl.fold (fun _ value total -> total + value) (Store.snapshot state.store) 0
    in
    let performed = Atomic.get state.increments in
    if stored <> performed
    then
      failwith
        (Printf.sprintf
           "increments were lost: store sums to %d after %d increments"
           stored
           performed)
  [@@invariant { always_check = true }]
  ;;
end

let%hegel_test concurrent_store tc =
  let init : Key_value_store.state =
    { store = Store.create ()
    ; keys = Stateful.Concurrent_pool.create tc
    ; increments = Atomic.make 0
    }
  in
  Key_value_store.run tc ~init ~max_concurrency:4
;;

The operations group allows registration, increments, and reads to overlap. The snapshot group runs separately. The invariant compares stored values with an atomic count of completed increments.

With max_concurrency > 1, failures are reported without shrinking, replay, database persistence, or reproduction blobs.

In the failure output, each rule execution is labeled with its worker and the time in milliseconds since the test case began to aid debugging.

--- Failure: concurrent_store (...) -------------------------

Concurrency level: 4
---------------- Round 1: group "operations" ----------------
[worker 2 +0.238ms] Rule: register
[worker 2 +0.246ms]   key = 0
...
[worker 2 +0.261ms] Rule: read
[worker 2 +0.263ms]   read 2 -> 0
...
[worker 3 +0.223ms] Rule: increment
...
---------------- Round 2: group "snapshot" ------------------
[worker 0 +0.363ms] Rule: snapshot
[worker 0 +0.365ms]   snapshot holds 3 keys
...
---------------- Round 3: group "operations" ----------------
[worker 0 +0.486ms] Rule: increment
...
[worker 1 +0.464ms] Rule: register
[worker 1 +0.468ms]   key = 1
...
[worker 2 +0.509ms] Rule: increment
...
[worker 3 +0.479ms] Rule: increment
Invariant no_lost_updates violated after round 3.

Exception: Failure("increments were lost: store sums to 11 after 12 increments")

OxCaml

On OxCaml only concurrent machines have modes.

A concurrent rule body is portable, so a generator it captures must be portable. See Mode-annotated API for every public signature that has a mode. A concurrent machine must use Concurrent_pool for pools.

The ~concurrency capability and context are @ local. See also hegel.jane.concurrent.

Submodules

Running stateful tests

let%hegel_test counter tc =
  Stateful.run tc (module Counter) ~init:(ref 0) ~step_count:200
;;

  state = 0
  Step 1: add
    by = 3
  state = 3
  Step 2: add
    by = 7
  state = 10
  Invariant small violated after step 2.