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

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

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

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

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

@threadcall

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

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

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

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

Ограничения

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

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

• Типы базовых коллекций требуют ручного блокирования, если они используются одновременно несколькими потоками, где хотя бы один поток изменяет коллекцию (общие примеры включают push! в массивах или вставку элементов в Dict).
• После того как задача начнёт выполняться в определённом потоке (например, с помощью @spawn), она всегда будет перезапускаться в том же потоке после блокировки. В будущем это ограничение будет устранено, и задачи будут перемещаться между потоками.
• @threads в настоящее время использует статическое расписание, используя все потоки и назначая одинаковое количество итераций каждому. В будущем значение расписания по умолчанию, скорее всего, изменится на динамическое.
• Расписание, используемое @spawn является недетерминированным и на нём не следует полагаться.
• Вычислительно-ёмкие задачи, не связанные с выделением памяти, могут препятствовать выполнению сбора мусора в других потоках, которые выделяют память. В таких случаях может потребоваться вставка явного вызова GC.safepoint() для запуска сбора мусора. Это ограничение будет устранено в будущем.
• Избегайте выполнения операций верхнего уровня, например, 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. Похожий третий подход заключается в использовании очереди без отдачи. В настоящее время в Base нет реализации бесключевой очереди, но Base.InvasiveLinkedListSynchronized{T} подходит. Это часто хороший подход для кода с циклами событий. Например, этот подход используется Gtk.jl для управления подсчетом ссылок на время жизни. В этом подходе мы не выполняем никакой явной работы внутри finalizer, а вместо этого добавляем её в очередь для выполнения в более безопасное время. Фактически, планировщик задач Julia уже использует это, поэтому определение финализатора как x -> @spawn do_cleanup(x) является одним примером такого подхода. Однако следует учитывать, что это не контролирует, на каком потоке do_cleanup будет выполнено, поэтому do_cleanup всё равно потребуется получить блокировку. Это не обязательно, если вы реализуете свою собственную очередь, поскольку вы можете явно считывать эту очередь только с вашего потока.