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

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

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

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

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

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 в любом доступном потоке. Для ожидания завершения задачи вызовите wait на результате этого макроса или вызовите fetch для ожидания и получения его возвращаемого значения.

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

Threads.threadid()

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

Threads.nthreads()

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

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 остается в сигнализированном состоянии, и задачи больше не будут блокироваться при ожидании его.

Threads.Atomic{T}

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

Только некоторые "простые" типы могут использоваться атомарно, а именно примитивные типы boolean, integer и float-point. Это 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 атомарно. Возвращает значение old. Не определено для 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

Атомарно выполняет побитовое AND x с val

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

Для получения более подробной информации см. инструкцию 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

Атомарно выполняет побитовое NAND (not-and) x с val

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

Для получения более подробной информации см. инструкцию 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

Атомарно выполняет побитовое OR x с val

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

Для получения более подробной информации см. инструкцию 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

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

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

Для получения более подробной информации см. инструкцию 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) атомарно. Возвращает значение old.

Для получения более подробной информации см. инструкцию 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) атомарно. Возвращает значение old.

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