Think Stitch
PRINCIPIA  最近の更新


並行プログラミング言語 2:チャネル通信とプロセスパラメータ

以前に作った並行プログラミング言語を拡張してみたいと思います。拡張するポイントは次の3点です。

チャネル通信の原理

チャネルというオブジェクトがあって,これを共有するプロセスの間で通信を行うことができると考えます。一方のプロセスがチャネルに値を送信すると,もう一方のプロセスはその値をチャネルから受信することができます。値の授受は同期的に行われるものとします。

チャネルを通して渡すことのできる値の範囲はあらかじめ決まっているとします。この値の集合をチャネルの定義域と呼ぶことにします。

原理的には,チャネル通信はイベント同期とまったく同じものと考えることができます。例えばチャネル ch の定義域が {0, 1, 2} だったとします。このとき,チャネル ch から値を受信するとは,各値に対応するイベントがあって,それらイベントうち,どれが発生するのかを選択として待つのと同じことです。より具体的にいうと,各値 0, 1, 2 に対応するイベント che0, che1, che2 というようなものがあって,次のプロセス式を実行するのと同じことです。

(alt
  (! che0 P0)
  (! che1 P1)
  (! che2 P2))

同様に,チャネル ch を通して値 2 を送信するというのは,単に対応するイベント che2 で同期するということと同じです。そのように解釈することができるということです。

このように,各チャネルには,定義域の値それぞれに対応するイベントの集合が付随していると考えます。この集合の要素をチャネルイベントと呼ぶことにします。

この考え方を使ってチャネル通信を実現することにします。

プロセスパラメータ

チャネルを通して値を受信できるようになっても,値を保持しておく方法がなければあまり有意義な仕事はできません。そこでプロセスが変数を持てるようにします。これをプロセスパラメータと呼ぶことにします。

いままでプロセスはシンボルで表してきました。これを手続きと同じように引数をとる形に拡張します。

(P a ...)

プロセス定義

プロセスパラメータは define-process で指定できるようにします。これも手続きと同様です。

(define-process (プロセス名 プロセスパラメータ ...) プロセス式)

パラメータを持たないプロセスについては,いままでどおりシンボル単体で指定できるようにします。これは互換性を保つためで,深い意味はありません。

プロセス式の拡張と変更

構文を3つ追加します。* のついた部分です。

プロセス式 ::=
      SKIP
    | (! イベント プロセス式)
*   | (? チャネル (パラメータ) プロセス式)
    | (alt プロセス式 ...)
    | (par イベントリスト プロセス式1 プロセス式2 プロセス式3 ...)
    | プロセス名
*   | (プロセス名 引数 ...)
*   | (if 条件式 プロセス式1 プロセス式2)

1つ目はチャネルからの受信を行うための構文です。受信した値に束縛されるパラメータをシンボルで指定します。

2つ目はすでに説明したもので,引数をとるプロセスのための構文です。

3つ目は条件文です。条件式を評価し,結果が真であればプロセス式1で示されるプロセスに移行します。偽であればプロセス式2で示されるプロセスに移行します。プロセス式2を省略することはできません。

チャネルへの送信は,イベント同期の構文においてチャネルイベントを指定することによって表すことにします。チャネルイベントを手に入れる方法はつぎのように設計することにします。まずチャネルを表すオブジェクトは関数(クロージャ)であるとします。どういう関数かというと,送信値,つまり定義域の値を受け取ると,それに対応するチャネルイベントを返す関数です。こうすると,例えばチャネルを ch するとき,値 2 に対応するチャネルイベントは (ch 2) と表すことができます。したがってチャネル送信の構文は次のようになります。

(! (ch x) P)

以上に加えて,2点,意味を変更します。

1つはイベント同期の構文で,イベントの部分を評価することにします。これはチャネルイベントを得るためです。代わりにイベントを単なるシンボルからオブジェクトに変更します。これについては後述します。

もう1つは並行合成 par のイベントリストも評価するようにします。これもチャネルイベントを含めることができるようにするためです。加えて,リストにはチャネルを含めることができるようにします。チャネルが指定された場合には,それに付随するすべてのチャネルイベントが指定されたと見なすことにします。こうすると同期の指定がかんたんになるからです。

チャネル定義

チャネルは define-channel で定義することとします。構文は次のとおりです。

(define-channel チャネル名 定義域)

チャネル名にはシンボルを指定します。このシンボルがチャネルを表すクロージャに束縛されます。定義域にはチャネルの定義域に含まれるすべての値をリストとして指定します。定義域は評価されます。

イベント定義

チャネル導入によってイベント同期構文のイベント部が評価されることになったので,イベントにも定義構文を用意することにします。

(define-event イベント名)

イベント名にはシンボルを指定します。このシンボルがイベントオブジェクトに束縛されます。

define-event で定義したイベントをチャネルイベントと区別するときは,アトミックイベントと呼ぶことにします。

続いて実装の話に入ります。

チャネルイベント

チャネルイベントはレコードで表すことにします。必要な要素は,チャネルイベントが属するチャネルと,対応する定義域の中の値です。これに加えて,識別を容易にするために,チャネルの名前を表すシンボルを入れておくことにします。

(define-record-type channel-event #t #t name (ch) arg)

チャネルとチャネルイベント集合との対応

チャネル受信の構文を処理するときに,チャネルに対応するチャネルイベントの集合を求める必要が出てきます。そこで,チャネルイベントの集合をリストで表し,チャネルとの対応を表で管理することにします。ここでは簡単に連想リストを使います。

(define ch->chev-list '())

指定されたチャネルから対応するチャネルイベントのリストを取得する関数です。

(define (lookup-chev-list ch)
  (let ((p (assq ch ch->chev-list)))
    (if p
        (cdr p)
        (error "unknown channel" ch))))

チャネルイベントのリストを登録する関数です。

(define (register-chev-list ch chev-list)
  (set! ch->chev-list
        (cons (cons ch chev-list)
              ch->chev-list)))

チャネル定義

チャネル定義のマクロ define-channel を定義します。 基本的には与えられた値に対応するチャネルイベントを返す関数定義に展開するのですが,ちょっと込み入っています。 込み入っている理由は,返すチャネルイベントの eq? 同値を保証するためです。こうするとカーネルでの同期判定が簡単に効率よくできるからです。定義域の値は equal? で比較することにします。

(define-macro (define-channel ch domain-expr)
  `(define ,ch
     (let ((domain ,domain-expr)
           (chev-list #f))
       (let ((ch
              (lambda (x)
                (let ((chev
                       (find (lambda (chev)
                               (equal? x (channel-event-arg chev)))
                             chev-list)))
                  (if chev
                      chev
                      (error "out of domain" ',ch x))))))
         (set! chev-list
               (map (lambda (arg)
                      (make-channel-event ',ch ch arg))
                    domain))
         (register-chev-list ch chev-list)
         ch))))

イベント定義

イベント定義のマクロ define-event を定義します。イベントは前回同様シンボルで表すことにします。

(define-macro (define-event e)
  `(define ,e ',e))

プロセス定義

プロセスパラメータを持つプロセスの定義を可能にします。パラメータを持たないプロセスは,いままでどおりシンボル単体で指定できるようにしておきます。

(define-macro (define-process pspec pexpr)
  (if (symbol? pspec)
      `(define (,pspec) ,(expand-process pexpr))
      `(define ,pspec ,(expand-process pexpr))))

プロセスの展開

プロセスの展開にあたっての大きな変更は,csp-alt に渡すクロージャを,引数のない thunk から引数を1つとるクロージャに変えることです。この引数は同期したイベントを表します。カーネルがプロセスの継続に対してイベントを渡してくれるということです。これによって受信した値がわかるようになります。イベント同期の場合も対称性のためにパラメータを追加しますが,値は使いません。

チャネル受信の場合は,指定されたチャネルに対応するチャネルイベントのリストを取得し,それらすべてについての選択を csp-alt に渡します。継続には同期したチャネルイベントが渡ってくるので,そこから引数を取り出して受信パラメータに束縛します。

if 式については,見てのとおり単純に変換するだけで済みます。

イベント同期とイベント部分と,並行合成のイベントリストの部分については,評価するために quote を外してあります。加えて並行合成では関数 make-sync-list を呼び出しています。これはイベントリストに含まれるチャネルを,対応するチャネルイベントのリストに変換するためのものです。

match の最後の節は,引数をとるプロセスのためのものです。

(define (expand-process pexpr)
  (match pexpr
    ('STOP '(csp-alt '()))
    ('SKIP '(csp-skip))
    (('! event pexpr2)
     (let ((e (gensym)))
       `(csp-alt
         (list (cons ,event
                     (lambda (,e)
                       ,(expand-process pexpr2)))))))
    (('? ch (z) pexpr2)
     (let ((e (gensym)))
       `(let ((proc
               (lambda (,e)
                 (let ((,z (channel-event-arg ,e)))
                   ,(expand-process pexpr2)))))
          (csp-alt
           (map (lambda (chev) (cons chev proc))
                (lookup-chev-list ,ch))))))
    (('alt . process-list)
     `(csp-alt
       (append ,@(map (lambda (p) (expand-process-in-alt p))
                      process-list))))
    (('par sync-list . process-list)
     `(csp-par (make-sync-list ,sync-list)
               (list ,@(map (lambda (p)
                              `(lambda () ,(expand-process p)))
                            process-list))))
    (('if test pexpr1 pexpr2)
     `(if ,test
          ,(expand-process pexpr1)
          ,(expand-process pexpr2)))
    ((? symbol?) `(,pexpr))
    (_ pexpr)))

選択 alt の引数となるプロセスについても同様の拡張と変更を行います。

(define (expand-process-in-alt pexpr)
  (match pexpr
    ('STOP '())
    (('! event pexpr2)
     (let ((e (gensym)))
       `(list (cons ,event (lambda (,e) ,(expand-process pexpr2))))))
    (('? ch (z) pexpr2)
     (let ((e (gensym)))
       `(let ((proc
               (lambda (,e)
                 (let ((,z (channel-event-arg ,e)))
                   ,(expand-process pexpr2)))))
          (map (lambda (chev) (cons chev proc))
               (lookup-chev-list ,ch)))))
    (('alt . process-list)
     `(append ,@(map (lambda (p) (expand-process-in-alt p))
                     process-list)))))

関数 make-sync-list を定義します。チャネルの識別はやや手抜きです。

(define (make-sync-list syncobj-list)
  (append-map
   (lambda (so)
     (if (procedure? so)
         (lookup-chev-list so)
         (list so)))
   syncobj-list))

カーネルの修正

カーネルの修正はわずか2カ所だけです。csp-alt に渡されたプロセスの継続がイベントを受け取るようになったので,プロセスコントロールブロックの継続スロットにセットするときに,thunk に変換するだけです。:)

(define (csp-alt sync-list)
  (let loop ((xs sync-list))
    (if (pair? xs)
        (let ((event (caar xs))
              (proc (cdar xs)))
          (let ((ps (find-signaling-processes event)))
            (if ps
                (begin
                  (set! trace (cons event trace))
                  (release-processes event ps)
                  (process-state-set! current-process 'ready)
                  (process-cont-set! current-process
                                     (lambda () (proc event)))
                  (set! ready-queue
                        (cons current-process ready-queue))
                  (sched))
                (loop (cdr xs)))))
        (begin
          (process-state-set! current-process 'wait-for-sync)
          (process-sync-set! current-process sync-list)
          (sched)))))
(define (release-process e p)
  (let ((proc (cdr (assq e (process-sync p)))))
    (process-cont-set! p (lambda () (proc e)))
    (process-state-set! p 'ready)
    (process-sync-set! p '())
    (set! ready-queue (cons p ready-queue))))

ユーティリティ

トレースに含まれるチャネルイベントを見やすくするために,次の関数を定義しておきます。チャネルイベントを,チャネル名と値のリストに変換する関数です。

(define (de e)
  (if (channel-event? e)
      (list (channel-event-name e)
            (channel-event-arg e))
      e))

実行例1: プロセスパラメータの使用

まずプロセスパラメータだけを使ってみます。プロセス P はパラメータ n をもち,n 回だけイベント a を発生させたら終了します。

(define-event a)

(define-process (P n)
  (if (= n 0)
      SKIP
      (! a (P (- n 1)))))

関数 start-process は thunk だけを受け取れるので,変換してから渡します。

(begin
  (start-process (lambda () (P 7)))
  (reverse trace))

=> (a a a a a a a)

実行例2: チャネル送信

チャネル送信の例です。

(define-event done)
(define-channel ch '(0 1 2 3 4 5))

(define-process (P n)
  (if (= n 6)
      (! done SKIP)
      (! (ch n) (P (+ n 1)))))
(begin
  (start-process (lambda () (P 0)))
  (map de (reverse trace)))

=> ((ch 0) (ch 1) (ch 2) (ch 3) (ch 4) (ch 5) done)

実行例3: チャネル通信

上のプロセス P と通信をするプロセス Q を定義します。受け取った値に基づいて計算を行い,その結果を別のチャネル out に出力します。並行合成のイベントリストにはチャネル ch が指定できます。

(define-channel out '(0 1 2))

(define-process Q
  (alt
    (! done SKIP)
    (? ch (z)
       (! (out (mod z 3)) Q))))

(define-process R
  (par (list done ch) (P 0) Q))
(begin
  (start-process R)
  (map de (reverse trace)))

=> ((ch 0) (out 0) (ch 1) (out 1) (ch 2) (out 2)
    (ch 3) (out 0) (ch 4) (out 1) (ch 5) (out 2)
    done)
2014/01/11

追記

チャネル受信にガードをつけられるようにしてみました。

2014/01/18
© 2013,2014,2015 PRINCIPIA Limited