CSPは、並行オブジェクト相互間での複雑なやりとりをモデル化するための仕組みです。CSPを使うことによる主な利点として、プログラムの各段階で関係するオブジェクトの振る舞いを、正確に規定、検証するための機能がCSPに備わっている点が挙げられます。CSPの理論と習慣は、並行性の設計やプログラミングに大きな影響を与えてきました。CSPは、occamのようなプログラミング言語の基礎となっており、その他にも、Adaなどの設計にも影響を与えてきました。第1回で簡単に触れた通り、Javaプラットフォーム上でCSPを使うと、安全かつ優雅なマルチスレッド・プログラミングができるため、CSPはJava開発者にとっても貴重なものです。

Javaプラットフォーム上でのCSPプログラミングを紹介する3回シリーズの第2回目である今回は、CSPの理論と習慣に焦点を当て、特にJava言語でのマルチスレッド・プログラミングに応用する上での注意点を中心に話を進めます。まずCSPの理論の概要から始め、次に、CSPが取り入れられている、JavaベースのJCSPライブラリーの実装を紹介します。

CSPの基礎

CSPでの基本的な構成体は、プロセスと、プロセス間における様々な形式の通信です。CSPにおいては全てがプロセスであり、（サブ）プロセスのネットワークまであります。しかし、プロセスの間には直接の相互動作はありません。すべての相互動作は、CSP同期オブジェクト、つまり一群のプロセスがサブスクライブする通信チャネルやイベント・バリヤーなど通してのみ行われるのです。

CSPのプロセスは、プロセス・コンポーネントにカプセル化されたデータも、そのデータを操作するアルゴリズムも共にプライベートである、という意味において、通常のJavaオブジェクトとは異なります。つまり、プロセスは外部から呼び出し可能なメソッドを持っておらず（もちろん、そのプロセスを起動するために呼び出す必要のある1つのメソッドは別です）、アルゴリズムの実行は、プロセスが、そのプロセス独自のコントロール・スレッドでのみ行います。これを、Java言語でのメソッド呼び出しと比較してみれば、CSPでは明示的なロックが必要ないことが、すぐに理解できるでしょう。

Java言語では、あるオブジェクトに対して呼び出されたメソッドは、常に呼び出し側のスレッドで実行します。ある特定なコントロール・スレッドが、システム中の複数のオブジェクトをぬって動作するのです。オブジェクトは大部分の場合、独自の生命は持ちません。実行スレッドがオブジェクトを呼び出す、そのわずかの間だけ生きるのです。そのため第1回で議論したように、別々な実行スレッドが、同じオブジェクトに対する同じメソッドを同時に呼び出そうとすることが起こります。当然のことですが、こうした事態はCSPでは絶対に起こりません。

通信チャネルとプロセス・ネットワーク

最も単純なプロセス間通信は、チャネルにまたがってデータを読み書きすることです。CSPでの基本的なチャネル構成体は、同期（synchronous）とpoint-to-pointです。つまり、何も内部バッファリング無しにプロセスとプロセスをつなぎます。この基本チャネルを初めとして、マルチ・リーダー/ライターのチャネル（つまり1対多、多対1、多対多）を構成することもできます。

CSPでのプロセスは、複雑なシステムのための基本ブロックとなります。つまり1つのプロセスが、1つ以上の他のプロセス（すべてが並列に実行するように設定されています）とつながり、プロセスのネットワークを構成するのです。このネットワーク自体をまた、プロセスと考えることができます。このプロセスを、再帰的に他のプロセス（それ自体がネットワークの場合もあります）と組み合わせることによって、対象とする問題を最もうまく解決するための、複雑な仕組みにと構成できるのです。

個別に考えれば、プロセスというのは、外部のI/Oデバイスとしか相互動作しない、単なる独立なシリアル・プログラムです。このプログラムは、I/Oチャネルの対向側にあるプロセスの存在や、そのプロセスの素性について関心を持つ必要はありません。

CSPの理論は、幾つかのJavaベースのフレームワークで実装されていますが、その1つがJCSP（Communicating Sequential Processes for Java）ライブラリーです。

上に戻る

JCSPライブラリー

JCSPライブラリーは、イギリスのUniversity of Kent at Canterbury（ケント大学カンタベリー校） の、Peter Welch教授とPaul Austinによって開発されました（参考文献）。この記事のこれから先では、CSPの概念が、どのようにJCSPで実装されているかに焦点を当てます。Java言語ではCSP構成体をネイティブでサポートしていないため、JCSPライブラリーは内部的に、synchronizedやwait、notifyなど、Java言語が実際にネイティブでサポートしている並行性のための構成体を使っています。JCSPの具体的な動作を分かりやすく説明するために、幾つかのJCSPライブラリー・クラスの内部実装を、こうしたJava構成体の面から見ることにしましょう。

最初にお断りしておきますが、これから先のセクションで取り上げる例は、JCSPライブラリー用のjavadocsに書かれているものや、JCSPのホームページにある、プレゼンテーション・スライドからとったもの、またはそれらを元にしたものです。

上に戻る

JSCPでのプロセス

JCSPでは、プロセスというのは基本的に、CSProcessインターフェースを実装するクラスです。リスト1は、このインターフェースを示しています。

                
package jcsp.lang;

public interface CSProcess
{
    public void run();
}

CSProcessインターフェースが、Java言語でのRunnableインターフェースと全く同じに見えること、そして役割も似ていることに注意してください。JCSPは現在、標準のJava APIを使って実装されていますが、必ずしもそうである必要はなく、将来は恐らく違ってくるでしょう。この理由から、JCSPではRunnableインターフェースを直接使ってはいません。

JCSPではチャネルへの読み書きのために、2つのインターフェースを定義しています。チャネルから読み取るためのインターフェースはChannelInputと呼ばれ、read()と呼ばれる1つのメソッドで構成されます。このメソッドを（ChannelInputインターフェースを実装するオブジェクトに対して）呼ぼうとするプロセスは、チャネルの対向側にあるプロセスによって、あるオブジェクトが実際にチャネル上に書かれるまでブロックされます。そうしたオブジェクトがチャネル上で利用できるようになると、そのオブジェクトは呼び出し側のプロセスに返されます。同じように、ChannelOutputインターフェースは、write(Object o)と呼ばれる1つのオブジェクトから構成されます。このメソッドを（ChannelOutputインターフェースを実装するオブジェクトに対して）呼ぼうとするプロセスは、そのオブジェクトがチャネルに受け付けられるまでブロックされます。先に触れたように、最も単純なタイプのチャネルは、何らバッファリングを行いません。ですから対向側（つまり読み取り側）のプロセスがread()を呼ぶまで、チャネルはそのオブジェクトを受け付けないのです。

ここから先では、コード例を使いながら、これら、およびその他の構成体の動作を説明して行きます。リスト2は、1から100までの間の偶数すべてを出力する、非常に単純なプロセスを示しています。

                
import jcsp.lang.*;

public class SendEvenIntsProcess implements CSProcess 
{
    private ChannelOutput out;

    public SendEvenIntsProcess(ChannelOutput out)
    {
      this.out = out;
    }

    public void run()
    {
      for (int i = 2; i <= 100; i = i + 2)
      {
        out.write (new Integer (i));
      }
    }
}

各書き込みプロセスには、それぞれに対応して、読み取りプロセスが必要です。そうしたプロセスが無い場合には、ChannelOutputオブジェクトのoutへの最初の書き込みの直後から、SendEvenIntsProcessは永遠にブロックされます。リスト3は、リスト2の書き込みプロセスに対応する、単純な読み取りプロセスを示しています。

                
import jcsp.lang.*;

public class ReadEvenIntsProcess implements CSProcess
{
    private ChannelInput in;
    public ReadEvenIntsProcess(ChannelInput in)
    {
      this.in = in;
    }

    public void run()
    {
      while (true)
      {
        Integer d = (Integer)in.read();
        System.out.println("Read: " + d.intValue());
      }
    }
}

上に戻る

JCSPでのチャネル

この時点では、2つの独立なプロセスがあるだけです。次のステップは、ある共通チャネルを共有同期機構として2つをつなぎ合わせ、次にそれぞれを起動することです。チャネル・インターフェースというのは、JCSPのChannelInputインターフェースとChannelOutputインターフェースのサブ・インターフェースであり、オブジェクトの読み取り、書き込みに対する共通インターフェースです。このインターフェースの実装には様々なものが考えられますが、その一例を下記に示します。

• クラス、One2OneChannelは、その名前の通り、1対1、つまり「single-writer-single-reader」型のチャネルを実装します。
• クラス、One2AnyChannelは、1対多、つまり「single-writer-multiple-readers」オブジェクト・チャネルを実装します（これはブロードキャスト機構とは異なり、そのチャネルから読み取るために、複数のリーダーが、お互いに実際に競争します。ある任意の時間にチャネルを使えるのは、1つのリーダーと1つのライターのみです）。
• クラス、Any2OneChannelは、多対1、つまり「multiple-writers-single-reader」オブジェクト・チャネルを実装します。上の場合と同じく、チャネルを使うために、複数の書き込みプロセスが互いに競争します。ある任意の時間に実際にチャネルを使えるのは、リーダーと、（複数ライターのうち）1つのライターのみです。
• クラス、Any2AnyChannelは、多対多、つまり「multiple-writers-multiple-readers」オブジェクト・チャネルを実装します。（書き込みプロセスの場合と同じように）読み取りプロセスも書き込みプロセスも、チャネルを使うために互いに競争します。ある任意の時間に実際にチャネルを使えるのは、1つのリーダーと1つのライターのみです。

リスト3の例では、1つのリーダーと1つのライターのプロセスしかありません。ですから、One2OneChannelクラスで十分です。リスト4は、このドライバー・プログラム用のサンプル・コードを示しています。

                
import jcsp.lang.*;

public class DriverProgram
{
    public static void main(String[] args)
    {
      One2OneChannel chan = new One2OneChannel();
      new Parallel
      (
        new CSProcess[]
	    {
	      new SendEvenIntsProcess (chan),
	      new ReadEvenIntsProcess (chan)
	    }
      ).run ();
    }
}

このコードを見ると分かるように、最初に新しいOne2OneChannelオブジェクトをインスタンス化し、それをSendEvenIntsProcessプロセスとReadEventIntsProcessプロセスのコンストラクターに渡しています。この方法でうまく行く理由は、One2OneChannelが両方のインターフェース、つまりChannelInputとChannelOutputの両方を実装しているためです。

チャネルの内部

JCSPでのチャネルは重要な概念なので、先に進む前にチャネルがどのように動作するのかを理解しておきましょう。先に触れた通り、デフォルトではチャネルはバッファリングを行いませんが、バッファリングするようにすることもできます。そうするためには、それ自身ではバッファリング特性を扱わないチャネルを使い、バッファリングの責任を別のクラスに委任するのです。このクラスは、ChannelDataStoreというインターフェースを実装する必要があります。JCSPでは、このインターフェース用の組み込み実装を、下記のように幾つか用意しています。

• ZeroBufferは、デフォルトの非バッファリング特性に対応します。
• Bufferは、バッファーを関連付ける対象となるチャネルに対して、ブロックFIFOバッファーを持つ意味体系を提供します。
• InfiniteBufferもFIFOを持つ意味体系を提供しますが、バッファーが空の時にブロックされるのはリーダーだけです。ライターがブロックされることはありません。これは、バッファー容量は無限に、あるいは少なくとも基礎となっているメモリー・システムによって課せられる制限に達するまで、拡張できるためです。

上に戻る

チャネルの実際の動作

チャネルが実際に動作している場合の例を考えて見ましょう。リスト4でOne2OneChannelインスタンスを作成した時には、このインスタンスの内部ChannelDatasourceを、新しいZeroBufferのインスタンスに設定しました。ZeroBufferには1つのオブジェクト（あるいは整数）しか保存できません。ZeroBufferは、値がEMPTYから始まる内部状態変数を持っています。この値は、あるオブジェクトが中に入った途端にFULLになります。

SendEvenIntsProcessプロセスが、その出力チャネルにwriteを行ったら、どうなるでしょう。One2OneChannelクラスのwrite()メソッドは、synchronized()メソッドです。従って、送信側のプロセスが実行しているスレッド（このすぐ後で、送信側と読み取り側のプロセスが別々のスレッドで実行することを説明します）は、このチャネル・インスタンスに関連したモニター・ロックを取得し、write()メソッドを進めます。このメソッド内での最優先の課題は、内部的に保持しているZeroBufferインスタンスに対して（putメソッドを呼ぶことによって）、オブジェクト（あるいは、この場合であれば整数）を書き込むことです。これによってバッファーの状態がFULLに変わります。この時点で、呼び出し側のスレッドはwaitを呼び出し、これによって呼び出し側スレッドはモニターのwait setに入ります。それに続いて、モニター・ロックが解放され、スレッドがブロックされます。

この後のある時点で、読み取り側スレッドはチャネル上でread操作を呼び出します（これも同期メソッドです。ですから読み取り側スレッドは、前に進む前にモニター・ロックを取得する必要があります）。内部バッファーの状態はFULLなので、そこにあるデータが返され、notify()が発行されます。このnotify()によって、送信側スレッドが起き上がります（wake up）。これで送信側スレッドはモニターのwait setから抜け出し、モニター・ロックを取り返します。

対話動作のシナリオでは、もし読み取り側のスレッドが、内部バッファーがEMPTY状態であるチャネル上でreadメソッドを呼び出してしまうと、waitせざるを得なくなります。この場合には、送信側スレッドが、データ・オブジェクトを内部バッファーに書き込んだ後、読み取り側スレッドに通知を行います。

上に戻る

Parallel構成体

皆さんはリスト4を見て、ドライバー・プログラムがParallelという新しいクラスを導入していることに気がついたかも知れません。Parallelクラスは、事前定義のCSProcessとしてJCSPが提供するものです。CSProcessは、個々のCSProcessインスタンスの配列を「並列に」実行します（最後のスレッドを除いて、すべてのスレッドは別個のスレッドで実行します。最後のスレッドは、Parallelオブジェクトが、独自のコントロール・スレッドで実行します）。Parallelプロセスのrunは、すべてのコンポーネント・プロセスが終了した、その時にのみ終了します。つまりParallelプロセスは、複数ある個々のプロセスを、（ドライバー・プログラムの例では）チャネルを「針金」のように使って、まとめ上げるための機構なのです。

Parallel構成体に対する別の見方として、Parallel構成体を使うことによって、小さな、単純なコンポーネントから、高位レベルのプロセスを構成できるようになる、ということもできます。実際Parallel構成体を使うと、何度か繰り返しを行い、前の繰り返しで作成されたコンポーネントを順次つなぎ合わせることによって、任意の複雑さを持った連結プロセス・ネットワークを構成できます。その結果できあがるプロセス・ネットワークは、また別のCSProcessオブジェクトとして、エクスポーズし、使用することができるのです。

上に戻る

Parallelの例

JCSPライブラリーでは、プラグ・アンド・プレイのコンポーネントを提供しています。これらは教育用の目的にしか使えないものですが、この記事での目的には全く問題ありません。こうしたコンポーネントの幾つかの内部実装に入り込んでみると、ネットワーク化された並行プロセスが、JSCPではどのように構成されているかを知ることができます。ここでは次のプロセス例を使って、JCSPにおけるParallel構成体の内部動作を説明することにします。

• PlusIntは、その2つの入力ストリームに整数を取り込み、2つを加算し、その結果を出力ストリームに出力します。
• Delta2Intは、その入力ストリームに到着するすべての整数を、並列に、その2つの出力チャネルにブロードキャストします。
• PrefixIntは、その整数入力ストリームに対して、（ユーザー設定の）整数を接頭辞として付加します。（つまり、入力チャネルに何か整数が入る前には、このプロセスからの最初の出力は、設定された整数そのもの、ということです。その後に続く出力は、入力ストリームから取られた整数そのまま、となります。）
• IntegrateIntは、Parallel構成体を使った上の3つから構成されるプロセスです。このプロセスの機能は、入力チャネルに入ってくる整数の瞬時合計を出力することです。

IntegrateIntクラスのrunメソッドを、リスト5に示します。

                
import jcsp.lang.*;

public class IntegrateInt implements CSProcess 
{
  private final ChannelInputInt in;
  private final ChannelOutputInt out;

  public IntegrateInt (ChannelInputInt in, ChannelOutputInt out)
  {
    this.in = in;
    this.out = out;
  }

  public void run()
  {
      One2OneChannelInt a = new One2OneChannelInt ();
      One2OneChannelInt b = new One2OneChannelInt ();
      One2OneChannelInt c = new One2OneChannelInt ();

      new Parallel 
      (
        new CSProcess[]
        {
          new PlusInt (in, c, a),
          new Delta2Int (a, out, b),
          new PrefixInt (0, b, c)
        }
      ).run ();
  }
}

この例では、リスト4に示したものとは異なる種類のチャネルが使われていることに注意してください。IntegrateIntクラスは、ChannelInputIntチャネルとChannelOutputIntチャネルを使っています。これらは名前からも分かる通り、int型の整数を転送するために使われます。これと対照的に、リスト4のドライバー・プログラムは、ChannelInputとChannelOutputを使っています。これらは、送信側から受信側に任意のオブジェクトを送るための、オブジェクト・チャネルです。そのためリスト4では、int値を転送する前には、int値をIntegerオブジェクトとしてラップする必要がありました。

リスト5のコードを見て、他にどんなことに気がつくでしょう。基本的に、PrefixIntプロセスからの最初の出力は0です。この出力はPlusIntプロセスによって、入力チャネルに到着する最初の整数に追加されます。この結果は、Delta2Intプロセスへの入力チャネルを構成する、チャネルaに書き込まれます。Delta2Intプロセスは、結果である整数をout（プロセス全体に対する出力チャネル）に書き込み、PrefixIntプロセスに送信します。そうするとPrefixIntプロセスは、その整数を、そのままPlusIntプロセスへの入力として送信し、これが今度はストリーム中にある2番目の整数に加算され、以下これが同じように繰り返されます。

IntegrateIntプロセスの構成を図式的に表現したものが、図1です。

上に戻る

ネットワーク内部でのネットワーク

このようにIntegrateIntプロセスは、3つの小さなプロセスから構成されていますが、IntegrateIntプロセス自体を構成プロセスとして使うこともできます。JCSPライブラリーには、SquaresIntというプロセスが用意されています。これは名前からも分かる通り、自然数（1、2、3、4など）の平方である整数のストリームを生成します。このプロセスのコードをリスト6に示します。

                
public class SquaresInt implements CSProcess 
{
  private final ChannelOutputInt out;

  public SquaresInt (ChannelOutputInt out)
  {
    this.out = out;
  }

  public void run()
  {
      One2OneChannelInt a = new One2OneChannelInt ();
      One2OneChannelInt b = new One2OneChannelInt ();

      new Parallel 
      (
        new CSProcess[]
        {
          new NumbersInt (a),
          new IntegrateInt (a, b),
          new PairsInt (b, out)
        }
      ).run ();
  }
}

恐らく皆さんは、リスト6にある2つの新しいプロセスに気がついたと思います。NumbersIntは、0から始まる自然数を単純に出力チャネルに出力する、組み込みのプロセスです。PairsIntは、入力値の対を次々と加算し、その結果を出力するプロセスです。これら2つの新しいプロセスとIntegrateIntが、SquaresIntプロセスを構成しています。これを図2に示します。

SquaresIntはどのように動作するか

先に進む前に、SquaresIntプロセスの内部動作を考えて見ましょう。下記を見ると、SquaresInt内の個々のチャネルで、どのようにトラフィックが流れているかが分かると思います。

Channel "a":	[0, 1, 2, 3, 4, 5, 6, 7, 8, ...ad infinitum]
Channel "b":	[0, 1, 3, 6, 10, 15, 21, 28, 36, ...ad infinitum]
Channel "out":	[1, 4, 9, 16, 25, 36, 49, 64, 81 ...ad infinitum]

チャネルaに書かれる整数が、チャネルbに書かれる整数になり、やがてチャネルoutに出力される、という様子が分かるでしょうか。一番最初の「刻み(tick)」では、NumbersIntプロセスが、整数0をチャネルaに書き込みます。また、IntegrateIntプロセスも、整数0（これがつまり、瞬間合計の現在の値です）をチャネルbに書き込みます。PairsIntプロセスは、（2つの入力が必要なため）この刻みでは何も生成しません。2番目の刻みでは、NumbersIntプロセスは、整数1を出力チャネルに書き込みます。これによってIntegrateIntプロセスが瞬間合計を0+1=1に変更し、その結果、整数1がチャネルbに書き込まれます。

この時点で、PairsIntには、作業するために必要な2つの整数入力があることになります。つまり、この前の刻みからの0と、現在の刻みでの1です。PairsIntは2つを加算し、出力0+1=1をチャネルoutに出力します。ここで、1は1の平方であることに注意すると、どうやら正しい道を進んでいるようです。この手順を次（3番目）の刻みに進めると、NumbersIntプロセスが、整数2をチャネルaに書き込みます。これによってIntegrateIntプロセスは瞬時合計を、1（前の合計）＋2（新しい値）= 3へと変更し、この整数をチャネルbに書き出します。

では、PairsIntプロセスが見る最後の2つの整数は何でしょう。それは、1（前の刻みの期間中）と、3（現在の刻みの期間中）です。従ってPairsIntプロセスは、この2つの整数を加算し、1+3=4をチャネルoutに書き出します。4は2の平方ですから、SquaresIntは望ましい姿で動作していることになります。実際、このプログラムの実行を続け、好きな回数だけ刻みを繰り返すと、チャネルoutに書き出される整数は常に、その後ろに並んでいる数の平方であることを検証できるのです。次のセクションでは、正にこの検証を行います。

ちょっと数学的な寄り道を

よく理解できない読者もいるかも知れないので、これらの平方がどのように生成されるかについて、数学的な基本を説明しておきましょう。NumbersIntプロセスが既に、ある整数n-1まで出力し終わっている時に、このプロセスの中を覗いたとしましょう。そうすると、IntegrateIntプロセスが最後に生成した（そして共有チャネルbを通してプロセスに与えられた）瞬時合計は、[1+2+3+...+(n-1)] = (n-1)(n-2)/2となります。

次の刻みでは、NumbersIntはnを出力します。これによってIntegrateIntプロセスの瞬時合計は、(1+2+3+...+n) = n(n-1)/2に増加します。この和が、今度は共有チャネルbを通してPairsIntプロセスに与えられます。PairsIntは、この2つの数字を加算して、[(n-1)(n-2)/2 + n(n-1)/2] = [(n-2) + n](n-1)/2 = (2n-2)(n-1)/2 = (n-1)exp2を生成します。

次に、NumbersIntプロセスが、(n+1)を生成します。これに対応して、IntegrateIntプロセスはn(n+1)/2をPairsIntプロセスに与えます。そうするとPairsIntは[n(n-1)/2 + n(n+1)/2] = nexp2を生成します。これを全てのnに対して一般化すると、望む通り、全てのnの平方が生成されます。

上に戻る

JSCPでの決定論

上の例は、CSPの意味体系の構成、つまりParallel構成体を使って、キメの細かいステートレス・コンポーネントから階層化ネットワークを構成できること、を示しています。こうした、並列プロセス同士が通信を行う階層化ネットワークの良いところは、完全に決定論的（deterministic）であることです。ここでの『決定論的』とは、一体何を意味するのでしょうか。『決定論的』ということは、こうした階層化ネットワークからの出力が、そのネットワークに提供される入力にのみ依存し、ネットワークを実行するランタイム環境（JVM）の特性に依存しない、ということなのです。つまり、このプロセス・ネットワークは、JVMのスケジューリング・ポリシーからも、JVMが複数プロセッサーに分散していることからも、独立しています。（ここでは単一のノードを想定していますが、様々なプロセスが線路を経由して通信し合うような、複数ノードに渡って物理的に分散したプロセス・ネットワークであっても、何ら状況が異なるわけではありません。）

決定論は、ツールキットの中に備えておくべき強力なツールです。決定論によれば、プログラムの振る舞いを、ランタイム環境の影響を考慮する必要なく明確に推論できるのです。それと同時に、並行プログラミングに対する手法は決定論のみ、あるいは決定論が必須、というわけでもありません。次の（そして最後の）例で示すように、JCSPライブラリーの中では、非決定論も、同じくらい強力な実用的概念なのです。

上に戻る

JCSPでの非決定論

現実の多くのアプリケーションでは、非決定論が要素となっています。非決定論では、目に見える結果というのは、イベントが発生する順序の関数です。言い方を変えると、非決定論が登場するのは、並行アプリケーションの結果が、偶発的な要素ではなく、設計によるスケジューリングに依存するような場合なのです。これから説明するように、JSCPでは、こうした問題を明示的に処理します。

例えば、あるプロセスが、次にすべきこととして、いくつかの選択肢を持っていると考えてください。それぞれの選択肢は、関連して「ガード（guard）」を持っており、その選択肢が実際に選択可能であるためには、このガードは「レディー（ready）」状態である必要があります。プロセスは、利用可能な（つまりレディー状態の）選択肢から1つを選択します。この選択自体は、ランダムな選択や優先度順の選択、公正な選択など、様々な戦略に基づく可能性があります。

イベント選択戦略

JCSPでは具体的に、Guardという抽象クラスが提供されています。Guardは、プロセスによって選択されるべく競争しあっているイベント・オブジェクトによって、サブクラス化される必要があります。プロセス自体は、別途事前提供された、Alternativeというクラスを使います。Alternativeのコンストラクターに対して、こうしたガード・オブジェクトをオブジェクト配列として渡す必要があるのです。Alternativeクラスは、イベント選択戦略に対して、3つのタイプのメソッドを提供しています。

Alternativeクラスのselect()メソッドは、ランダムな選択という戦略に対応します。select()メソッド・コールは、1つ以上のガードがレディー状態になるまでブロックされます（Alternativeクラスは、競争しあっている全部のガードを知っていることを思い出してください）。レディー状態のガードのうちの1つがランダムに選択され、そのインデックス（渡されたガードの、配列のインデックス）が返されます。

priSelect()メソッドは、優先度順での選択という戦略に対応します。つまり、1つ以上のガードがレディー状態である場合には、最もインデックスの小さなガードが返されます（ただし、Alternativeのコンストラクターに渡された配列の中にあるガードは、既に優先度の降順にソートされているという前提です）。

最後に、fairSelectメソッドは、1つ以上のガードがレディー状態である場合には、公正に選択を行います。つまり、このメソッドを連続的に呼び出した場合、ある1つのガードだけが2度選択され、他のガードが選択されずに終わる、ということがありません。従って、レディー状態にあるガードの総数がnである場合には、最悪の場合でも、n回以上の連続的な選択操作を行えば、選択されずに終わるガードはありません。

複数のガードがレディー状態の場合にどれを選択するかについて、プロセスが関知しないのであれば、ランダム選択戦略が最もうまく行きます。逆に、リアルタイム・システムなど、不足無しの保証やワーストケースでのサービス時間を要求するようなプロセスに対しては、あまりうまく行きません。前者にはfairSelectが向いており、後者にはpriSelect()が最適です。

ガードのタイプ

広い意味で言うと、JCSPには3つのタイプのガードが用意されています。

• チャネル・ガードは常に、プロセスがデータを読み取ろうとしている対象のチャネルに対応します。つまり、チャネルの対向側にあるプロセスがチャネルへの出力を持っているものの、そのプロセスが、まだそのデータを入力していない時に、そしてその時にのみ、このガードはレディー状態となります。
• タイマー・ガードは常に、（絶対的な）タイムアウト設定に対応します。つまりタイマー・ガードは、そのタイムアウトが失効した時にレディー状態になります。
• スキップ・ガードは、常にレディー状態です。

JCSPでのチャネル・ガードには、次のようなタイプがあります。AltingChannelInput/AltingChannelInputIntはそれぞれ、オブジェクトあるいは整数データが、対応するチャネルで保留状態の場合には、常にレディーになります。AltingChannelAcceptは、受け入れられなかった「CALL」がチャネル上で保留状態の場合にレディーになります（これについては、後でさらに説明します）。これらは抽象クラスであり、One2OneやAny2Oneというタイプのチャネルの形で、具体的な実装があります。JCSPでのタイマー・ガードはCSTimer型ですが、スキップ・ガードは、Skipクラスとして提供されています。

上に戻る

ガードの実際

ではここで、並行アプリケーションでの非決定論を実現するためのJSCPのガードの簡単な使い方を例に挙げて、JCSPの紹介を終えたいと思います。まず仮定として、乗算（スケーリング）デバイスを開発することを考えてみてください。固定した速度で入力チャネルに到着する整数を読み取り、ある係数でそれらを乗算し、出力チャネルに書き出すのです。このデバイスは、ある初期係数で開始しますが、この係数は5秒毎に自動的に倍になります。

この話の展開は次の通りです。このシステムには、2番目のコントローラー・プロセスが存在し、専用チャネルを通して、このデバイスにsuspend operation信号を送るのです。これによってデバイスは自分でサスペンド状態になり、乗算係数の現在の値を、2番目のチャネルを使ってコントローラーに送ります。

デバイスは、サスペンド状態にある場合には、単純に入ってくる整数を変更せず出力チャネルに流す必要があります。コントローラー・プロセスは（例えば、そのプロセスに送られた乗算係数を入力に使って何らかの計算を行った後）、専用チャネルを通して新しい係数をデバイスに送ります。（デバイスは、サスペンド状態にある場合には、この係数を受け取る義務があることに注意してください。）

更新された係数をデバイスに注入すると、これがデバイスを起き上がらせる信号としても動作します。これでデバイスはスケーリング・アップ操作を再開し、入力された整数を、新たに更新された係数で乗算します。タイマーも、この時点でリセットされます。ですから、新しい乗算係数は5秒後に倍になるように設定され、以下、これが繰り返されます。

図3は、このスケーリング・デバイスを表しています。

上に戻る

ScaleIntプロセス

このスケーリング・デバイスのソースコードをリスト7に示します。この例では、出力の値はinストリームとinjectストリーム上の値（それに加えて、こうした値が到着する順序）に基づくため、非決定論的です。

                
import jcsp.lang.*;
import jcsp.plugNplay.ints.*;

public class ScaleInt implements CSProcess
{
  private int s;
  private final ChannelOutputInt out, factor;
  private final AltingChannelInputInt in, suspend, inject;

  public ScaleInt (int s, AltingChannelInputInt suspend, AltingChannelInputInt in, 
    ChannelOutputInt factor, AltingChannelInputInt inject, ChannelOutputInt out)
  {
    this.s = s;
	this.in = in;
	this.out = out;
	this.suspend = suspend;
	this.factor = factor;
	this.inject = inject;
  }

  public void run()
  {
	final long second = 1000;               // Java timings are in millisecs
	final long doubleInterval = 5*second;
	final CSTimer timer = new CSTimer ();

	final Alternative normalAlt = new Alternative (new Guard[] {suspend, timer, in});
	
	final int NORMAL_SUSPEND=0, NORMAL_TIMER=1, NORMAL_IN = 2;

	final Alternative suspendedAlt = new Alternative (new Guard[] {inject, in});
	
	final int SUSPENDED_INJECT=0, SUSPENDED_IN = 1;
	
	long timeout = timer.read () + doubleInterval;
	timer.setAlarm (timeout);

	while (true)
	{
	  switch (normalAlt.priSelect ())
	  {
		case NORMAL_SUSPEND:
		  suspend.read ();              // don't care what's sent
		  factor.write (s);             // reply with the crucial information
		  boolean suspended = true;
		  while (suspended)
		  {
		    switch (suspendedAlt.priSelect ())
			{
			  case SUSPENDED_INJECT:    // this is the resume signal as well
			    s = inject.read ();     // get the new scaling factor
				suspended = false;      // and resume normal operations
				timeout = timer.read () + doubleInterval;
				timer.setAlarm (timeout);
				break;
			  case SUSPENDED_IN:
			    out.write (in.read ());
				break;
			}
		  }
		  break;
		case NORMAL_TIMER:
		  timeout = timer.read () + doubleInterval;
		  timer.setAlarm (timeout);
		  s = s*2;
		  break;
		case NORMAL_IN:
		  out.write (s * in.read ());
		  break;
	  }
    }
  }
}

import jcsp.lang.*;
import jcsp.plugNplay.ints.*;

public class Controller implements CSProcess
{
  private long interval;
  private final ChannelOutputInt suspend, inject;
  private final ChannelInputInt factor;

  public Controller (long interval, ChannelOutputInt suspend, ChannelOutputInt inject, 
    ChannelInputInt factor)
  { 
    this.interval = interval;
    this.suspend = suspend;
    this.inject = inject;
    this.factor = factor;
  }

  public void run ()
  {
	int currFactor = 0;
	final CSTimer tim = new CSTimer ();
	long timeout = tim.read ();
	while (true)
	{
	  timeout += interval;
	  tim.after (timeout);        // blocks until timeout reached
	  suspend.write (0);          // suspend signal (value irrelevant)
	  currFactor = factor.read ();			
	  currFactor ++;              // compute new factor
	  inject.write (currFactor);  // inject new factor
	}
  }
}

import jcsp.lang.*;
import jcsp.plugNplay.ints.*;

public class DriverProgram
{
  public static void main(String args[])
  {
	try
	{
	  final One2OneChannelInt temp = new One2OneChannelInt ();
	  final One2OneChannelInt in = new One2OneChannelInt ();
	  final One2OneChannelInt suspend = new One2OneChannelInt ();
	  final One2OneChannelInt factor = new One2OneChannelInt ();
	  final One2OneChannelInt inject = new One2OneChannelInt ();
	  final One2OneChannelInt out = new One2OneChannelInt ();
		
	  new Parallel
	  (
		new CSProcess[]
		{
		  new NumbersInt (temp),
		  new FixedDelayInt (1000, temp, in),
		  new ScaleInt (2, suspend, in, factor, inject, out),
		  new Controller (6000, suspend, inject, factor),
		  new PrinterInt (out, "--> ", "\n")
		}
	  ).run ();
	}
	catch (Exception e)
	{
		e.printStackTrace();
	}
  }
}

上のクラス、ScaleIntは、スケーリング・デバイスに対応します。先に触れた通り、このクラスは、CSProcessインターフェースを実装する必要があります。上記のコードには多くの概念が含まれているため、それぞれの側面を1つずつ説明して行くことにします。

2つのAlternative

ScaleIntクラスで最初に注目すべきメソッドがrun()です。run()メソッドの中で最初に行っているのは、Alternativeクラスのインスタンスを2つ作り、それぞれが別々のGuardsオブジェクト配列を持つようにすることです。

Alternativeインスタンスの最初の方（変数normalAltという名前です）は、デバイスが正常に動作している場合に使うことを想定しています。これに関連したガードのリストは次の通りです。

• suspendは、One2OneChannelIntのインスタンスです。先に触れた通り、One2OneChannelIntは、読み取り側/書き込み側が1つの整数チャネル（ゼロ・バッファーで完全同期）を実装します。コントローラー・プロセスがサスペンド信号をデバイスに送るのが、このチャネルです。
• timerは、CSTimerのインスタンスです。5秒毎にトリガーされるように設定されており、トリガーされると、デバイスは乗算係数の現在の値を倍にします。
• inは、One2OneChannelIntのインスタンスです。デバイスはこの上で入力整数を受信します。

2番目のAlternativeインスタンスには、suspendedAltという名前が付いていますが、このインスタンスは、デバイスがそれまでControllerによってサスペンドされている場合に使われることを想定しています。これに関連付したガードのリストは次の通りです。

• injectは、One2OneChannelIntのインスタンスであり、コントローラー・プロセスが新しい乗算係数（ウェイクアップ信号としての役割も果たします）をデバイスに送信するときに使われます。
• inは、先に見たOne2OneChannelIntのインスタンスと同じです。デバイスは、このチャネル上で入力整数を受信します。

2つのAlternativeインスタンスは、ガードがレディー状態になるのを待つために、それぞれ異なる状況で使われており、リストの順序は暗黙的な優先順位になっています。例えば、normalAltのsuspendガードとtimerガードが同時にレディー状態になると、suspendガードに対応するイベントが最初に処理されます。

ガードします！

次に面白い点として、それぞれのガードがレディー状態の場合に何が起きるか、という問題があります。最初に、デバイスが正常に動作している（つまり、まだサスペンドされていない）という前提で、normalSelectを取り上げます。

• もしコントローラーがデバイスに対してサスペンド信号を送ると、このイベントは最優先で処理されます。これに応答してデバイスは、乗算係数の現在の値を、factorというチャネルを通してコントローラーに送ります。そしてsuspendedという内部フラグをtrueに設定してループに入り、再開信号が送られてくるのを待ちます。デバイスはこのループ内で、2番目のAlternativeインスタンス（suspendedAlt）上のpriSelect()メソッドを呼び出します。
  
  このAlternativeインスタンスは、2つのガードから構成されます。最初のガードはイベントを表し、このイベントの中で、コントローラーが新しい乗算係数をデバイスに送ります。2番目のガードは、デバイスの入力チャネルに整数が到着したことを表します。前者の場合、デバイスは（変数sの中に保持されている）係数を、injectチャネルから読み取った新しい値で更新し、suspendedフラグをfalseに戻し（これによって、次の繰り返しで内部ループから出られることが保証されます）、現在のタイマー値をベースに使ってアラームをリセットして終了します。後者の場合は、デバイスは単純に入力チャネルから整数を読み取り、それを出力チャネルに書き出します（デバイスのサスペンド中は乗算係数を使わない、という要求によるものです）。
• その次に優先して処理されるのは、アラームを発するイベントです。このイベントによってデバイスは現在の乗算係数を倍にし、現在のタイマー値をベースとして使ってアラームをリセットし、次のイベントを待つ状態に戻ります。
• 3番目に優先して処理されるのは、デバイスの入力チャネルで整数を受信するイベントです。デバイスはこれに対応して、整数を読み取り、現在の係数sを掛け、結果をそのデバイスの出力チャネルに書き出します。

Controllerクラス

次に考慮すべきクラスは、Controllerクラスです。思い出して欲しいのですが、コントローラー・クラスの仕事というのは、新しい乗算係数値を（恐らく複雑な計算に基づいて）デバイス・プロセスに周期的に注入することです。周期的といっても、この例では単にタイマーが、正規の、設定可能な時間間隔に従って動作するだけです。タイマーが動作する度に、コントローラーはサスペンド・チャネルに0を書き込み（つまりデバイスをサスペンドし）、入力チャネル上にある、factorと呼ばれる現在の乗算係数を読み取ります。

この時点では、コントローラーは単にこの値を1増加し、それを、この目的専用の（injectという）1対1チャネルを通してデバイスに再度注入するだけです。この信号によってデバイスは再開し、この時点でタイマーは、適当な時間間隔の後に動作するようにリセットされます。

DriverProgramクラス

残っているクラスは、ドライバー・クラス、DriverProgramだけです。このクラスは適当なチャネルと、CSProcessインスタンスの配列を作成します。そしてJCSPが提供するクラス、NumbersIntを使って、自然数の連続を生成します。これらの自然数はtempチャネルを通して、（FixedDelayIntという）別の組み込みクラスに供給されます。FixedDelayIntは名前からも分かる通り、その入力チャネルから入ってくるデータを、一定の遅延時間（この例のコードでは1秒）の後、出力チャネルに渡します。

こうした1秒間隔の自然数ストリームは、ScaleIntプロセスのinチャネルに供給されます。ScaleIntプロセスのoutチャネルからの出力は、JCSPが提供するPrinterIntプロセスに供給され、このプロセスが今度は整数値をSystem.outに出力されるのです。

上に戻る

第2回のまとめ

JavaプログラマーにCSPを紹介するための3回シリーズの第2回目である今回は、並行プログラミングにおけるCSPの理論に関して、解説と実例を示しました。ここではCSP構成体の概要を説明した後、JavaベースのCSPライブラリーとして最も一般的なJCSPを紹介しました。Java言語ではCSP構成体をネイティブではサポートしていないため、JCSPライブラリーは内部的に、synchronized()やwait()、notify()など、Javaが実際にネイティブでサポートしている、並行性のための構成体を使っています。この記事ではさらに、JCSPの動作を具体的に理解できるように、一部のJCSPライブラリー・クラスの内部実装をJava構成体の面から説明し、その使い方を幾つかの実例を使って示しました。

この記事で行った議論は、このシリーズ最後の記事を導入するための基礎ともなっています。最後の記事では、CSPとAOPとの類似点を説明します。またCSPでの並行性と、新しいjava.util.concurrentパッケージで並行性を比較し、最後にJCSPでの高度な同期化手法の幾つかを紹介する予定です。

謝辞

このシリーズを執筆するに当たって、Peter Welch教授から頂いた励ましに深く感謝致します。教授は、それでなくても多忙な中で、私の下書きを見て下さり、記事の質や精確さを高める上で多くの貴重な助言を下さいました。残りの間違いは、全て私の責任です。私が記事の中で扱った例はどれも、JCSPライブラリーのJavadocsの中に文書化されている資料、あるいはJCSPのWebサイトにあるPowerpointのスライドを元にしたものです。どちらも豊富な情報を提供しています。

参考文献

• Brian Goetzによる3回シリーズの記事「Threading lightly」（developerWorks, 2001年7月）は、Javaプラットフォーム上での同期問題を解決する上で非常に賢明な、方法論的手法を紹介しています。
  
  
• Allen Holubによる「If I were king: A proposal for fixing the Java programming language's threading problems」（developerWorks, 2000年10月）は、Javaプラットフォーム上でのマルチスレッド・プログラミングの何が悪いかを概説した啓蒙的な記事として、今でも通用するものです。
  
  
• C.A.R. Hoareによる「Communicating Sequential Processes」は、通信しあう連続プロセスの並列合成を、基本的なプログラム構成方法として紹介しています（Communications of the ACM Archive, 1978年刊）。
  
  
• C.A.R. Hoareによるbook on CSPは、PDFで無料で入手することができます。
  
  
• Bill RoscoeによるTheory and Practice of Concurrency（Prentice Hall, 1997刊）は、平行性とCSPの話題に関する最近の本です。
  
  
• Oxford University Computer Laboratory（オックスフォード大学コンピューター研究所）がホストするCSP Archiveは、WoTUG homepageと並んで、CSPについて学ぶための素晴らしい情報源です。
  
  
• Professor Peter WelchとJeremy Martinによる「Formal Analysis of Concurrent Java Systems」（IOS Press, 2000年）は、JavaプログラミングでCSPを練習するための出発点として最適です。
  
  
• JCSP homepageは、イギリスのUniversity of Kent at Canterbury（ケント大学カンタベリー校）がホストしています。
  
  
• FDR2 (Failures-Divergence Refinement)は、CSPベースのプログラムに対する商用のモデル・チェック・ツールの中の１つです。
  
  
• CSPの実装には、他の言語用のものもあります。C++CSPはC++用の実装であり、J#.Netは .Net用の実装です。
  
  
• Occam-piは、occam 言語としてのCSPの概念を、pi-calculusの機動性で拡張することを目的とした言語プラットフォームです。この最先端研究に関して、occam-pi homepageで学んでください。
  
  
• そのサイトを訪ねたら、occamコンパイラーに関する様々な拡張についても調べてください。
  
  
• developerWorksのJava technologyゾーンには、Javaプログラミングのあらゆる面を網羅した豊富な記事が用意されています。
  
  
• Developer Bookstoreには、Java関連の技術書をはじめ、広範な話題を網羅した書籍が取り揃えられていますので、ぜひご利用ください。
  
  
• Java technologyゾーンのチュートリアル・ページには、developerWorksで提供する、Javaに焦点を当てた無料チュートリアルのリストがありますので、ぜひご覧ください。
  
  

著者について

この記事を評価する

コメント

上に戻る

目次

• CSPの基礎
• JCSPライブラリー
• JSCPでのプロセス
• JCSPでのチャネル
• チャネルの実際の動作
• Parallel構成体
• Parallelの例
• ネットワーク内部でのネットワーク
• JSCPでの決定論
• JCSPでの非決定論
• ガードの実際
• ScaleIntプロセス
• 第2回のまとめ
• 参考文献
• 著者について
• コメント