はじめに
今回はGCDの内容を中心に並列処理についてまとめています。
マルチコアと並列処理
コンピュータの黎明期にはCPUのクロック速度で決定されていた最大作業量ですが、CPUのコア数を増やすことで格段に改善されました。2006年頃にコアが2つのデュアルコア、2007年以降はコアが4つあるクアッドコアが登場し、近年ではヘキサコア、オクタコアが標準になりつつあります。これらのマルチコアの優位性を活かすために、複数の処理を同時に実行ソフトウェアが必要となります。実際iOS, OSXのような最新マルチタスクOSでは数百ものプログラムを同時に実行できます。
これらの並列処理を実現するためには、単にコアと同数のスレッドを生成すればいいという単純な話ではありません。アプリケーションが独自に使用可能なコア数を計算し、それを効率的に干渉させずに動作させることは非常に困難になります。
Appleのオペレーションシステムには、コア数、システムの状況に応じてアプリケーションの処理量を動的、かつ単純に扱える手段が組み込まれています。
プロセッサのベンチマークスコア: TechTastic Apple A10X prestaties
並列処理のAPI
伝統的なスレッド方式による並列処理のアプローチは、開発者の実装コストやメンテナンス性が高いことが問題でしたが、iOS, OSXでは非同期設計のアプローチで並列処理を実現し、スレッドを管理することなく、あらゆるタスクを非同期に実行できる技術を提供しています。その一つとしてGrand CentralDispatch（GCD）があります。
GCDはタスクをキューに追加するだけで、必要なスレッドを生成し、タスクを適切にスケジューリングしてくれるので、効率的に並列処理を実現してくれます。
従来のGCDは関数でインターフェイスを提供しているのでSwiftだけではなく、C言語でも使用できました。Swift3からはGCDの仕様が大幅に変更され、CベースなインターフェイスからSwiftライクなインターフェイスに変更されました $^1$
二つ目にオペレーションキューというものがあります。これはGCDをラップしたものになっていてGCDと同様、オペレーションキューもスレッド管理に必要な処理をすべて行ってくれます $^2$
- swift-evolution / Modernize libdispatch for Swift 3 naming conventions
- API Reference / OperationQueue
ディスパッチキュー
タスクを直列にも並列にも実行でき、順序はFIFO方式(キューに追加されたのと同じ順序でタスクを取り出す)で決定します。直列ディスパッチキューでは同時に1つのタスクのみを実行し、並列ディスパッチキューは、タスクの完了を待たずに、できるだけ多くのタスクを起動します。
オペレーションキュー
ディスパッチキューがタスクをFIFO方式で実行するのに対し、オペレーションキューはほかの因子も考慮して実行順序を決めます。タスクが完了してからでないと実行できないなどの依存関係を、タスクを定義する際に設定できます。常にタスクを並列実行しますが、必要ならば依存関係を適切に設定することにより、直列に実行させることも可能になっています。
非同期設計
並列処理プログラミングガイドでは非同期処理の設計をするにあたり、いくつかのヒントが記述されています。
アプリケーションに期待される振る舞いの定義
アプリケーションが実行するべきタスク,それに対する振る舞いや状態の変化を明確に規定し、各タスクの遂行に必要な、個々の手順に分割します。その結果データ構造どうしの依存関係や状態にどんな影響があるかを検討し、独立に変更できるのであれば並列処理が可能かどうか検討するべきです。
実行可能な処理単位の抽出
タスクを実行する順序によって結果が変わるようであれば直列実行を検討し、結果が変わらないのであれば並列実行を検討します。
従来型のスレッドよりも生成コストが格段に小さいため、処理単位が小さくても問題はなく効率的に実行できます。ただ、実際の処理性能を測定し、必要ならばタスクの大きさを調整することは必要です。
キューの使い分けについて
作成したタスクがどのような順序で実行すれば、当該タスクが正常に処理されるか検討し、該当するディスパッチキューに追加、もしくは複数用意して使い分ける必要があります。
オペレーションオブジェクトの場合はオブジェクトの設定が重要になります。
スレッドが向いている状況
実時間処理を要するコードの実装手段としては、今なおスレッドが優れているため、バックグラウンドで動作するコードについて、事前の振る舞いを精度よく予測したい場合は、スレッドは向いています。スレッドプログラミングでは慎重に、どうしても必要な箇所に限定してスレッドを使います。
その他のヒント
- 値を直接計算する際には、あるプロセッサコアのレジスタやキャッシュを使うことになるが、こ れはメインメモリよりもはるかに高速なため、タスク内で直接値を計算することを検討する
- ディスパッチキューやオペレーションキューを利用すれば、ロックが必要になる状況はほとんどないため、競合を回避するためにはタスクの順序を意識する
Grand Central Dispatch
今回は、主にGrand Central Dispatch（GCD）のディスパッチキューについてまとめていきたいと思います。
ディスパッチキューは、タスク群を非同期かつ並列に実行する手段として容易に利用できる強力なAPIです。基本的にはタスクを関数やクロージャーの形式でキュー追加していく方式で扱います。
ディスパッチキューの型
ディスパッチキューは、追加された順にタスクを実行する先入れ先出し方式のデータ構造になっています。目的に応じて扱うディスパッチキューの型は以下3つになります。
- 直列キュー
- タスクを同時に1つずつ、追加された順に実行する。プライベートディスパッチキューも直列キューに分類されます
- 並列キュー
- 複数のタスクを並列に実行するが、各タスクはキューに追加された順序で実行をされる。グローバルディスパッチキューの一種もこちらに分類される
- メインディスパッチキュー
- アプリケーションが起動した時に自動的に1つだけ作成される特別なキュー。アプリケーションのどこからでも利用できる直列キューで、アプリケーションのメインスレッド上でタスクを実行する。主にUIの更新処理はここで行う
キューにタスクを追加する方法
生成されたディスパッチキューに対してタスクを追加して、処理を実行していくわけですが、追加する方法として主に同期 / 非同期の2通りがあります。基本的に非同期で追加する方法が推奨されているようです。
タスクをキューに追加する方法には、非同期/同期の2とおりがあります。可能な限り、非同期に追加する方法を推奨します Apple, 並列プログラミングガイド
同期的に追加しなければならない場合は、デッドロックを回避するため同じキュー上で動作しているタスクから呼び出してはいけないようです。
同期、非同期の挙動の違い
// 同期でタスクを追加
DispatchQueue.global().sync {
sleep(1)
print("Excute Task")
}
print("Qiita")
// Excute Task
// Qiita
//非同期でタスクを追加
DispatchQueue.global().async {
sleep(1)
print("Excute Task")
}
print("Qiita")
// Qiita
// Excute Task
直列、並列の挙動の違い
(0...10).forEach { index in
// 直列で実行
serialQueue.async {
print("Index: \(index)")
}
// Index: 0
// Index: 1
// Index: 2
.....
// 並列で実行
concurrentQueue.async {
print("Index: \(index)")
}
// Index: 1
// Index: 0
// Index: 2
.....
注意点として、直列に実行されるのは同じディスパッチキューに登録されたタスクの範囲内だけです。
使い方
ディスパッチキューは非常に簡潔に扱えるAPIとなっています。主にやることは二つだけです。
- キューを生成する
- タスクを追加する
メインディスパッチキュー
public class var main: DispatchQueue { get }
DispatchQueue.main.async {
// excute in main thread
}
- メインスレッド上で動作
- 直列
- システム作成されているので取得するだけ
グローバルディスパッチキュー
public class func global(qos: DispatchQoS.QoSClass = default) -> DispatchQueue
DispatchQueue.global().async {
// excute
}
- 並列
- 並列に実行できる複数のタスクがある場合に有用
- システム作成されているので取得するだけで使用可能
- 優先度を指定することが可能
- 同時に実行可能なタスク数は可変で、プロセッサのコア数や他のプロセスが処理した処理量など様々な要因が関与する
Quality of Service
public enum QoSClass {
case background
case utility
case `default`
case userInitiated
case userInteractive
case unspecified
}
Quality of Serviceはシステムのタスクのスケジューリングに一貫性を持たせるための概念で、タスク優先度を示すサービスレベルが定義されています。
優先度の指定は、次のような用途別に指定します。
| 用途 | サービスレベル |
|---|---|
| ユーザのインタラクションによって、即座にUIの更新を求めたい | userInteractive |
| ユーザのインタラクションをもとに、UIに結果が反映したい | userInitiated |
| どのサービスレベルにも当てはまらなそう | default |
| 定時的なコンテンツのアップデート、ビデオファイルの取り込み、プログレスバー表示など即時反映することを期待しないもの | utility |
| 適切な時間に延期可能なタスクで、ユーザを起点としないbackgrondで動作するもの | background |
プライベートディスパッチキュー
// 直列
DispatchQueue(label: "com.GCD.privateSerialQueue").async {
// excute
}
// 並列
DispatchQueue(label: "com.GCD.privateConcurrentQueue", attributes: .concurrent).async {
// excute
}
- QoSや直列・並列など用途によって細かい指定ができる。
- ラベルを明示的に指定する
逆引きディスパッチキュー
X秒後に実行
DispatchQueue.main.asyncAfter(deadline: .now() + .seconds(5)) {
// excute 5 second after
}
処理の完了を待つ
DispatchGroup
let group = DispatchGroup()
let queue1 = DispatchQueue(label: "com.GCD.groupQueue1")
let queue2 = DispatchQueue(label: "com.GCD.groupQueue2")
let queue3 = DispatchQueue(label: "com.GCD.groupQueue3")
queue1.async(group: group) {
sleep(4)
print("excute queue1")
}
queue2.async(group: group) {
sleep(2)
print("excute queue2")
}
queue3.async(group: group) {
sleep(1)
print("excute queue3")
}
group.notify(queue: DispatchQueue.main) {
print("All task done")
}
excute queue3
excute queue2
excute queue1
All task done
DispatchSemaphore
-
signal()カウントを1増やす -
wait()カウントを1減らす
let semaphone = DispatchSemaphore(value: 0)
let queue = DispatchQueue.global()
queue.async {
print("Excute sleep")
sleep(2)
print("End sleep")
semaphone.signal()
}
print("Wait task")
semaphone.wait()
print("Task finished")
カウントが1増やされるまで処理を待つことができます。
結果は次のようになります。
Wait task
Excute sleep
End sleep
Task finished
タスク終了時にクロージャを実行
タスク終了時に完了処理を実行する例です。[並列プログラミングガイド リスト 3-4]を参考にしています。
// @convention(c)はつけなくても良い
typealias CallBack = @convention(c) () -> Void
let closure: (DispatchQueue, @escaping CallBack) -> Void = { queue, callBack in
//3秒後にサブスレッドで実行
DispatchQueue.global().asyncAfter(deadline: .now() + .seconds(3)) {
queue.async(execute: callBack)
}
}
let queue = DispatchQueue.global()
closure(queue) {
print("excute 3 second after")
}
queue.async {
print("excute")
}
excute
excute 3 second after
Objective-Cの例ではcallBackに戻り値を与えていますが、Swift3のasyncの定義を見てみると、処理を実行するexecute引数の型がC言語のAPIのためのシンタックス(@convention(block) () -> Swift.Void)で定義されてますね。よって、callBackのためのクロージャに引数を与えたい場合は、自作で処理を書く必要があると思います。
public func async(group: DispatchGroup? = default, qos: DispatchQoS = default, flags: DispatchWorkItemFlags = default, execute work: @escaping @convention(block) () -> Swift.Void)
繰り返しタスクを実行
// 並列で実行
DispatchQueue.concurrentPerform(iterations: 5) { index in
print("index: \(index)")
}
index: 1
index: 3
index: 2
index: 0
index: 5
並列のキューで直列処理を行う
let queue = DispatchQueue(label: "com.GCD.barrier", attributes: .concurrent)
(0...100).forEach { index in
// 直列で実行
if 30...50 ~= index {
queue.async(flags: .barrier) {
print("excute index: \(index)")
}
} else {
// 並列で実行
queue.async {
print("excute index: \(index)")
}
}
}
0~100まで並列キューでループ処理を行っています。barrierのフラグを指定することで直列で処理を実行することができます。よって並列のキューにのみ有効です。上の処理は30~50の範囲は直列で処理が実行されることになります。
excute index: 4
excute index: 0
excute index: 6
...
excute index: 30
excute index: 31
excute index: 32
excute index: 33
excute index: 34
...
excute index: 75
excute index: 81
excute index: 77
キューの一時停止と再開
let queue = DispatchQueue(label: "com.GCD.privateQueue")
// 一時停止
queue.suspend()
var number = 30
(0...10).forEach { index in queue.async { print("number: \(number)") } }
// デバックをわかりやすくするためにsleep
sleep(1)
number = 10
print("change number to 10")
// 再開
queue.resume()
通常であれば、number: 30, number: 30..と出力された後に、change number to 10と主力されますが、suspend()とresume（）によってキューの処理を停止、再開させています。これにより、次のような出力になります。
change number to 10
number: 10
number: 10
number: 10
...
resume()の後でキューの処理が実行されるため、number: 10と繰り返し処理されます。
同時に実行される処理の制限
let semaphone = DispatchSemaphore(value: 2)
let queue = DispatchQueue.global()
(0...10).forEach { index in
queue.async {
semaphone.wait()
print("Excute sleep: \(index)")
sleep(2)
print("End sleep: \(index)")
semaphone.signal()
}
}
並列で同時に実行される処理を二つに制限しています。
一度だけ処理を実行する
Swift3からは、dispatch_onceが使えなくなったので代わりに以下の方法で実行することが推奨されています。
let myGlobal = { … global contains initialization in a call to a closure … }()
_ = myGlobal // using myGlobal will invoke the initialization code only
参考: Migrating to Swift 2.3 or Swift 3 from Swift 2.2
lazy var once: Void = { print("excute only once") }()
once
once
once
once
//結果
//excute only once
ディスパッチソースについて
ディスパッチソースについては、別記事でまとめていきたいと思います。
ディスパッチソースは、システム関係のイベント処理によく使われる、非同期コールバック関数の代
替となるものです。ディスパッチソースを生成した後、監視対象のイベント、ディスパッチキュー、
イベントを処理するコードを設定します。このコードは、ブロックオブジェクトまたは関数の形で定義
します。該当するイベントが届くと、ディスパッチソースはこのブロックまたは関数を所定のディス
パッチキューに登録し、実行を委ねます。
サンプルコードについて
上記のソースコードはshoheiyokoyama/ GCDにまとめてありますので、参考にしてください