今回は「継続渡しスタイル (Continuation Passing Style : CPS) 」という手法について説明します。Scheme には「継続」という他の言語 [*1] にはない強力な機能がありますが、使いこなすのはちょっと難しいといわれています。継続渡しスタイルはクロージャを使った汎用的な方法で、クロージャがあるプログラミング言語であれば、継続渡しスタイルでプログラムを作成することができます。
最初に継続について簡単に説明します。継続は「次に行われる計算」のことです。たとえば、次のプログラムを例に考えてみましょう。
リスト 1 : 逐次実行 let foo () = print_string "foo\n" let bar () = print_string "bar\n" let baz () = print_string "baz\n" let test () = foo (); bar (); baz ()
# test ();; foo bar baz - : unit = ()
関数 test は関数 foo, bar, baz を順番に呼び出します。foo の次に実行される処理は bar, baz の関数呼び出しです。この処理が foo を呼び出したあとの「継続」になります。同様に、bar のあとに実行されるのは baz の呼び出しで、この処理がこの時点での「継続」になります。また、baz を呼び出したあと、test の中では次に実行する処理はありませんが、test は関数呼び出しされているので、関数呼び出しから元に戻る処理が baz を呼び出したあとの「継続」になります。
このように、あるプログラムを実行しているとき、そのプログラムを終了するまでには「次に実行する処理 (計算) 」が必ず存在します。一般に、この処理 (計算) のことを「継続」といいます。Scheme の場合、次の計算を続行するための情報を取り出して、それを保存することができます。Scheme では、この保存した情報を「継続」といって、通常のデータ型と同様に取り扱うことができます。つまり、継続を変数に代入したり関数の引数に渡すことができるのです。継続を使うとプログラムの実行を途中で中断し、あとからそこに戻ってプログラムの実行を再開することができます。
一般のプログラミング言語では、Scheme のように継続を取り出して保存することはできません。そこで、継続 (次に行う処理) を関数 (クロージャ) で表して、それを引数に渡して実行することにします。これを「継続渡しスタイル (CPS) 」といいます。たとえば、次の例を見てください。
リスト 2 : 継続渡しスタイル let test_cps cont = foo (); bar (); cont ()
# test_cps baz;; foo bar baz - : unit = ()
関数 test_cps は foo, bar を呼び出したあと、引数 cont に渡された処理 (継続) を実行します。関数 baz を渡せば foo, bar, baz と表示されますし、他の処理を渡せばそれを実行することができます。
もう一つ簡単な例を示しましょう。継続に値を渡して処理を行うこともできます。
# let add_cps a b cont = cont (a + b);; val add_cps : int -> int -> (int -> 'a) -> 'a = <fun> # add_cps 1 2 (fun x -> x);; - : int = 3 # add_cps 1 2 (fun x -> print_int x);; 3- : unit = ()
関数 add_cps は引数 a と b を加算して、その結果を継続 cont に渡します。cont に fun x -> x を渡せば、計算結果を返すことができます。また、cont で print_int x を呼び出せば、計算結果を表示することができます。
CPS を使うと再帰呼び出しを末尾再帰に変換することができます。たとえば、階乗の計算を CPS でプログラムすると次のようになります。
リスト 3 : 階乗の計算 (CPS) let rec fact_cps n cont = if n = 0 then cont 1 else fact_cps (n - 1) (fun x -> cont (n * x))
引数 cont が継続を表します。n = 0 のときは、cont に階乗の値 1 を渡します。それ以外の場合は、階乗の計算を継続の処理にまかせて fact_cps を再帰呼び出します。ここで、fact_cps の呼び出しは末尾再帰になることに注意してください。
継続の処理 fun x -> cont (n * x) では、継続の引数 x と fact_cps の引数 n を掛け算して、その結果を cont に渡します。たとえば、fact_cps 4 (fun x -> x) の呼び出しを図に示すと、次のようになります。
fact 4 (fun x -> x)
=> 4 (fun x1 -> (fun x -> x) (4 * x1))
=> 3 (fun x2 -> (fun x1 -> (fun x -> x) (4 * x1)) (3 * x2))
=> 2 (fun x3 -> (fun x2 -> (fun x1 -> (fun x -> x) (4 * x1)) (3 * x2)) (2 * x3))
=> 1 (fun x4 -> (fun x3 -> (fun x2 -> (fun x1 -> (fun x -> x) (4 * x1)) (3 * x2))
(2 * x3)) (1 * x4))
=> 0 (fun x4 -> (fun x3 -> (fun x2 -> (fun x1 -> (fun x -> x) (4 * x1)) (3 * x2))
(2 * x3)) (1 * x4)) 1
継続の評価
(fun x4 -> (fun x3 -> (fun x2 -> (fun x1 -> (fun x -> x) (4 * x1)) (3 * x2))
(2 * x3)) (1 * x4)) 1
=> (fun x3 -> (fun x2 -> (fun x1 -> (fun x -> x) (4 * x1)) (3 * x2)) (2 * x3)) 1
=> (fun x2 -> (fun x1 -> (fun x -> x) (4 * x1)) (3 * x2)) 2
=> (fun x1 -> (fun x -> x) (4 * x1)) 6
=> (fun x -> x) 24
=> 24
図 1 : fact_cps の実行
このように、継続の中で階乗の式が組み立てられていきます。そして、n = 0 のとき継続 cont に引数 1 を渡して評価すると、今までに組み立てられた式が評価されて階乗の値を求めることができます。つまり、n の階乗を求めるとき、継続 fun x -> cont (n * x) の引数 x には n - 1 の階乗の値が渡されていくわけです。そして、最後に継続 fun x -> x に n の階乗の値が渡されるので、階乗の値を返すことができます。
それでは実際に実行してみましょう。
val fact_cps : int -> (int -> 'a) -> 'a = <fun>
# for i = 1 to 10 do (fact i (fun x -> print_int x)); print_newline () done;; 1 2 6 24 120 720 5040 40320 362880 3628800 - : unit = ()
次はフィボナッチ数列を求める関数を CPS で作りましょう。次のリストを見てください。
リスト 4 : フィボナッチ関数 (* 二重再帰 *) let rec fibo n = if n = 0 || n = 1 then 1 else fibo (n - 1) + fibo (n - 2) (* CPS *) let rec fibo_cps n cont = if n = 0 || n = 1 then cont 1 else fibo_cps (n - 1) (fun x -> fibo_cps (n - 2) (fun y -> cont (x + y)))
val fibo : int -> int = <fun> val fibo_cps : int -> (int -> 'a) -> 'a = <fun>
関数 fibo_cps は、引数 n が 0 または 1 のとき cont 1 を評価します。それ以外の場合は fibo_cps を再帰呼び出しします。fibo_cps (n - 1) が求まると、その値は継続の引数 x に渡されます。継続の中で、今度は fibo_cps (n - 2) の値を求めます。すると、その値は fibo_cps (n - 2) の継続の引数 y に渡されます。したがって、fibo_cps n の値は x + y で求めることができます。この値を fibo_cps n の継続 cont に渡せばいいわけです。
fibo_cps の実行を図に示すと、次のようになります。
f(5) ┬ f(4) ┬ f(3) ┬ f(2) ┬ f(1)
│ │ │ │
cont cont cont cont
│ │ │ └ f(0)
│ │ └ f(1)
│ └ f(2) ┬ f(1)
│ │
│ cont
│ └ f(0)
│
└ f(3) ┬ f(2) ┬ f(1)
│ │
cont cont
│ └ f(0)
└ f(1)
図 2 : fibo_cps の実行
cont は継続を表します。fibo_cps は末尾再帰になっているので、n - 1 の値を求めるために左から右へ処理が進みます。このとき、n - 2 の値を求める継続 cont が生成されていくことに注意してください。そして、f(1) の実行が終了すると継続が評価され、n - 2 の値が求められます。すると、2 番目の継続が評価されて n - 1 の値 x と n - 2 の値 y を加算して、その値を継続 cont に渡します。こうして、次々と継続が評価されてフィボナッチ関数の値を求めることができます。
それでは実際に実行してみましょう。
# for i = 1 to 10 do fibo i (fun x -> print_int x); print_newline () done;; 1 2 3 5 8 13 21 34 55 89 - : unit = ()
正常に動作していますね。
ところで、fibo_cps は末尾再帰になっていますが、関数の呼び出し回数は二重再帰の場合と同じです。したがって、実行速度は二重再帰の場合とほとんどかわりません。また、二重再帰の場合は関数呼び出しによりスタックが消費されますが、CPS の場合はクロージャが生成されるのでメモリ (ヒープ領域) が消費されます。このように、再帰呼び出しを CPS に変換したからといって、効率の良いプログラムになるとは限りません。ご注意くださいませ。
階乗やフィボナッチ関数の場合、CPS に変換するメリットはほとんどありませんが、場合によっては CPS に変換した方が簡単にプログラムできることもあります。たとえば、リストを平坦化する関数 flatten で、リストの要素に空リストが含まれていたら空リストを返すようにプログラムを修正することを考えてみましょう。次のリストを見てください。
リスト 5 : リストの平坦化 (間違い)
let rec flatten ls =
match ls with
[] -> []
| x::_ when x = [] -> []
| x::xs -> x @ flatten xs
val flatten : 'a list list -> 'a list = <fun>
関数 flatten は空リストを見つけたら空リストを返していますが、これでは正常に動作しません。実際に試してみると次のようになります。
# flatten [[1; 2]; [3; 4]; [5; 6]];; - : int list = [1; 2; 3; 4; 5; 6] # flatten [[1; 2]; [3; 4]; []; [5; 6]];; - : int list = [1; 2; 3; 4]
2 番目の例が空リストを含む場合です。この場合、空リストを返したいのですが、その前の要素を連結したリストを返しています。空リストを見つける前にリストの連結処理を行っているので、空リストを見つけたらその処理を廃棄しないといけないのです。
このような場合、CPS を使うと簡単です。次のリストを見てください。
リスト 6 : リストの平坦化 (CPS)
let rec flatten_cps ls cont =
match ls with
[] -> cont []
| x::_ when x = [] -> []
| x::xs -> flatten_cps xs (fun y -> cont (x @ y))
val flatten_cps : 'a list list -> ('a list -> 'b list) -> 'b list = <fun>
flatten を CPS に変換するのは簡単です。リストの先頭の要素 x と平坦化したリストの連結を継続で行うだけです。平坦化したリストは継続の引数 y に渡されるので、x @ y でリストを連結して、それを継続 cont に渡せばいいわけです。
引数のリストが空リストになったら継続 cont に空リストを渡して評価します。これで、リストの連結処理が行われます。もしも、途中で空リストを見つけた場合は、空リストをそのまま返します。この場合、継続 cont は評価されないので、リストの連結処理は行われず、空リストをそのまま返すことができます。
それでは実行してみましょう。
# flatten_cps [[1; 2]; [3; 4]; [5; 6]] (fun x -> x);; - : int list = [1; 2; 3; 4; 5; 6] # flatten_cps [[1; 2]; [3; 4]; []; [5; 6]] (fun x -> x);; - : int list = []
正常に動作していますね。
次は二分木を巡回するプログラムを CPS で作ってみましょう。最初に、拙作のページ ファンクタ で作成した二分木の構造と二分木を巡回する関数 iter を再掲します。二分木の詳細は プログラムリスト をお読みください。
リスト 7 : 二分木の巡回 (* 節の定義 *) type 'a tree = Nil | Node of 'a * 'a tree * 'a tree (* 巡回 *) let rec iter f = function Nil -> () | Node (x, left, right) -> iter f left; f x; iter f right
iter は二重再帰になっています。そこで、f x の評価と右部分木の巡回は継続で行うことにします。プログラムは次のようになります。
リスト 8 : 二分木の巡回 (CPS)
let rec iter_cps f node cont =
match node with
Nil -> cont ()
| Node(x, left, right) ->
iter_cps f left (fun () -> f x; iter_cps f right (fun () -> cont ()))
val iter_cps : ('a -> 'b) -> 'a tree -> (unit -> 'c) -> 'c = <fun>
iter_cps は副作用が目的なので、継続に値を渡す必要はありません。そこで、cont には unit を渡すことにします。左部分木をたどったら継続 cont を呼び出します。その中で f x を評価し、そのあと右部分木をたどります。このときの継続は cont () を評価するだけです。これで生成された継続を呼び出して、木を巡回することができます。
それでは実際に試してみましょう。
# let a = tree_of_list [5; 3; 1; 4; 7; 6; 8; 2; 9];; val a : int tree = Node (5, Node (3, Node (1, Nil, Node (2, Nil, Nil)), Node (4, Nil, Nil)), Node (7, Node (6, Nil, Nil), Node (8, Nil, Node (9, Nil, Nil)))) # iter (fun x -> print_int x) a;; 123456789- : unit = () # iter_cps (fun x -> print_int x) a (fun () -> ());; 123456789- : unit = ()
このように、iter_cps で二分木を通りがけ順で巡回することができます。
二分木の巡回を CPS に変換すると、遅延ストリームに対応するのも簡単です。次のリストを見てください。
リスト 9 : 巡回 (遅延ストリーム版)
let rec stream_of_tree node cont =
match node with
Nil -> cont ()
| Node (x, left, right) ->
stream_of_tree left
(fun () -> Cons (x, lazy (stream_of_tree right (fun () -> cont ()))))
val stream_of_tree : 'a tree -> (unit -> 'a stream) -> 'a stream = <fun>
stream_of_tree は二分木を巡回してその要素を順番に出力する遅延ストリームを生成します。iter_cps は継続の中で関数 f を呼び出しましたが、stream_of_tree は継続の中で遅延ストリーム Cons (x, <lazy>) を返します。そして、遅延オブジェクトの中で右部分木をたどり、その継続の中で cont () を呼び出します。
ここで継続 cont の型は unit -> 'a stream になることに注意してください。stream_of_tree を呼び出すときに渡す継続が一番最後に呼び出されるので、遅延ストリームの終端 Nils を返すように定義してください。
簡単な実行例を示しましょう。
# #use "lazy_stream.ml";; ... 省略 ... # #use "tree.ml";; ... 省略 ... # let a = tree_of_list [5; 3; 1; 4; 7; 6; 8; 2; 9];; val a : int tree = Node (5, Node (3, Node (1, Nil, Node (2, Nil, Nil)), Node (4, Nil, Nil)), Node (7, Node (6, Nil, Nil), Node (8, Nil, Node (9, Nil, Nil)))) # let s = stream_of_tree a (fun () -> Nils);; val s : int stream = Cons (1, <lazy>) # stream_take s 9;; - : int list = [1; 2; 3; 4; 5; 6; 7; 8; 9] # stream_take s 10;; Exception: Empty_stream.
stream_of_tree を使うと、2 つの二分木が等しいか判定する述語 isequal を簡単に作ることができます。二分木の要素がすべて等しい場合、isequal は true を返し、そうでなければ false を返すことにします。つまり、二分木を集合として扱うわけです。プログラムは次のようになります。
リスト 10 : 同値の判定
let isequal tree1 tree2 =
let rec isequal_sub s1 s2 =
match (s1, s2) with
(Nils, Nils) -> true
| (Cons (x, t1), Cons(y, t2)) when x = y -> isequal_sub (force t1) (force t2)
| _ -> false
in
isequal_sub (stream_of_tree tree1 (fun () -> Nils))
(stream_of_tree tree2 (fun () -> Nils))
実際の処理は局所関数 isequal_sub で行います。isequal_sub には二分木の遅延ストリームを渡します。あとは、遅延ストリームから要素を一つずつ取り出して、それが等しいかチェックするだけです。
それでは実行例を示します。
# let a = tree_of_list [5; 2; 1; 3; 4; 8; 7; 9; 6];;
val a : int tree =
Node (5, Node (2, Node (1, Nil, Nil), Node (3, Nil, Node (4, Nil, Nil))),
Node (8, Node (7, Node (6, Nil, Nil), Nil), Node (9, Nil, Nil)))
# let b = tree_of_list [1; 2; 3; 4; 5; 6; 7; 8; 9];;
val b : int tree =
Node (1, Nil,
Node (2, Nil,
Node (3, Nil,
Node (4, Nil,
Node (5, Nil,
Node (6, Nil, Node (7, Nil, Node (8, Nil, Node (9, Nil, Nil)))))))))
# isequal a b;;
- : bool = true
# let c = tree_of_list [5; 2; 1; 3; 4; 8; 7; 10; 6];;
val c : int tree =
Node (5, Node (2, Node (1, Nil, Nil), Node (3, Nil, Node (4, Nil, Nil))),
Node (8, Node (7, Node (6, Nil, Nil), Nil), Node (10, Nil, Nil)))
# isequal a c;;
- : bool = false
変数 a, b に二分木をセットします。a と b では二分木の形状は異なりますが要素はすべて同じです。したがって、isequal a b は true を返します。変数 c にセットされた二分木は要素が一つだけ異なっているので、isequal a c は false を返します。
部分集合を判定する関数 issubset も簡単です。次のリストを見てください。
リスト 11 : 部分集合の判定
let issubset tree1 tree2 =
let rec issubset_sub s1 s2 =
match (s1, s2) with
(Nils, _) -> true
| (Cons (x, t1), Cons(y, t2)) ->
if x = y then issubset_sub (force t1) (force t2)
else if x > y then issubset_sub s1 (force t2)
else false
| _ -> false
in
issubset_sub (stream_of_tree tree1 (fun () -> Nils))
(stream_of_tree tree2 (fun () -> Nils))
実際の処理は局所関数 issubset_sub で行います。遅延ストリーム s1 が s2 の途中で終了した場合、tree1 の要素はすべて tree2 にあるので tree1 は tree2 の部分集合です。issubset は true を返します。そうでなければ、遅延ストリームから要素を一つずつ取り出します。x = y ならば次の要素を調べます。 x > y の場合、x と等しい要素が s2 に存在するかもしれないので、x と s2 の次の要素を比較します。x < y の場合は、x と等しい要素は s2 に存在しないことがわかるので false を返します。
それでは実行例を示します。
# let a = tree_of_list [5; 6; 4; 7; 3; 8; 2; 9; 1];; val a : int tree = Node (5, Node (4, Node (3, Node (2, Node (1, Nil, Nil), Nil), Nil), Nil), Node (6, Nil, Node (7, Nil, Node (8, Nil, Node (9, Nil, Nil))))) # let b = tree_of_list [6; 4; 8; 2];; val b : int tree = Node (6, Node (4, Node (2, Nil, Nil), Nil), Node (8, Nil, Nil)) # issubset a b;; - : bool = false # issubset b a;; - : bool = true
ところで、遅延ストリームを使わなくても、クロージャを使って同様のことを行うことができます。次のリストを見てください。
リスト 12 : 遅延ストリームを使わない場合
(* 継続を表すデータ型 *)
type 'a continue = Nilc | Cont of 'a * 'a continue
(* 二分木の巡回 *)
let rec continue_of_tree node cont =
match node with
Nil -> cont ()
| Node (x, left, right) ->
continue_of_tree left
(fun () -> Cont (x, fun () -> continue_of_tree right
(fun () -> cont ())))
(* 二分木：同値の判定 *)
let isequal tree1 tree2 =
let rec isequal_sub s1 s2 =
match (s1, s2) with
(Nilc, Nilc) -> true
| (Cont (x, t1), Cont(y, t2)) when x = y -> isequal_sub (t1 ()) (t2 ())
| _ -> false
in
isequal_sub (continue_of_tree tree1 (fun () -> Nilc))
(continue_of_tree tree2 (fun () -> Nilc))
(* 二分木：部分集合の判定 *)
let issubset tree1 tree2 =
let rec issubset_sub s1 s2 =
match (s1, s2) with
(Nilc, _) -> true
| (Cont (x, t1), Cont(y, t2)) ->
if x = y then issubset_sub (t1 ()) (t2 ())
else if x > y then issubset_sub s1 (t2 ())
else false
| _ -> false
in
issubset_sub (continue_of_tree tree1 (fun () -> Nilc))
(continue_of_tree tree2 (fun () -> Nilc))
val continue_of_tree : 'a tree -> (unit -> 'a continue) -> 'a continue = <fun> val isequal : 'a tree -> 'a tree -> bool = <fun> val issubset : 'a tree -> 'a tree -> bool = <fun>
継続を表すデータ型として 'a continue を定義します。Nilc は継続が終了したことを表し、Cont が継続の本体を表します。Cont の第 2 要素にはクロージャを格納し、このクロージャを評価すると次の要素を求めることができます。この場合、遅延オブジェクト (lazy_t) のかわりにクロージャを使っているだけなので、計算結果はキャッシュされないことに注意してください。
それでは実行例を示します。
# #use "tree1.ml";; ... 省略 ... # let a = tree_of_list [5; 6; 4; 7; 3; 8; 2; 9; 1];; val a : int tree = Node (5, Node (4, Node (3, Node (2, Node (1, Nil, Nil), Nil), Nil), Nil), Node (6, Nil, Node (7, Nil, Node (8, Nil, Node (9, Nil, Nil))))) # let b = tree_of_list [2; 4; 6; 8];; val b : int tree = Node (2, Nil, Node (4, Nil, Node (6, Nil, Node (8, Nil, Nil)))) # isequal a a;; - : bool = true # issubset b a;; - : bool = true # issubset a b;; - : bool = false
このように、クロージャを使ってプログラムの実行を中断したり、あとから再開することもできます。
(*
* tree.ml : 二分木
*
* Copyright (C) 2008 Makoto Hiroi
*)
(* 節の定義 *)
type 'a tree = Nil | Node of 'a * 'a tree * 'a tree
(* データの探索 *)
let rec search x = function
Nil -> false
| Node (y, _, _) when x = y -> true
| Node (y, left, _) when x < y -> search x left
| Node (_, _, right) -> search x right
(* データの挿入 *)
let rec insert x = function
Nil -> Node (x, Nil, Nil)
| (Node (y, _, _)) as node when x = y -> node
| Node (y, left, right) when x < y -> Node (y, (insert x left), right)
| Node (y, left, right) -> Node (y, left, (insert x right))
(* 最小値を求める *)
let rec search_min = function
Nil -> raise (Failure "search_min")
| Node (x, Nil, _) -> x
| Node (_, left, _) -> search_min left
(* 最小値を削除する *)
let rec delete_min = function
Nil -> raise (Failure "delete_min")
| Node (x, Nil, right) -> right
| Node (x, left, right) -> Node (x, (delete_min left), right)
(* 削除 *)
let rec delete x = function
Nil -> raise Not_found
| Node(y, left, right) ->
if x = y then
if left = Nil then right
else if right = Nil then left
else
let min_data = search_min right in
Node (min_data, left, (delete_min right))
else if x < y then
Node (y, (delete x left), right)
else
Node (y, left, (delete x right))
(* 巡回 *)
let rec iter f = function
Nil -> ()
| Node (x, left, right) -> iter f left; f x; iter f right
(* CPS スタイル *)
let rec iter_cps f node k =
match node with
Nil -> k ()
| Node(x, left, right) ->
iter_cps f left (fun () -> f x; iter_cps f right (fun () -> k ()))
(* 遅延ストリーム *)
let rec stream_of_tree node cont =
match node with
Nil -> cont ()
| Node (x, left, right) ->
stream_of_tree left
(fun () -> Cons (x, lazy (stream_of_tree right (fun () -> cont ()))))
(* 同値の判定 *)
let isequal tree1 tree2 =
let rec isequal_sub s1 s2 =
match (s1, s2) with
(Nils, Nils) -> true
| (Cons (x, t1), Cons(y, t2)) when x = y -> isequal_sub (force t1) (force t2)
| _ -> false
in
isequal_sub (stream_of_tree tree1 (fun () -> Nils))
(stream_of_tree tree2 (fun () -> Nils))
(* 部分集合の判定 *)
let issubset tree1 tree2 =
let rec issubset_sub s1 s2 =
match (s1, s2) with
(Nils, _) -> true
| (Cons (x, t1), Cons(y, t2)) ->
if x = y then issubset_sub (force t1) (force t2)
else if x > y then issubset_sub s1 (force t2)
else false
| _ -> false
in
issubset_sub (stream_of_tree tree1 (fun () -> Nils))
(stream_of_tree tree2 (fun () -> Nils))
(* リスト -> 二分木 *)
let tree_of_list xs =
List.fold_left (fun x y -> insert y x) Nil xs
(* 二分木 -> リスト *)
let list_of_tree bt =
let rec iter a = function
Nil -> a
| Node (x, left, right) -> iter (x::(iter a right)) left
in
iter [] bt