Multi-Threading

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

for ループを並列に実行するためのマクロです。反復空間は粗粒度のタスクに分配されます。このポリシーは schedule 引数によって指定できます。ループの実行は、すべての反復の評価が完了するのを待ちます。

参照: @spawn および Distributed の pmap。

拡張ヘルプ

セマンティクス

スケジューリングオプションによってより強力な保証が指定されない限り、@threads マクロによって実行されるループは以下のセマンティクスを持ちます。

@threads マクロは、ループ本体を不特定の順序で、かつ潜在的に同時に実行します。タスクとワーカースレッドの正確な割り当ては指定されていません。割り当ては各実行ごとに異なる場合があります。ループ本体のコード（それから遷移的に呼び出されるコードを含む）は、反復のタスクへの分配や、それらが実行されるワーカースレッドについての仮定をしてはいけません。各反復のループ本体は、他の反復に依存せずに前進することができ、データ競合から解放されている必要があります。そのため、反復間での無効な同期はデッドロックを引き起こす可能性があり、同期されていないメモリアクセスは未定義の動作を引き起こす可能性があります。

例えば、上記の条件は以下を示唆します：

• 反復内で取得されたロックは、同じ反復内で解放されなければなりません。
• Channel のようなブロッキングプリミティブを使用して反復間で通信することは不正です。
• 反復間で共有されていない場所にのみ書き込みを行います（ロックまたは原子操作が使用されていない限り）。
• :static スケジュールが使用されない限り、threadid() の値は単一の反復内でも変わる可能性があります。Task Migrationを参照してください。

スケジューラ

スケジューラ引数がない場合、正確なスケジューリングは指定されておらず、Julia のリリースによって異なります。現在、スケジューラが指定されていない場合は :dynamic が使用されます。

:dynamic (デフォルト)

:dynamic スケジューラは、利用可能なワーカースレッドに対して反復を動的に実行します。現在の実装は、各反復の作業負荷が均一であると仮定しています。ただし、この仮定は将来的に削除される可能性があります。

このスケジューリングオプションは、基盤となる実行メカニズムへの単なるヒントです。ただし、いくつかの特性が期待できます。:dynamic スケジューラによって使用される Task の数は、利用可能なワーカースレッドの数の小さな定数倍に制限されています（Threads.threadpoolsize()）。各タスクは反復空間の連続した領域を処理します。したがって、@threads :dynamic for x in xs; f(x); end は、length(xs) がワーカースレッドの数よりも大幅に大きく、f(x) の実行時間がタスクの生成と同期のコスト（通常は10マイクロ秒未満）よりも相対的に小さい場合、通常は @sync for x in xs; @spawn f(x); end よりも効率的です。

:greedy

:greedy スケジューラは、最大で Threads.threadpoolsize() タスクを生成し、生成された値に対して貪欲に作業を行います。1つのタスクが作業を終えると、次の値をイテレータから取得します。個々のタスクが行う作業は、イテレータからの連続した値である必要はありません。与えられたイテレータは永遠に値を生成する可能性があり、イテレータインターフェースのみが必要です（インデックス付けは不要です）。

このスケジューリングオプションは、個々の反復の作業負荷が均一でない/大きなばらつきがある場合に一般的に良い選択です。

:static

:static スケジューラは、スレッドごとに1つのタスクを作成し、反復を均等に分割し、各タスクを特定のスレッドに割り当てます。特に、threadid() の値は1つの反復内で一定であることが保証されています。別の @threads ループ内またはスレッド1以外から使用する場合、:static を指定することはエラーです。

例

異なるスケジューリング戦略を示すために、指定された秒数間実行される非待機のタイムループを含む関数 busywait を考えます。

julia> function busywait(seconds)
            tstart = time_ns()
            while (time_ns() - tstart) / 1e9 < seconds
            end
        end

julia> @time begin
            Threads.@spawn busywait(5)
            Threads.@threads :static for i in 1:Threads.threadpoolsize()
                busywait(1)
            end
        end
6.003001 seconds (16.33 k allocations: 899.255 KiB, 0.25% compilation time)

julia> @time begin
            Threads.@spawn busywait(5)
            Threads.@threads :dynamic for i in 1:Threads.threadpoolsize()
                busywait(1)
            end
        end
2.012056 seconds (16.05 k allocations: 883.919 KiB, 0.66% compilation time)

:dynamic の例は2秒かかります。なぜなら、占有されていないスレッドの1つが2つの1秒の反復を実行してforループを完了できるからです。 ```

Threads.foreach(f, channel::Channel;
                schedule::Threads.AbstractSchedule=Threads.FairSchedule(),
                ntasks=Threads.threadpoolsize())

foreach(f, channel)と似ていますが、channelの反復処理とfへの呼び出しは、Threads.@spawnによって生成されたntasksタスクに分割されます。この関数は、内部で生成されたすべてのタスクが完了するのを待ってから戻ります。

scheduleがFairScheduleである場合、Threads.foreachは、Juliaのスケジューラがスレッド間で作業項目をより自由にロードバランスできるようにタスクを生成しようとします。このアプローチは一般的にアイテムごとのオーバーヘッドが高くなりますが、他のマルチスレッドワークロードと同時に実行する場合、StaticScheduleよりもパフォーマンスが向上する可能性があります。

scheduleがStaticScheduleである場合、Threads.foreachは、FairScheduleよりもアイテムごとのオーバーヘッドが低くなるようにタスクを生成しますが、ロードバランスにはあまり適していません。このアプローチは、細かい粒度の均一なワークロードにより適している可能性がありますが、他のマルチスレッドワークロードと同時に実行する場合、FairScheduleよりもパフォーマンスが低下する可能性があります。

例

julia> n = 20

julia> c = Channel{Int}(ch -> foreach(i -> put!(ch, i), 1:n), 1)

julia> d = Channel{Int}(n) do ch
           f = i -> put!(ch, i^2)
           Threads.foreach(f, c)
       end

julia> collect(d)
collect(d) = [1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 196, 225, 256, 289, 324, 361, 400]

Threads.@spawn [:default|:interactive] expr

Task を作成し、指定されたスレッドプール（指定しない場合は :default）の任意の利用可能なスレッドで実行するように schedule します。タスクは、利用可能なスレッドが見つかると、そのスレッドに割り当てられます。タスクが完了するのを待つには、このマクロの結果に対して wait を呼び出すか、fetch を呼び出して待機し、その戻り値を取得します。

値は $ を介して @spawn に補間することができ、これにより値が構築された基盤となるクロージャに直接コピーされます。これにより、変数の値を挿入でき、現在のタスク内での変数の値の変更から非同期コードを隔離できます。

例

julia> t() = println("Hello from ", Threads.threadid());

julia> tasks = fetch.([Threads.@spawn t() for i in 1:4]);
Hello from 1
Hello from 1
Hello from 3
Hello from 4

Threads.threadid() -> Int

現在の実行スレッドのID番号を取得します。マスタースレッドのIDは 1 です。

例

julia> Threads.threadid()
1

julia> Threads.@threads for i in 1:4
          println(Threads.threadid())
       end
4
2
5
4

Threads.maxthreadid() -> Int

Juliaプロセスで利用可能なスレッドの数（すべてのスレッドプールを通じて）の下限を取得します。これは、アトミック取得セマンティクスを持ちます。結果は常にthreadid()およびthreadid(task)（maxthreadidを呼び出す前に観察できたタスクに対して）以上になります。

Threads.nthreads(:default | :interactive) -> Int

指定されたスレッドプール内の現在のスレッド数を取得します。:interactive のスレッドは 1:nthreads(:interactive) のID番号を持ち、:default のスレッドは nthreads(:interactive) .+ (1:nthreads(:default)) のID番号を持ちます。

また、LinearAlgebra 標準ライブラリの BLAS.get_num_threads および BLAS.set_num_threads、および Distributed 標準ライブラリの nprocs() と Threads.maxthreadid() も参照してください。

Threads.threadpool(tid = threadid()) -> Symbol

指定されたスレッドのスレッドプールを返します。：default、：interactive、または：foreignのいずれかです。

Threads.nthreadpools() -> Int

現在構成されているスレッドプールの数を返します。

Threads.threadpoolsize(pool::Symbol = :default) -> Int

デフォルトのスレッドプール（または指定されたスレッドプール）で利用可能なスレッドの数を取得します。

関連情報: LinearAlgebra 標準ライブラリの BLAS.get_num_threads および BLAS.set_num_threads、および Distributed 標準ライブラリの nprocs()。

Threads.ngcthreads() -> Int

現在構成されているGCスレッドの数を返します。これには、マークスレッドと同時スイープスレッドの両方が含まれます。

Unsafe pointer operations are compatible with loading and storing pointers declared with _Atomic and std::atomic type in C11 and C++23 respectively. An error may be thrown if there is not support for atomically loading the Julia type T.

See also: unsafe_load, unsafe_modify!, unsafe_replace!, unsafe_store!, unsafe_swap!

@atomic var
@atomic order ex

var または ex を原子的に実行されるものとしてマークします。order が指定されていない場合、デフォルトは :sequentially_consistent です。

@atomic a.b.x = new
@atomic a.b.x += addend
@atomic :release a.b.x = new
@atomic :acquire_release a.b.x += addend

右側で表現されたストア操作を原子的に実行し、新しい値を返します。

= の場合、この操作は setproperty!(a.b, :x, new) 呼び出しに変換されます。演算子がある場合も、この操作は modifyproperty!(a.b, :x, +, addend)[2] 呼び出しに変換されます。

@atomic a.b.x max arg2
@atomic a.b.x + arg2
@atomic max(a.b.x, arg2)
@atomic :acquire_release max(a.b.x, arg2)
@atomic :acquire_release a.b.x + arg2
@atomic :acquire_release a.b.x max arg2

右側で表現された二項操作を原子的に実行します。結果を最初の引数のフィールドに格納し、値 (old, new) を返します。

この操作は modifyproperty!(a.b, :x, func, arg2) 呼び出しに変換されます。

詳細については、マニュアルの Per-field atomics セクションを参照してください。

例

julia> mutable struct Atomic{T}; @atomic x::T; end

julia> a = Atomic(1)
Atomic{Int64}(1)

julia> @atomic a.x # a のフィールド x を取得し、順序の一貫性を持たせる
1

julia> @atomic :sequentially_consistent a.x = 2 # a のフィールド x を設定し、順序の一貫性を持たせる
2

julia> @atomic a.x += 1 # a のフィールド x をインクリメントし、順序の一貫性を持たせる
3

julia> @atomic a.x + 1 # a のフィールド x をインクリメントし、順序の一貫性を持たせる
3 => 4

julia> @atomic a.x # a のフィールド x を取得し、順序の一貫性を持たせる
4

julia> @atomic max(a.x, 10) # a のフィールド x を最大値に変更し、順序の一貫性を持たせる
4 => 10

julia> @atomic a.x max 5 # 再度 a のフィールド x を最大値に変更し、順序の一貫性を持たせる
10 => 10

@atomicswap a.b.x = new
@atomicswap :sequentially_consistent a.b.x = new

newをa.b.xに格納し、a.b.xの古い値を返します。

この操作はswapproperty!(a.b, :x, new)呼び出しに変換されます。

詳細については、マニュアルのPer-field atomicsセクションを参照してください。

例

julia> mutable struct Atomic{T}; @atomic x::T; end

julia> a = Atomic(1)
Atomic{Int64}(1)

julia> @atomicswap a.x = 2+2 # aのフィールドxを4に置き換え、逐次的整合性を持つ
1

julia> @atomic a.x # aのフィールドxを取得し、逐次的整合性を持つ
4

@atomicreplace a.b.x 期待される => 希望する
@atomicreplace :sequentially_consistent a.b.x 期待される => 希望する
@atomicreplace :sequentially_consistent :monotonic a.b.x 期待される => 希望する

条件付き置換をペアで原子的に実行し、値 (old, success::Bool) を返します。ここで success は置換が完了したかどうかを示します。

この操作は replaceproperty!(a.b, :x, 期待される, 希望する) の呼び出しに相当します。

詳細についてはマニュアルの Per-field atomics セクションを参照してください。

例

julia> mutable struct Atomic{T}; @atomic x::T; end

julia> a = Atomic(1)
Atomic{Int64}(1)

julia> @atomicreplace a.x 1 => 2 # aのフィールドxを1であれば2に置換し、逐次的整合性を持つ
(old = 1, success = true)

julia> @atomic a.x # aのフィールドxを取得し、逐次的整合性を持つ
2

julia> @atomicreplace a.x 1 => 2 # aのフィールドxを1であれば2に置換し、逐次的整合性を持つ
(old = 2, success = false)

julia> xchg = 2 => 0; # aのフィールドxを2であれば0に置換し、逐次的整合性を持つ

julia> @atomicreplace a.x xchg
(old = 2, success = true)

julia> @atomic a.x # aのフィールドxを取得し、逐次的整合性を持つ
0

@atomiconce a.b.x = value
@atomiconce :sequentially_consistent a.b.x = value
@atomiconce :sequentially_consistent :monotonic a.b.x = value

値が以前に未設定であった場合に、条件付きで値を原子的に割り当て、success::Boolを返します。ここで、successは割り当てが完了したかどうかを示します。

この操作はsetpropertyonce!(a.b, :x, value)呼び出しに変換されます。

詳細については、マニュアルのPer-field atomicsセクションを参照してください。

例

julia> mutable struct AtomicOnce
           @atomic x
           AtomicOnce() = new()
       end

julia> a = AtomicOnce()
AtomicOnce(#undef)

julia> @atomiconce a.x = 1 # aのフィールドxを1に設定します。未設定の場合、順次一貫性を持って
true

julia> @atomic a.x # aのフィールドxを取得します。順次一貫性を持って
1

julia> @atomiconce a.x = 1 # aのフィールドxを1に設定します。未設定の場合、順次一貫性を持って
false

AtomicMemory{T} == GenericMemory{:atomic, T, Core.CPU}

固定サイズの DenseVector{T}。その要素へのアクセスは原子的に行われます（:monotonic 順序で）。要素の設定は @atomic マクロを使用し、明示的に順序を指定する必要があります。

詳細については、Atomic Operationsを参照してください。

Threads.Atomic{T}

型 T のオブジェクトへの参照を保持し、それが原子的に、つまりスレッドセーフな方法でのみアクセスされることを保証します。

原子的に使用できるのは、特定の「単純な」型のみで、具体的には原始的なブール型、整数型、および浮動小数点型です。これらは 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を原子的に実行します。古い値を返します。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

xをvalと原子的にビット単位でANDします。

x[] &= valを原子的に実行します。古い値を返します。

詳細については、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

原子的に x と val のビット単位の NAND（非論理積）を行います。

x[] = ~(x[] & val) を原子的に実行します。古い値を返します。

詳細については、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

xをvalと原子的にビット単位で論理和を取ります。

x[] |= valを原子的に実行します。古い値を返します。

詳細については、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

原子的に x と val のビット単位の排他的論理和（XOR）を計算します。

x[] $= val を原子的に実行します。古い値を返します。

詳細については、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)を原子的に実行します。古い値を返します。

詳細については、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)を原子的に実行します。古い値を返します。

詳細については、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 と同じ方法で呼び出されますが、異なるスレッドで作業を行います。これは、現在の julia スレッドをブロックさせることなく、ブロッキング C 関数を呼び出したい場合に便利です。並行性は libuv スレッドプールのサイズによって制限され、デフォルトでは 4 スレッドですが、UV_THREADPOOL_SIZE 環境変数を設定し、julia プロセスを再起動することで増やすことができます。

呼び出された関数は、決して Julia にコールバックしてはいけないことに注意してください。

SpinLock()

再入不可のテスト・アンド・テスト・アンド・セットスピンロックを作成します。再帰的な使用はデッドロックを引き起こします。この種のロックは、実行にかかる時間が短く、ブロックしないコード（例：I/Oを実行する）周辺でのみ使用するべきです。一般的には、ReentrantLock を代わりに使用するべきです。

各 lock は unlock と対になっている必要があります。もし !islocked(lck::SpinLock) が成り立つ場合、trylock(lck) は、他のタスクが「同時に」ロックを保持しようとしていない限り成功します。

テスト・アンド・テスト・アンド・セットスピンロックは、約30程度の競合スレッドまでは最も速いです。それ以上の競合がある場合は、異なる同期アプローチを検討する必要があります。