高速なキューを作る話（前編）

私が以前Haskellで作成した（割と自己満足な）[kazura-queue][]という比較的高速なキューのライブラリがあります。
このライブラリを作るときに考えたアレコレを書いてみたいと思います。

# 前提
## ここでいうキューって何？

データ構造としてのキュー（[Sequence][]みたいな）や分散処理用のキュー？（[RabbitMQ][]みたいな）ではなく、並行処理のためにスレッド間の部品として使うようなキュー（[Chan][]（や[TQueue][]や[TChan][]）みたいな）を意図しています。

ここでは[Chan][]が持っているキューとしての性質をざっと書き出してみましょう。

* キューに入るデータの型は任意
  * ``Chan a``という型で表現できる範囲で好きなデータを入れることができる。
* FIFO
  * キューなのだから当たり前といえば当たり前かもしれませんが一応。
* スレッドセーフ
  * 複数のスレッドが同時に書き込もうとしても問題が起きない。[^1]
  * 複数のスレッドが同時に読み出そうとしても問題が起きない。[^1]
  * 複数のスレッドにより書き込みと読み出しが同時に行おうとしても問題が起きない。[^1]
* 読み出しを「待つ」ことが可能
  * キューが空の時に読み出しを行おうとすると、そのスレッドはブロックされる。その状態で新たなデータが書き込まれるとブロックされたスレッドが起きる。
  * （[Chan][]は提供していませんが）読み出しを待たずに失敗して返るAPIもある方が望ましい。
* 非同期例外への耐性
  * 書き込み／読み出し／読み出し待ち中のスレッドに対して非同期例外が飛んでも問題が起きない。[^1]
* 不要な参照を持たない
  * キューから読み出したデータへの不要な参照は持たず、（他の参照がなければ）すぐにGC可能になる。
* 待ち順の公平性
  * 複数のスレッドが読み出しを待っているときにデータを書き込むと、「待ち」に入った順番にデータを読み出す。

[^1]: 「問題が起きない」と書いているものは、厳密な言い方が面倒なのでさぼった記述です。

[Chan][]はキューとして使うこともできますが、単純なキューではなく「チャネル」として利用する機能も持っています。ここでいう「チャネル」は必ずしも一般的な言葉ではない気がしますが「書き込まれたデータを複数のキューに書き込まれたかのように読み出す機能を持っているもの」としておきます。（言い換えると[dupChan][Chan.dupChan]が使えるものが「チャネル」）
しかし、本記事では単純なキューとしてだけ考えることにします。上に列挙した性質も、
単純なキューとして一連の値を読み出すケースを想定して性質を書き出しています。

これらの性質がいつでも全て必要になるわけではありません。（実際[TQueue]や[TChan]にはない性質もあります）
しかし、これらの性質は多くの場合あった方が望ましいです。また、使う側としても、使おうとしているキューがどの性質を持っているのかは考えておいた方がよいでしょう。

## 高速って何？

ひとくちに高速といっても、速度には複数の尺度があり得ます。例えば以下です。

* 単位時間当たりに処理できるアイテムが多い。
* 書き込みから読み出しまでの時間が短い。

一つの尺度に絞って最高速を目指すということもできますが、高速さを求めるときに欲しいものは、往々にして「どの尺度でもそこそこ速い」ものであったりします。しかし、一方で尺度を定めなければ速度を計測することも難しくなります。
ここでは最終的にどのようなベンチマークで測定するか、という形で以下のように表現しておきます。
ここでは欲しい性能を以下のようにしておきます。

* 以下の状況で一定（大量）のデータを処理するのにかかる時間が短いこと
  * 書き込みの頻度が高く、キューの中身は空ではないことが多い状況
  * キューはひとつではなく、複数のキューが同時に稼働しているような状況
* ただし、他の状況でもできるだけ速くしたい
* 比較対象は[Chan][]や[TQueue][]や[TChan][]

また、ここでもうひとつしておかなければいけない話があります。
それは、 **多くの場合、[Chan][]や[TQueue][]や[TChan][]は十分に高速なので、より高速なキューを使用する必要はほとんどない** ということです。
そもそもキューが律速因子になることはそんなにありません。必要以上の高速さを求めてバグに悩まされるよりも、手軽に確実なものを利用する方がよいと思います。

## バージョンや環境など

本記事の内容は主に以下の環境で確認しています。

* stackage: lts-7.13 (ghc-8.0.1)
* OS: Windows10

# [Chan][]と[MVar][]について

高速なキューの話をする前に、まずは比較対象として[Chan][]及びその重要な構成要素である[MVar][]について簡単に説明します。
詳しく知りたい方は[『Haskellによる並列・並行プログラミング』][1]などを読むとよいでしょう。
キューの例としては[TQueue][]や[TChan][]もあげていますが、この二つは[stm][]を使用しているため、今回扱うライブラリとは勝手が違います。このため解説からは省略します。

## [MVar][]

[MVar][]はロックと「待つ」機能のついたデータの入れ物です。データが入っている状態と空の状態があり、[putMVar][MVar.putMVar]でデータを書き込み、[takeMVar][MVar.takeMVar]でデータを取り出して空にします。
空の状態で[takeMVar][MVar.takeMVar]、もしくはデータがある状態で[putMVar][MVar.putMVar]すると、それができる状態になるまでスレッドがブロックして待ち状態になります。
待ちには公平性があり複数のスレッドがブロックした場合、ブロックした順番で起きます。
その他、データがある状態を待って空にはせずにデータを読み出すのみの[readMVar][MVar.readMVar]といった操作も可能です。

[MVar][]に対する操作は、（例えば[IORef][]のような）単純なメモリの読み書きと比較すると、重い処理になります。

GHCにおいてスレッド間の待ち合わせを行う方法はいくつか存在しますが、基本的には[MVar][]か[stm][][^2]を使います。[stm][]は非常に便利ですが、他の部品も含めて[stm][]を多用するのでなければ、[MVar][]を用いることが多いようです。

[^2]: [stm][]についてはここでは説明しません。[『Haskellによる並列・並行プログラミング』][1]を読んでください。

## [Chan][]

まずは[Chanの型][code:Chan.Chan] を見てみましょう。
UNPACKやBangやderivingを取り除くと以下のようになっています。

```haskell
data Chan a = Chan (MVar (Stream a)) (MVar (Stream a))
type Stream a = MVar (ChItem a)
data ChItem a = ChItem a (Stream a)
```

キューに書き込まれたデータを保持するのは``Stream``（[MVar][]）の中の``ChItem``です。``ChItem``は続く``Stream``への参照を持っており、一方向リストになっています。
``Chan``は``Stream``を格納した[MVar][]を二つ保持しており、二つの[MVar][]は一連の``Stream``の別の場所（もしくは同じ場所）を参照しています。一つ目の[MVar][]は読み出し用で、``Stream``の次に読み出す位置への参照を持っています。二つ目の[MVar][]は書き込み用で、``Stream``の次に書き込む位置（``Stream``の先頭）への参照を持っています。

[書き込み処理（writeChan）][code:Chan.writeChan] では、読み出し用[MVar][]から書き込む位置の``Stream``を取り出し、``Stream``の[MVar][]に``ChItem``を書き込みます。
[読み出し処理（readChan）][code:Chan.readChan] では、書き込み用[MVar][]から読み出す位置の``Stream``を取り出し、``Stream``の[MVar][]から``ChItem``を読み出します。

処理をよく理解すると、書き込み用、読み出し用、及び``Stream``内の各[MVar][]はそれぞれ別のスレッドセーフティのために使われていることがわかります。

* 複数のスレッドによる同時の[読み出し（readChan）][Chan.readChan]を制御するために、読み出し用``Stream``は読み出し用の[MVar][]に格納されています。
* 複数のスレッドによる同時の[書き込み（writeChan）][Chan.writeChan]を制御するために、書き込み用``Stream``は書き込み用の[MVar][]に格納されています。
* 複数のスレッドによる同時の[書き込み（writeChan）][Chan.readChan]と[読み出し（readChan）][Chan.readChan]を制御するために、``Stream``の各``ChItem``は[MVar][]に格納されています。

書き込みと読み出しの操作は、それぞれ2回[MVar][]に対する操作が必要になっており、これが[Chan][]の処理に必要な時間の大部分を占めていると考えられます。

# [unagi-chan][]

次に[Chan][]などより高速なキューを実装した[unagi-chan][]（うなぎちゃん）というなんだか可愛げな名前のライブラリについて紹介します。

[unagi-chan][]の提供するキューは、[hackageのページ][unagi-chan]のグラフを見ればわかる通り速度に関してはかなり優れています。ただし、後述する通りいくつかの難を抱えています。
[unagi-chan][]はデータの種類や用途に応じて最適化された複数の種類のキューを提供しているのですが、以降では汎用的に使用できる[Control.Concurrent.Chan.Unagi][Chan.Unagi]（以降はChan.Unagiと記載）について見ていきます。

## [Chan.Unagi][]が早い理由

[Chan.Unagi][]は多数の高速化のためのアイデアを実装しています。
しかし、その中で最も効果を発揮しているのは以下の点です。

* [CAS（Compare-and-Swap）][Compare-and-Swap]ではなく[Fetch-and-Add][]を使用し、できるだけ「処理のやり直し」が発生しないようにしている

### [CAS][Compare-and-Swap]と[Fetch-and-Add][]

リンク先のWikipediaを見てもらった方が早いかもしれませんが、[CAS][Compare-and-Swap]と[Fetch-and-Add][]について軽く紹介します。

* [CAS（Compare-and-Swap）][Compare-and-Swap]は、対象のメモリの値が指定された値と同じ場合にのみ書き換える操作をアトミックに行うCPU命令です。値が異なる場合には書き換えず、処理は失敗します。
* [Fetch-and-Add][]は、対象のメモリの値に指定された値を足すという操作をアトミックに行うCPU命令です。

まず前提として、複数のスレッド間での並行処理を問題が起きないように行うには、何らかの特別な操作を行うCPU命令が必要になります。
通常は[CAS][Compare-and-Swap]の方が便利かつ普及しているなので、こちらを使います。
[CAS][Compare-and-Swap]は、複数スレッドの処理が競合した場合には失敗しますが、その場合、新しい値に基づいて処理をやりなおす必要があります。
これに対して、[Fetch-and-Add][]は失敗せず、必ず値を更新することができるので、やりなおしのコストが発生しません。
このため、処理内容によっては[Fetch-and-Add][]を使うことで性能が向上することがあります。
ただし、CPU命令なので、[Fetch-and-Add][]をサポートしていないCPUでは使えません。x86系CPUでは[Fetch-and-Add][]をサポートしているため、利用することができます。

### GHCにおける[CAS][Compare-and-Swap]と[Fetch-and-Add][]の利用

GHCのランタイム実装では並行処理のための基本的な部分で[CAS][Compare-and-Swap]を利用しています。[MVar][]の操作内でも[CAS][Compare-and-Swap]が利用されています。
しかし、[atomic-primops][]というライブラリを使用することで、[Fetch-and-Add][]をHaskellのコードから利用することができるようになります。（x86以外のCPUでは、[Fetch-and-Add][]が使えないため、[CAS][Compare-and-Swap]を利用して[Fetch-and-Add][]と同様の処理を実装したものが自動的に使われるようになっています）
**追記：GHC7.8以降は[atomic-primops][]を使わなくても[GHC.Prim][]で[Fetch-and-Addを利用できる（fetchAddIntArray）][fetchAddIntArray]ようです。**

## [Chan.Unagi][]の仕組み

では、[Chan.Unagi][]が[Fetch-and-Add][]をどのように使用するのか見ていきましょう。

[Chan.Unagi][]では書き込まれたデータを保持するための領域として[MutableArray][]を用意しています。（[ソースコード][code:Chan.Unagi.StreamSegment]）
[MutableArray][]の長さは1024で、領域が足りなくなる前に次の[MutableArray][]を確保していきます。

[MutableArray][]とは別に書き込み用のカウンタと読み出し用のカウンタを保持しており、このカウンタの値が、一連の[MutableArray][]上の座標を表しています。
書き込み及び読み出しの操作ではそれぞれのカウンタを[Fetch-and-Add][]でインクリメントしつつカウンタの値を取得し、カウンタの値の示す場所にデータを書き込み／読み出しに行きます。

* 書き込み操作同士の競合
  * 書き込みを行う各スレッドは異なるカウンタ値を取得するため、書き込みスレッド同士が[MutableArray][]へのデータの書き込みで競合することはありません。
  * 後続の[MutableArray][]の確保するのを待つ処理で競合することはあり得ますが、書き込みの頻度と比較すると少ないため、あまり問題にはなりません。
* 読み出し操作同士の競合
  * [Chan.Unagi][]をチャネルではなくキューとしてのみ見た場合、読み出し操作についても書き込み操作と同様、[MutableArray][]の異なる場所を見に行くため、競合することはありません。
  * キューではなくチャネルとして見た場合、複数のスレッドが同時に[MutableArray][]の同じ場所の値を読み出そうとすることはありえます。
* 書き込み操作と読み出し操作の競合
  * 書き込みスレッドと読み出しスレッドの処理が競合することはあります。詳しくは後述します。

[MutableArray][]に格納するデータは以下のように定義されています。[ソースコード][code:Chan.Unagi.Cell]

```haskell
data Cell a = Empty | Written a | Blocking !(MVar a)
```

それぞれ、以下の状態を表しています。

* ``Empty``: まだ値が書き込まれていない状態
* ``Written``: （読み出しスレッドが読み出そうとするより先に）値が書き込まれた状態
* ``Blocking``: （書き込みスレッドが書き込むより先に）読み出しスレッドが値を読み出そうとした状態
  * 値が書き込まれる前は[MVar][]が空になっていて、読み出しを待っているスレッドが[MVar][]に値が書き込まれるのを待っています。値が書き込まれた後も``Written``にはならず、値の入った[MVar][]を保持する``Blocking``のままとなります。

書き込み及び読み出しは以下のように行われます。

* 書き込み時の動作
  * 書き込み先の内容が``Empty``なら、``Written``にデータを詰めて書き込みます。
  * 書き込み先の内容が``Blocking``なら、中の[MVar][]にデータを[putMVar][MVar.putMVar]します。
  * 書き込み処理同士は競合しないため、書き込み先の内容が``Written``であることはありません。
* 読み出し時の動作
  * 読み出し元の内容が``Written``なら、中の値を読み出します。
  * 読み出し元の内容が``Empty``なら、``Blocking``に新しい空の[MVar][]を生成して詰め込んだものに書き換えます。[MVar][]の値を[readMVar][MVar.readMVar]して値が書き込まれるのを待ちます。
  * 読み出し元の内容が``Blocking``なら、中に入っている[MVar][]の値を[readMVar][MVar.readMVar]します。既に値が書き込まれていれば、即座に値が読み出され、まだ書き込まれていない場合には値が書き込まれるのを待つことになります。[^3]

[^3]: [Chan.Unagi][]をチャネルとして利用した場合、他のスレッドにより読み出し操作で``Blocking``に書き換えられることによりこのケースが発生します。単純にキューとしてのみ利用した場合にはこのケースは発生しえません。

書き込みスレッドだけでなく、読み出しスレッドも[MutableArray][]に値を書き込むことになります。同時に書き込みを行ってしまわないようにするため、[MutableArray][]への書き込みは[CAS][Compare-and-Swap]で行います。

実のところCellではなく、直接[MVar][]が[MutableArray][]の中身であっても機能的には問題はないと思われます。しかし、[MVar][]の操作は単純な[MutableArray][]への書き込み／読み出し操作と比較して遅いため、できる限り[MVar][]を使用しないようにしているのではないかと思われます。

## [Chan.Unagi][]の欠点

上記のような仕組みによって高速に動作する[Chan.Unagi][]ですが、以下の欠点があります。

* 読み出しを行うスレッドに対する非同期例外の扱いが面倒
* データがすぐにGCされない

それぞれについて詳しく見ていきます。

### 読み出しを行うスレッドに対する非同期例外の扱いが面倒

書き込み・読み出しの各動作と非同期例外が絡んだシーンについて考えていきます。

#### [mask][]

まず前提知識として[mask][]について軽く説明します。より詳細に知りたい場合は[Control.Exceptionのドキュメント][Control.Exception]もしくは[『Haskellによる並列・並行プログラミング』][1]を読んでください。

* [mask][]された処理を行っている最中は、非同期例外が投げられても、即座に処理を中止することはせず、[mask][]が解除された時点で例外が処理される。
* ただし、[mask][]された処理の最中であっても、当該スレッドがブロック状態（例えば[MVar][]の値を待っている状態など）であれば、例外が処理される。
  * ブロック状態でも例外を処理したくない場合は[uninterruptibleMask][]を使用する。

#### [書き込み（writeChan）][Chan.Unagi.writeChan]

[writeChan][Chan.Unagi.writeChan]は[Chan.Unagi][]のへの書き込みを行う関数です。
[writeChan][Chan.Unagi.writeChan]の中身の処理は[mask][]されており、ブロックもしないため特に問題は発生しません。[ソースコード][code:Chan.Unagi.writeChan]（実際に使われているのは[mask][]ではなく[mask_][]になっています）

#### [読み出し（readChan）][Chan.Unagi.readChan]

[readChan][Chan.Unagi.readChan]は[Chan.Unagi][]の読み出しを行う関数です。
[readChanのドキュメント][Chan.Unagi.readChan]には「[readChan][Chan.Unagi.readChan]では非同期例外が投げられるとデータがロストする」と書かれています。
[ソースコード][code:Chan.Unagi.readChan] で確認すると、なんと[mask][]されていません。
しかし、ロストする原因は[mask][]をさぼっているからではありません。[mask][]したとしてもロストは発生します。

例えば以下のようなケースを考えてみましょう。

1. 空の[Chan.Unagi][]を用意する
2. 読み出し用のスレッドにて、[mask][]した状態で[readChan][Chan.Unagi.readChan]を呼び出す。読み出し用のスレッドは１つ目のデータの書き込みを待ってブロック状態になる
3. 読み出し用のスレッドに非同期例外を投げる。読み出し用スレッドはブロック状態のため、[mask][]された状態でも非同期例外が処理される
4. 書き込み用のスレッドにて[writeChan][Chan.Unagi.writeChan]で書き込みを行う。１つ目のデータが書き込まれる
5. 読み出し用スレッドにて、再び[readChan][Chan.Unagi.readChan]を呼び出す。手順２で読み出し用カウンタは既に回った後であるため、この[readChan][Chan.Unagi.readChan]は２つ目のデータの書き込みを待ってブロック状態になる

読み出し用カウンタは進む一方なので、手順４にて書き込んだ１つ目のデータは二度と読み出されることはありません。このデータはロストしたように観測されてしまいます。

[Chan.Unagi][]はカウンタを[Fetch-and-Add][]で回しつつ値を取得することで、読み出しを試みた処理ごとに異なる座標の値を読み出そうとすることで競合をなくして高速化する仕組みでした。つまり、同じ座標を読み出しに行こうとする操作はたった一つしか存在しないのです。このため、読み出し座標が決定された以降の処理で非同期例外を受けると、その座標の値を読み出すことができなくなってしまいます。

[readChan][Chan.Unagi.readChan]が[mask][]を使用しないのは、仮に[mask][]したとしてもロストは発生し、maskせずともそれ以外の不整合が発生しないため、無駄なコストの発生する[mask][]を使用していないのだと考えられます。

#### [readChanOnException][Chan.Unagi.readChanOnException]

データがロストしては困るという場合に用意されているのが、[readChanOnException][Chan.Unagi.readChanOnException]です。
非同期例外が投げられた場合は、その読み出し処理が読み出すはずだった値を読み出すアクション（IO a）を受け取るハンドラを渡すことで非同期例外時にもデータがロストしないようにすることができるというものです。

[readChanOnException][Chan.Unagi.readChanOnException]を使うことで大抵の場合は対応可能なように思われます。しかし、そもそもこのようなデータのロストが発生しない[Chan][]や[TQueue][]や[TChan][][^4]などと比較すると、より多くのことを考えて使用する必要があります。

[^4]: [Chan][]や[TQueue][]や[TChan][]であっても、ちょうど読み出した直後のタイミングで非同期例外が投げられた場合に値がロストしないようにするには[mask][]などを使う必要があります。

### データがすぐにGCされない

[Chan][]の性質として「不要な参照を持たず、（他の参照がなければ）すぐにGC可能」というものをあげました。しかし[Chan.Unagi][]では、そうではありません。他の参照がなくなってもGCできないケースがあります。

[Chan.Unagi][]では[MutableArray][]にデータを書き込みます。書き込まれたデータは読み出された後も[MutableArray][]に残り続けます。このため、[MutableArray][]自体が誰にも参照されなくなってGCされるまで、書き込まれたデータはGCできない状態になります。
[MutableArray][]への参照がなくなるのは、読み出し側がその[MutableArray][]の全てのデータを読み終えて次の[MutableArray][]まで到達したときです。
[Chan.Unagi][]で使用している[MutableArray][]の長さは1024なので、最悪の場合1000個以上のデータが書き込まれて読み出されるまで、書き込まれたデータはGCされなくなってしまいます。
定常的に大量のデータが書き込まれ、読み出されるような使い方であれば、これが問題になるようなことはないかもしれません。しかし、データの書き込み頻度が一定でない場合などには、[Chan.Unagi][]の利用によって、不要にメモリを圧迫してしまう可能性があるのです。

ところで、なぜ[Chan.Unagi][]では[MutableArray][]が書き込まれたデータを保持し続けているのでしょうか。単純なキューとして使うのであれば、読み出し処理の最後で[MutableArray][]内のデータを削除してしまえばよいはずです。しかし、[Chan.Unagi][]はチャネルであるため、同じデータを読み出そうとするスレッドが複数存在しえます。全ての読み出しスレッドがそのデータを読み終えたならば削除しても問題ないはずですが、[Chan.Unagi][]では[MutableArray][]の各座標のデータを誰が読み終えたかという管理はしていません。もしそのような管理を行うとその管理コストのために、[Chan.Unagi][]のメリットである高速さを失ってしまうことになります。
なお、[Chan][]では、[MutableArray][]は使用せず、１つのデータにつき１つの[MVar][]を使用しているためこの問題は発生しません。（[Chan.Unagi][]の[MutableArray][]の長さが1である状態とほぼ同じと考えるとわかりやすいと思います）

# まとめ

前編では、キューの要件や、比較的高速なキューである[Chan.Unagi][]の仕組み、およびその欠点について解説しました。
[後編][]では、[Chan.Unagi][]の欠点を克服したキューを作る方法について解説する予定です。

[1]:http://www.oreilly.co.jp/books/9784873116891/ "Haskellによる並列・並行プログラミング"

[Compare-and-Swap]:https://ja.wikipedia.org/wiki/%E3%82%B3%E3%83%B3%E3%83%9A%E3%82%A2%E3%83%BB%E3%82%A2%E3%83%B3%E3%83%89%E3%83%BB%E3%82%B9%E3%83%AF%E3%83%83%E3%83%97
[Fetch-and-Add]:https://ja.wikipedia.org/wiki/%E3%83%95%E3%82%A7%E3%83%83%E3%83%81%E3%83%BB%E3%82%A2%E3%83%B3%E3%83%89%E3%83%BB%E3%82%A2%E3%83%83%E3%83%89
[RabbitMQ]:http://www.rabbitmq.com/

[Sequence]:http://hackage.haskell.org/package/containers-0.5.8.1/docs/Data-Sequence.html
[TQueue]:https://hackage.haskell.org/package/stm-2.4.4.1/docs/Control-Concurrent-STM-TQueue.html
[TChan]:https://hackage.haskell.org/package/stm-2.4.4.1/docs/Control-Concurrent-STM-TChan.html
[IORef]:https://hackage.haskell.org/package/base-4.9.0.0/docs/Data-IORef.html
[MutableArray]:https://hackage.haskell.org/package/primitive-0.6.2.0/docs/Data-Primitive-Array.html#t:MutableArray

[MVar]:https://hackage.haskell.org/package/base-4.9.0.0/docs/Control-Concurrent-MVar.html#t:MVar
[MVar.putMVar]:https://hackage.haskell.org/package/base-4.9.0.0/docs/Control-Concurrent-MVar.html#v:putMVar
[MVar.takeMVar]:https://hackage.haskell.org/package/base-4.9.0.0/docs/Control-Concurrent-MVar.html#v:takeMVar
[MVar.readMVar]:https://hackage.haskell.org/package/base-4.9.0.0/docs/Control-Concurrent-MVar.html#v:readMVar

[Chan]:https://hackage.haskell.org/package/base-4.9.0.0/docs/Control-Concurrent-Chan.html
[Chan.dupChan]:https://hackage.haskell.org/package/base-4.9.0.0/docs/Control-Concurrent-Chan.html#v:dupChan
[Chan.readChan]:https://hackage.haskell.org/package/base-4.9.0.0/docs/Control-Concurrent-Chan.html#v:readChan
[Chan.writeChan]:https://hackage.haskell.org/package/base-4.9.0.0/docs/Control-Concurrent-Chan.html#v:writeChan
[code:Chan.Chan]:https://hackage.haskell.org/package/base-4.9.0.0/docs/src/Control.Concurrent.Chan.html#Chan
[code:Chan.writeChan]:https://hackage.haskell.org/package/base-4.9.0.0/docs/src/Control.Concurrent.Chan.html#writeChan
[code:Chan.readChan]:https://hackage.haskell.org/package/base-4.9.0.0/docs/src/Control.Concurrent.Chan.html#readChan

[Control.Exception]:https://hackage.haskell.org/package/base-4.9.0.0/docs/Control-Exception.html
[mask]:https://hackage.haskell.org/package/base-4.9.0.0/docs/Control-Exception.html#v:mask
[mask_]:https://hackage.haskell.org/package/base-4.9.0.0/docs/Control-Exception.html#v:mask_
[uninterruptibleMask]:https://hackage.haskell.org/package/base-4.9.0.0/docs/Control-Exception.html#v:uninterruptibleMask

[Chan.Unagi]:http://hackage.haskell.org/package/unagi-chan-0.4.0.0/docs/Control-Concurrent-Chan-Unagi.html
[Chan.Unagi.writeChan]:https://hackage.haskell.org/package/unagi-chan-0.4.0.0/docs/Control-Concurrent-Chan-Unagi.html#v:writeChan
[Chan.Unagi.readChan]:https://hackage.haskell.org/package/unagi-chan-0.4.0.0/docs/Control-Concurrent-Chan-Unagi.html#v:readChan
[Chan.Unagi.readChanOnException]:https://hackage.haskell.org/package/unagi-chan-0.4.0.0/docs/Control-Concurrent-Chan-Unagi.html#v:readChanOnException
[code:Chan.Unagi.StreamSegment]:http://hackage.haskell.org/package/unagi-chan-0.4.0.0/docs/src/Control-Concurrent-Chan-Unagi-Internal.html#StreamSegment
[code:Chan.Unagi.Cell]: http://hackage.haskell.org/package/unagi-chan-0.4.0.0/docs/src/Control-Concurrent-Chan-Unagi-Internal.html#Cell
[code:Chan.Unagi.writeChan]:http://hackage.haskell.org/package/unagi-chan-0.4.0.0/docs/src/Control-Concurrent-Chan-Unagi-Internal.html#writeChan
[code:Chan.Unagi.readChan]:http://hackage.haskell.org/package/unagi-chan-0.4.0.0/docs/src/Control-Concurrent-Chan-Unagi-Internal.html#readChan

[stm]:https://hackage.haskell.org/package/stm
[unagi-chan]:http://hackage.haskell.org/package/unagi-chan
[atomic-primops]:https://hackage.haskell.org/package/atomic-primops
[kazura-queue]:http://hackage.haskell.org/package/kazura-queue

[後編]:http://qiita.com/asukamirai/items/481b22d9a8a410990828

[GHC.Prim]:https://hackage.haskell.org/package/ghc-prim-0.5.0.0/docs/GHC-Prim.html
[fetchAddIntArray]:https://hackage.haskell.org/package/ghc-prim-0.5.0.0/docs/GHC-Prim.html#v:fetchAddIntArray-35-