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

Посетите эту статью блога для ознакомления с возможностями многопоточности в Julia.

Запуск Julia с несколькими потоками

По умолчанию Julia запускается с одним потоком выполнения. Это можно проверить с помощью команды Threads.nthreads():

julia> Threads.nthreads()
1

Количество потоков выполнения контролируется либо с помощью аргумента командной строки -t/--threads , либо с помощью переменной окружения JULIA_NUM_THREADS. Если оба параметра указаны, то -t/--threads имеет приоритет.

Количество потоков может быть указано либо как целое число (--threads=4) , либо как auto (--threads=auto), где auto устанавливает количество потоков, равное количеству локальных потоков процессора.

Давайте запустим Julia с 4 потоками:

$ julia --threads 4

Давайте проверим, что у нас есть 4 потока.

julia> Threads.nthreads()
4

Но мы сейчас находимся в главном потоке. Чтобы проверить это, используем функцию Threads.threadid

julia> Threads.threadid()
1

export JULIA_NUM_THREADS=4

set JULIA_NUM_THREADS=4

$env:JULIA_NUM_THREADS=4

Безопасность от гонок данных

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

Лучший способ гарантировать это — получить блокировку вокруг любого доступа к данным, которые могут наблюдаться из нескольких потоков. Например, в большинстве случаев вы должны использовать следующую схему кода:

julia> lock(lk) do
           use(a)
       end

julia> begin
           lock(lk)
           try
               use(a)
           finally
               unlock(lk)
           end
       end

где lk — это блокировка (например, ReentrantLock()), а a — данные.

Кроме того, Julia не является безопасной в отношении памяти при наличии гонок данных. Будьте очень осторожны при чтении любых данных, если другой поток может их записать! Вместо этого всегда используйте схему блокировки выше при изменении данных (например, при присваивании глобальной или переменной замыкания), к которым обращаются другие потоки.

Thread 1:
global b = false
global a = rand()
global b = true

Thread 2:
while !b; end
bad_read1(a) # it is NOT safe to access `a` here!

Thread 3:
while !@isdefined(a); end
bad_read2(a) # it is NOT safe to access `a` here

Макрос @threads

Давайте рассмотрим простой пример с использованием наших родных потоков. Давайте создадим массив нулей:

julia> a = zeros(10)
10-element Vector{Float64}:
 0.0
 0.0
 0.0
 0.0
 0.0
 0.0
 0.0
 0.0
 0.0
 0.0

Давайте одновременно обработаем этот массив с использованием 4 потоков. Каждый поток будет записывать свой идентификатор потока в каждую ячейку.

Julia поддерживает параллельные циклы с помощью макроса Threads.@threads. Этот макрос ставится перед циклом for для указания Julia того, что цикл представляет собой многопоточную область:

julia> Threads.@threads for i = 1:10
           a[i] = Threads.threadid()
       end

Пространство итераций делится между потоками, после чего каждый поток записывает свой идентификатор потока в свои выделенные ячейки:

julia> a
10-element Vector{Float64}:
 1.0
 1.0
 1.0
 2.0
 2.0
 2.0
 3.0
 3.0
 4.0
 4.0

Обратите внимание, что Threads.@threads не имеет необязательного параметра сокращения, как @distributed.

Атомарные операции

Julia поддерживает доступ к значениям и их изменение атомарно, то есть безопасным для потоков способом, чтобы избежать гонок. Значение (которое должно быть примитивного типа) может быть обернуто как Threads.Atomic для указания того, что к нему нужно обращаться таким образом. Вот пример:

julia> i = Threads.Atomic{Int}(0);

julia> ids = zeros(4);

julia> old_is = zeros(4);

julia> Threads.@threads for id in 1:4
           old_is[id] = Threads.atomic_add!(i, id)
           ids[id] = id
       end

julia> old_is
4-element Vector{Float64}:
 0.0
 1.0
 7.0
 3.0

julia> i[]
 10

julia> ids
4-element Vector{Float64}:
 1.0
 2.0
 3.0
 4.0

Если бы мы попытались выполнить сложение без атомарной метки, мы, возможно, получили бы неправильный ответ из-за гонки. Пример того, что бы произошло, если бы мы не избежали гонки:

julia> using Base.Threads

julia> nthreads()
4

julia> acc = Ref(0)
Base.RefValue{Int64}(0)

julia> @threads for i in 1:1000
          acc[] += 1
       end

julia> acc[]
926

julia> acc = Atomic{Int64}(0)
Atomic{Int64}(0)

julia> @threads for i in 1:1000
          atomic_add!(acc, 1)
       end

julia> acc[]
1000

Атомарные операции для полей

Мы также можем использовать атомарные операции на более мелком уровне с помощью макросов @atomic, @atomicswap и @atomicreplace.

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

Любое поле в объявлении структуры может быть помечено с помощью @atomic, а затем любая запись должна быть помечена с помощью @atomic также, и должна использовать одну из определённых атомарных упорядочений (:monotonic, :acquire, :release, :acquire_release, или :sequentially_consistent). Любое чтение атомарного поля также может быть аннотировано атомарным ограничением упорядочения, или будет выполнено с монотонным (упрощённым) упорядочением, если не указано иное.

Побочные эффекты и изменяемые аргументы функций

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

@threadcall

Внешние библиотеки, такие как те, к которым обращаются с помощью ccall, создают проблему для механизма ввода-вывода Julia, основанного на задачах. Если библиотека C выполняет блокирующую операцию, это предотвращает планировщик Julia от выполнения каких-либо других задач до возврата вызова. (Исключения — вызовы в пользовательский код C, который делает обратные вызовы в Julia, которые затем могут уступить, или код C, который вызывает jl_yield(), эквивалент C yield.)

Макрос @threadcall предоставляет способ избежать остановки выполнения в такой ситуации. Он планирует функцию C для выполнения в отдельном потоке. Для этого используется пул потоков с размером по умолчанию 4. Размер пула потоков контролируется через переменную окружения UV_THREADPOOL_SIZE. При ожидании свободного потока и во время выполнения функции, как только поток становится доступным, запрашиваемая задача (в основном цикле событий Julia) уступает другим задачам. Обратите внимание, что @threadcall не возвращается до завершения выполнения. С точки зрения пользователя это, следовательно, блокирующий вызов, как и другие API Julia.

Крайне важно, чтобы вызываемая функция не делала обратных вызовов в Julia, так как это приведёт к аварийному завершению.

@threadcall может быть удалено/изменено в будущих версиях Julia.

Ограничения

В настоящее время большинство операций в среде выполнения Julia и стандартных библиотеках могут использоваться безопасным для потоков способом, если код пользователя не содержит гонок. Однако в некоторых областях ведутся работы по стабилизации поддержки потоков. Многопоточное программирование имеет много присущих трудностей, и если программа, использующая потоки, демонстрирует необычное или нежелательное поведение (например, аварийное завершение или загадочные результаты), прежде всего следует подозревать взаимодействие потоков.

Существует несколько конкретных ограничений и предупреждений, которые необходимо учитывать при использовании потоков в Julia:

• Типы базовых коллекций требуют ручной блокировки, если они используются одновременно несколькими потоками, где, по крайней мере, один поток изменяет коллекцию (общие примеры включают push! для массивов или вставка элементов в Dict).
• Расписание, используемое @spawn , является непредсказуемым и на нём нельзя полагаться.
• Вычислительно интенсивные задачи, не требующие выделения памяти, могут препятствовать выполнению сбора мусора в других потоках, которые выделяют память. В этих случаях может потребоваться вставка явного вызова GC.safepoint() для запуска GC. Это ограничение будет устранено в будущем.
• Избегайте запуска операций верхнего уровня, например include или eval определения типов, методов и модулей параллельно.
• Обратите внимание, что финализаторы, зарегистрированные библиотекой, могут сломаться, если потоки включены. Это может потребовать некоторой переходной работы по всему экосистеме до того, как многопоточность будет широко принята с уверенностью. Подробности см. в следующем разделе.

Безопасное использование финализаторов

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

1. При однопоточном выполнении код может вызывать внутреннюю jl_gc_enable_finalizers функцию C, чтобы предотвратить планирование финализаторов внутри критической области. Внутренне это используется внутри некоторых функций (таких как наши блокировки C) для предотвращения рекурсии при выполнении определенных операций (инкрементная загрузка пакетов, генерация кода и т.д.). Сочетание блокировки и этого флага может использоваться для обеспечения безопасности финализаторов.

2. Второй подход, используемый Base в нескольких местах, заключается в явном откладывании финализатора до момента, когда он сможет получить свою блокировку нерекурсивно. Следующий пример демонстрирует, как этот подход можно применить к Distributed.finalize_ref:

   function finalize_ref(r::AbstractRemoteRef)
       if r.where > 0 # Check if the finalizer is already run
           if islocked(client_refs) || !trylock(client_refs)
               # delay finalizer for later if we aren't free to acquire the lock
               finalizer(finalize_ref, r)
               return nothing
           end
           try # `lock` should always be followed by `try`
               if r.where > 0 # Must check again here
                   # Do actual cleanup here
                   r.where = 0
               end
           finally
               unlock(client_refs)
           end
       end
       nothing
   end

3. Связанный третий подход заключается в использовании очереди без использования yield. У нас в настоящее время нет реализации бесблокировочной очереди в Base, но Base.InvasiveLinkedListSynchronized{T} подходит. Это часто может быть хорошим подходом для кода с циклами событий. Например, этот подход используется Gtk.jl для управления подсчётом ссылок на время существования. В этом подходе мы не выполняем никакой явной работы внутри finalizer, а вместо этого добавляем его в очередь для выполнения в более безопасное время. Фактически, планировщик задач Julia уже использует это, поэтому определение финализатора как x -> @spawn do_cleanup(x) является примером этого подхода. Однако следует отметить, что это не контролирует, на каком потоке do_cleanup будет выполняться, поэтому do_cleanup по-прежнему должен получить блокировку. Это не обязательно должно выполняться, если вы реализуете свою собственную очередь, так как вы можете явно осуществить дренаж этой очереди только из своего потока.