Think Stitch
PRINCIPIA  最近の更新


並行プログラミング言語 7:逐次合成

並行プログラミング言語に逐次合成 seq を追加します。逐次合成があるとプロセスを再利用できるようになります。 この点については逐次合成の利用2例メモプロセスに SyncStitch を使った解説がありますので,よろしければご参照ください。

過去の記事は以下のとおりです。

尚,今回は話を簡単にするために,改名と隠蔽を入れる前の実装を土台にします。

逐次合成の構文

逐次合成の構文を以下のように定めます。順番に実行するプロセスを単純に並べるだけです。

(seq プロセス式 ...)

例えばこんな感じです。

(seq (! a SKIP) (! b SKIP))

プロセス式を1つも指定しない場合は SKIP と同じになります。つまり即座に終了します。

(seq) = SKIP

これは終了プロセス SKIP が seq の単位元だからです。実際 seq の中に直接出現する SKIP は振る舞いとしては何の効果もありません。遅くなりますけど。

(seq SKIP SKIP (! a SKIP) SKIP SKIP (! b SKIP) SKIP)
= (seq (! a SKIP) (! b SKIP))

プロセス式を1つだけ指定した場合は,seq を使わないものと同じです。

(seq P) = P

逐次合成に対する CSP カーネルのシステムコール

逐次合成に対応するため CSP の実行カーネルに専用のシステムコール csp-seq を用意します。

(csp-seq thunk-list)

thunk-listは seq の引数である各プロセスに対応する引数なしの手続きのリストです。

expand-process の拡張

プロセス定義マクロ define-process で使用するヘルパ expand-process の修正は1カ所だけです。 ご覧のとおり,システムコール csp-seq の呼び出しに単純に置き換えるだけです。

(define (expand-process pexpr)
  (match pexpr
    ...
    (('seq . process-list)
     `(csp-seq
       (list ,@(map (lambda (p)
                      `(lambda () ,(expand-process p)))
                    process-list))))
    ...))

par や rename と同様に expand-process-in-alt は修正しません。つまり alt の引数として直接に seq を書くことはできません。

CSP カーネルの修正と csp-seq の実装

逐次合成を実現する方法について概要を説明します。

  1. csp-seq が呼び出されたら thunk-list のうち,あとで実行する2番目以降の thunk を保存しておきます。そのためにプロセスコントロールブロック PCB に保存用のフィールド seq-stack を用意します。
  2. csp-skip が呼び出されたら,seq-stack を調べます。もし空ならばいままで通り終了します。空でなかったら先頭の thunk を取り出して実行を継続します。

なぜスタックかというと,seq がネストされたとき,あとから実行した csp-seq で指定された thunk が前に追加されるからです。取り出しは前からですから,したがって LIFO,つまりスタックになります。

では順に細かく見ていくことにします。

レコード process

プロセスコントロールブロック(PCB)に csp-seq で指定されたthunk-list を保存するためのフィールド seq-stack を追加します。

(define-record-type process make-process #t
  (state)         ; running ready wait-for-sync wait-for-children omega
  parent
  (children)      ; leaf: #f, node: list of child processes
  (sync)          ; leaf: event-cont-alist, node: sync-list
  (cont)
  (seq-stack))    ; ★ 追加

逐次合成システムコール csp-seq

システムコール csp-seq が基本的に行うことは,指定された thunk-list のうち2番目以降を PCB の seq-stack に保存し,実行を継続することです。

(define (csp-seq thunk-list)
  (if (null? thunk-list)
      (csp-skip) ; *1
      (begin
        (process-seq-stack-set!
         current-process
         (append (cdr thunk-list)
                 (process-seq-stack current-process))) ; *2
        (process-cont-set!
         current-process
         (car thunk-list)) ; *3
        (set! ready-queue (cons current-process ready-queue))  ; *4
        (sched))))  ;*5
  1. もし thunk-list が空の場合は csp-skip を呼び出して即座終了します。
  2. 空でなければ,thunk-list の2番目以降 (cdr) を seq-stack の前に連結します。

あとは実行を継続するだけなので thunk-list の先頭を呼び出すだけでもよいのですが,ちょっと遊んでみることにしました。システムコール csp-seq が呼び出されたときに,プロセススイッチが起こるようにしてみます。そのために thunk-list の先頭を継続フィールド cont にセットし(*3),一度 ready-queue に戻し(*4),次に実行するプロセスの選択をスケジューラに委ねます(*5)。この効果についてはあとで見ます。

csp-skip の修正

csp-skip を修正します。 おおまかにいうと,いままで終了としていたところ(*1)で seq-stack をチェックし,もし空でなかったら先頭のthunk を取り出して実行を継続します。

(define (csp-skip)
  (let loop ((p current-process))
    (if p
        (let ((children (process-children p)))
          (if (or (not children) (every omega? children))
              (let ((ss (process-seq-stack p))) ; *1
                (if (null? ss) ; *2
                    (begin
                      (process-state-set! p 'omega)
                      (loop (process-parent p)))
                    (begin
                      (process-seq-stack-set! p (cdr ss))     ; *3
                      (process-children-set! p #f)            ; *4
                      (process-sync-set! p '())               ; *5
                      (process-cont-set! p (car ss))          ; *6
                      (set! ready-queue (cons p ready-queue)) ; *7
                      (sched))))   ; *8
              (sched)))
        ;; root process is terminated
        'done)))

いままでは,プロセス木の葉にあたるプロセスが csp-skip を呼び出した場合と,そこから波及してすべての子プロセスが終了した親プロセスの場合はそのまま終了していました。今回はここで seq-stack を調べます(*1)。 もし空だったらいままで通りに終了します(*2)。 空でなかったら先頭の thunk を取り出して実行を継続することになります(*3)。

ここでちょっと注意が必要です。実行を継続するプロセスは par で子プロセスを生成していた親プロセスだった可能性があるので,children が #f でなく sync にはイベントの同期リストが入っているかもしれません。そこで葉のプロセスとして改めて実行するためには,この2つのフィールドを再初期化する必要があります(*4, *5)。

続いて継続に制御を移すのですが,csp-seq のときと同じように,ここでもスケジューラを挟むことにしました(*6, *7, *8)。

修正は以上です。

実行例

実際に動かしてみます。

単純な例

(define-process P
  (seq (! a SKIP) (! b SKIP)))
(start-process P)
(map de (reverse trace))
=> (a b)

SKIP があっても遅くなるだけで,イベント発生という意味での振る舞いには影響しません。

(define-process P
  (seq SKIP SKIP (! a SKIP) SKIP SKIP (! b SKIP) SKIP))
(start-process P)
(map de (reverse trace))
=> (a b)

もちろんプロセスが3つ以上あってもかまいません。

(define-process P
  (seq (! a SKIP) (! b SKIP) (! c SKIP)))
(start-process P)
(map de (reverse trace))
=> (a b c)

入れ子 の seq

seq-stack はスタックになっているので,seq を入れ子にすることができます。

(define-process P
  (seq (! a SKIP)
       (seq (! b SKIP) (! c SKIP))
       (! d SKIP)))
(start-process P)
(map de (reverse trace))
=> (a b c d)

並行プロセスの合流

seq を使うと par で子プロセスを作ったあと,再び合流することができます。いわゆる join でしょうか。

(define-process P
  (seq
    (par '()
      (! a SKIP)
      (! b SKIP))
    (! c SKIP)))
(start-process P)
(map de (reverse trace))
=> (a b c) or (b a c)

ランダムスケジューラを使っている場合,a と b の発生順序は非決定的になります。

システムコール csp-seq でのプロセススイッチ

csp-seq でスケジューラを呼び出すようにした効果を見てみます。

(define-process P
  (par '()
    (seq (! a SKIP) (! b SKIP))
    (! c SKIP)))
(start-process P)
(map de (reverse trace))
=> (a b c) or (c a b) or (a c b)

seq でプロセスが (! a SKIP) から (! b SKIP) に移行する際に,プロセスが切り替わる可能性があります。 ランダムスケジューラの元で何度も実行してみると,(a c b) というトレースが発生することが確認できます。

プロセスの再利用

逐次合成の利用2例で説明したプロセス再利用の例を見てみます。繰り返し行われる仕事を実行するプロセス Q を別に定義して,主であるプロセス P から seq を使って複数回呼び出すことができます。

(define-process Q
  (! c SKIP))

(define-process P
  (seq
    (! a SKIP)
    Q
    (! b SKIP)
    Q))
(start-process P)
(map de (reverse trace))
=> (a c b c)

分岐と合流

先ほどは並行プロセスが合流する例を見ましたが,今度は if による分岐の合流です。これも 逐次合成の利用2例で説明したものです。 独立したプロセスを定義するほどではないけれどもコピーするのもいや,というときに seq を使うことができます。

ここでは (! c SKIP) の部分が1度書くだけで済んでいる部分です。

(define-process (Q x)
  (seq
    (if (= x 0)
        (! a SKIP)
        (! b SKIP))
    (! c SKIP)))

(define-process P
  (seq (Q 0) (Q 1)))
(start-process P)
(map de (reverse trace))
=> (a c b c)

メモプロセス

メモプロセスの例です。 プロセス SQUARE は渡された値の2乗を計算するプロセスです。このプロセスは逐次合成によって利用されることを想定しています。そのために結果を MEMO プロセスを通じて呼び出し元プロセスに渡します。

(define N 100)
(define D (iota N))
(define-channel rd D)
(define-channel wr D)
(define-channel out D)
(define-event terminate)

(define-process (MEMO m)
  (alt
    (! (rd m) (MEMO m))
    (? wr (x) (MEMO x))
    (! terminate SKIP)))

(define-process (SQUARE x)
  (if (< (* x x) N)
      (! (wr (* x x)) SKIP)
      STOP))

(define-process P
  (par (list rd wr terminate)
    (MEMO 0)
    (seq
      (SQUARE 3)
      (? rd (x)
         (seq
           (SQUARE 4)
           (? rd (y)
              (! terminate
                 (! (out (sqrt (+ x y))) SKIP))))))))
(start-process P)
(map de (reverse trace))
=> ((wr 9) (rd 9) (wr 16) (rd 16) terminate (out 5))
2014/05/10
© 2013,2014,2015 PRINCIPIA Limited