c# - C# での並列 A* 検索 - スレッド同期

Question

2 つの異なる関数を使用する 2 つのスレッドがあります。1 つ目は最初から最後まで検索し、2 つ目は最後から最後まで検索します。

現在Thread.Sleep(10)同期に使用していますが、時間がかかりすぎて、そのような状態ではテストできません。

2 つのスレッドを異なる機能で同期するにはどうすればよいですか?

score 5 · Accepted Answer

それはあなたが何をしたいかによって少し異なります。

2 つのスレッドがあり、もう一方が「成功」に達したときに一方を終了したい場合 (またはn 個のスレッドがあり、一方が最初に「成功」に達したときにすべてを終了したい場合) は、各スレッドの成功を定期的にチェックするだけです。
- ロックやその他のメカニズムを使用Interlockedせずにこれを行うために使用します (以下を参照)。
- キャンセル可能Taskなオブジェクトを使用する
各スレッドが何かを実行し、他のスレッドが追いつくのを待つ段階で検索を行う必要がある場合は、別のアプローチが必要です。
- 使用するBarrier

A* 検索を行っていることを考えると、とにかく 2 つまたは 3 つすべての組み合わせが必要になる可能性があります。

Barrierステップを調整し、ステップ間のオープンセットを更新する
別のスレッドが成功した場合にいつスレッドを終了するかを解決するための成功シグナル
TaskオブジェクトをCancellationToken使用して、呼び出し元が検索をキャンセルできるようにします。

別の回答が提案されましたSemaphore-これはあなたのニーズにはあまり適していません（以下のコメントを参照）。

Barrier次の方法で、このような検索に使用できます。

アルゴリズムのステップ 0 に入る
n 個のスレッドが現在のレベルを等分し、半分ずつ処理します。それぞれが完了すると、信号を送って他のスレッドを待ちます
すべてのスレッドの準備ができたら、次のステップに進み、検索を繰り返します

出口の簡単なチェック -`Interlocked`

最初の部分は、成功を確認することです。「ロックレス」のままにしたい場合は、Interlockedこれを行うために使用できます。一般的なパターンは次のとおりです。

// global success indicator
private const int NotDone = 0;
private const int AllDone = 1;
private int _allDone = NotDone;

private GeneralSearchFunction(bool directionForward) {
  bool iFoundIt = false;
  ... do some search operations that won't take much time
  if (iFoundIt) {
    // set _allDone to AllDone!
    Interlocked.Exchange(ref _allDone, AllDone);
    return;
  }
  ... do more work
  // after one or a few iterations, if this thread is still going
  //   see if another thread has set _allDone to AllDone
  if (Interlocked.CompareExchange(ref _allDone, NotDone, NotDone) == AllDone) {
    return; // if they did, then exit
  }
  ... loop to the top and carry on working
}

// main thread:
  Thread t1 = new Thread(() => GeneralSearchFunction(true));
  Thread t2 = new Thread(() => GeneralSearchFunction(false));
  t1.Start(); t2.Start(); // start both
  t1.Join(); t2.Join(); 
  // when this gets to here, one of them will have succeeded

これは、あらゆる種類の成功トークンまたはキャンセルトークンの一般的なパターンです。

いくつかの作業を行います
成功した場合は、他のすべてのスレッドが定期的にチェックするシグナルを設定します
まだ成功していない場合は、その作業の途中で、すべての反復または数回の反復ごとに、このスレッドが終了するかどうかを確認してください

したがって、実装は次のようになります。

class Program
{
    // global success indicator
    private const int NotDone = 0;
    private const int AllDone = 1;
    private static int _allDone = NotDone;

    private static int _forwardsCount = 0;    // counters to simulate a "find"
    private static int _backwardsCount = 0;   // counters to simulate a "find"
    static void Main(string[] args) {
        var searchItem = "foo";
        Thread t1 = new Thread(() => DoSearchWithBarrier(SearchForwards, searchItem));
        Thread t2 = new Thread(() => DoSearchWithBarrier(SearchBackwards, searchItem));
        t1.Start(); t2.Start();
        t1.Join(); t2.Join();
        Console.WriteLine("all done");
    }
    private static void DoSearchWithBarrier(Func<string, bool> searchMethod, string searchItem) {
        while (!searchMethod(searchItem)) {
            // after one or a few iterations, if this thread is still going
            //   see if another thread has set _allDone to AllDone
            if (Interlocked.CompareExchange(ref _allDone, NotDone, NotDone) == AllDone) {
                return; // if they did, then exit
            }
        }
        Interlocked.Exchange(ref _allDone, AllDone);
    }
    public static bool SearchForwards(string item) {
        //  return true if we "found it", false if not
        return (Interlocked.Increment(ref _forwardsCount) == 10);
    }
    public static bool SearchBackwards(string item) {
        //  return true if we "found it", false if not
        return (Interlocked.Increment(ref _backwardsCount) == 20); // make this less than 10 to find it backwards first
    }
}

同じ目的でタスクを使用する

もちろん、これを使用しないと .NET 4.5 にはなりませんTask。

class Program
{
    private static int _forwardsCount = 0;    // counters to simulate a "find"
    private static int _backwardsCount = 0;   // counters to simulate a "find"
    static void Main(string[] args) {
        var searchItem = "foo";
        var tokenSource = new CancellationTokenSource();
        var allDone = tokenSource.Token;
        Task t1 = Task.Factory.StartNew(() => DoSearchWithBarrier(SearchForwards, searchItem, tokenSource, allDone), allDone);
        Task t2 = Task.Factory.StartNew(() => DoSearchWithBarrier(SearchBackwards, searchItem, tokenSource, allDone), allDone);
        Task.WaitAll(new[] {t2, t2});
        Console.WriteLine("all done");
    }
    private static void DoSearchWithBarrier(Func<string, bool> searchMethod, string searchItem, CancellationTokenSource tokenSource, CancellationToken allDone) {
        while (!searchMethod(searchItem)) {
            if (allDone.IsCancellationRequested) {
                return;
            }
        }
        tokenSource.Cancel();
    }
    ...
}

ただし、CancellationToken間違ったことにを使用してしまいました。実際には、これは検索の呼び出し元が検索をキャンセルするために保持する必要があるCancellationTokenため、要求されたキャンセル (呼び出し元のみが必要tokenSourceとする) を確認するためにを使用し、別の成功を確認する必要があります。同期 (Interlocked上記のサンプルなど) を終了します。

位相/ステップ同期

これは多くの理由で難しくなりますが、簡単な方法があります。(.NET 4 の新機能) を終了シグナルと組み合わせて使用Barrierすると、次のことができます。

現在のステップで割り当てられたスレッドの作業を実行し、別の反復を行う前に他のスレッドが追いつくのを待ちます
どちらかが成功したら両方のスレッドを終了する

達成したい内容に応じて、スレッド同期にはさまざまなアプローチがあります。いくつかは次のとおりです。

Barrier: 前方検索と後方検索の両方を同時に実行する場合は、おそらくこれが最も適しています。また、「すべてのスレッドがバリアに到達するまで、すべてのスレッドを続行できません」という意図を叫びます。
ManualResetEvent- 1 つのスレッドがシグナルを解放すると、他のすべてのスレッドはシグナルが再設定されるまで続行できます。AutoResetEventは似ていますが、再度ブロックする前に 1 つのスレッドのみを続行できる点が異なります。
Interlocked- これと組み合わせることSpinWaitで、実行可能なロックレスソリューションとなります
Semaphore- 使用可能ですが、シナリオにはあまり適していません

Barrierあなたの場合に最も適していると思われるため、ここでは完全なサンプルのみを提供しました。( ref. albahariBarrier )に次ぐ最もパフォーマンスの高いものの 1 つですが、使用するにはより複雑なコードが必要です。ManualResetEventSlimManualResetEvent

上記のいずれも機能しない場合(現在はロックを使用しています)、およびタスクの継続Monitor.Waitを確認するための他の手法を参照してください。Monitor.Pulse後者は、ある非同期操作から別の非同期操作にデータを渡すために使用されますが、シナリオには使用できます。また、回答の上部にあるサンプルと同様に、一方を他方の代わりに使用Taskするよりも、組み合わせる可能性が高くなります。Barrierタスクの継続は、A* 検索でオープンセットのステップ後のリビジョンを実行するために使用できますBarrierが、とにかく簡単に使用できます。

このコードは、Barrier作品を使用しています。本質的にDoSearchWithBarrierは、同期を行う唯一のビットであり、残りはすべてセットアップと破棄のコードです。

class Program {
    ...
    private static int _forwardsCount = 0;    // counters to simulate a "find"
    private static int _backwardsCount = 0;   // counters to simulate a "find"
    static void Main(string[] args) {
        Barrier barrier = new Barrier(numThreads, 
            b => Console.WriteLine("Completed search iteration {0}", b.CurrentPhaseNumber));
        var searchItem = "foo";
        Thread t1 = new Thread(() => DoSearchWithBarrier(SearchForwards, searchItem, barrier));
        Thread t2 = new Thread(() => DoSearchWithBarrier(SearchBackwards, searchItem, barrier));
        t1.Start(); Console.WriteLine("Started t1");
        t2.Start(); Console.WriteLine("Started t2");
        t1.Join(); Console.WriteLine("t1 done");
        t2.Join(); Console.WriteLine("t2 done");
        Console.WriteLine("all done");
    }
    private static void DoSearchWithBarrier(Func<string, bool> searchMethod, string searchItem, Barrier barrier) {
        while (!searchMethod(searchItem)) {
            // while we haven't found it, wait for the other thread to catch up
            barrier.SignalAndWait(); // check for the other thread AFTER the barrier
            if (Interlocked.CompareExchange(ref _allDone, NotDone, NotDone) == AllDone) {
                return;
            }
        }
        // set success signal on this thread BEFORE the barrier
        Interlocked.Exchange(ref _allDone, AllDone);
        // wait for the other thread, and then exit (and it will too)
        barrier.SignalAndWait();
    }
    ...
}

ここでは次の 2 つのことが行われています。

Barrier2 つのスレッドを同期するために使用されるため、他のスレッドが追いつくまで次のステップを実行できません。
Interlocked最初に説明したように、終了シグナルはを使用します。

これを A* 検索に実装することは、上記のサンプルと非常によく似ています。すべてのスレッドがバリアに到達して続行すると、ManualResetEvent またはシンプルlockを使用して、1 つ (および 1 つだけ) にオープンセットを変更させることができます。

注意事項`Semaphore`

リソースのプールが限られており、リソースを所有しているよりも多くのリソースユーザーがアクセスを必要としている場合に最もよく使用されるため、これはおそらくあなたが望むものではありません。

社員食堂の隅にある CoD を搭載した PlayStation を考えてみてください。コントローラーが 4 つ、20 人WaitOneが使用を待っReleaseています。特定の FIFO/LIFO 順序付けが強制されることはなく、実際Release、避けられない戦いを防ぐために使用する用心棒によって呼び出すことができます (つまり、スレッド ID は強制されません)。

終了の簡単なチェック - その他のアプローチ

`lock`単純な成功表示のための使用

ロックでも同じことができます。両方ともInterlocked、lockスレッド間の共通変数の読み取りでメモリキャッシュの問題が発生しないようにします。

private readonly object _syncAllDone = new object();
...
  if (iFoundIt) {
    lock (_syncAllDone) { _allDone = AllDone };
    return;
  }
  ...
  //   see if another thread has set _allDone to AllDone
  lock (_syncAllDone) {
    if (_allDone == AllDone) {
      return; // if they did, then exit
    }
  }

これの欠点は、ロックが遅くなる可能性があることですが、状況をテストする必要があります。利点は、ロックを使用してスレッドから結果を書き出すなどの他のことを行う場合、余分なオーバーヘッドがないことです。

`ManualResetEvent`単純な成功表示のための使用

これは実際にはリセットイベントの使用目的ではありませんが、機能する可能性があります。(.NET 4 以降を使用している場合は、ManualResetEventSlim代わりに使用しますManualResetEvent):

private ManualResetEvent _mreAllDone = new ManualResetEvent(true); // will not block a thread
...
  if (iFoundIt) {
    _mreAllDone.Reset(); // stop other threads proceeding
    return;
  }
  ...
  //   see if another thread has reset _mreAllDone by testing with a 0 timeout
  if (!_mreAllDone.WaitOne(0)) {
      return; // if they did, then exit
  }

位相同期 - その他のアプローチ

他のアプローチはすべて、競合状態や永久にブロックされたスレッドを防ぐために双方向の継続チェックを行う必要があるため、はるかに複雑になります。私はそれらをお勧めしませんので、ここではサンプルを提供しません (長くて複雑です)。

参考文献:

score 1 · Accepted Answer

thread.Join()

おそらくあなたの後です。これにより、他のスレッドが終了するまで現在のスレッドがブロックされます。

すべてのスレッドを 1 点に同期することで、そこで複数のスレッドを結合することができます。

List<Thread> threads = new List<Thread>();
threads.Add(new Thread(new ThreadStart(<Actual method here>)));
threads.Add(new Thread(new ThreadStart(<Another method here>)));
threads.Add(new Thread(new ThreadStart(<Another method here>)));

foreach(Thread thread in threads)
{
  thread.Start();
}
//All your threads are now running
foreach(Thread thread in threads)
{
  thread.Join();
}
//You wont get here until all those threads have finished

c# - C# での並列 A* 検索 - スレッド同期

3 に答える 3

出口の簡単なチェック -Interlocked

同じ目的でタスクを使用する

位相/ステップ同期

注意事項Semaphore

終了の簡単なチェック - その他のアプローチ

lock単純な成功表示のための使用

ManualResetEvent単純な成功表示のための使用

位相同期 - その他のアプローチ

Related

Reference

出口の簡単なチェック -`Interlocked`

注意事項`Semaphore`

`lock`単純な成功表示のための使用

`ManualResetEvent`単純な成功表示のための使用