お気楽Ｃ＃プログラミング超入門

●並列ループの例外処理

並列ループで実行されているスレッドで例外が送出された場合、並列ループで実行しているすべてのスレッドで実行中の処理が終了したあと、次の処理 (繰り返し) に進まないでスレッドを終了します。このとき、別のスレッドで例外がさらに送出される可能性があります。.NET (C#) には、複数の例外を捕捉するためのクラス AggregateException が用意されています。以下に構文を示します。

try { ... } catch (AggregateException ae) { ... }

捕捉した例外はプロパティ InnerExceptions で取得することができます。

ae.InnerExceptions => 例外を格納したコレクション

通常は、foreach などで InnerExceptions から例外を順番に取り出し、それを処理していくことになります。簡単な例を示しましょう。

リスト : 並列ループの例外処理

void test3() {
  try {
    Parallel.For(1, 10, i => {
      Console.WriteLine("start {0}", i);
      if (i == 5) {
        throw new Exception("oops!! 5");
      } else if (i == 7) {
        throw new Exception("oops!! 7");
      }
      Thread.Sleep(50);
      Console.WriteLine("end {0}", i);
    });
  }
  catch (AggregateException ae) {
    foreach (var ex in ae.InnerExceptions) {
      Console.WriteLine(ex.Message);
    }
  }
}

Parallel.For で並列に処理 (ラムダ式) を実行し、インデックス i が 5 または 7 の場合に例外 Exception を送出します。それを catch で捕捉します。実行結果は次のようになりました。

> test3()
start 1
start 3
start 5
start 7
start 9
start 2
end 9
end 3
end 1
end 2
oops!! 7
oops!! 5

インデックスが 1, 3, 5, 7, 9, 2 の処理が並列に実行され、5 と 7 で例外が送出されます。1, 2, 3, 9 の処理が終了したあと、次の処理に進まずに並列ループを終了します。そして、送出した例外を carch で捕捉し、foreach で順番に取り出して処理 (エラーメッセージの表示) します。

●並列ループのキャンセル

.NET で並列ループの処理をキャンセルしたい場合、メソッド Stop() のほかにも「キャンセルトークン (CancellationToken)」という方法が用意されています。以下に基本的な関数を示します。

1. Parallel.For(start, end, options, var => { ... })
2. Parallel.For(start, end, options, (var, state) => { ... })
3. Parallel.ForEach(collection, options, var => { ... })
4. Parallel.ForEach(collection, options, (var, state) => { ... })
5. Parallel.Invoke(options, params Action[] actions)

引数 options はクラス ParallelOptions のインスタンスです。ParallelOptions のコンストラクタとプロパティを示します。

ParallelOptions() => options, コンストラクタ
options.CancellationToken, キャンセルトークン (構造体 CancellationToken)
options.MaxDegreeOfParallelism, 同時実行タスクの最大値
options.TaskScheduler, タスクスケジューラ

キャンセルの検出は構造体 CancellationToken を、キャンセルの通知はクラス CancellationTokenSource を使います。主なプロパティとメソッドを以下に示します。

CancellationTokenSource:
CancellationTokenSource() => source
source.Token => キャンセルトークン (構造体 CancellationToken)
source.Cancel() => void

CancellationToken:
token.CanBeCanceled => bool, 
キャンセルできる場合は true, できない場合は false
token.IsCancellationRequested => bool, 
キャンセルが通知されると true, そうでなければ false
token.ThrowIfCancellationRequested(), 
キャンセルが通知されていれば例外 OperationCanceledException を送出する

キャンセルはメソッド Cancel() で通知します。これは Token にセットされているキャンセルトークンに通知されます。ParallelOptions() はデフォルトでキャンセルできない CancellationToken を生成します。また、キャンセルの通知を受け取るためにも CancellationTokenSource() で生成したキャンセルトークンに置き換えておく必要があります。

簡単な例を示しましょう。

リスト : 並列ループのキャンセル

void test4(ParallelOptions options) {
  try {
    Parallel.For(1, 10, options, i => {
      Console.WriteLine("start {0}", i);
      Thread.Sleep(50);
      if (options.CancellationToken.IsCancellationRequested) {
        Console.WriteLine("cancel {0}", i);
        options.CancellationToken.ThrowIfCancellationRequested();
      }
      Console.WriteLine("end {0}", i);
    });
  }
  catch (AggregateException ae) {
    foreach (var ex in ae.InnerExceptions) {
      Console.WriteLine(ex.Message);
    }
  }
  catch (OperationCanceledException e) {
    Console.WriteLine(e.Message);
  }
}

void test5() {
  var source = new CancellationTokenSource();
  var options = new ParallelOptions(){ CancellationToken = source.Token };
  Console.WriteLine("main start");
  var a = Task.Run(() => test4(options));
  var b = Task.Run(() => {
    Thread.Sleep(50);
    Console.WriteLine("Cancel!!");
    source.Cancel();
  });
  Task.WaitAll(a, b);
  Console.WriteLine("main end");
}

関数 test4 は Parallel.For で処理を並列に実行します。キャンセルの通知を確認したら、メッセージを表示して例外を送出します。Prallel.For のあと catch で例外を捕捉します。Parallel.For の引数で options を指定すると、例外 OperationCanceledException を送出するのは 1 回だけになります。AggregateException では捕捉できないので注意してください。

関数 test5 では、CancellationTokenSource() で source を生成し、ParallelOptions() で options を生成します。このとき、キャンセルトークンを source.Token に設定します。次に、test4 を Task.Run で実行します。次に、50 msec 後に Cancel() を実行するタスクを Task.Run で実行します。あとは、2 つのタスクが終了するのを WaitAll で待つだけです。

実行結果は次のようになりました。

> test5()
main start
start 1
start 3
start 5
start 7
end 3
Cancel!!
end 7
start 8
start 4
end 1
end 5
cancel 8
cancel 4
The operation was canceled.
main end

インデックス 1, 3, 5, 7 の処理が並列に実行されます。そのあと、4, 8 の処理が開始されますが、タスク b で Cancel() を実行したことにより 4 と 8 の処理がキャンセルされ、まだ開始していない処理もキャンセルされていることがわかります。また、タスク a の最後で例外を捕捉していますが、送出される例外は 1 回だけであることがわかります。

●排他制御

最初に説明しましたが、スレッドは同じプロセス内に存在するので、メモリ空間を共有することができます。これを共有メモリといいます。スレッド間の通信は共有メモリを使って簡単に行うことができますが、共有メモリのアクセス時に発生する「競合」が問題になります。このため、あるスレッドがデータにアクセスしている間、他のスレッドからアクセスできないように制限します。これを「排他制御」といいます。

具体的には、鍵を担当するオブジェクト (同期オブジェクト) を用意します。データにアクセスする場合、最初にオブジェクトに鍵をかけます。この操作を「ロック (lock)」といいます。逆に、鍵を外す操作を「アンロック (unlock)」といいます。

オブジェクトがすでにロックされている場合、それがアンロックされるまでスレッドは待機します。オブジェクトをロックしたスレッドは、データにアクセスしたあと、オブジェクトをアンロックします。そのあと、待機していたスレッドがオブジェクトをロックして、データにアクセスすることになります。

C# には排他制御を行うためのクラスがいくつか用意されていますが、一番簡単な方法は lock 文を使うことです。lock 文の構文を示します。

lock (lock_obj) {
  // クリティカルセクション
  処理;
  ...;
}

lock_obj は同期オブジェクトです。データ型は参照型でなければいけません。C# のリファレンス「lock ステートメント」では Object のインスタンスを使っています。オブジェクトをロックしたあと、ブロック { ... } の処理を実行します。これを「クリティカルセクション」といいます。ここには複数のスレッドが重なり合ってはいけない処理を記述します。ブロックの実行が終了すると、ロックは自動的に解除されます。

簡単な例を示しましょう。次のリストを見てください。

リスト : データの競合 (排他制御なし)

using System.Threading;
using System.Threading.Tasks;

int test_bad(int n) {
  int cnt = 0;
  Parallel.For(0, n, _ => cnt += 1);
  return cnt;
}

関数 test_bad はループの回数を cnt でカウントします。逐次処理で行うと、当然ですが cnt の値は引数 n と同じになりますが、Parallel.For で並列に行うと、cnt のアクセスで競合が発生するため、cnt の値は n と等しくなりません。実際に試してみましょう。

> test_bad(1000000)
818046
> test_bad(1000000)
968616
> test_bad(1000000)
958903

cnt += 1 は 1 つの文ですが、低レベルの操作では次の 3 つの処理に分けられます。

1. cnt の値を読み込む
2. その値に 1 を加算する
3. 結果を cnt に書き込む

処理が分かれていると、途中で他のスレッドが実行されることがあるのです。たとえば、スレッド A が cnt の値を読み込み、その値が 1000 だとしましょう。そして、A が cnt の値を書き換える前にスレッド B が実行されたとします。すると、スレッド B が読み込む cnt の値は 1000 のままです。これではループの回数を正確に数えることはできませんね。

ちなみに、1, 2, 3 のような低レベルの操作は、その途中で他のスレッドが実行されることはなく、処理は最後まで実行されます。このような操作を「不可分操作」とか「アトミック操作 (atomic operation)」といいます。インクリメントやデクリメントのような操作は、1 命令で実行できる CPU もあるので、その命令を使えば不可分操作になります。

test_bad では、cnt +=1 で競合が発生しているので、lock 文で排他制御すると正常に動作します。次のリストを見てください。

リスト : lock 文による排他制御

int test1(int n) {
  int cnt = 0;
  var lc = new Object();
  Parallel.For(0, n, _ => {
    lock (lc) { cnt += 1; };
  });
  return cnt;
}

変数 lc に Object のインスタンスをセットします。これが lock 文で使用する同期オブジェクトになります。Parallel.For のラムダ式で、cnt += 1 を lock 文で包みます。これで cnt += 1 を実行している間、他のスレッドが重なり合うことはありません。

それでは実際に試してみましょう。

> test1(1000000)
1000000
> test1(1000000)
1000000

正常に動作していますね。ところで、lock 文による排他制御はけっこう重たい処理なので、今回のように変数の値を書き換えるだけであれば、クラス System.Threading.Interlocked の関数を使うことができます。整数の加算、インクリメント、デクリメントは以下の関数で行うことができます。

Interlocked.Add(ref var, value) => 結果
Interlocked.Increment(ref var) => 結果
Interlocked.Decrement(ref var) => 結果

関数 Add は var += value を不可分操作として実行し、その結果を返します。Increment は var += 1 を、Decrement は var -= 1 を不可分操作として実行し、その結果を返します。test1 を Increment を使って書き直すと次のようになります。

リスト : クラス Interlocked による不可分操作

int test2(int n) {
  int cnt = 0;
  Parallel.For(0, n, _ => {
    Interlocked.Increment(ref cnt);
  });
  return cnt;
}

> test2(1000000)
1000000
> test2(1000000)
1000000

lock 文よりも Increment のほうが処理が軽いので、実行時間はこちらのほうが速くなります。このほかにも Interlocked には便利な関数が用意されています。詳細は C# リファレンス「Interlocked クラス」をお読みください。

●哲学者の食事

今回は「哲学者の食事」という並行プログラミングでは有名な問題を解いてみましょう。

[哲学者の食事]

5 人の哲学者が丸いテーブルに座っています．テーブルの中央にはスパゲッティが盛られた大皿があり、哲学者の間には 5 本のフォークが置かれています。哲学者は思索することとスパゲッティを食べることを繰り返します。食事のときには 2 本のフォークを持たなければなりません。食事が終わると 2 本のフォークを元の位置に戻します。

詳しい説明は食事する哲学者の問題 -- Wikipedia をお読みください。

●フォークと操作関数

それではプログラムを作りましょう。最初にフォークを管理する関数を定義します。

リスト : フォークと操作関数

// フォーク
bool[] forks;
// 同期オブジェクト
Object lockobj;

// フォークを取る
async Task getFork(int n) {
  bool fork = false;
  while (!fork) {
    lock (lockobj) {
      if (forks[n]) {
        forks[n] = false;
        fork = true;
      }
    }
    await Task.Delay(100);
  }
}

// フォークを置く
async Task putFork(int n) {
  lock (lockobj) {
    forks[n] = true;
  }
  await Task.Delay(100);
}

フォークは bool 型の配列 forks で表します。フォークが置いてある状態を true で、使用中を false で表します。変数 lockobj は同期オブジェクトとして使います。これらの変数はタスクを実行する前に初期化します。

関数 getFork は n 番目のフォークを取得します。lock 文で lockobj をロックしてから配列 forks にアクセスします。フォークが使用中の場合は 100 msec 待ってから再度取得を試みます。フォークを取得するまで無限ループになることに注意してください。関数 putFork は n 番目のフォークを元に戻します。lockobj をロックしてから forks[n] を false に書き換えるだけです。

●哲学者の動作

次は哲学者の動作をプログラムします。

リスト : 哲学者の動作

async Task personAsync0(int n) {
  for (int i = 0; i < 2; i++) {
    Console.WriteLine("Philosopher {0} is thinking", n);
    await Task.Delay(1000);
    await getFork(n);           // 右のフォーク
    await getFork((n + 1) % 5); // 左のフォーク
    Console.WriteLine("Philosopher {0} is eating", n);
    await Task.Delay(100);
    await putFork(n);
    await putFork((n + 1) % 5);
  }
  Console.WriteLine("Philosopher {0} is sleeping", n);
}

関数 personAsync0 の引数 n は哲学者の番号を表します。哲学者が食事をする場合、最初に getFork で右側のフォーク (n 番目) を取り、次に左側のフォーク ((n + 1) % 5 番目) を取ります。哲学者は円形に並んでいるので、5 人目の左側のフォークが 1 人目の右側のフォークになります。食事を終えたら putFork で右側のフォークを返却し、次に左側のフォークを返却します。

このように、タスクを使うと哲学者の動作を簡単にプログラムできますが、実は並行プログラミング特有の大きな問題点があるのです。これはプログラムを実行してみるとわかります。

●実行結果 (1)

プログラムの実行は関数 test0 で行います。

リスト : 実行

void test0() {
  forks = new bool[] { true, true, true, true, true };
  lockobj = new Object();
  var p0 = personAsync0(0);
  var p1 = personAsync0(1);
  var p2 = personAsync0(2);
  var p3 = personAsync0(3);
  var p4 = personAsync0(4);
  Task.WaitAll(p0, p1, p2, p3, p4);
}

test0 を実行する前に、forks と lockobj を初期化します。あとは 5 人の哲学者を表すタスク (personAsync0) を実行し、Task.WaitAll でタスクがすべて終了するのを待ちます。実行結果は次のようになります。

> test0()
Philosopher 0 is thinking
Philosopher 1 is thinking
Philosopher 2 is thinking
Philosopher 3 is thinking
Philosopher 4 is thinking

このように、すべてのタスクが待ち状態となり先へ進むことができなくなります。これを「デッドロック (deadlock)」といいます。哲学者全員が右側のフォークを取り、左側のフォークが置かれるのを待つときにデッドロックとなるわけです。

●デッドロックの防止

デッドロックを防止する簡単な方法は、右側のフォークを取っても左側のフォークを取れないときは、右側のフォークを元に戻すことです。プログラムは次のようになります。

リスト : デッドロックの防止 (1)

// 左側のフォークを取る
async Task<bool> getForkL(int n) {
  bool fork = false;
  lock (lockobj) {
    if (forks[n]) {
      forks[n] = false;
      fork = true;
    }
  }
  await Task.Delay(100);
  return fork;
}

async Task personAsync1(int n) {
  int i = 0;
  while (i < 2) {
    Console.WriteLine("Philosopher {0} is thinking", n);
    await Task.Delay(1000);
    await getFork(n);
    if (await getForkL((n + 1) % 5)) {
      Console.WriteLine("Philosopher {0} is eating", n);
      await Task.Delay(100);
      await putFork(n);
      await putFork((n + 1) % 5);
      i++;
    } else {
      await putFork(n);
    }
  }
  Console.WriteLine("Philosopher {0} is sleeping", n);
}

右側のフォークを取ったあと、関数 getForkL で左側のフォークを要求します。フォークを受け取った場合は true を返すので、食事をすることができます。false の場合は右側のフォークを返却して思索に戻ります。

Lua のようなノンプリエンプティブなコールチンの場合、これでデッドロックを解消して正常に動作するのですが、プリエンプティブなタスク (スレッド) では新たな問題が発生します。

●実行結果 (2)

実行結果は次のようになります。

リスト : 実行 (2)

void test1() {
  forks = new bool[] { true, true, true, true, true };
  lockobj = new Object();
  var p0 = personAsync1(0);
  var p1 = personAsync1(1);
  var p2 = personAsync1(2);
  var p3 = personAsync1(3);
  var p4 = personAsync1(4);
  Task.WaitAll(p0, p1, p2, p3, p4);
}

> test1()
Philosopher 0 is thinking
Philosopher 1 is thinking
Philosopher 2 is thinking
Philosopher 3 is thinking
Philosopher 4 is thinking
Philosopher 1 is thinking
Philosopher 0 is thinking
Philosopher 2 is thinking
Philosopher 3 is thinking
Philosopher 4 is thinking
Philosopher 0 is thinking
Philosopher 3 is thinking
Philosopher 1 is thinking
Philosopher 2 is thinking
Philosopher 4 is thinking
Philosopher 3 is thinking
Philosopher 0 is thinking
Philosopher 1 is thinking
Philosopher 2 is thinking
Philosopher 4 is thinking

・・・ CTRL-C で中断 ・・・

哲学者全員が右側のフォークを受け取っては返却することを繰り返すため、次の状態へ進むことができません。デッドロックではありませんが、無限ループに陥っているわけです。このような状態を「ライブロック (livelock)」といいます。

●ライブロックの防止

哲学者の食事問題の場合、ライブロックを解消する簡単な方法があります。フォークが残り 1 本の場合、右側のフォークを要求されたらそれを待たせることにするのです。左側のフォークであれば、その要求を受け付けます。4 人の哲学者が右側のフォークを持ったとき、5 人目の哲学者は右側のフォークを持つことができません。次に、4 人のうちの誰かが左側のフォークを要求し、それが受け付けられるので、最低でもひとりの哲学者が食事をすることができます。

プログラムは次のようになります。

リスト : ライブロックの解消

// フォークの本数
int countFork() {
  int c = 0;
  for (int i = 0; i < 5; i++) {
    if (forks[i]) c++;
  }
  return c;
}

// 右側のフォークを取る
async Task getForkR(int n) {
  bool fork = false;
  while (!fork) {
    lock (lockobj) {
      if (forks[n] && countFork() > 1) {
        forks[n] = false;
        fork = true;
      }
    }
    await Task.Delay(100);
  }
}

async Task personAsync1a(int n) {
  int i = 0;
  while (i < 2) {
    Console.WriteLine("Philosopher {0} is thinking", n);
    await Task.Delay(1000);
    await getForkR(n);
    if (await getForkL((n + 1) % 5)) {
      Console.WriteLine("Philosopher {0} is eating", n);
      await Task.Delay(100);
      await putFork(n);
      await putFork((n + 1) % 5);
      i++;
    } else {
      await putFork(n);
    }
  }
  Console.WriteLine("Philosopher {0} is sleeping", n);
}

関数 getForkR は右側のフォークを取得します。基本的には getFork と同じですが、関数 countFork でフォークの残数を数え、それが 1 よりも多ければフォークを渡します。personAsync1a は personAsync1 の getFork を getForkR に変更するだけです。

●実行結果 (3)

それでは実行してみましょう。

リスト : 実行 (3)

void test1a() {
  forks = new bool[] { true, true, true, true, true };
  lockobj = new Object();
  var p0 = personAsync1a(0);
  var p1 = personAsync1a(1);
  var p2 = personAsync1a(2);
  var p3 = personAsync1a(3);
  var p4 = personAsync1a(4);
  Task.WaitAll(p0, p1, p2, p3, p4);
}

> test1a()
Philosopher 0 is thinking
Philosopher 1 is thinking
Philosopher 2 is thinking
Philosopher 3 is thinking
Philosopher 4 is thinking
Philosopher 4 is eating
Philosopher 2 is thinking
Philosopher 3 is thinking
Philosopher 1 is thinking
Philosopher 4 is thinking
Philosopher 0 is eating
Philosopher 0 is thinking
Philosopher 3 is eating
Philosopher 1 is thinking
Philosopher 2 is thinking
Philosopher 3 is thinking
Philosopher 4 is eating
Philosopher 4 is sleeping
Philosopher 0 is eating
Philosopher 0 is sleeping
Philosopher 2 is eating
Philosopher 1 is thinking
Philosopher 2 is thinking
Philosopher 3 is eating
Philosopher 3 is sleeping
Philosopher 1 is eating
Philosopher 1 is thinking
Philosopher 2 is eating
Philosopher 2 is sleeping
Philosopher 1 is eating
Philosopher 1 is sleeping

どの哲学者も 2 回食事をして睡眠まで到達しています。

●デッドロックの防止 (2)

もうひとつ簡単な方法を紹介しましょう。奇数番目の哲学者は、まず左側のフォークを取り上げてから右側のフォークを取り、偶数番目の哲学者は、今までのように右側のフォークを取り上げてから左側のフォークを取ります。こんな簡単な方法で動作するのは不思議なように思います。たとえば、哲学者が 2 人の場合を考えてみてください。

哲学者 0 の右側のフォークを A、左側のフォークを B とします。哲学者 1 からみると、B が右側のフォークで、A が左側のフォークになります。デッドロックは、哲学者 0 が A を取り、哲学者 1 が B を取ったときに発生します。ここで、哲学者 1 が左側のフォーク A から取るようにします。先に哲学者 0 が A を取った場合、哲学者 1 は A があくまで待つことになるので、哲学者 0 はフォーク B を取って食事をすることができます。哲学者 1 が先にフォーク A を取った場合も同じです。これでデッドロックを防止することができます。