Модуль Atomic

module Atomic: sig .. end

Атомарные ссылки.

См. примеры ниже. См. главу «Модель памяти: сложные аспекты» в руководстве.

С версии 4.12

type !'a t

Атомарная (изменяемая) ссылка на значение типа 'a.

val make : 'a -> 'a t

Создать атомарную ссылку.

val make_contended : 'a -> 'a t

Создать атомарную ссылку, размещённую отдельно в строке кэша. Она занимает в 4–16 раз больше памяти, чем ссылка, выделенная с помощью make v.

Основная цель — предотвратить ложное совместное использование и вызванное им снижение производительности. Когда ЦП выполняет атомарную операцию, он временно получает владение всей строкой кэша, содержащей атомарную ссылку. Если несколько атомарных ссылок находятся в одной строке кэша, одновременно изменять эти непересекающиеся области памяти становится невозможно, что может создать узкое место. Поэтому в качестве общего правила: если атомарная ссылка испытывает конкуренцию, выделение ей собственной строки кэша может повысить производительность.

С версии 5.2

val get : 'a t -> 'a

Получить текущее значение атомарной ссылки.

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

Установить новое значение атомарной ссылки.

val exchange : 'a t -> 'a -> 'a

Установить новое значение атомарной ссылки и вернуть текущее значение.

val compare_and_set : 'a t -> 'a -> 'a -> bool

compare_and_set r seen v устанавливает новое значение r равным v только в том случае, если его текущее значение физически равно seen — сравнение и установка выполняются атомарно. Возвращает true, если сравнение успешно (то есть установка выполнена), и false в противном случае.

val fetch_and_add : int t -> int -> int

fetch_and_add r n атомарно увеличивает значение r на n и возвращает текущее значение (до увеличения).

val incr : int t -> unit

incr r атомарно увеличивает значение r на 1.

val decr : int t -> unit

decr r атомарно уменьшает значение r на 1.

module Loc: sig .. end

Атомарные «ячейки», например поля записей.

Примеры

Базовая координация потоков

Один из простых вариантов использования — глобальные счётчики, которые обновляются потокобезопасным способом, например для сбора метрик по операциям ввода-вывода, выполняемым программой. Другой простой вариант — координация завершения потоков в программе, например когда один поток находит ответ или когда пользователь завершает программу.

Например, попробуем найти число, хеш которого удовлетворяет простому условию. Для этого запустим несколько потоков, которые будут перебирать случайные числа, пока не найдут подходящее.

Разумеется, приведённый ниже результат — пример; он будет меняться при каждом запуске программы.

    (* use for termination *)
    let stop_all_threads = Atomic.make false

    (* total number of individual attempts to find a number *)
    let num_attempts = Atomic.make 0

    (* find a number that satisfies [p], by... trying random numbers
       until one fits. *)
    let find_number_where (p:int -> bool) =
      let rand = Random.State.make_self_init() in
      while not (Atomic.get stop_all_threads) do

        let n = Random.State.full_int rand max_int in
        ignore (Atomic.fetch_and_add num_attempts 1 : int);

        if p (Hashtbl.hash n) then (
          Printf.printf "found %d (hash=%d)\n%!" n (Hashtbl.hash n);
          Atomic.set stop_all_threads true; (* signal all threads to stop *)
        )
      done;;


    (* run multiple domains to search for a [n] where [hash n <= 100] *)
    let () =
      let criterion n = n <= 100 in
      let threads =
        Array.init 8
          (fun _ -> Domain.spawn (fun () -> find_number_where criterion))
      in
      Array.iter Domain.join threads;
      Printf.printf "total number of attempts: %d\n%!"
        (Atomic.get num_attempts) ;;

    - : unit = ()
    found 1651745641680046833 (hash=33)
    total number of attempts: 30230350

Стек Трайбера

Ещё один пример — простой стек Трайбера (потокобезопасный стек), которым можно безопасно пользоваться из разных потоков.

Обратите внимание, что и push, и pop являются рекурсивными, поскольку пытаются заменить старый стек новым (с одним дополнительным элементом или на один элемент короче). Это оптимистичная конкурентность: на каждой итерации, например, push stack x получает старый стек l и надеется, что к моменту попытки заменить l на x::l никто другой не успеет изменить список. Если compare_and_set завершается неудачей, значит, мы были слишком оптимистичны и должны повторить попытку.

    type 'a stack = 'a list Atomic.t

    let rec push (stack: _ stack) elt : unit =
      let cur = Atomic.get stack in
      let success = Atomic.compare_and_set stack cur (elt :: cur) in
      if not success then
        push stack elt

    let rec pop (stack: _ stack) : _ option =
      let cur = Atomic.get stack in
      match cur with
      | [] -> None
      | x :: tail ->
        let success = Atomic.compare_and_set stack cur tail in
        if success then Some x
        else pop stack

    # let st = Atomic.make []
    # push st 1
    - : unit = ()
    # push st 2
    - : unit = ()
    # pop st
    - : int option = Some 2
    # pop st
    - : int option = Some 1
    # pop st
    - : int option = None