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

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

Макрос для выполнения цикла for параллельно. Пространство итераций распределяется по крупным задачам. Этот параметр может быть задан аргументом schedule. Выполнение цикла ждёт завершения всех итераций.

См. также: @spawn и pmap в Distributed.

Расширенная справка

Семантика

Если более сильные гарантии не заданы опцией планирования, цикл, выполняемый макросом @threads, имеет следующую семантику.

Макрос @threads выполняет тело цикла в неопределённом порядке и потенциально одновременно. Он не определяет точные задания задач и потоков. Присваивания могут отличаться в каждом выполнении. Код тела цикла (включая любой код, вызываемый им транзитивно) не должен делать никаких предположений о распределении итераций по задачам или рабочему потоку, в котором они выполняются. Код тела цикла для каждой итерации должен иметь возможность прогрессировать независимо от других итераций и быть свободным от гонок данных. Таким образом, некорректные синхронизации между итерациями могут привести к тупику, а несинхронизированный доступ к памяти может привести к неопределённому поведению.

Например, вышеуказанные условия подразумевают, что:

• Блокировка, взятая в итерации, должна быть освобождена в той же итерации.
• Обмен информацией между итерациями с использованием блокирующих примитивов, таких как Channel , некорректен.
• Записи должны производиться только в локации, не используемой в других итерациях (если не используется блокировка или атомная операция).
• Если не используется планирование :static, значение threadid() может измениться даже внутри одной итерации. См. Task Migration.

Планировщики

Без аргумента планировщика точное планирование не определено и меняется между версиями Julia. В настоящее время используется :dynamic, когда планировщик не задан.

:dynamic (по умолчанию)

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

Эта опция планирования — лишь подсказка для механизма выполнения. Однако можно ожидать нескольких свойств. Количество Task , используемых планировщиком :dynamic, ограничено небольшой константой, умноженной на количество доступных рабочих потоков (Threads.threadpoolsize()). Каждая задача обрабатывает смежные области пространства итераций. Таким образом, @threads :dynamic for x in xs; f(x); end обычно более эффективен, чем @sync for x in xs; @spawn f(x); end, если length(xs) значительно больше, чем количество рабочих потоков, а время выполнения f(x) относительно меньше, чем стоимость создания и синхронизации задачи (обычно менее 10 микросекунд).

:greedy

Планировщик :greedy создаёт до Threads.threadpoolsize() задач, каждая из которых жадно работает с заданными значениями итераций по мере их получения. Как только одна задача завершает свою работу, она берёт следующее значение из итератора. Работа, выполняемая любой отдельной задачей, необязательно выполняется на смежных значениях из итератора. Данный итератор может генерировать значения бесконечно, требуется только интерфейс итератора (без индексирования).

Эта опция планирования, как правило, является хорошим выбором, если нагрузка отдельных итераций не однородна или имеет большой разброс.

:static

Планировщик :static создаёт по одной задаче на поток и делит итерации поровну между ними, назначая каждую задачу конкретному потоку. В частности, значение threadid() гарантированно является постоянным внутри одной итерации. Указание :static является ошибкой, если оно используется внутри другого цикла @threads или из потока, отличного от 1.

Примеры

Чтобы проиллюстрировать различные стратегии планирования, рассмотрим следующую функцию busywait, содержащую неблокируемый цикл с указанным временем выполнения.

julia> function busywait(seconds)
            tstart = time_ns()
            while (time_ns() - tstart) / 1e9 < seconds
            end
        end

julia> @time begin
            Threads.@spawn busywait(5)
            Threads.@threads :static for i in 1:Threads.threadpoolsize()
                busywait(1)
            end
        end
6.003001 seconds (16.33 k allocations: 899.255 KiB, 0.25% compilation time)

julia> @time begin
            Threads.@spawn busywait(5)
            Threads.@threads :dynamic for i in 1:Threads.threadpoolsize()
                busywait(1)
            end
        end
2.012056 seconds (16.05 k allocations: 883.919 KiB, 0.66% compilation time)

Пример :dynamic занимает 2 секунды, так как один из неиспользуемых потоков может выполнить две из 1-секундных итераций для завершения цикла.

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

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

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

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

Примеры

julia> n = 20

julia> c = Channel{Int}(ch -> foreach(i -> put!(ch, i), 1:n), 1)

julia> d = Channel{Int}(n) do ch
           f = i -> put!(ch, i^2)
           Threads.foreach(f, c)
       end

julia> collect(d)
collect(d) = [1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 196, 225, 256, 289, 324, 361, 400]

Threads.@spawn [:default|:interactive] expr

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

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

Примеры

julia> t() = println("Hello from ", Threads.threadid());

julia> tasks = fetch.([Threads.@spawn t() for i in 1:4]);
Hello from 1
Hello from 1
Hello from 3
Hello from 4

Threads.threadid() -> Int

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

Примеры

julia> Threads.threadid()
1

julia> Threads.@threads for i in 1:4
          println(Threads.threadid())
       end
4
2
5
4

Threads.maxthreadid() -> Int

Получить нижнюю границу количества потоков (по всем пулам потоков) доступных для процесса Julia, с семантикой атомарного захвата. Результат всегда будет больше или равен threadid(), а также threadid(task) для любой задачи, которую вы могли наблюдать до вызова maxthreadid.

Threads.nthreads(:default | :interactive) -> Int

Получите текущее количество потоков в указанном пуле потоков. Потоки в :interactive имеют идентификаторы 1:nthreads(:interactive), а потоки в :default имеют идентификаторы в nthreads(:interactive) .+ (1:nthreads(:default)).

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

Threads.threadpool(tid = threadid()) -> Symbol

Возвращает указанный пул потоков; либо :default, либо :interactive, либо :foreign.

Threads.nthreadpools() -> Int

Возвращает количество пулов потоков, которые в настоящее время настроены.

Threads.threadpoolsize(pool::Symbol = :default) -> Int

Получить количество потоков, доступных для стандартного пула потоков (или для указанного пула потоков).

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

Threads.ngcthreads() -> Int

Возвращает количество потоков GC, которые в настоящее время настроены. Это включает как маркерные потоки, так и конкурирующие потоки сброса.

Небезопасные операции с указателями совместимы с загрузкой и сохранением указателей, объявленных с типом _Atomic и std::atomic соответственно в C11 и C++23. Может быть выброшено исключение, если нет поддержки атомарной загрузки типа Julia T.

См. также: unsafe_load, unsafe_modify!, unsafe_replace!, unsafe_store!, unsafe_swap!

@atomic var
@atomic order ex

Отметить var или ex как выполняемые атомарно, если ex является поддерживаемым выражением. Если order не указано, по умолчанию используется :sequentially_consistent.

@atomic a.b.x = new
@atomic a.b.x += addend
@atomic :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 2, 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

@atomiconce a.b.x = value
@atomiconce :sequentially_consistent a.b.x = value
@atomiconce :sequentially_consistent :monotonic a.b.x = value

Выполнить условное присваивание значения атомарно, если оно ранее не было установлено, вернув значение success::Bool. Где success указывает, была ли выполнено присваивание.

Эта операция переводится в вызов setpropertyonce!(a.b, :x, value).

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

Примеры

julia> mutable struct AtomicOnce
           @atomic x
           AtomicOnce() = new()
       end

julia> a = AtomicOnce()
AtomicOnce(#undef)

julia> @atomiconce a.x = 1 # set field x of a to 1, if unset, with sequential consistency
true

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

julia> @atomiconce a.x = 1 # set field x of a to 1, if unset, with sequential consistency
false

AtomicMemory{T} == GenericMemory{:atomic, T, Core.CPU}

Массив фиксированного размера DenseVector{T}. Доступ к любому из его элементов выполняется атомарно (с порядком :monotonic). Установка любого из элементов должна быть выполнена с помощью макроса @atomic и явным указанием порядка.

Для получения подробностей см. Атомарные операции.

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. Если !islocked(lck::SpinLock) выполняется, trylock(lck) успешно завершится, если другие задачи не пытаются захватить блокировку "одновременно".

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