キューの効率的な実装 または私は如何にしてHaskellを止めてF#を愛するようになったか

この記事は [F# Advent Calendar 2017](https://qiita.com/advent-calendar/2017/fsharp) と [Haskell (その2) Advent Calendar 2017](https://qiita.com/advent-calendar/2017/haskell2) のために書かれています。

# はじめに

最近『[純粋関数型データ構造](https://www.amazon.co.jp/dp/4048930567)』の勉強会に参加していて、いろいろと学ぶことが多かったので、その一端でも残しておこうと思いました。
一言で言うと、『純粋関数型データ構造で効率を追求するなら遅延評価と正格評価の両方が必要』だという話です。
あらためて [kinabaさんの有名なスライド](http://www.kmonos.net/pub/Presen/PFDS.pdf) を読んだら、ぜんぶ書いてあったので、そちらを理解している方は読む必要ありません。
私はネタとしては知っていましたが、今回本を読んで初めて内容を理解しました。

# キューと償却計算量

キューってありますよね。
末尾に要素を追加したり（`snoc`）、先頭の要素を取り出したり（`head`、`tail`）できるデータ構造です。
純粋関数型で（破壊的更新なく）実装する場合は、2つのリストを利用するのが一般的です。先頭側リストと末尾側リストを用意して、追加するときには末尾側リストに追加、取り出すときには先頭側リストから取り出し、先頭側リストが空になったら末尾側リストを反転して先頭側リストにする、という方法です。
リストを反転するときに _O(n)_ の時間がかかりますが、他の場合は _O(1)_ で実行できます。また反転操作は _O(n)_ ですが、 _n_ 回 `snoc` してから実行されるので、平均すれば _O(1)_ とみなすことができます。これが償却計算量（Amortized Cost）という考え方です。

![BatchedQueue.png](https://raw.githubusercontent.com/rst76/pfds/master/ch05/BatchedQueue.png)

## Haskell による実装

```haskell
type Queue a = ([a], [a])

empty :: Queue a
empty = ([], [])

check :: [a] -> [a] -> Queue a
check [] r = (reverse r, [])
check f  r = (f, r)

snoc :: Queue a -> a -> Queue a
snoc (f, r) x = check f (x : r)

head :: Queue a -> a
head ([], _) = error "empty queue"
head (x : _, _) = x

tail :: Queue a -> Queue a
tail ([], _) = error "empty queue"
tail (_ : f, r) = check f r
```

## F# による実装

```fsharp
type Queue<'a> = 'a list * 'a list

let empty<'a> : Queue<'a> = ([], [])

let check (q : Queue<'a>) : Queue<'a> =
    match q with
    | ([], r) -> (List.rev r, [])
    | _ -> q
 
let snoc ((f, r) : Queue<'a>, x : 'a) : Queue<'a> =
    check (f, x :: r)

let head (q : Queue<'a>) : 'a =
    match q with
    | ([], _) -> failwith "empty queue"
    | (x :: _, _) -> x

let tail (q : Queue<'a>) : Queue<'a> =
    match q with
    | ([], _) -> failwith "empty queue"
    | (_ :: f, r) -> check (f, r)
```

## 永続化にともなう課題

ただし、_O(1)_ という償却計算量をかならず満たせるわけではありません。
以下のように同じキューを使いまわして別々に操作すると、それぞれに反転操作が走ってしまい、平均しても _O(1)_ になりません。（分岐の数を _k_ とすると、_O(k)_ になる）

```haskell
let q0 = foldl snoc empty [1 .. n]
let q1 = snoc q0 1
let q2 = snoc q0 2
let q3 = snoc q0 3
let q1' = tail q1
let q2' = tail q2
let q3' = tail q3
```

![BatchedQueue1.png](https://raw.githubusercontent.com/rst76/pfds/master/ch05/BatchedQueue1.png)

# 永続化（データの使いまわし）と銀行家法

このような問題を避けるために、事前に借金をして実際に操作が走る前に完済する、という考え方をとります（銀行家法）。
具体的には、先頭側リストが空になったら反転ではなく、先頭側リストより末尾側リストが長くなったら反転、とします。また、その時点では反転操作を予約（停止計算を生成）するだけで、実際に反転操作を実行するのは先頭側リストが空になった時点とします。それまでには _n_ 回の `tail` が必要なので、反転操作の計算量を償却することが可能です。

![BankersQueue.png](https://raw.githubusercontent.com/rst76/pfds/master/ch06/BankersQueue.png)

ちなみに銀行家法の他に物理学者法というものも本には出てくるのですが、説明が大変なので割愛します。

## Haskell による実装

Haskell による実装はそれほど難しくありません。
2つのリストの長さ（`lenf`、`lenr`）を保持しておき、`check` 関数で比較、反転しています。

```haskell
type Queue a = (Int, [a], Int, [a])

empty :: Queue a
empty = (0, [], 0, [])

check :: Queue a -> Queue a
check q@(lenf, f, lenr, r) =
  if lenr <= lenf then q
  else (lenf + lenr, f ++ reverse r, 0, [])

snoc :: Queue a -> a -> Queue a
snoc (lenf, f, lenr, r) x = check (lenf, f, lenr + 1, x : r)

head :: Queue a -> a
head (_, [], _, _) = error "empty queue"
head (_, x : _, _, _) = x

tail :: Queue a -> Queue a
tail (_, [], _, _) = error "empty queue"
tail (lenf, _ : f, lenr, r) = check (lenf - 1, f, lenr, r)
```

## F# による実装

`(++)` や `reverse` を自前で定義しているため若干煩雑になっていますが、F# による実装も考え方は同じです。
`check` 関数の下の行で `f ++ reverse r` としており、この計算は遅延されて、必要になったタイミングで実行されます。

```fsharp
type StreamCell<'a> = Nil | Cons of 'a * Stream<'a>
and Stream<'a> = Lazy<StreamCell<'a>>
type Queue<'a> = int * Stream<'a> * int * Stream<'a>

let rec (++) (xs : Stream<'a>) (ys : Stream<'a>) =
    match xs with
    | Lazy Nil -> ys
    | Lazy (Cons (x, xs')) -> lazy (Cons (x, xs' ++ ys))

let reverse (xs : Stream<'a>) : Stream<'a> =
    let rec reverse' xs ys =
        match xs with
        | Lazy Nil -> ys
        | Lazy (Cons (x, xs')) -> reverse' xs' (lazy (Cons (x, ys)))
    reverse' xs (lazy Nil)

let empty<'a> : Queue<'a> = (0, lazy Nil, 0, lazy Nil)

let check ((lenf, f, lenr, r) as q : Queue<'a>) : Queue<'a> =
    if lenr <= lenf then q
    else (lenf + lenr, f ++ reverse r, 0, lazy Nil)

let snoc ((lenf, f, lenr, r) : Queue<'a>, x : 'a) =
    check (lenf, f, lenr + 1, lazy (Cons (x, r)))

let head (q : Queue<'a>) : 'a =
    match q with
    | (_, Lazy Nil, _, _) -> failwith "empty queue"
    | (_, Lazy (Cons (x, _)), _, _) -> x

let tail (q : Queue<'a>) : Queue<'a> =
    match q with
    | (_, Lazy Nil, _, _) -> failwith "empty queue"
    | (lenf, Lazy (Cons (_, f)), lenr, r) -> check (lenf - 1, f, lenr, r)
```

## 永続化にともなう課題の解決

同じキューに異なる操作を行うとき、分岐してから停止計算を生成する場合は、前述の考え方を適用できます。

![BankersQueue1.png](https://raw.githubusercontent.com/rst76/pfds/master/ch06/BankersQueue1.png)

また停止計算を生成してから分岐する場合は、同じ停止計算を共有するので結果をメモ化できて、やはり計算量を _O(n)_ に抑えることが可能です。

![BankersQueue2.png](https://raw.githubusercontent.com/rst76/pfds/master/ch06/BankersQueue2.png)

# 実時間（最悪時間）の改善

ここまでは、『実際はこれだけ時間かかることもあるけど、平均したらこれしか時間かからないよ！』という（償却計算量を保証する）考え方で進めてきましたが、やっぱり最悪時の計算量を保証したいという要求はあります。そんな要求に対応するためには、最悪時に実行される計算（反転操作）を事前に少しずつ進めておく必要があります。そのためには遅延評価と正格評価をうまく使い分ける必要があるのです。

![RealTimeQueue.png](https://raw.githubusercontent.com/rst76/pfds/master/ch07/RealTimeQueue.png)

## Haskell による実装

正格評価（`seq`）をうまく使えば不可能ではないと思うのですが、実装も検証も大変そうなので諦めました。
だれかできたら教えてください。

追記：@autotaker1984さんが、書いてくださいました！

http://autotaker.hatenablog.com/entry/2017/12/21/125153

## F# による実装

以下のコードのとおり、キューは `(先頭側ストリーム, 末尾側リスト, 停止計算用ストリーム)` というデータ構造になっており、 `exec` 関数がポイントです。
`exec` の下の行は `rotate` 関数を呼び出していますが、この計算は即座に実行されず遅延されます（停止計算を作っている）。
逆に上の行は一見何もしていないように見えますが、パターンマッチで `Cons` であるか `Nil` であるか判定する際に、 `rotate` で作った停止計算を進行させています。そしてこのストリームは先頭側ストリームと同一のデータを指しているので、結果的に先頭側ストリームの計算を進めていることになります。
先頭側と末尾側の長さを見ていませんが、停止計算用ストリームが空になったとき（先頭側リストを評価しきったとき）に `rotate` を呼び出すので、結果的には同じような振る舞いになります。

```fsharp
type StreamCell<'a> = Nil | Cons of 'a * Stream<'a>
and Stream<'a> = Lazy<StreamCell<'a>>
type Queue<'a> = Stream<'a> * 'a list * Stream<'a>

let empty<'a> : Queue<'a> = (lazy Nil, [], lazy Nil)

let rec rotate (q : Queue<'a>) : Stream<'a> =
    match q with
    | (Lazy Nil, y :: _, a) -> lazy (Cons (y, a))
    | (Lazy (Cons (x, xs)), y :: ys, a) ->
        lazy (Cons (x, rotate (xs, ys, lazy (Cons (y, a)))))

let exec (q : Queue<'a>) : Queue<'a> =
    match q with
    | (f, r, Lazy (Cons (_, s))) -> (f, r, s)
    | (_, _, Lazy Nil) -> let f' = rotate q in (f', [], f')

let snoc ((f, r, s) : Queue<'a>, x : 'a) : Queue<'a> =
    exec (f, x :: r, s)

let head (q : Queue<'a>) : 'a =
    match q with
    | (Lazy Nil, _, _) -> failwith "empty queue"
    | (Lazy (Cons (x, _)), _, _) -> x

let tail (q : Queue<'a>) : Queue<'a> =
    match q with
    | (Lazy Nil, _, _) -> failwith "empty queue"
    | (Lazy (Cons (_, f)), r, s) -> exec (f, r, s)
```

## 最悪時間の確認

具体的には以下のように評価計算が逐次進行し、一度に進むことがありません。これにより最悪時間も _O(1)_ となっています。
遅延評価と正格評価を組み合わせると、ここまできめ細かな制御ができるんですね。

![RealTimeQueue1.png](https://raw.githubusercontent.com/rst76/pfds/master/ch07/RealTimeQueue1.png)

# おわりに

今回ご紹介した話は本に書かれていることのごく一部で、他にも面白いトピックがたくさんあります。（まだ読んでいないけれど『暗黙再帰減速』とか厨二心をくすぐられますね）
第2章、第3章を読むだけでも楽しく、かつ勉強になると思うので、みなさん読みましょう！
私は以下のリポジトリをがんばって埋めていきます。
[rst76/pfds: Purely Functional Data Structures](https://github.com/rst76/pfds/blob/master/README.md)