Многопоточность

Threads.@threads [schedule] for ... end

Макрос для распараллеливания цикла for для выполнения с несколькими потоками. Разбивает пространство итераций между несколькими задачами и выполняет эти задачи в потоках в соответствии с политикой планирования. В конце цикла устанавливается барьер, который ожидает завершения всех задач.

Аргумент schedule может использоваться для запроса определённой политики планирования. Единственное в настоящее время поддерживаемое значение — :static, которое создаёт по одной задаче на поток и равномерно распределяет итерации между ними. Указание :static является ошибкой, если оно используется изнутри другого цикла @threads или из потока, отличного от 1.

По умолчанию используется политика планирования (если аргумент schedule отсутствует), которая может изменяться.

См. также: @spawn, nthreads(), threadid(), pmap в Distributed, BLAS.set_num_threads в LinearAlgebra.

Threads.foreach(f, channel::Channel;
                schedule::Threads.AbstractSchedule=Threads.FairSchedule(),
                ntasks=Threads.nthreads())

Аналогично foreach(f, channel), но итерации по channel и вызовы f распределяются между ntasks задачами, порождёнными Threads.@spawn. Эта функция подождёт, пока все внутренне порождённые задачи не завершатся, прежде чем вернуть значение.

Если schedule isa FairSchedule, Threads.foreach попытается породить задачи таким образом, чтобы планировщик Julia мог более свободно распределять задачи по потокам. Этот подход, как правило, имеет более высокую накладные расходы на элемент, но может работать лучше, чем StaticSchedule при одновременной работе с другими многопоточными рабочими нагрузками.

Если schedule isa StaticSchedule, Threads.foreach будет порождать задачи таким образом, чтобы накладные расходы на элемент были ниже, чем у FairSchedule, но он менее подходит для распределения нагрузки. Таким образом, этот подход может быть более подходящим для мелкозернистых, однородных рабочих нагрузок, но может работать хуже, чем FairSchedule при одновременной работе с другими многопоточными рабочими нагрузками.

Threads.@spawn expr

Создать Task и schedule его для выполнения в любом доступном потоке. Задача назначается потоку после его освобождения. Для ожидания завершения задачи вызовите wait на результате этого макроса или вызовите fetch для ожидания и получения его возвращаемого значения.

Значения можно интерполировать в @spawn с помощью $, которая копирует значение непосредственно в созданный базовый закрытый блок. Это позволяет вставлять значение переменной, изолируя асинхронный код от изменений значения переменной в текущей задаче.

Threads.threadid()

Получить идентификатор текущего потока выполнения. Главный поток имеет ID 1.

Threads.nthreads()

Получить количество потоков, доступных для процесса Julia. Это верхняя граница, включающая threadid().

См. также BLAS.get_num_threads и BLAS.set_num_threads в стандартной библиотеке LinearAlgebra, и nprocs() в стандартной библиотеке Distributed.

Threads.Condition([lock])

Потокобезопасная версия Base.Condition.

Чтобы вызвать wait или notify на Threads.Condition, необходимо сначала вызвать lock на ней. При вызове wait, блокировка атомарно освобождается во время блокировки, и она будет повторно приобретена перед тем, как wait вернёт значение. Поэтому типичное использование Threads.Condition c выглядит следующим образом:

lock(c)
try
    while !thing_we_are_waiting_for
        wait(c)
    end
finally
    unlock(c)
end

Event()

Создать источник событий с уровнем срабатывания. Задачи, которые вызывают wait на Event, приостанавливаются и помещаются в очередь до вызова notify на Event. После вызова notify, Event остаётся в сигнализированном состоянии, и задачи больше не будут блокироваться при ожидании её.

@atomic var
@atomic order ex

Отметить var или ex как выполняемые атомарно, если ex является поддерживаемым выражением.

@atomic a.b.x = new
@atomic a.b.x += addend
@atomic :acquire_release a.b.x = new
@atomic :acquire_release a.b.x += addend

Выполнить операцию записи, выраженную справа, атомарно и вернуть новое значение.

С =, эта операция преобразуется в вызов setproperty!(a.b, :x, new). С любым оператором также, эта операция преобразуется в вызов modifyproperty!(a.b, :x, +, addend)[2].

@atomic a.b.x max arg2
@atomic a.b.x + arg2
@atomic max(a.b.x, arg2)
@atomic :acquire_release max(a.b.x, arg2)
@atomic :acquire_release a.b.x + arg2
@atomic :acquire_release a.b.x max arg2

Выполнить бинарную операцию, выраженную справа, атомарно. Сохранить результат в поле в первом аргументе и вернуть значения (old, new).

Эта операция преобразуется в вызов modifyproperty!(a.b, :x, func, arg2).

См. раздел «Атомарные операции по полю» в руководстве для получения более подробной информации.

julia> mutable struct Atomic{T}; @atomic x::T; end

julia> a = Atomic(1)
Atomic{Int64}(1)

julia> @atomic a.x # fetch field x of a, with sequential consistency
1

julia> @atomic :sequentially_consistent a.x = 2 # set field x of a, with sequential consistency
2

julia> @atomic a.x += 1 # increment field x of a, with sequential consistency
3

julia> @atomic a.x + 1 # increment field x of a, with sequential consistency
3 => 4

julia> @atomic a.x # fetch field x of a, with sequential consistency
4

julia> @atomic max(a.x, 10) # change field x of a to the max value, with sequential consistency
4 => 10

julia> @atomic a.x max 5 # again change field x of a to the max value, with sequential consistency
10 => 10

@atomicswap a.b.x = new
@atomicswap :sequentially_consistent a.b.x = new

Сохраняет new в a.b.x и возвращает старое значение a.b.x.

Эта операция преобразуется в вызов swapproperty!(a.b, :x, new).

См. раздел «Атомарные операции по полю» в руководстве для получения более подробной информации.

julia> mutable struct Atomic{T}; @atomic x::T; end

julia> a = Atomic(1)
Atomic{Int64}(1)

julia> @atomicswap a.x = 2+2 # replace field x of a with 4, with sequential consistency
1

julia> @atomic a.x # fetch field x of a, with sequential consistency
4

@atomicreplace a.b.x expected => desired
@atomicreplace :sequentially_consistent a.b.x expected => desired
@atomicreplace :sequentially_consistent :monotonic a.b.x expected => desired

Выполняет условную замену, выраженную парой атомарно, возвращая значения (old, success::Bool). Где success указывает, была ли замена выполнена.

Эта операция преобразуется в вызов replaceproperty!(a.b, :x, expected, desired).

См. раздел «Атомарные операции по полю» в руководстве для получения более подробной информации.

julia> mutable struct Atomic{T}; @atomic x::T; end

julia> a = Atomic(1)
Atomic{Int64}(1)

julia> @atomicreplace a.x 1 => 2 # replace field x of a with 2 if it was 1, with sequential consistency
(old = 1, success = true)

julia> @atomic a.x # fetch field x of a, with sequential consistency
2

julia> @atomicreplace a.x 1 => 2 # replace field x of a with 2 if it was 1, with sequential consistency
(old = 2, success = false)

julia> xchg = 2 => 0; # replace field x of a with 0 if it was 1, with sequential consistency

julia> @atomicreplace a.x xchg
(old = 2, success = true)

julia> @atomic a.x # fetch field x of a, with sequential consistency
0

Threads.Atomic{T}

Содержит ссылку на объект типа T, гарантируя, что к нему обращаются только атомарно, то есть в потокобезопасном режиме.

Для атомарного доступа могут использоваться только некоторые «простые» типы, а именно примитивные булевы, целочисленные и числа с плавающей точкой. Это Bool, Int8...Int128, UInt8...UInt128, и Float16...Float64.

Новые атомарные объекты могут быть созданы из значения, не являющегося атомарным; если не указано иное, атомарный объект инициализируется нулём.

К атомарным объектам можно получить доступ с помощью обозначения []:

Примеры

julia> x = Threads.Atomic{Int}(3)
Base.Threads.Atomic{Int64}(3)

julia> x[] = 1
1

julia> x[]
1

Атомарные операции используют префикс atomic_, такой как atomic_add!, atomic_xchg! и т. д.

Threads.atomic_cas!(x::Atomic{T}, cmp::T, newval::T) where T

Атомарное сравнение и установка x

Атомарно сравнивает значение в x с cmp. Если они равны, записывает newval в x. В противном случае оставляет x без изменений. Возвращает старое значение в x. Сравнивая возвращённое значение с cmp (через ===) можно узнать, было ли x модифицировано и теперь содержит новое значение newval.

Для более подробной информации см. инструкцию LLVM cmpxchg.

Эта функция может использоваться для реализации транзакционных семантик. Перед транзакцией записывается значение в x. После транзакции новое значение сохраняется только в том случае, если x не было модифицировано в это время.

Примеры

julia> x = Threads.Atomic{Int}(3)
Base.Threads.Atomic{Int64}(3)

julia> Threads.atomic_cas!(x, 4, 2);

julia> x
Base.Threads.Atomic{Int64}(3)

julia> Threads.atomic_cas!(x, 3, 2);

julia> x
Base.Threads.Atomic{Int64}(2)

Threads.atomic_xchg!(x::Atomic{T}, newval::T) where T

Атомарный обмен значением в x

Атомарно меняет значение в x на newval. Возвращает старое значение.

Для более подробной информации см. инструкцию LLVM atomicrmw xchg.

Примеры

julia> x = Threads.Atomic{Int}(3)
Base.Threads.Atomic{Int64}(3)

julia> Threads.atomic_xchg!(x, 2)
3

julia> x[]
2

Threads.atomic_add!(x::Atomic{T}, val::T) where T <: ArithmeticTypes

Атомарно прибавить val к x

Выполняет x[] += val атомарно. Возвращает старое значение. Не определено для Atomic{Bool}.

Для более подробной информации см. инструкцию LLVM atomicrmw add.

Примеры

julia> x = Threads.Atomic{Int}(3)
Base.Threads.Atomic{Int64}(3)

julia> Threads.atomic_add!(x, 2)
3

julia> x[]
5

Threads.atomic_sub!(x::Atomic{T}, val::T) where T <: ArithmeticTypes

Атомарно вычесть val из x

Выполняет x[] -= val атомарно. Возвращает старое значение. Не определено для Atomic{Bool}.

Для более подробной информации см. инструкцию LLVM atomicrmw sub.

Примеры

julia> x = Threads.Atomic{Int}(3)
Base.Threads.Atomic{Int64}(3)

julia> Threads.atomic_sub!(x, 2)
3

julia> x[]
1

Threads.atomic_and!(x::Atomic{T}, val::T) where T

Атомарно побитовое И x с val

Выполняет x[] &= val атомарно. Возвращает старое значение.

Для более подробной информации см. инструкцию LLVM atomicrmw and.

Примеры

julia> x = Threads.Atomic{Int}(3)
Base.Threads.Atomic{Int64}(3)

julia> Threads.atomic_and!(x, 2)
3

julia> x[]
2

Threads.atomic_nand!(x::Atomic{T}, val::T) where T

Атомарно побитовое НЕ-И x с val

Выполняет x[] = ~(x[] & val) атомарно. Возвращает старое значение.

Для более подробной информации см. инструкцию LLVM atomicrmw nand.

Примеры

julia> x = Threads.Atomic{Int}(3)
Base.Threads.Atomic{Int64}(3)

julia> Threads.atomic_nand!(x, 2)
3

julia> x[]
-3

Threads.atomic_or!(x::Atomic{T}, val::T) where T

Атомарно побитовое ИЛИ x с val

Выполняет x[] |= val атомарно. Возвращает старое значение.

Для более подробной информации см. инструкцию LLVM atomicrmw or.

Примеры

julia> x = Threads.Atomic{Int}(5)
Base.Threads.Atomic{Int64}(5)

julia> Threads.atomic_or!(x, 7)
5

julia> x[]
7

Threads.atomic_xor!(x::Atomic{T}, val::T) where T

Атомарно побитовое исключающее ИЛИ x с val

Выполняет x[] $= val атомарно. Возвращает старое значение.

Для более подробной информации см. инструкцию LLVM atomicrmw xor.

Примеры

julia> x = Threads.Atomic{Int}(5)
Base.Threads.Atomic{Int64}(5)

julia> Threads.atomic_xor!(x, 7)
5

julia> x[]
2

Threads.atomic_max!(x::Atomic{T}, val::T) where T

Атомарно сохранить максимальное значение из x и val в x

Выполняет x[] = max(x[], val) атомарно. Возвращает старое значение.

Для более подробной информации см. инструкцию LLVM atomicrmw max.

Примеры

julia> x = Threads.Atomic{Int}(5)
Base.Threads.Atomic{Int64}(5)

julia> Threads.atomic_max!(x, 7)
5

julia> x[]
7

Threads.atomic_min!(x::Atomic{T}, val::T) where T

Атомарно сохранить минимальное значение из x и val в x

Выполняет x[] = min(x[], val) атомарно. Возвращает старое значение.

Для более подробной информации см. инструкцию LLVM atomicrmw min.

Примеры

julia> x = Threads.Atomic{Int}(7)
Base.Threads.Atomic{Int64}(7)

julia> Threads.atomic_min!(x, 5)
7

julia> x[]
5

Threads.atomic_fence()

Вставить барьер последовательной согласованности памяти

Вставляет барьер памяти с семантикой последовательной согласованности. В некоторых алгоритмах это необходимо, то есть там, где порядка приобретения/освобождения недостаточно.

Вероятно, это очень дорогая операция. Учитывая, что все остальные атомарные операции в Julia уже имеют семантику приобретения/освобождения, явные барьеры в большинстве случаев не нужны.

Для более подробной информации см. инструкцию LLVM fence.

@threadcall((cfunc, clib), rettype, (argtypes...), argvals...)

Макрос @threadcall вызывается так же, как и ccall, но выполняет работу в другом потоке. Это полезно, когда вы хотите вызвать блокирующую C-функцию, не заставляя основной julia поток блокироваться. Конкурентность ограничена размером пула потоков libuv, который по умолчанию равен 4 потокам, но может быть увеличен путём установки переменной среды UV_THREADPOOL_SIZE и перезапуска процесса julia.

Обратите внимание, что вызываемая функция никогда не должна вызывать Julia.

SpinLock()

Создаёт нерекурсивную блокировку с принципом «проверить и установить». Рекурсивное использование приведёт к тупику. Этот тип блокировки следует использовать только для кода, выполняющегося за короткое время и не блокирующего (например, выполняющего ввод-вывод). В целом, следует использовать ReentrantLock.

Каждый вызов lock должен быть сопоставлен с вызовом unlock.

Блокировки «проверить и установить» быстрее всего при примерно 30 и более конкурирующих потоках. Если у вас больше конкурирующих потоков, следует рассмотреть альтернативные подходы к синхронизации.