11 Objects and Concurrency

11.1ロック(lock)

ロックの目的はスレッド間で共有リソースへの排他アクセスを調停する事です。その様なメカニズムは典型的に、クリティカル領域への排他アクセスを制限する事により、より安全で頑健に出来ます。クリティカル領域への進入する際、ロックはセキュアになりスレッドはリソースへの排他アクセス権を与えられます、そして実行が領域を出る時、通常か例外発生によってかに関わらず、ロックは解放されます。同じロックを取得しようとする並行な試みは、ロックを保持している現在のスレッドがそれを解放するまでブロックされます。

11.1.1シンプルロック(simple lock)

シンプルロックの場合、ロックにより保護されたクリティカルセクションの動的スコープの間の同じスレッドによるネストした同じロックの要求の試みは、ブロックされるでしょう。これを再入(reentrancy)がサポートされていないと呼びます。シンプルロックは Oz では以下の様にモデル化されます、ここで Code はクリティカルセクションで行われる計算をカプセル化した引数無しの手続きです。ロックは手続きとして表現されます: 同じコードに適用される時、それは Old が unit に束縛されるまで待つ事によりロックを取得しようと試みます。通常終了と異常終了のどちらでもロックが解放される事に注意して下さい。

proc {NewSimpleLock ?Lock}
   Cell = {NewCell unit}
in 
   proc {Lock Code}
      Old New in  
      try 
         {Exchange Cell Old New}
         {Wait Old} {Code}  
      finally New=unit end 
   end 
end

11.1.2オブジェクトの属性でのアトミックな交換

オブジェクトを使ってロックを実装する他の実装が下に示されます。構造体を使っている事に注目して下さい:

Old = lck := New

セルの Exchange 操作と似て、これはオブジェクトの属性のアトミックな交換です。

class SimpleLock 
   attr lck:unit 
   meth init skip end 
   meth 'lock'(Code)
     Old New in 
     try  
        Old = lck := New
        {Wait Old} {Code}
     finally New= unit end 
   end 
end

11.2スレッド再入ロック(Thread-Reentrant Lock)

Oz では、計算の単位はスレッドです。それゆえ、適切なロック機構はスレッドに排他アクセスの権利を与えるでしょう。上で表現された非再入のシンプルロック機構の結果は不適当です。スレッド再入ロックは同じスレッドにロック部分への再入を許します、つまり、動的に同じロックによって保護されたネストしたクリティカル領域に入れるという事です。その様はロックは一度に高々1スレッドに取得され得ます。同じロックを取得しようとする並行スレッドはキューされます。ロックが解放された時、ラインetcに最初に並んでいるスレッドに権利が与えられます。スレッド再入ロックは Oz では以下の様にモデル化され得ます:

class ReentrantLock from SimpleLock 
   attr Current:unit 
   meth 'lock'(Code)
      ThisThread = {Thread.this} in 
      if ThisThread == @Current then 
         {Code}
      else 
         proc {Code1}
            try 
               Current := ThisThread
               {Code}
            finally 
               Current := unit 
            end 
         end 
      in 
         SimpleLock, 'lock'(Code1)
      end 
   end 
end

スレッド再入ロックは Oz で構文と実装のサポートが与えられています。それらはチャンクのサブ型として実装されています。Oz は保護されたクリティカル領域のために以下の構文を提供しています:

lock E then S end

E はロックに評価される式です。この構造体は S が実行されるまでブロックします。E がロックされていなければ、型エラーが発生します。

• {NewLock L} は新しいロック L を生成します。

• {IsLock E} は E がロックされている時に true を返します。

11.2.1配列(array)

Oz はチャンクのサブ型として配列を持っています。配列の操作はモジュール Array で定義されています。

• {NewArray +L +H +I ?A} は配列 A を作ります、ここで L はインデックスの下限で、H はインデックスの上限で、I は配列要素の初期値です。

• {Array.low +A ?L} はインデックスの下限を返します。

• {Array.high +A ?L} はインデックスは上限を返します。

• R:=A.I は R の A[I] を返します。

• A.I:=X は X にエントリ A[I] を割り当てます。

ロックの使用の簡単な図示として Figure 11.1 のプログラムを考えましょう。手続き Switch は配列の負の要素を正に変更し、0要素をアトム zero に変更します！手続き Zero は全ての要素を0にリセットします。

declare A L in 
A = {NewArray 1 100 ~5}
L = {NewLock}
proc {Switch A}
   {For {Array.low A} {Array.high A} 1
    proc {$ I}  
       X := A.I in 
       if X<0 then A.I := ~X
       elseif X == 0  then A.I := zero end 
       {Delay 100}
    end}
end 
proc {Zero A}
   {For {Array.low A} {Array.high A} 1
    proc {$ I} A.I := 0 {Delay 100} end}
end   

以下のプログラムを試してみて下さい。

local X Y in 
   thread {Zero A} X = unit end   
   thread {Switch A} Y = X end  
   {Wait Y}  
   {For 1 10 1 proc {$ I} {Browse A.I} end}  
end

配列の要素は 0 と zero が混ざったものになるでしょう。

私達は手続き Zero と Switch にアトミックでしかし任意の順番で振る舞って欲しいと想定しましょう。そのためには私達は以下の例での様にロックを使用できます。

local X Y in 
   thread  
      {Delay 100}  
      lock L then {Zero A} end   
      X = unit  
   end  
   thread  
      lock L then {Switch A} end  
      Y = X  
   end  
   {Wait Y}  
   {For 1 10 1 proc {$ I} {Browse  A.I} end}  
end

上で最初と2番目のスレッド間で delay 文に変更する事によって、私達は配列の全ての要素が値 zero か 0 のどちらかを得る事を見る事になります。私達は混ざった値を持っていません。

注意:*** 複数のロックを使った複数のオブジェクトにおいてのアトミックなトランザクションの例を書いて下さい。

11.3オブジェクトのロック

オブジェクトでの相互排他を保証するためには、前の節で記述されたロックを使うかもしれません。代替として、クラスの中で、オブジェクトが生成された時に存在するデフォルトのロックによってインスタンスオブジェクトをロック出来ます。暗黙のロックをともなったクラスは以下の様に宣言されます:

class C from .... 
   prop locking
  .... 
end

これはオブジェクトのメソッドの一つが呼び出された時に、自動的にオブジェクトをロックします。代わりに構造体を使わなければいけません:

lock S end

S が実行される時には内部のどのメソッドも排他アクセスである事を保証します。私達のロックが再入可能である事を思い出して下さい。これは次の事を暗に言っています:

• もし私達が、私達が構築し、各メソッド本体部を lock ... end で閉じたオブジェクトを全て取り、

• 私達のプログラムを1スレッドでしか実行しないのであれば、

• プログラムは以前の様に正確に振る舞うでしょう

もちろん、複数のオブジェクトのメソッドを複数のスレッドで呼び出すなら、そこに循環的依存があれば私達はデッドロックに出会うでしょう。自明でない並行プログラムを書く事はスレッド間の依存パターンの注意深い理解を要します。その様なプログラムにおけるデッドロックはロックが使われているかどうかに関わらず発生します。循環的コミュニケーションパターンを持つ事はデッドロックの発生のために有害です。

Figure 10.3 のプログラムは以下の様に洗練させる事で並行環境でも働く事が出来るように洗練させる事が出来ます:

class CCounter from Counter 
   prop locking
   meth inc(Value)
      lock Counter,inc(Value) end 
   end 
   meth init(Value)
      lock Counter,init(Value) end 
   end 
end

さあ、スレッドがオブジェクトでただアトミックなトランザクションを行うだけでなく、オブジェクトを通して同期を行うような、数々の興味深い例について学びましょう。

11.4並行FIFOチャネル

最初の例は任意の数のスレッド間で共有される並行チャネルを示します。どの生産者スレッドも情報をチャネルに非同期にputする事が出来ます。消費者スレッドはチャネルに情報が表れるまで待たなければいけません。スレッドの待機は公平に提供されます。Figure 11.2 は現実化としてありえるうちの一つを示しますこのプログラムは望まれる同期を達成するために論理変数の使用に頼っています。メソッド put/1 は要素をチャネルに挿入します。メソッド get/1 を実行するスレッドはチャネルに要素が挿入されるまで待ちます。複数の消費者スレッドはチャネルに自らの場所を予約し、それにより公平さを達成します。{Wait I} が排他領域の外側で行われる事に注目して下さい。待機が lock ... end の内側で行われると、プログラムはデッドロックするでしょう。それで、ルールの概要は次の様になります:

• 起床動作(waking-up action)が同じロックを必要とする時は、排他領域の内側で待機してはいけない。

class Channel from BaseObject 
   prop locking
   attr f r
   meth init 
      X in f := X r := X
   end 
   meth put(I)
      X in lock @r=I|X r:=X end 
   end 
   meth get(?I)
      X in lock @f=I|X f:=X end {Wait I}
   end 
end

11.5モニタ(monitor)

次の例はモニタ(monitor)を書く伝統的な方法を示します。私達はイベントの記法と Channel を特化したモニタ操作 notify(Event) と wait(Event) を定義するクラスの定義を開始します。

class Event from Channel 
   meth wait 
      Channel , get(_)
   end 
   meth notify 
      Channel , put(unit)
   end 
end

私達はここで伝統的なモニタスタイルでのunitバッファの例を示しています。unitバッファは消費者に来る時のチャネルにとても似た方法で振る舞います。各消費者はバッファが一杯になるまで待機します。生産者の場合は一つのみがアイテムを空のバッファに挿入する事が許されています。他の生産者はアイテムが消費されるまで一時停止しなければなりません。Figure 11.3 のプログラムは単一バッファモニタを示しています。ここで私達は生産者と消費者のためにシグナルのメカニズムをプログラムしなければなりません。put/1 と get/1 メソッドでのパターンを観察して下さい。ほとんどの実行は排他領域で行われます。待機が必要ならそれは排他領域の外側で行われます。これは yes への束縛を得る付属の変数 X を使って行われます。get/1 メソッドは一つの生産者に empty フラグをセットする時に知らせ、一つの生産者(もしあれば)に知らせます。これはアトミックなステップで行われます。put/1 メソッドは相互アクションを行います。

class UnitBufferM 
   attr item empty psignal csignal
   prop locking
   meth init 
      empty := true 
      psignal := {New Event init}
      csignal := {New Event init}
   end 
   meth put(I)
      X in 
      lock 
         if @empty then 
            item := I
            empty := false 
            X = yes
            {@csignal notify}
         else X = no end 
      end 
      if X == no then 
         {@psignal wait}
         {self put(I)}
      end 
   end 
   meth get(I)
      X in 
      lock 
         if {Not @empty} then 
            I = @item
            empty := true 
            {@psignal notify}
            X = yes
         else X = no end 
      end 
      if X == no then 
         {@csignal wait}
         {self get(I)}
      end 
   end 
end

上の例を以下のコードを走らせて試してみて下さい:

local 
  UB = {New UnitBufferM init} in 
  {For 1 15 1
   proc{$ I} thread {UB put(I)} {Delay 500} end end}
  {For 1 15 1
   proc{$ I} thread {UB get({Browse}}{Delay 500} end end}
end

11.5.1Oz スタイルの有界バッファ(bounded buffer)

Oz では、上で示されたモニタスタイルのプログラムを書くのはとても稀な事です。通常、それは不格好です。伝統的ではない UnitBuffer クラスを書くためのより簡単な方法があります。これはオブジェクトと論理変数の組み合わせによるもので、Figure 11.4 に簡単な定義を示しています。直接的なロックは不要です。

class UnitBuffer from BaseObject 
   attr prodq buffer
   meth init 
      buffer := {New Channel init}
      prodq := {New Event init}
      {@prodq notify}
   end 
   meth put(I)
      {@prodq wait}
      {@buffer put(I)}
   end 
   meth get(?I)
      {@buffer get(I)}
      {@prodq notify}
   end 
end

上のプログラムのシンプルな一般化は任意のサイズの有界バッファクラスに導きます。これは下で示されます。put と get メソッドは前と同じです。初期化メソッドだけが変更されています。

class BoundedBuffer from UnitBuffer 
   attr prodq buffer
   meth init(N)
      buffer := {New Channel init}
      prodq := {New Event init}
      {For 1 N 1 proc {$ _} {@prodq notify} end}
   end 
end      

11.6動的オブジェクト(active object)

動的オブジェクトはクラス定義によって記述される振る舞いを持つスレッドです。動的オブジェクトとのコミュニケーションは非同期メッセージパッシングを通して行われます。動的オブジェクトは受け取ったメッセージに関連するクラスの対応するメソッドを実行する事によって反応します。動的オブジェクトは一度に一つのメソッドを実行します。それゆえ動的オブジェクトによって実行されるメソッドのためにロックは不要です。動的オブジェクトへのインターフェースは Oz のポートを通したものです。動的オブジェクトのクライアントは関連するポートにメッセージを送る事により、オブジェクトにメッセージを送ります。私達は一般的にこの抽象をどの様に生成するかを示します。動的オブジェクトはネットワークを通してクライアントからメッセージを受け取るサーバに似ているので、私達はこの抽象をサーバ抽象と呼びます。クラス Class からサーバ S を生成するために次を実行します:

S = {NewServer Class init}

ここで init は初期オブジェクト構築メソッドです。基本的な概念を得るために、最初に NewServer 関数の簡単な形式を示します。以下の関数:

• ポート生成 Port,

• オブジェクト生成 Object, そして最後に

• 対応するクラスのメソッドを適用する事による、ポートにメッセージを送るサーバのスレッド生成。

fun {NewServer Class Init}
   S    % ポートのストリーム
   Port = {NewPort S}
   Object = {New Class Init}
in 
   thread {ForAll S  
           proc{$ M} {Object M} end}
   end 
   Port
end

私達は Class の中のメソッドにアクセス可能なプロテクテッドメソッド Close を作ることによってスレッド終了の能力を追加したいと思います。これは私達を上の関数の以下の拡張に導きます。私達は受け取りループの外側にジャンプするために例外制御機構を使います。

local 
   class Server 
     attr close:Close
     meth Close raise closeException end end 
   end 
in 
fun {NewServer Class Init}
   S    % ポートのストリーム
   Port = {NewPort S}
   Object = {New class $ from Server Class end Init}
in 
   thread  
     try  {ForAll S  
           proc{$ M} {Object M} end}
     catch closeException then skip end 
   end 
   Port
end