Posted at
ElixirDay 5

Elixirでの継続渡しマクロと非決定性オペレータの実装

More than 3 years have passed since last update.


はじめに

この記事は on lisp を参考にelixir向けにカスタマイズしています。

まず、継続の説明を行った後、継続渡しマクロを実装し、その後、非決定性オペレータの実装を行っています。


継続(continuation)

継続 とは、ある時点での、計算の未来といえます。例えば、REPLから式a + double(b) を評価しているとすると、double(b)の評価を完了してその値がretだとして、その継続は a+(ret) となります。つまり、fn(x) -> a + x end が 式a + double(b) の double(b)の評価が完了した時点 での継続となります。

schemeでは、任意の時点で継続をとりだし、一変数の無名関数としてあつかうことができますが、Elixirでは残念ながら継続をサポートしていません。

ところで、クロージャはある時点での計算に必要な資源をまとめたものでした。継続はクロージャに加えて関数の呼出し履歴もまとめたものといえます。もしクロージャに関数の呼び出し履歴もまとめることができたなら、それは継続と同じことができるようになります。

そして、ある特定の形式でプログラムを書くことでクロージャに関数の呼び出し履歴も保存させることができます。その特定の形式は、継続渡しスタイル(CPS) と呼ばれます。


継続渡しスタイル(CPS)とは

    def double(b) do

b * 2
end
def foo(a, b) do
a + double(b)
end

を例にとり、継続渡しスタイルへ書き換えながら説明してみます。double/1は、fooから呼ばれるので、double/1の仕事が終ったらfooから依頼された仕事を行うようにします。つまり、自分の仕事が終わったら、「次にする仕事」をc として、以下のように書き換えることができます。

    def double(x, c) do

result = x * 2
c.(result)
end
def foo(a, b, c) do
double(b, fn(x) ->
result = a + x
c.(result)
end)
end

これを使うためには、foo(3, 4, fn(x) -> x end)などとします。各関数は、自分自身の計算が終ったら次に行う関数(継続)を引数として渡してもらうことから継続渡しスタイル(Continuation-passing Style、以下CPS)と呼ばれています。ではElixirでCPSを支援するマクロを実装してみましょう。


On Elixir

CPSは書きづらく、読みづらいことと、全てをCPSで畫く必要はないため、必要な部分だけをマクロの力を借りて実現してみます。継続用の変数をcont_として、それをマクロで包み隠すことで通常の関数と同じように使えるようにしたい、というのが基本的な考えかたです。

いつものように、どのように展開してほしいかを考えてみます。CPSな関数を定義する def_/2 は、 cont_を使った関数への呼出しを定義するマクロ定義へ展開されます。つまり、

    def_ double(x) do

values_(x * 2)
end

は、

    defmacro double(x) do

quote do
double_(var!(cont_), unquote_splicing(x))
end
end
def double_(cont_, x) do
values_(x * 2)
end

のように展開されるようにしたい。この例ではcont_の意味がないが、この中でさらにdef_/2で定義された関数をbind_/2の中で呼び出すと継続の効果を得られます。def_/2の定義は

    defmacro def_(name, exp) do

c = Macro.var(:cont_, nil)
{n, m, a} = name
arg = [c|a]
nm = :"#{n}_"
name_ = {nm, m, arg}
quote do
defmacro unquote(n)(unquote_splicing(a)) do
snm = unquote(nm)
a_ = unquote(a)
quote do
unquote(snm)(var!(cont_), unquote_splicing(a_))
end
end
Kernel.def(unquote(name_), unquote(exp))
end
end

のようになります。

values_/1 は、現在の継続cont_へ値を渡して、処理を継続させます。つまり、

    defmacro values_(x) do

quote do
var!(cont_).(unquote(x))
end
end

となります。先の例を続けると、

    def double_(cont_, x) do

values_(x * 2)
end

の部分は、更に

    def double_(cont_, x) do

cont_.(x * 2)
end

となり、現在の継続cont_を呼出していることがわかります。

bind_/2 は、パラメタpとその処理exp(def_で定義された処理)、および継続処理ブロックblockを引数にとり、CPSの処理を実現します。bind_/2の定義は、

    defmacro bind_([{param, exp}], block) do

p = Macro.var(param, nil)
block = Keyword.get(block, :do, nil)
quote do
fn(cont_) ->
unquote(exp)
end.(fn(unquote(p)) -> unquote(block) end)
end |> Macro.prewalk(fn(x) ->
case x do
{:cont_, m, _atom} ->
{:cont_, m, nil}
x -> x
end end)
end

のようになるでしょう。quoteしたあとでprewalkしているのは、cont_変数についてのコンテキストの書き換えであるので、無視してよいです。bind_/2の動きを例で説明すると、例えば、

    bind_ p: double(x) do

values_(p + 1)
end

となっていたら、

    (fn(cont_) ->

double(x)
end).(fn(p) -> values_(p + 1) end)

と展開されます。double(x)は更に

    double_(cont_, x)

に展開されるので、

    (fn(cont_) ->

double_(cont_, x)
end).(fn(p) -> values_(p + 1) end)

となるのですが、values_(p + 1)の部分は更に

    cont_.(p + 1)

のように展開され、

    (fn(cont_) ->

double_(cont_, x)
end).(fn(p) -> cont_.(p + 1) end)

となります。この中で、ここが大事なところなのですが、cont_.(p + 1)の部分のcont_だけが、外側のコンテキスト から束縛されたもので、他のcont_は無名関数の仮引数となっていることに注意してください。この外部から与えられたcont_と、自由変数xを決めると、このbind_ブロックは実行可能になります。例えば、xを5とし、cont_をfn(y) -> y endとすると、上は

    (fn(cont_) ->

double_(cont_, 5)
end).(fn(p) -> (fn(y) -> y end).(p + 1) end)

となります。これを(fn(y) -> y end).(p + 1)の部分だけ簡約すると

    (fn(cont_) ->

double_(cont_, 5)
end).(fn(p) -> p + 1 end)

となり、さらに無名関数の仮引数cont_に引数、fn(p) -> p + 1 end を代入すると

    double_(fn(p) -> p + 1 end, 5)

となります。double_の定義は、

    def double_(cont_, x) do

cont_.(x * 2)
end

でしたので、double_の内部では、

    (fn(p) -> p + 1 end).(5 * 2)

結局

    10 + 1

ここでようやくdouble_およびその継続から戻ることになります。なんでこんな面倒なことをしているのか理解できないかもしれませんが、継続渡しスタイルというのは、このように、関数の呼出しもその戻った後の処理も関係なく、常に「次に行う関数」を伴うというスタイルだということです。もとのbind_ブロックをもう一度みてもらうと、double_(cont_, 5)に対して、継続Cはfn(p) -> cont_.(p + 1) end です。そしてこのCの中での継続Dつまりcont_は、外部から束縛された継続関数です。次々に継続関数が渡され、決して帰って来ることはない(最後の最後でまとめて戻って来る)。Elixir/Erlangの基盤であるBEAM実行時システムは末尾関数呼びだしの最適化を行うのでスタックを食い潰すことはないでしょう。

このように、cont_はほとんどの場合、生の値が使われることはなく、次の継続関数に渡される仮引数となりますが、def_で定義された関数を使う(通常の)関数では、cont_自体の定義をどこかでしておかなければなりません。これをするのは、use_contです。これは先程の例でも見たように、

    defmacro use_cont() do

quote do
var!(cont_) = fn(x) -> x end
end
end

とかで構いません。

実はこれらのCPSマクロを使うには条件があります。


  • def_で定義された関数は、最後にvalue_/1、bind_/2か、
    def_で定義された他の関数を呼び出さなければならない。

  • bind_/2は、処理の最後に呼び出されなければならない。

この制限は、最初に説明した継続渡しスタイル的書き方をプログラマに強制させるための制限ですが、手に入る継続という抽象化能力と比較すれば、実際にはそれほど気にはならないでしょう。


非決定性(non-deterministic)

非決定性とは、いくつかの選択肢から選択したとき、選択したあとでfail()しないように選択するということです。つまり、非決定性オペレータは未来予測をしているかのように振る舞います。

説明のため、仮にこれをしてくれるマクロをchoose_bind/2とします。第一引数は選択肢のリストで、第二引数は選択された要素を引数に取る継続ブロックです。例えば、

    choose_bind [1,2,3,4,5], fn(x) ->

case rem(x, 2) do
0 -> values_(x)
1 -> fail()
end
end

などとなっていたとします。fail()が呼ばれる結果には進まないので、このブロックの実行結果は、

2か4

になります(リスト[2,4]ではないことに注意してください)。これだけではどちらかには決められませんが、どちらも条件には合っています。このように、非決定性オペレータは、可能な解(の集合)を求めます。状態遷移機械を使ったアルゴリズムなどは非決定性有限オートマトン(NFA)を使うと記述が単純になるため多用されます。NFAから決定性有限オートマトン(DFA)への変換自体も機械的に行うことができますが、現実の問題を解決する場合にはNFAで考えたほうが断然楽です。


非決定性オペレータのCPSマクロでの実現

CPSマクロで定義した関数の継続は、先述した通り、cont_というアナフォラでアクセスできます。このcont_を保存することで、あとで呼び出すことができるので、これを利用してchoose_bind/2の実装を考えてみましょう。choose_bind/2の第一引数は選択肢で、第二引数は選択された要素を引数にとり、処理を行う継続ブロックでした。この継続ブロック中では、values_/1かfail/0を返さねばなりません。choose_bind/2はまだ選択していない選択肢とその継続をスタックに保存します。スタックはErlang/Elixirの中では禁断ともいえる プロセス変数 として保存することにします。fail/0は、スタックから継続と残りの選択肢を取り出して再開します(そのためfail/0に至る選択は「選択しなかったこと」になる)。これによって、fail/0を通らないルートを通った結果が得られます。

      def fail(_result \\ nil) do

case ppop(@choose_stack) do
nil ->
nil # %{result: false, values: result}
{f,t} ->
f.(t)
end
end
def do_choise_bind(c, f) do
case c do
[] ->
fail()
[h|t] ->
if (length(t) > 0) do
ppush(@choose_stack, {fn(x) ->
do_choise_bind(x, f) end,
t})
end
f.(h)
_ ->
fail()
end
defmacro choose_bind(x, f) do
quote bind_quoted: [x: x, f: f] do
do_choise_bind(x, f)
end
end

これらをCpsモジュールとしてまとめておきます。

この非決定的選択の効果はchoose_bind/2を呼び出した関数から返った後でさえも有効であることに注意してください。以下にその例をあげます。

      def_ two_numbers() do

choose_bind([0,1,2,3,4,5],
fn(x) ->
choose_bind([0,1,2,3,4,5],
fn(y) ->
values_([x,y])
end)
end)
end
def_ parlor_trick(sum) do
bind_ [p: two_numbers()] do
[x,y] = p
if (x + y == sum) do
values_([:sum, x, y])
else
fail()
end
end
end

これをTestモジュールで定義して、use Cpsしてからiexから動かすと、

iex(7)> import Test

nil
iex(8)> use_cont
#Function<6.90072148/1 in :erl_eval.expr/5>
iex(9)> parlor_trick(3)
[:sum, 0, 3]
iex(10)> fail
[:sum, 1, 2]
iex(11)> fail
[:sum, 2, 1]
iex(12)> fail
[:sum, 3, 0]
iex(13)> fail
nil
iex(14)>

このように和が3となるような選択肢だけが結果として選択されたように振る舞います。各数は、choose_bindを使ったtow_numbers/0から帰って来てから、parlor_trick/1中のbind_/2の中でfail()を呼ばなければならないような選択はされていないことに注意してください(実際はfail()を呼ぶルートでは選択が巻き戻されて、別の選択から再開されます)。つまり、深さ優先探索をしながらバックトラックの動作をしているのですが、CPSマクロのお陰でバックトラックの詳細は殆ど隠されています。

また、REPLから明示的にfailを呼ぶことで、強制的にバックトラックを起こして別の選択肢を選択にいくことがわかります。そして、failを呼ぶまでは、継続がスタックに積まれているままであるため、真の意味で関数から戻っているわけではないこともわかるでしょう。


非決定性マクロの利用例

せっかくなので 8 Queen 問題を解いてみましょう。8 Queen問題とはチェスのQueenを互いの効き筋に掛からないように8個並べる配置を求めるという問題です。駒の縦位置1から8の数字とし、リスト上の位置を駒の横位置で表現します。listに対してxだけ離れた位置にqを置いた時の駒の効き筋に掛かるかのチェックはnot_reachable(q, x, list)で行います。

defmodule Q8e do

use Cps
def not_reachable(q, t) do
not_reachable(q, 1, t)
end
def not_reachable(q, x, []) do
x
end
def not_reachable(q, x, [h|t]) do
cond do
(q == h - x) -> false
(q == h + x) -> false
true -> not_reachable(q, x + 1, t)
end
end
def_ solve(l, a) do
choose_bind l,
fn (p) ->
t = l -- [p]
case t do
[] ->
values_([p|a])
t ->
if (not_reachable(p, a)) do
solve(t, [p|a])
else
fail()
end
end
end
end
def solve() do
use_cont
s = [1,2,3,4,5,6,7,8]
solve(s, [])
end
end

これを "examples/q8e.ex" として保存し、iexから実行してみます。

iex(2)> c( "examples/q8e.ex")

[Q8e]
iex(3)> Q8e.solve
[8, 6, 4, 2, 7, 5, 3, 1]
iex(4)> use Cps
nil
iex(5)> fail
[7, 6, 4, 2, 8, 5, 3, 1]
iex(6)> fail
[7, 5, 2, 4, 6, 8, 3, 1]
iex(7)> Process.get(:choose)
[{#Function<1.42871611/1 in Cps.do_choise_bind/2>, '\a'},
{#Function<1.42871611/1 in Cps.do_choise_bind/2>, [5,7]},
{#Function<1.42871611/1 in Cps.do_choise_bind/2>, [5,7]},
{#Function<1.42871611/1 in Cps.do_choise_bind/2>, '\a'},
{#Function<1.42871611/1 in Cps.do_choise_bind/2>, [4, 5, 6, 7, 8]},
{#Function<1.42871611/1 in Cps.do_choise_bind/2>, [2, 3, 4, 5, 6, 7, 8]}]
iex(8)> fail
[7, 3, 5, 2, 8, 6, 4, 1]
...

きちんと動いていますね。Process.get(:choose)で積まれた他の選択肢が継続として見えています。


終わりに

Elixirでの継続渡しマクロと、それを使った非決定性オペレータの説明をしました。

非決定性オペレータがあると、探索問題についてはループとチェックの山から解放され、問題空間を記述するだけで、幾つあるかわからない解のうち一つを比較的すぐに取り出すことができるようになります。

今回のコードは

https://github.com/k1complete/cps

に置いてあります。