マルチスレッディング

julialang.org のブログ記事 Announcing composable multi-threaded parallelism in Julia にも Julia のマルチスレッディングに関する機能の解説があります。

Julia で複数のスレッドを使う

デフォルトの Julia は一つの実行スレッドを持って起動します。この事実は Threads.nthreads() を呼び出すと確認できます:

julia> Threads.nthreads()
1

実行スレッドの個数はコマンドライン引数 -t/--threads または環境変数 JULIA_NUM_THREADS で設定できます。両方とも指定されると -t/--threads が優先されます。

Julia 1.5

コマンドライン引数 -t/--threads を使うには最低でも Julia 1.5 が必要です。それよりも古いバージョンでは環境変数だけが使用できます。

Julia を四つのスレッドで開始してみましょう:

$ julia --threads 4

本当に四つのスレッドが利用できるかを確認します:

julia> Threads.nthreads()
4

Julia を開始すると、実行はマスタースレッドで始まります。この事実は Threads.threadid を使えば確認できます:

julia> Threads.threadid()
1

情報

環境変数を設定する方法を示します:

Bash (Linux/macOS):
```
export JULIA_NUM_THREADS=4
```
Linux/macOS の C shell および Windows の CMD:
```
set JULIA_NUM_THREADS=4
```
Windows の Powershell:
```
$env:JULIA_NUM_THREADS=4
```

これらのコマンドは Julia を開始する前に実行してください。

情報

-t/--threads で指定するスレッドの個数はコマンドライン引数 -p/--procs あるいは --machine-file を通して起動されるワーカープロセスに伝播されます。例えば julia -p2 -t2 とすると一つのメインプロセスと二つのワーカープロセスが起動し、三つのプロセス全てで二つのスレッドが有効化されます。ワーカースレッドをさらに細かく調整するには、addprocs 関数を使った上で exeflags に -t/--threads を渡してください。

データ競合

プログラムがデータ競合を持たないことを保証する責任は全てあなたにあります。この要件が満たされないとき、Julia が保証する種々の性質は一切成り立たたず、直感とは大きく異なる結果が生じる可能性があります。

データ競合を無くす一番の方法は、複数のスレッドから観測されるデータへの任意のアクセスをロックの取得と解放で囲むことです。例えば、多くの場合で次のパターンのいずれかを使うべきです:

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

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

またデータ競合があると、Julia はメモリ安全でなくなります。グローバル変数 (あるいはクロージャ変数) を読むときは、他のスレッドがその変数に書き込んでいないか注意してください！複数のスレッドに見えている任意のデータを変更するとき (例えばグローバル変数に代入するとき) は、必ず上記のロックパターンを使ってください。

スレッド 1:
global b = false
global a = rand()
global b = true

スレッド 2:
while !b; end
bad(a) # この a へのアクセスは安全でない！

スレッド 3:
while !@isdefined(a); end
use(a) # この a へのアクセスは安全でない！

@threads マクロ

Julia のネイティブスレッドを使う簡単な例を見てみましょう。最初にゼロ埋めされた配列を作ります:

julia> a = zeros(10)
10-element Array{Float64,1}:
 0.0
 0.0
 0.0
 0.0
 0.0
 0.0
 0.0
 0.0
 0.0
 0.0

これから四つのスレッドを使ってこの配列を処理します。それぞれのスレッドは自身のスレッド ID を配列の自身が受け持つ部分に書き込みます。

Julia は Threads.@threads マクロを使った並列ループをサポートします。このマクロの後に for ループを続けると、そのループがマルチスレッドで実行する領域であることが Julia に伝わります:

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

反復空間 (1:10) が自動的に分割されて四つのスレッドに割り当てられ、各スレッドが自身のスレッド ID を割り当てられた場所に書き込みます:

julia> a
10-element Array{Float64,1}:
 1.0
 1.0
 1.0
 2.0
 2.0
 2.0
 3.0
 3.0
 4.0
 4.0

Threads.@threads は縮約に関するパラメータを持たないことに注意してください。@distributed は省略可能なパラメータとして縮約関数を受け取ります。

不可分操作

Julia は値に対する不可分な (atomic) アクセスと更新をサポートします。これは競合状態を避けスレッドセーフに値を操作する方法の一つです。不可分操作の対象となる値はプリミティブ型でなければなりません。値を Threads.Atomic で包むと、その値には不可分操作でのみアクセス可能であることを指示できます。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 Array{Float64,1}:
 0.0
 1.0
 7.0
 3.0

julia> ids
4-element Array{Float64,1}:
 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 で包めるわけではありません。サポートされている型は Int8, Int16, Int32, Int64, Int128, UInt8, UInt16, UInt32, UInt64, UInt128, Float16, Float32, Float64 です。ただし AAarch32 と ppc64le では Int128 と UInt128 はサポートされません。

可変な関数引数の副作用

マルチスレッディングで実行しているときに純粋でない関数を使うときは、気を付けないと答えが間違った値となります。例えば ! で終わる関数は慣習として引数を変更するので、純粋ではありません。

@threadcall

外部ライブラリ (例えば ccall を通して呼び出す関数) は Julia が使うタスクベースの IO 機構と上手く噛み合いません。C ライブラリがブロックしても、Julia のスケジューラはその C 関数が返るまで他のタスクを実行できないためです (例外は Julia コードを呼び出す C 関数で、このときは呼び戻された Julia コードから yield できます。また Julia の yield と等価な C 関数 jl_yield() を呼び出せば C コードから直接 yeild できます)。

@threadcall は C 関数の実行を個別のスレッドでスケジュールし、このような状況で実行のストールを避ける方法を提供します。そのときはサイズ 4 のスレッドプールがデフォルトで使われますが、プールの大きさは環境変数 UV_THREADPOOL_SIZE で制御できます。外部関数を呼び出すタスクが (Julia のメインイベントループで) フリースレッドを待っている間、およびスレッドが利用可能になった後の関数実行中には、このタスクはすぐに他のタスクへ yield を行います。@threadcall は呼び出した関数の実行が終わるまで返らないことに注意してください。そのためユーザーの視点からは、このマクロは Julia が持つ他の API と同様のブロックする呼び出しとなります。

非常に重要な注意点として、@threadcall で呼び出した関数から Julia コードを呼び出してはいけません。セグメンテーションフォルトが発生します。

@threadcall は将来のバージョンの Julia で変更/削除される可能性があります。

注意点

現在のバージョンでは、Julia ランタイムと標準ライブラリに含まれる大部分の操作はスレッドセーフに使うことができます (ユーザーのコードがデータ競合を起こさないことが前提です)。しかし、一部の領域ではスレッドサポートを安定化させる作業がまだ進行中となっています。マルチスレッドプログラミングには特有の難しさが多く存在するので、もしプログラムが通常とは異なる動作や望ましくない動作 (例えばクラッシュや間違った返り値) を見せたときはスレッドの取り扱いを最初に調べるべきです。

Julia でスレッドを使うとき特に気を付けるべき制限や注意点を示します:

Base に含まれるコレクション型の値を複数のスレッドが利用し、かつ一つ以上のスレッドがコレクションを変更するときは手動のロックが必要となります (配列への push! や Dict への要素の挿入がよくある例です)。
タスクが (@spawn などを通して) 特定のスレッドで実行を開始すると、ブロックの後は必ず同じスレッドで実行が再開されます。将来この制限は取り除かれ、タスクはスレッド間を移動できるようになる予定です。
現在 Threads.@threads は静的なスケジュールを利用し、全てのスレッドに同じ回数の反復を割り当ています。将来デフォルトのスケジュールは動的なものにおそらく変更されます。
@spawn が使うスケジュールは非決定的であり、スケジュールに依存した処理を行ってはいけません。
メモリをアロケートしない計算バウンドなタスクがあると、メモリのアロケートを行う他のスレッドのガベージコレクションが実行されなくなることがあります。このような状況では GC.safepoint() を手動で挿入する必要があります。将来この制限は取り除かれる予定です。
型・メソッド・モジュールの定義の eval や include といったトップレベルの操作を並列に実行しないでください。
ライブラリが登録するファイナライザはスレッドを有効にしていると正常に動かない可能性があります。スレッディングが安定したものとして広く採用されるまでの間は、エコシステムのあちらこちらで移行のための処理が必要になるかもしれません。詳細はマニュアルの次の節を見てください。

ファイナライザの安全な使用

ファイナライザは任意のコードの実行に割り込めるので、ファイナライザからグローバルな状態を操作するときは厳重な注意が必要です。残念ながらファイナライザが使われる主な理由はグローバルな状態を更新するためである (純粋な関数は普通ファイナライザとして意味がない) ので、これは難しい問題となります。この問題に対処するアプローチをいくつか示します:

シングルスレッドのときは、コードから内部 C 関数 jl_gc_enable_finalizers を呼び出すことでファイナライザがクリティカルリージョン内にスケジュールされないようにできます。この関数は Julia 内部関数の一部 (C のロックに関する関数など) で使われており、特定の操作 (漸進的なパッケージの読み込みやコード生成など) が再帰的に行われないようにします。このフラグとロックを使えばファイナライザを安全に実行できます。

二つ目のアプローチは Base などのライブラリが採用しているもので、非再帰的にロックを取得できるまで明示的にファイナライザを遅らせるというものです。Distributed.finalize_ref に対してこのアプローチを適用した様子を次に示します¹:

function finalize_ref(r::AbstractRemoteRef)
    if r.where > 0 # ファイナライザが既に実行されていないかを確認する。
        if islocked(client_refs) || !trylock(client_refs)
            # ロックを取得できないなら、ファイナライザを遅らせる。
            finalizer(finalize_ref, r)
            return nothing
        end
        try # lock の後には必ず try を使うべきである。
            if r.where > 0 # もう一度チェックが必要。
                # 実際のクリーンアップを行う。
                r.where = 0
            end
        finally
            unlock(client_refs)
        end
    end
    nothing
end

これと似た三つ目のアプローチは yield が起こらないキューを使うものです。現在ロックフリーのキューは Base に実装されていませんが、Base.InvasiveLinkedListSynchronized{T} が適しています。イベントループを持つコードではこのアプローチが優れている場合が多くあり、例えば Gtk.jl では寿命に関する参照カウントの管理にこのアプローチが使われています。このアプローチでは finalizer の中では実際の処理は行われず、安全になったときに実行するキューにファイナライザが入れられるだけです。

実は Julia のタスクスケジューラもこのアプローチを使うので、x -> @spawn do_cleanup(x) とファイナライザを設定すればある意味でこれを使ったことになります。ただしこうすると do_cleanup を実行するスレッドを指定できないので、いずれにせよ do_cleanup でロックが必要になることに注意してください。キューを独自に実装して特定のスレッドからだけキューを消費するようにすればロックは必要ありません。

訳注: 以前のバージョンのコードであり、現在のコードとは少し異なる。^[return]