System.Threading.Semaphore

セマフォとは

セマフォ(Semaphore)とは排他制御を行う上で用いられる概念のひとつ。　セマフォは、セマフォ(=排他区間)へ入ることを要求するP操作、セマフォから出ることを通知するV操作、現在セマフォ内にある処理を計上するカウンタの三つの要素から構成される。

セマフォの動作は次のとおり。

カウンタの初期値nのセマフォは、同時にn個までの処理がセマフォに入ることを許可する
P操作によりカウンタをデクリメントし、V操作によりカウンタをインクリメントする
P操作によりカウンタが0になった場合は、以降V操作が行われない限り、処理がセマフォに入ることを許可しない(処理がブロックされる)

セマフォを使用することで排他区間(ある処理を実行している間は他の処理の実行を制限する区間)を定義できる。　セマフォによって、定義した排他区間に対して同時に行うことができる処理の数を制限することができる。

ちなみに、カウンタの初期値が1の場合は、同時に1個の処理しかセマフォに入ることを許可しないため、この場合ミューテックス(Mutex)と同義となる

.NET FrameworkにおけるSemaphore

名前付きでSemaphoreインスタンスを生成すると、オペレーティングシステム全体から参照できる(プロセス間で共有できる)システムセマフォとなる。　この場合、複数のプロセスをまたがってひとつの排他区間を定義することになり、複数のプロセスが同時に行うことができる処理の数を制限することができる。

名前なしでSemaphoreインスタンスを生成すると、プロセス内でのみ参照できるローカルセマフォとなる。　この場合、Semaphoreインスタンスを生成したプロセスのみが参照できる排他区間を定義することになり、プロセス内で同時に行うことができる処理の数を制限することができる。

セマフォに入る(WaitOne)ときブロックされる処理(カウンタが0になっていてセマフォに入れず待たされる処理)は、セマフォから出て(Release)ほかの処理がセマフォに入れるようになっても、その実行順序は保障されない。　つまり、最初にブロックされた処理が最初に入れる、つまりFIFOやLIFOのような保障された順序にはならない為、状況によっては最初にブロックされた処理がいつまでも待たされることもあり得る。

主なコンストラクタ

Semaphore(int, int): カウンタの初期値と最大値を指定してローカルセマフォのインスタンスを生成する。
Semaphore(int, int, string, out bool): カウンタの初期値と最大値、セマフォの名前を指定してシステムセマフォのインスタンスを生成する。　最後のoutパラメータによって、生成したインスタンスが新しく作成したシステムセマフォなのか、それとも既に作成されているシステムセマフォなのか判断できる。

主なメソッド

bool WaitOne(): セマフォのカウンタをデクリメントする(P操作)。　カウンタが0になった場合は、Release()が呼ばれる(V操作が行われる)まで処理がブロックされる。
int Release(): セマフォのカウンタをインクリメントする(V操作)。　インクリメントする前のカウンタの値が戻り値として返される。
void Close: インスタンスが保持しているリソースを解放する。

主な呼び出し順序

new Semaphore()
WaitOne()
Release()
Close()

使用例(ローカルセマフォ)

ローカルセマフォを生成し、排他区間に対して複数のスレッドから同時にアクセスを試みる。

行番号を表示する

using System;
using System.Threading;

class Sample {
  // リソースへのアクセス制御
  private Semaphore semaphore = null;

  // ワーカースレッドのインスタンス格納用の配列
  private Thread[] threads = new Thread[5];

  // 各ワーカースレッドがアクセスする共有リソース
  private string[] resource = new string[] { "sem", "aph", "ore" };

  private Random random = null;

  //
  // エントリーポイント
  //
  static void Main()
  {
    Console.WriteLine( "semaphore sample start" );

    ( new LocalSemaphoreSample() ).Run();
  }

  //
  // 主スレッド
  //
  public void Run()
  {
    random = new Random();

    // 初期値3、最大値3のセマフォ・インスタンスを作成
    semaphore = new Semaphore( 3, 3 );

    // 共有リソースにアクセスするワーカースレッドを5本作成
    for ( int index = 0; index < threads.Length; index++ ) {
      threads[index] = new Thread( new ParameterizedThreadStart( RunWorkerThread ) );
    }

    // 各ワーカースレッドを50ミリ秒おきに起動
    for ( int index = 0; index < threads.Length; index++ ) {
      // ワーカースレッドに渡すパラメータとして、
      // スレッドが参照するリソースのインデックスを指定する
      threads[index].Start( index % resource.Length );

      Thread.Sleep( 50 );
    }

    // 主スレッドは3秒間待機する
    Thread.Sleep( 3000 );

    // 全ワーカースレッドを中断
    for ( int index = 0; index < threads.Length; index++ ) {
      threads[index].Abort();
    }

    // セマフォを開放
    semaphore.Close();

    // プロセス終了
    return;
  }

  //
  // ワーカースレッド
  //
  private void RunWorkerThread( object param )
  {
    Console.WriteLine( "ThreadID: {0:X8}, started", Thread.CurrentThread.ManagedThreadId );

    try {
      int index = (int)param;

      // 250〜1000ミリ秒おきにリソースにアクセスする
      for ( ;; ) {
        Thread.Sleep( random.Next( 250, 1000 ) );

        GetResource( index );
      }
    }
    catch ( ThreadAbortException ) {
      // スレッドの中断が要求されたら、その時点でスレッドの処理を終了する
      return;
    }
    finally {
      Console.WriteLine( "ThreadID: {0:X8}, finished", Thread.CurrentThread.ManagedThreadId );
    }
  }

  //
  // リソースへアクセスするメソッド
  //
  private void GetResource( int index )
  {
    Console.WriteLine( "ThreadID: {0:X8}, wait for  resource #{1}",
                Thread.CurrentThread.ManagedThreadId, index );

    // セマフォに入る(P操作、セマフォのカウンタをデクリメントする)
    semaphore.WaitOne();

    // リソースへアクセス
    Console.WriteLine( "ThreadID: {0:X8}, access to resource #{1} '{2}'",
                Thread.CurrentThread.ManagedThreadId, index, resource[index] );

    // リソースへのアクセス時に250ミリ秒遅延させる
    Thread.Sleep( 250 );

    Console.WriteLine( "ThreadID: {0:X8}, release   resource #{1}",
                Thread.CurrentThread.ManagedThreadId, index );

    // セマフォから出る(V操作、セマフォのカウンタをデクリメントする)
    int count = semaphore.Release();

    Console.WriteLine( "->previous semaphore count: {0}", count );
  }
}

実行結果例(Mono 1.2.3.1で実行したときのもの)

semaphore sample start
ThreadID: B719BB90, started
ThreadID: B708AB90, started
ThreadID: B6F89B90, started
ThreadID: B6E88B90, started
ThreadID: B6D87B90, started
ThreadID: B719BB90, wait for  resource #0
ThreadID: B719BB90, access to resource #0 'sem'
ThreadID: B708AB90, wait for  resource #1
ThreadID: B708AB90, access to resource #1 'aph'
ThreadID: B719BB90, release   resource #0
->previous semaphore count: 1
ThreadID: B708AB90, release   resource #1
->previous semaphore count: 2
ThreadID: B6D87B90, wait for  resource #1
ThreadID: B6D87B90, access to resource #1 'aph'
ThreadID: B6E88B90, wait for  resource #0
ThreadID: B6E88B90, access to resource #0 'sem'
ThreadID: B6D87B90, release   resource #1
->previous semaphore count: 1
ThreadID: B6F89B90, wait for  resource #2
ThreadID: B6F89B90, access to resource #2 'ore'
ThreadID: B6E88B90, release   resource #0
->previous semaphore count: 1
ThreadID: B708AB90, wait for  resource #1
ThreadID: B708AB90, access to resource #1 'aph'
ThreadID: B6F89B90, release   resource #2
->previous semaphore count: 1
ThreadID: B708AB90, release   resource #1
->previous semaphore count: 2
ThreadID: B6E88B90, wait for  resource #0
ThreadID: B6E88B90, access to resource #0 'sem'
ThreadID: B6D87B90, wait for  resource #1
ThreadID: B6D87B90, access to resource #1 'aph'
ThreadID: B719BB90, wait for  resource #0
ThreadID: B719BB90, access to resource #0 'sem'
ThreadID: B6E88B90, release   resource #0
->previous semaphore count: 0
ThreadID: B6D87B90, release   resource #1
->previous semaphore count: 1
ThreadID: B719BB90, release   resource #0
->previous semaphore count: 2
ThreadID: B6D87B90, wait for  resource #1
ThreadID: B6D87B90, access to resource #1 'aph'
ThreadID: B708AB90, wait for  resource #1
ThreadID: B708AB90, access to resource #1 'aph'
ThreadID: B6F89B90, wait for  resource #2
ThreadID: B6F89B90, access to resource #2 'ore'
ThreadID: B6D87B90, release   resource #1
->previous semaphore count: 0
ThreadID: B708AB90, release   resource #1
->previous semaphore count: 1
ThreadID: B6F89B90, release   resource #2
->previous semaphore count: 2
ThreadID: B719BB90, wait for  resource #0
ThreadID: B719BB90, access to resource #0 'sem'
ThreadID: B6E88B90, wait for  resource #0
ThreadID: B6E88B90, access to resource #0 'sem'
ThreadID: B6D87B90, wait for  resource #1
ThreadID: B6D87B90, access to resource #1 'aph'
ThreadID: B6F89B90, wait for  resource #2
ThreadID: B719BB90, release   resource #0
->previous semaphore count: 0
ThreadID: B6F89B90, access to resource #2 'ore'
ThreadID: B6E88B90, release   resource #0
->previous semaphore count: 0
ThreadID: B6D87B90, release   resource #1
->previous semaphore count: 1
ThreadID: B6F89B90, release   resource #2
->previous semaphore count: 2
ThreadID: B719BB90, finished
ThreadID: B708AB90, finished
ThreadID: B6F89B90, finished
ThreadID: B6E88B90, finished
ThreadID: B6D87B90, finished

使用例(システムセマフォ)

システムセマフォを生成し、排他区間に対して複数のプロセスから同時にアクセスを試みる。

行番号を表示する

using System;
using System.Diagnostics;
using System.Threading;

class Sample {
  // システムセマフォ(プロセス間で共有する)
  private Semaphore semaphore = null;

  //
  // エントリーポイント
  //
  static void Main()
  {
    Console.WriteLine( "semaphore sample start" );

    ( new SystemSemaphoreSample() ).Run();
  }

  public void Run()
  {
    Process proc = Process.GetCurrentProcess();

    // 初期値2、最大値2のセマフォ・インスタンスを作成
    bool created = false;

    semaphore = new Semaphore( 2, 2, "SystemSemaphoreSample", out created );

    if ( created ) {
      // このプロセスでシステムセマフォを生成した
      Console.WriteLine( "new semaphore created by PID:{0:X8}.", proc.Id );
    }

    // 他のプロセスが起動するまで適当な時間だけ待ち合わせる
    Thread.Sleep( new Random( proc.Id ).Next( 500, 5000 ) );

    Console.WriteLine( "ProcessID {0:X8}, wait for  semaphore", proc.Id );

    // セマフォに入る
    semaphore.WaitOne();

    Console.WriteLine( "ProcessID {0:X8}, access to semaphore", proc.Id );

    // 適当な時間だけ待ち合わせる
    Thread.Sleep( new Random( proc.Id ).Next( 100, 500 ) );

    Console.WriteLine( "ProcessID {0:X8}, release   semaphore", proc.Id );

    // セマフォを出る
    int count = semaphore.Release();

    Console.WriteLine( "->previous semaphore count: {0}", count );

    // プロセス終了
    return;
  }
}

上記プログラムを10プロセス同時に実行した場合の例(Mono 1.2.3.1で実行したときのもの)

$ for i in `seq 1 10` ; do mono sample.exe &  done

semaphore sample start
new semaphore created by PID:000048CF.
semaphore sample start
semaphore sample start
semaphore sample start
semaphore sample start
semaphore sample start
semaphore sample start
semaphore sample start
semaphore sample start
ProcessID 000048D1, wait for  semaphore
ProcessID 000048D1, access to semaphore
ProcessID 000048D1, release   semaphore
->previous semaphore count: 1
ProcessID 000048CF, wait for  semaphore
ProcessID 000048CF, access to semaphore
ProcessID 000048CF, release   semaphore
->previous semaphore count: 1
ProcessID 000048DF, wait for  semaphore
ProcessID 000048DF, access to semaphore
ProcessID 000048D4, wait for  semaphore
ProcessID 000048D4, access to semaphore
ProcessID 000048DF, release   semaphore
->previous semaphore count: 0
ProcessID 000048D4, release   semaphore
->previous semaphore count: 1
ProcessID 000048D2, wait for  semaphore
ProcessID 000048D2, access to semaphore
ProcessID 000048D0, wait for  semaphore
ProcessID 000048D0, access to semaphore
ProcessID 000048D2, release   semaphore
->previous semaphore count: 0
ProcessID 000048D0, release   semaphore
->previous semaphore count: 1
ProcessID 000048E0, wait for  semaphore
ProcessID 000048E0, access to semaphore
ProcessID 000048D5, wait for  semaphore
ProcessID 000048D5, access to semaphore
ProcessID 000048E0, release   semaphore
->previous semaphore count: 0
ProcessID 000048D3, wait for  semaphore
ProcessID 000048D3, access to semaphore
ProcessID 000048DE, wait for  semaphore
ProcessID 000048D5, release   semaphore
->previous semaphore count: 0
ProcessID 000048DE, access to semaphore
ProcessID 000048D3, release   semaphore
->previous semaphore count: 0
ProcessID 000048DE, release   semaphore
->previous semaphore count: 1