Модуль Атомарный

module Atomic: sig .. end

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

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

Since 4.12

type !'a t

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

val make : 'a -> 'a t

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

val make_contended : 'a -> 'a t

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

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

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.

Примеры

Основные координации потоков

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

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

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

    (* 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