Multi-Threading

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 секунде, чтобы завершить цикл for. ```

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

!!! примечание Поток, на котором выполняется задача, может измениться, если задача приостановится, что известно как Миграция задач. По этой причине в большинстве случаев небезопасно использовать threadid() для индексации, скажем, вектор буферов или объектов с состоянием.

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).

Смотрите раздел Per-field atomics в руководстве для получения дополнительных сведений.

Примеры

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

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

julia> @atomic a.x # получить поле x из a, с последовательной согласованностью
1

julia> @atomic :sequentially_consistent a.x = 2 # установить поле x из a, с последовательной согласованностью
2

julia> @atomic a.x += 1 # увеличить поле x из a, с последовательной согласованностью
3

julia> @atomic a.x + 1 # увеличить поле x из a, с последовательной согласованностью
3 => 4

julia> @atomic a.x # получить поле x из a, с последовательной согласованностью
4

julia> @atomic max(a.x, 10) # изменить поле x из a на максимальное значение, с последовательной согласованностью
4 => 10

julia> @atomic a.x max 5 # снова изменить поле x из a на максимальное значение, с последовательной согласованностью
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).

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

Примеры

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

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

julia> @atomicswap a.x = 2+2 # заменить поле x у a на 4 с последовательной согласованностью
1

julia> @atomic a.x # получить поле x у a с последовательной согласованностью
4

@atomicreplace a.b.x ожидаемое => желаемое
@atomicreplace :sequentially_consistent a.b.x ожидаемое => желаемое
@atomicreplace :sequentially_consistent :monotonic a.b.x ожидаемое => желаемое

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

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

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

Примеры

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

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

julia> @atomicreplace a.x 1 => 2 # заменить поле x у a на 2, если оно было 1, с последовательной согласованностью
(old = 1, success = true)

julia> @atomic a.x # получить поле x у a, с последовательной согласованностью
2

julia> @atomicreplace a.x 1 => 2 # заменить поле x у a на 2, если оно было 1, с последовательной согласованностью
(old = 2, success = false)

julia> xchg = 2 => 0; # заменить поле x у a на 0, если оно было 2, с последовательной согласованностью

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

julia> @atomic a.x # получить поле x у a, с последовательной согласованностью
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).

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

Примеры

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

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

julia> @atomiconce a.x = 1 # установить поле x у a в 1, если не установлено, с последовательной согласованностью
true

julia> @atomic a.x # получить поле x у a, с последовательной согласованностью
1

julia> @atomiconce a.x = 1 # установить поле x у a в 1, если не установлено, с последовательной согласованностью
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.

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

Эта функция может быть использована для реализации транзакционной семантики. Перед транзакцией записывается значение в 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. Возвращает старое значение.

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

Примеры

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}.

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

Примеры

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}.

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

Примеры

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 атомарно. Возвращает старое значение.

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

Примеры

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

Атомарно выполняет побитовую nand (не-и) x с val

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

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

Примеры

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 атомарно. Возвращает старое значение.

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

Примеры

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

Атомарно выполняет побитовую операцию XOR (исключающее ИЛИ) x с val

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

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

Примеры

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) атомарно. Возвращает старое значение.

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

Примеры

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) атомарно. Возвращает старое значение.

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

Примеры

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

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

julia> x[]
5

Threads.atomic_fence()

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

Вставляет барьер памяти с семантикой последовательной согласованности. Существуют алгоритмы, где это необходимо, т.е. где порядок acquire/release недостаточен.

Это, вероятно, очень дорогостоящая операция. Учитывая, что все другие атомарные операции в Julia уже имеют семантику acquire/release, явные барьеры, как правило, не должны быть необходимы.

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

@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 конкурирующих потоках. Если у вас больше конкуренции, следует рассмотреть другие подходы к синхронизации.