こんにちは。NaCl松江本社のyharaです。最近BiwaSchemeにdelay-forceという機能を追加しました。 BiwaSchemeは筆者の個人的なプロジェクトで、JavaScriptで書かれたScheme処理系です。 delay-forceはSchemeの最新規格であるR7RS (small language)から標準として取り込まれた機能で、delayおよびforceと併用してScheme上で遅延評価を実現するためのものですが、それがどういうものなのかについては順を追って見ていく必要があるでしょう。
本記事では以下の順で説明を行います。
手短にいうと、(delay-force expr)はおおよそ(delay (force expr))と同じですが、より効率よく実行されます。具体的な例は以下で見ていきますが、末尾再帰のアルゴリズムを実装した場合でも、delay-forceなしではメモリを際限なく消費するような例があります。
まず最初に、delayとforceについて確認しておきましょう。
delayはある式の評価を遅延させるための構文で、式をラップしたオブジェクト、「プロミス(promise)」を返します。プロミスをforceに渡すと、ラップされている式の評価が行われます。
以下の例では、(+ 1 2)という式の評価を遅延させています。delayでプロミスを作った時点では何も起こりませんが、それをforceで実行すると、1 + 2という計算が行われます。
(define p (delay (+ 1 2)))
(display (force p)) ;=> 3これだけだと、delayとforceはlambdaによる無名関数の生成と、その実行で置き換えられそうな気もします。実際、delayとforceは以下のように置き換えてもだいたい同じように動きます。
(delay expr) => (lambda () expr)
(force p) => (p)しかしプロミスには一つ重要な特性があります。それは、計算結果がキャッシュされるということです。上の例では(+ 1 2)という式をラップしたプロミスpを作りましたが、この足し算は最初にforceしたときのみ実行され、それ以降は(force p)を何度実行しても、プロミス内部に保存された計算済みの値(3)が返ります。
(+ 1 2)程度ではあまり意味がありませんが、delayの引数がもっと時間のかかる計算だった場合は、この特性が効いてきます。
遅延評価を利用したプログラムの例として、遅延ストリームがあります。遅延ストリームは「データの列を表す」という点でリストに似ていますが、リストと違い、先頭以外の要素の計算は必要になるまで遅延されます。そのため、「ユーザがマウスを操作したイベントの列」のように終わりが分からないデータ列や、フィボナッチ数列のように無限に続くようなデータ列も表現することができます。
delayとforceを利用して遅延ストリームを実現する方法として、odd streamとeven streamという2種類の形式が知られています(参考：srfi-40)。まずはodd streamの方から見ていきましょう。
odd streamでは、ストリームを「先頭の要素」と「次を得る計算(のプロミス)」の組で表します。例えば、先頭が1,2,3から始まるストリームは下図のようになります。
(1 . #<Promise>)
| force
v
(2 . #<Promise>)
| force
v
(3 . #<Promise>)
|
v
...
この定義は単純ですが、一方で「ストリームを生成した時点で先頭の要素が評価されてしまう」という問題があります。上のような単純なデータ列なら問題ないですが、例えば「ネットワークから1バイトずつ読むストリーム」を使おうとした場合、ストリームを作った瞬間に最初の1バイトが読み込まれてしまいます。評価を遅延したいから遅延ストリームを使うのに、先頭の要素だけ遅延できないというのは少し奇妙ですね。
これを解決するために、組全体をプロミスで囲うようにしたのがeven streamです。先頭が1,2,3から始まるeven streamを図示したのが以下です。先ほどの図と似ていますが、最初の要素の計算もプロミスによって遅延されていることが分かります。
#<Promise (1 . #<Promise>)>
| force
v
(1 . #<Promise>)
| force
v
(2 . #<Promise>)
| force
v
(3 . #<Promise>)
| force
v
...
(ちなみにodd, evenという名前の由来が気になる人は、こちらの記事を参照してください。)
次に、これらのストリームを操作する関数について考えます。例えばodd streamに対するmap関数は以下のように定義できます(ソース全体は記事末を参照してください)。
(define (odd-stream-map f st)
(if (odd-stream-null? st)
odd-stream-null
(odd-stream-cons (f (odd-stream-car st))
(odd-stream-map f (odd-stream-cdr st)))))odd-stream-mapの定義は、map関数を自分で定義してみたことがあるのであればとても読みやすいと思います。実際、以下のmy-map関数と見比べてみると、関数名を取り替えただけで同じ構造になっていることが分かります。
(define (my-map f st)
(if (null? st)
'()
(cons (f (car st))
(my-map f (cdr st)))))ではeven streamに対するmap関数はどのように書けるでしょうか。実は、odd-stream-mapの関数名を置き換えて全体を(delay (forceで包むと、even-stream-mapができ上がります。odd-stream-mapはmy-mapと同じ構造だったので、「リストに対する処理のように関数を書いて、全体を(delay (forceで包むとeven stream用の関数ができる」と言い換えることもできます。
(define (even-stream-map f st)
(delay (force
(if (even-stream-null? st)
even-stream-null
(even-stream-cons (f (even-stream-car st))
(even-stream-map f (even-stream-cdr st)))))))delayは評価を遅延する構文で、forceはそれを展開する関数なので、(delay (forceはなんだか意味のない二度手間をかけているようにも見えます。しかしeven streamの定義を考えると、delay、forceのいずれもが必要なものであることが分かります。
まず、even-stream-mapが(delay (forceで包まれていなかったらどうなるでしょうか。この場合、(even-stream-map f strm1)のようにして新しいストリームを得ようとした瞬間に、if式が評価されてしまいます。ifの条件式で使っているeven-stream-null?はストリームをforceするので(記事末を見て下さい)、ストリームの変換をしようとすると先頭の要素が計算されてしまうことになり、これではeven streamの意義に反します。先頭の要素の計算をさせないためには、(delayが必要になります。
しかし単純に(delayで包むだけだと、「変換後のストリーム(プロミス)」をforceしたらまたプロミスが出てきてしまいます。even streamの定義では、ストリームであるプロミスをforceしたら要素とプロミスの組が出てこないといけません。このため、(delayだけでなく(forceが必要になります。
even streamは先頭の要素の生成まで遅延できるという利点がありますが、一方で一つ問題点があります。それは、例の(delay (forceの部分の実行効率が必ずしも良くないという点です。
極端な例として、(delay (forceだけを無限に繰り返すことを考えます。以下のloopという関数はプロミスを返します。それをforceすると、Scheme処理系によっては無事無限ループになるのですが(例：Gauche)、そのような動作が規格で保証されているわけではなく、処理系によっては「out of stack space」のようなエラーを出して止まってしまいます(例：Chibi Scheme)。
(define (loop)
(delay (force (loop))))
(define p (loop))
(force p)その理由は、プロミスは計算結果をキャッシュしなければならない、という仕様にあります。上のプログラムを実行すると、まずプロミスpが作られます。それをforceすると、loop関数が何度も呼ばれ、そのたびにdelayにより新しいプロミスが作られます。しかし最初のプロミスpの値は、(force (loop))全体が終了するまで確定しません。2個目、3個目のプロミスも同様です。このようにして「計算結果確定待ち」のプロミスが無限に溜まっていくというのがエラーの原因です。
delay-forceは、この問題を解決するための構文です。(delay (forceの部分をdelay-forceに差し替えると、Chibi Schemeでもちゃんと無限ループになります。
(define (loop)
(delay-force (loop)))
(define p (loop))
(force p)delay-forceは意味的には(delay (forceと同じですが、効率良く動くことが規格で保証されています。しかし、どのような仕組みでそうなるのでしょうか？
方針の概略を説明します。delay-forceの引数はプロミスであると決まっています。またdelay-forceの返り値もプロミスです。つまり以下のようなプログラムにおいて、p1もp2もプロミスです。
(define p2 (delay-force p1))ここで、p2をforceするとどうなるでしょうか。(delay-force p1)は(delay (force p1))と同じなので、p2が表す計算は(force p1)です。よって、プロミスp1の計算結果が、プロミスp2の計算結果になります。
言い換えると、プロミスp1の計算結果が出れば、それがそのままプロミスp2の計算結果にもなります。(delay (forceでループを実現した場合に問題だったのは、「計算結果確定待ち」のプロミスが溜まってしまうことでした。delay-forceにおいては、プロミスp1とp2の計算結果が同一であることがわかっているので、Scheme処理系はプロミスの内部構造を工夫して、p1とp2で「プロミスの中身」を共有することができます。これがdelay-forceの基本的なアイデアです。
delay-forceの実装方法については、srfi-45(日本語訳)に参照実装があります。これはdelay, forceの定義をSchemeレベルで上書きすることで、delay-forceに対応していない処理系にdelay-forceを実装するものです。
個人的にはChibi Schemeの実装の方が、処理は同じですが読解しやすかったです。ということで、本記事ではそれをもとにアレンジしたコードで説明をします。最初に全体を貼っておきます。
; 補助関数(ユーザには見せない)
(define (make-promise done? value-or-gen)
(list (cons done? value-or-gen)))
(define (promise-done? p) (car (car p)))
(define (promise-value p) (cdr (car p)))
(define (promise-gen p) (cdr (car p)))
(define (promise-merge! new old)
(set-car! (car old) (promise-done? new))
(set-cdr! (car old) (promise-value new))
(set-car! new (car old)))
; 構文定義
(define-syntax delay-force
(syntax-rules ()
((delay-force promise-exp) (make-promise #f (lambda () promise-exp)))))
(define-syntax delay
(syntax-rules ()
((delay exp) (delay-force (make-promise #t exp)))))
(define (force promise)
(if (promise-done? promise)
(promise-value promise)
(let ((new-promise ((promise-gen promise))))
(if (promise-done? promise)
(promise-value promise)
(begin
(promise-merge! new-promise promise))
(force promise)))))このコードでは、delayはdelay-forceを使って定義されています。delayとdelay-forceはいずれもプロミスを作る構文ですが、delay-forceの方が引数の範囲が狭い(プロミスを返す式しか受け取れない)ので、delayの引数expをmake-promiseでプロミス化することでdelay-forceに渡しています。
(define-syntax delay-force
(syntax-rules ()
((delay-force promise-exp) (make-promise #f (lambda () promise-exp)))))
(define-syntax delay
(syntax-rules ()
((delay exp) (delay-force (make-promise #t exp)))))ただし単純に引数をプロミス化してdelay-forceに渡すだけだと、式を(lambda () ...)で包む処理が二重に走ってしまいます。これを避けるため、プロミスがdelay産であることをフラグ(#t)として記録しています。
このコードでは、プロミスをSchemeのコンスセルで表現しています。ひとつ目の値が計算済み(done)かどうかを表します。計算済みの場合はふたつ目の値が計算結果です。未計算の場合は、ふたつ目の値が「プロミスを返す式」になっています。
; 計算済みのプロミス
((#t . 3))
; 未計算のプロミス
((#f . (lambda () promise-exp)))またこの図をよく見ると、単に2つのデータを保持するだけでなく、それをもう一段コンスセルで包んでいることがわかります(一番外側の括弧)。これは、前述のように2つのプロミスを合体させる(中身を共有させる)ためです。promise-merge!が2つのプロミスを合体させる関数です。
(define (promise-merge! new old)
(set-car! (car old) (promise-done? new))
(set-cdr! (car old) (promise-value new))
(set-car! new (car old)))次にforceの定義を見ていきます。まずforceしたいpromiseがdoneである場合は、キャッシュされている値を返すだけです。
(define (force promise)
(if (promise-done? promise)
(promise-value promise)promiseがdoneでない場合は、まず「プロミスを作る計算」を実行しnew-promiseを得ます。
(let ((new-promise ((promise-gen promise))))new-promiseを得たら、それをpromiseと合体させ、forceを続けます。
(begin
(promise-merge! new-promise promise))
(force promise)))))forceの挙動はだいたい上記の通りなのですが、あと2行だけ、説明していない箇所があります。それは以下の条件分岐です。
(if (promise-done? promise)
(promise-value promise)new-promiseを得たあと、promiseがdoneになったかを再度確認しています。これは、特殊な例において、pをforceしている最中にpがforceされるケースがあるからです。 srfi-45のreentrancy test3というのがそれですが、シンプルにしたものを以下に示します。
(define flag #f)
(define p (delay (if flag
1
(begin
(set! flag #t)
(force p)
2))))
(display (force p)) ;=> should be 1pの定義の中に(force p)があるため、(force p)の計算中に再度(force p)が実行されることになります。そして、これらの計算結果はそれぞれ2と1という違う値になります。こんなときは、pのプロミスとしての値はどちらになるべきでしょうか。
結果からいうと、先に計算が終わった方が優先されます。上の例ではpの値は1になります。
BiwaSchemeではこのコミットでdelay-forceを足しています。基本的に、上で解説したコードをそのままJavaScriptに直しただけです。 (実はその前にdelayとforceだけ先に実装したのですが、ここまで説明したような挙動を実現するためにdelay, forceも含めて書きなおしています。)
even streamを用いたズンドコキヨシ問題の実装です。このページにbiwascheme.jsが埋め込まれており、Reloadボタンを押すたびにプログラムが再実行されます。
;; odd stream
(define odd-stream-null '())
(define (odd-stream-null? st)
(null? st))
(define (odd-stream-car st)
(car st))
(define (odd-stream-cdr st)
(force (cdr st)))
; BiwaSchemeには今のところdefine-syntaxが無いのでdefine-macroで代用
(define-macro (odd-stream-cons item strm-expr)
`(cons ,item (delay ,strm-expr)))
(define (odd-stream-map f st)
(if (odd-stream-null? st)
odd-stream-null
(odd-stream-cons (f (odd-stream-car st))
(odd-stream-map f (odd-stream-cdr st)))))
(define (odd-stream-from n)
(odd-stream-cons n (odd-stream-from (+ n 1))))
(define (odd-stream-nth st n)
(if (= n 0)
(odd-stream-car st)
(odd-stream-nth (odd-stream-cdr st) (- n 1))))
;; even stream
(define even-stream-null (delay '()))
(define (even-stream-null? st)
(null? (force st)))
(define (even-stream-car st)
(car (force st)))
(define (even-stream-cdr st)
(cdr (force st)))
(define-macro (even-stream-cons item-expr strm-expr)
`(delay (cons ,item-expr ,strm-expr)))
(define (even-stream-map f st)
(delay-force
(if (even-stream-null? st)
even-stream-null
(even-stream-cons (f (even-stream-car st))
(even-stream-map f (even-stream-cdr st))))))
(define (even-stream-from n)
(even-stream-cons n (even-stream-from (+ n 1))))
(define (even-stream-nth st n)
(if (= n 0)
(even-stream-car st)
(even-stream-nth (even-stream-cdr st) (- n 1))))
;; demo
(display (even-stream-nth (even-stream-from 3) 10)); 補助関数(ユーザには見せない)
(define (make-promise done? value-or-gen)
(list (cons done? value-or-gen)))
(define (promise-done? p) (car (car p)))
(define (promise-value p) (cdr (car p)))
(define (promise-gen p) (cdr (car p)))
(define (promise-merge! new old)
(set-car! (car old) (promise-done? new))
(set-cdr! (car old) (promise-value new))
(set-car! new (car old)))
; 構文定義
(define-syntax delay-force
(syntax-rules ()
((delay-force promise-exp) (make-promise #f (lambda () promise-exp)))))
(define-syntax delay
(syntax-rules ()
((delay exp) (delay-force (make-promise #t exp)))))
(define (force promise)
(if (promise-done? promise)
(promise-value promise)
(let ((new-promise ((promise-gen promise))))
(if (promise-done? promise)
(promise-value promise)
(begin
(promise-merge! new-promise promise))
(force promise)))))
;; demo
(define loop
(delay-force loop))
(force loop)