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

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

Многопотоковый сборщик мусора

Сборщик мусора (GC) может использовать несколько потоков. Количество используемых потоков равно либо половине числа потоков вычислений, либо настраивается с помощью аргумента командной строки --gcthreads или переменной среды JULIA_NUM_GC_THREADS.

Пулы потоков

Когда потоки программы заняты многими задачами, задачи могут испытывать задержки, что может негативно сказаться на отзывчивости и интерактивности программы. Чтобы решить эту проблему, вы можете указать, что задача является интерактивной, когда вы Threads.@spawn её:

using Base.Threads
@spawn :interactive f()

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

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

$ julia --threads 3,1

Переменная среды JULIA_NUM_THREADS также может быть использована аналогичным образом:

export JULIA_NUM_THREADS=3,1

Это запускает Julia с 3 потоками в :default пуле потоков и 1 потоком в :interactive пуле потоков:

julia> using Base.Threads

julia> nthreadpools()
2

julia> threadpool() # the main thread is in the interactive thread pool
:interactive

julia> nthreads(:default)
3

julia> nthreads(:interactive)
1

julia> nthreads()
3

Любое из чисел или оба можно заменить словом auto, что заставит Julia выбрать разумное значение по умолчанию.

Макрос @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 не имеет необязательного параметра reduction, в отличие от @distributed.

Использование @threads без гонок данных

Концепция гонок данных подробно описана в "Общении и гонках данных между потоками". На данный момент достаточно знать, что гонка данных может привести к некорректным результатам и опасным ошибкам.

Предположим, мы хотим сделать функцию sum_single ниже многопоточной.

julia> function sum_single(a)
           s = 0
           for i in a
               s += i
           end
           s
       end
sum_single (generic function with 1 method)

julia> sum_single(1:1_000_000)
500000500000

Просто добавление @threads приводит к гонке данных с несколькими потоками, одновременно читающими и записывающими s.

julia> function sum_multi_bad(a)
           s = 0
           Threads.@threads for i in a
               s += i
           end
           s
       end
sum_multi_bad (generic function with 1 method)

julia> sum_multi_bad(1:1_000_000)
70140554652

Обратите внимание, что результат не 500000500000, как это должно быть, и, скорее всего, будет меняться при каждом вычислении.

Для исправления этого можно использовать буферы, специфичные для задачи, для разделения суммы на блоки, свободные от гонок. Здесь sum_single используется повторно со своим внутренним буфером s. Вектор ввода a разбивается на nthreads() блока для параллельной работы. Затем мы используем Threads.@spawn для создания задач, которые индивидуально суммируют каждый блок. Наконец, мы суммируем результаты каждой задачи, используя sum_single снова:

julia> function sum_multi_good(a)
           chunks = Iterators.partition(a, length(a) ÷ Threads.nthreads())
           tasks = map(chunks) do chunk
               Threads.@spawn sum_single(chunk)
           end
           chunk_sums = fetch.(tasks)
           return sum_single(chunk_sums)
       end
sum_multi_good (generic function with 1 method)

julia> sum_multi_good(1:1_000_000)
500000500000

Другой вариант — использование атомных операций над переменными, общими для задач/потоков, что может быть более эффективным в зависимости от характеристик операций.

Общение и гонки данных между потоками

Хотя потоки Julia могут обмениваться данными через общую память, создание правильного и свободного от гонок многопоточного кода чрезвычайно сложно. Channel Julia являются потокобезопасными и могут использоваться для безопасного обмена данными. Также ниже представлены разделы, объясняющие, как использовать блокировки и атомарные операции для предотвращения гонок данных.

Свобода от гонок данных

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

Если гонки данных возникнут, 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

Использование блокировок для предотвращения гонок данных

Важным инструментом для предотвращения гонок данных и, следовательно, для написания потокобезопасного кода является концепция «блокировки». Блокировка может быть заблокирована и разблокирована. Если поток заблокировал блокировку и не разблокировал её, считается, что он «держит» блокировку. Если существует только одна блокировка и мы пишем код, требующий удержания блокировки для доступа к данным, мы можем гарантировать, что несколько потоков никогда не будут одновременно получать доступ к тем же данным. Обратите внимание, что связь между блокировкой и переменной устанавливается программистом, а не программой.

Например, мы можем создать блокировку my_lock, и заблокировать её, пока мы изменяем переменную my_variable. Это проще всего сделать с помощью макроса @lock:

julia> my_lock = ReentrantLock();

julia> my_variable = [1, 2, 3];

julia> @lock my_lock my_variable[1] = 100
100

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

Мы могли бы выполнить операцию выше с функциональной версией lock следующими двумя способами:

julia> lock(my_lock) do
           my_variable[1] = 100
       end
100

julia> begin
           lock(my_lock)
           try
               my_variable[1] = 100
           finally
               unlock(my_lock)
           end
       end
100

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

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

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> Threads.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 и макросы @atomiconce.

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

Любое поле в объявлении структуры может быть снабжено декоратором @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 определений типов, методов и модулей в параллельном режиме.
• Следите за тем, что финализаторы, зарегистрированные библиотекой, могут сломаться, если включены потоки. Это может потребовать некоторых переходных работ в экосистеме, прежде чем многопоточность сможет быть широко принята с уверенностью. Подробности см. в разделе о безопасном использовании финализаторов.

Перемещение задач

После запуска задачи в определённом потоке она может переместиться в другой поток, если задача уступит.

Такие задачи могут быть запущены с помощью @spawn или @threads, хотя опция планирования :static для @threads замораживает идентификатор потока.

Это означает, что в большинстве случаев threadid() не следует рассматривать как постоянную величину в рамках задачи и поэтому не следует использовать для индексирования в вектор буферов или состоятельных объектов.

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

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

1. В однопоточном режиме код может вызвать внутреннюю функцию C jl_gc_enable_finalizers для предотвращения планирования финализаторов внутри критической области. Внутренне это используется в некоторых функциях (таких как наши блокировки 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.IntrusiveLinkedListSynchronized{T} подходит. Это часто бывает хорошей стратегией для кода с циклами событий. Например, эта стратегия используется Gtk.jl для управления подсчётом ссылок на время жизни. В этом подходе мы не выполняем никаких явных действий внутри finalizer, а вместо этого добавляем его в очередь для выполнения в более безопасное время. Фактически, планировщик задач Julia уже использует это, поэтому определение финализатора как x -> @spawn do_cleanup(x) — это один из примеров этого подхода. Однако обратите внимание, что это не контролирует, в каком потоке do_cleanup выполняется, поэтому do_cleanup всё равно потребуется получить блокировку. Это не обязательно должно быть так, если вы реализуете свою собственную очередь, так как вы можете явно ограничить извлечение из этой очереди только из вашего потока.