Rill のチャネルと strand：並行処理を深掘りする

前回の記事では Rill を全体的に見て回りました。今回は、この言語でいちばん楽しい部分だと私が思っているひとつのテーマ、strand とチャネルに絞ります。ランデブーチャネルが正確に何を保証するのかからバッファ付きチャネルへ、close によるブロードキャストからパイプラインへ、並行素数ふるいから select によるタイムアウトへ、シグナルによるグレースフルシャットダウンから SDL スタイルのステートマシン、そして複数コアでの共有チェックまで進みます。

この記事のプログラムはすべて、書く前に私がコンパイルして実行しました。出力ブロックはその実行の実際の出力です。（プログラム内の文字列がトルコ語なのは、私が実行したプログラムそのものだからです。）特に断りがない限り、計測は Apple M4 上でワーカー1つで行いました。

strand とは何か？

strand は Rill の軽量スレッドです。Go ではゴルーチン、Erlang ではプロセス、Java では仮想スレッドと呼ばれているのと同じ考え方で、名前はロープの撚り糸（strand）から来ています。プログラムは、撚り合わされたたくさんの糸でできている、というわけです。（C++ のネットワークプログラマー向けの補足：Boost.Asio の strand はまったく別物で、直列化された実行コンテキストです。両者に関係はありません。）

strand を起動するにはキーワードひとつで十分です。

spawn worker(jobs, results)

spawn は式全体を新しい strand 上で実行します。この小さな詳細が、サーバーを書くときの古典的な落とし穴を説明してくれます。spawn handle(accept(server)) と書くと、accept も新しい strand 上で実行されます。メインループは一切待たず、毎回 accept で停止する新しい strand を生成しながら空回りします。正しいやり方は、先に束縛してから渡すことです。

fn serve(server) =
  conn = tcp_accept(server)
  spawn handle(conn)
  serve(server)

こうすると待つのはループ自身で、接続を strand に渡します。serve は末尾位置で自分自身を呼んでいるので、スタックが伸びることもありません。

strand が安価なのは、Rill が copy-on-block のスタックモデルを使っているからです。実行中の strand はワーカーのスタック上で動き、停止（park）するとき、使っていたスタックの部分がコピーされて保持されます。Rill リポジトリにある計測では、停止中の strand 1つあたり約 300 バイトです。

ランデブーチャネル：ハンドシェイク

channel() はバッファを持たないチャネルを作ります。このようなチャネルでは、送信側は受信側が来るまで待ち、受信側は送信側が来るまで待ちます。値は両者が出会ったときにだけ受け渡されます。これを確かめるために、2つの strand の間で3つのメッセージを送り、すべてのステップを表示してみましょう。

fn ping(ch, n) =
  if n == 0 then ()
  else
    println("ping: " + int_to_str(n) + " gönderiliyor")
    send(ch, n)
    println("ping: " + int_to_str(n) + " teslim edildi")
    ping(ch, n - 1)

fn pong(ch, k) =
  if k == 0 then ()
  else
    v = recv(ch)
    println("pong: " + int_to_str(v) + " alındı")
    pong(ch, k - 1)

fn main() =
  ch = channel()
  spawn ping(ch, 3)
  pong(ch, 3)

ping: 3 gönderiliyor
ping: 3 teslim edildi
ping: 2 gönderiliyor
pong: 3 alındı
pong: 2 alındı
ping: 2 teslim edildi
ping: 1 gönderiliyor
ping: 1 teslim edildi
pong: 1 alındı

行の順序は一見ばらばらに見えるかもしれませんが、規則を正確に反映しています。どの「teslim edildi」（受け渡し完了）の行も、その値が相手側で受信される前に起きることはありません。ハンドシェイクのあと、表示は両方の strand で独立に続くので、どちらが先に画面に出るかはスケジューラー次第です。保証されているのは順序ではなく同期です。send が戻ってきたことは、recv が完了したことの証明なのです。

バッファ付きチャネルとメールボックス

channel(n) は最大 n 個の値をバッファに保持します。送信側が待つのはバッファが満杯のときだけ、受信側が待つのは空のときだけです。プロデューサーが先行する様子を見るために、コンシューマーを 50 ms 眠らせてみましょう。

fn producer(ch, i, n) =
  if i > n then ()
  else
    send(ch, i)
    println("üretici: " + int_to_str(i) + " kuyruğa kondu")
    producer(ch, i + 1, n)

fn main() =
  ch = channel(3)
  spawn producer(ch, 1, 5)
  sleep_ms(50)
  println("tüketici uyandı")
  for _ in 0..5
    println("tüketici: " + int_to_str(recv(ch)))

üretici: 1 kuyruğa kondu
üretici: 2 kuyruğa kondu
üretici: 3 kuyruğa kondu
tüketici uyandı
tüketici: 1
üretici: 4 kuyruğa kondu
tüketici: 2
tüketici: 3
tüketici: 4
tüketici: 5

プロデューサーは最初の3つの値をコンシューマーをまったく待たずにキューに入れ、4つ目でバッファが満杯になったため停止しました。そしてコンシューマーが値を1つ取り出したところで再開しました。バッファはプロデューサーとコンシューマーの速度差を吸収すると同時に、自然なバックプレッシャーを与えます。プロデューサーはコンシューマーより際限なく先に進むことはできません。

3つ目の選択肢として mailbox() があります。メールボックスは上限なく伸びるので、送信側は決して待ちません。2者が同時に互いへ送信する場面に最適です。バッファ付きチャネルが2本だと、互いに相手が空くのを待ってデッドロックすることがあるからです。代償もはっきりしていて、送信側を減速させるものが何もありません。ステートマシンでメールボックスがデフォルトになっている理由は、このあとすぐ見ていきます。

終わらせる：close と recv_opt

recv で停止した strand は、値が届くまでそこに留まります。値を送るはずだったものを片付けても、strand はスタックを抱えたままその場に残ります。接続ごとに strand を開くサーバーでは、これはそれまでに受け付けた接続の数だけのリークになります。close はまさにこのためにあります。

fn produce(ch, n) =
  if n <= 0 then close(ch)
  else
    send(ch, n)
    produce(ch, n - 1)

fn drain(ch, acc) = match recv_opt(ch)
  None -> acc
  Some(v) -> drain(ch, acc + v)

fn main() =
  ch = channel(4)
  spawn produce(ch, 100)
  println(drain(ch, 0))
  println(closed(ch))

5050
true

この例には、いくつか重要な設計判断が表れています。

• close はブロードキャストです。 値を送ると1つの受信者に届きますが、close するとチャネルで停止しているすべての受信者に一度に届きます。終了しようとしているプログラムは、何個の strand が待っているかを知る必要がありません。
• バッファ内の値は失われません。 プロデューサーは言うことがなくなった瞬間に close でき、コンシューマーは送信途中のものをすべて受け取れます。合計がちょうど 5050 になっているのがその証拠です。
• 2回 close してもエラーではありません。 2方向から走るシャットダウン処理が、どちらが先に着くかを調整する必要はありません。
• recv_opt は終わりのあるチャネルのためのものです。 値がある間は Some(v) を返し、チャネルが close されて空になると None を返します。

では、なぜすべての recv が Option を返さないのでしょうか？ ほとんどのチャネルは決して終わらないからです。全部を Option にすると、終わるごく一部のチャネルのために、言語中のあらゆる受信の前に match を置くことになります。その代わりに、close されたチャネルからの素の recv はプログラムを停止させ、どうすればよいかを伝えます。

fn main() =
  ch = channel(1)
  close(ch)
  println(recv(ch))

rill: recv on a closed channel - use recv_opt

これはバッファの範囲外のインデックスが受けるのと同じ扱いで、理由も同じです。嘘にならない返せる値が存在しないのです。close されたチャネルへの送信もプログラムを停止させます。これを避ける規則は Go がたどり着いたものと同じで、送る側が close する、です。

パイプライン

チャネルと close を組み合わせると、段をつないだパイプラインをとても自然に組み立てられます。各段は入力が close されたら自分の出力を close するので、「完了」の情報は勝手にチェーンを流れていきます。1 から 1,000,000 までの数を生成し、3 の倍数だけを残して2乗し、合計してみましょう。

fn numbers(out, i, n) =
  if i > n then close(out)
  else
    send(out, i)
    numbers(out, i + 1, n)

fn keep(inp, out, p) = match recv_opt(inp)
  None -> close(out)
  Some(v) ->
    if p(v) then send(out, v) else ()
    keep(inp, out, p)

fn square(inp, out) = match recv_opt(inp)
  None -> close(out)
  Some(v) ->
    send(out, v * v)
    square(inp, out)

fn total(inp, acc) = match recv_opt(inp)
  None -> acc
  Some(v) -> total(inp, acc + v)

fn main() =
  a = channel(64)
  b = channel(64)
  c = channel(64)
  spawn numbers(a, 1, 1000000)
  spawn keep(a, b, \x -> x % 3 == 0)
  spawn square(b, c)
  println(total(c, 0))

111111277777611111

結果は独立に検算しました。3 の倍数の2乗の和は、n = 333,333 として公式 9 · n(n+1)(2n+1)/6 でちょうどこの数になります。keep 段がラムダを受け取っている点も重要です。各段は普通の関数なので、好きなように再利用したり組み合わせたりできます。段の間にある 64 のバッファのおかげで、strand 同士が値1つごとに待ち合うことがなくなります。

並行素数ふるい

Go を学ぶ人が誰でもいつか出会う美しい例があります。並行版のエラトステネスのふるいです。ある strand が 2 から順に数を生成します。メインの流れはチェーンの末端にある最初の数を素数として受け取り、「この素数の倍数を取り除け」という新しい strand をチェーンに追加します。Rill では次のようになります。

fn generate(ch, i) =
  send(ch, i)
  generate(ch, i + 1)

fn filter_out(inp, out, p) =
  v = recv(inp)
  if v % p != 0 then send(out, v) else ()
  filter_out(inp, out, p)

fn sieve(ch, k, acc) =
  if k == 0 then reverse(acc)
  else
    p = recv(ch)
    next = channel()
    spawn filter_out(ch, next, p)
    sieve(next, k - 1, Cons(p, acc))

fn main() =
  ch = channel()
  spawn generate(ch, 2)
  println(sieve(ch, 20, Nil))

Cons(2, Cons(3, Cons(5, Cons(7, Cons(11, Cons(13, Cons(17, Cons(19, Cons(23, Cons(29, Cons(31, Cons(37, Cons(41, Cons(43, Cons(47, Cons(53, Cons(59, Cons(61, Cons(67, Cons(71, Nil))))))))))))))))))))

最初の 20 個の素数のために、21 個の strand と 21 本のランデブーチャネルが動いています。共有メモリもロックもどこにもありません。各 strand が知っているのは自分の入力チャネルと出力チャネルだけです。generate と filter_out は無限ループですが、どちらも末尾呼び出しなのでスタックは伸びません。main が戻るとプログラムは終了し、残された strand は Go と同じように放置されます。

select：複数のチャネルを同時に待つ

実際のプログラムが1本のチャネルだけを待つことはめったにありません。select は複数のチャネルを同時に待ち、最初に準備ができたもののアームを実行します。アームは順番に試されるので、同着の場合は前にあるアームが勝ちます。send アームは送信が通るときに準備完了になります。アームは最大 64 個までです。

ブロックしないポーリング：default

default アームを付けると、select はブロックしないポーリングになります。準備のできたチャネルがなければ、そのアームが即座に実行されます。

fn poll(ch) =
  v = select
    x = recv(ch) -> "hazır: " + int_to_str(x)
    default -> "henüz bir şey yok"
  v

fn main() =
  ch = channel(1)
  println(poll(ch))
  send(ch, 42)
  println(poll(ch))

henüz bir şey yok
hazır: 42

時計とジョブを一緒に見張る

Rill には Ticker 型がありません。ティッカーは3行とチャネル1本で書けるからです。ティッカーとジョブを同じ select で見張り、長いジョブが終わるまで一定間隔で進捗メッセージを表示するウォッチャーを書いてみましょう。

fn ticker(ch, ms) =
  sleep_ms(ms)
  send(ch, 1)
  ticker(ch, ms)

fn job(done) =
  sleep_ms(230)
  send(done, "iş bitti")

fn watch(tick, done, n) =
  r = select
    msg = recv(done) -> msg + " (" + int_to_str(n) + " tik)"
    t = recv(tick) ->
      println("  ... hâlâ çalışıyor, tik " + int_to_str(n + t))
      watch(tick, done, n + t)
  r

fn main() =
  tick = channel(1)
  done = channel()
  spawn ticker(tick, 50)
  spawn job(done)
  println(watch(tick, done, 0))

  ... hâlâ çalışıyor, tik 1
  ... hâlâ çalışıyor, tik 2
  ... hâlâ çalışıyor, tik 3
  ... hâlâ çalışıyor, tik 4
iş bitti (4 tik)

ティッカーに channel(1) を与えているのは設計上の選択です。ウォッチャーが忙しい間ティックは1つ待てますが、ティックが溜まることはありません。眠っている strand はプロセッサーをまったく使いません。千個の strand が同時に眠っていても、プログラムはアイドル状態のままです。固定レートのループを書くなら、一定時間ではなく次の締め切りまでの時間だけ眠るようにしてください。そうしないと、OS のタイマーの誤差が積み重なっていきます。

締め切り：timeout(ms)

timeout(ms) アームは、その時間ほかに何も準備ができなかったときに準備完了になります。前回の記事でも見ましたが、ここで重要なエンジニアリング上の詳細は、このアームが strand もチャネルも消費しないことです。チャネルと並んでポーラーの中で待機しています。処理中のすべてのリクエストに締め切りを設けるサーバーでも、そのためにリクエストごとに strand を払う必要はありません。いくつかの規則があります。

• 最も短い締め切りが勝つので、1回の待機で警告と上限の両方を扱えます。
• すでに過ぎた締め切りは即座に準備完了になります。つまり timeout(0) はポーリングで、default の別の書き方です。
• default と timeout は同じ select に書けません。default は「待たない」という意味なので、締め切りに到達することがありえないからです。

グレースフルシャットダウン：シグナルと close

Ctrl-C でサーバーをきれいに終了させるのに必要な2つの部品は、すでに揃っています。wait_signal() はプログラムがシグナルを受け取るまで停止してどのシグナルかを返し、close はそれを全員にブロードキャストします。1つの strand が待って他に知らせます。待つ側は、他が何個あるかを知る必要がありません。

fn worker(id, stop, n) =
  r = select
    recv(stop) -> "işçi " + int_to_str(id) + " durdu, " + int_to_str(n) + " iş yapmıştı"
    timeout(20) -> worker(id, stop, n + 1)
  r

fn run(id, stop, report) = send(report, worker(id, stop, 0))

fn main() =
  stop = channel()
  report = channel(4)
  for id in 0..3
    spawn run(id, stop, report)
  println("hazır, Ctrl-C bekleniyor")
  s = wait_signal()
  println("sinyal: " + int_to_str(s))
  close(stop)
  for _ in 0..3
    println(recv(report))

プログラムを起動し、約 0.5 秒後に kill -INT を送りました。

hazır, Ctrl-C bekleniyor
sinyal: 2
işçi 0 durdu, 25 iş yapmıştı
işçi 1 durdu, 25 iş yapmıştı
işçi 2 durdu, 25 iş yapmıştı

微妙なポイントは、recv(stop) -> アームが何も束縛しないことです。close されたチャネルは受信アームを準備完了にしますが、生み出す値はありません。x = recv(stop) -> と書いたアームは値を生み出さなければならず、プログラムはその旨を告げて停止してしまいます。こうしておけば、もう一方のアームが何をしていても、close(stop) はすべてのワーカーに届きます。

シグナル側でも、いくつかの詳細がよく考えられています。シグナルは OS が選んだ任意のスレッドに、任意の2命令の間に届く可能性があります。そのため、プログラムが書いたものはそこでは何も実行されません。ハンドラーはパイプに1バイト書き込むだけで、それ以外はすべて strand 上で行われます。SIGUSR1 と SIGUSR2 は意図的に用意していません。その番号が macOS では 30/31、Linux では 10/12 だからです。書いた場所と実行する場所で意味が変わる番号は、ないほうがましです。設定の再読み込みには伝統的な SIGHUP があります。シグナルを待っているだけのプログラムはデッドロックとはみなされません。応答してくれるものを待っているからです。

デッドロックは黙っていない

全員が全員を待っているプログラムは、ハングする代わりにそう伝えます。2つの strand がそれぞれ、送信する前に値を受信しようと待つ場合です。

fn main() =
  a = channel()
  b = channel()
  spawn send(a, recv(b))
  send(b, recv(a))

rill: deadlock - all strands are blocked

ランタイムは、眠っている strand や、ソケットやシグナルを待っている strand をデッドロックとは数えません。応答してくれるものを待っているからです。全員がチャネル上で互いを待っているときにだけ、状況を報告してプログラムを停止します。

ステートマシン：machine

シグナルを待ち、今いる状態に応じてシグナルが求めることを行い、また待つ strand。これは電話交換機の世界で process と呼ばれ、SDL で書かれていたものです。Rill ではチャネルと状態の上の末尾再帰関数になります。machine は、その関数をそれが本来そうであるマシンとして書くための方法です。料金を払うと開き、誰かが通るか、3秒以内に誰も通らなければ再びロックする回転式ゲートを書いてみましょう。

type Signal
  Coin(kurus: Int)
  Push
  Report
  Status(text: Str)

type Gate
  Locked(credit: Int)
  Open

machine gate(me: Chan(Mail(Signal)), log: Chan(Str))
  state Locked(credit)
    Coin(k) if credit + k >= 500 ->
      send(log, "jeton tamam, kapı açıldı")
      Open
    Coin(k) ->
      send(log, "bakiye: " + int_to_str(credit + k))
      Locked(credit + k)
    Push ->
      send(log, "kilitli, önce ödeme")
      stay
  state Open
    Push ->
      send(log, "geçildi, kilitlendi")
      Locked(0)
    Coin(_) ->
      send(log, "zaten açık, para iade")
      stay
    after 3000 ->
      send(log, "kimse geçmedi, yeniden kilitlendi")
      Locked(0)
  state *
    Report ->
      reply(Status("rapor istendi"))
      stay

fn show_log(log, n) =
  if n == 0 then ()
  else
    println(recv(log))
    show_log(log, n - 1)

fn main() =
  env = mailbox()
  g = mailbox()
  log = channel(16)
  spawn gate(g, log, Locked(0))
  post(g, env, Push)
  post(g, env, Coin(250))
  post(g, env, Coin(250))
  post(g, env, Coin(100))
  post(g, env, Push)
  post(g, env, Coin(500))
  show_log(log, 6)
  println(recv(log))
  post(g, env, Report)
  println(recv(env).signal)

kilitli, önce ödeme
bakiye: 250
jeton tamam, kapı açıldı
zaten açık, para iade
geçildi, kilitlendi
jeton tamam, kapı açıldı
kimse geçmedi, yeniden kilitlendi
Status(rapor istendi)

このマシンはほとんど仕様書のように読めます。

• 状態は普通の型（Gate）のコンストラクターです。フィールドの型は一度だけ書きます。
• 各 state の下には、その状態が応答するシグナルが match のアームのように並びます。アームの値が次の状態です。stay は現在の状態に留まること、stop は strand を終了することを意味します。
• after 3000 -> は、その時間シグナルが届かなかったときに選ばれるアームです。
• state * にはすべての状態が応答するアームを置きます。ここでは、どの状態でも report できます。
• 届くものはすべて Mail、つまりシグナルとそれを送ったメールボックスです。reply(sig) は送信者に応答します。シグナルが返信用チャネルを運ぶ必要はありません。main が外部から話しかけるときは、post(to, from, sig) で応答が返ってくるメールボックス（env）を指定します。
• どのアームにも名前が出てこないシグナルは消費され、状態は維持されます。SDL が予期しないシグナルに対して行うのと同じです。型のすべての状態がマシンに登場しなければならず、そうでなければコンパイラーが足りない状態の名前を挙げます。

コンパイラーはこのマシンを、手で書くのとまったく同じ関数に変換します。状態に対する match、各状態での recv（after がある場所では timeout 付きの select）、そして次の状態で自分自身を呼ぶ末尾呼び出しです。待機以外のコストは一切かかりません。

さらに良いことに、マシンはコードである前に図でもあります。rill diagram turnstile.rill --svg は、宣言から直接、下の SDL プロセス図を描きます。図はコードの横で別に管理されるのではなくコードから読み取られるので、決して古くなりません。

仮想時間：モデルをテストする

回転式ゲートの最後の行は、3秒の after のせいで実際に3秒待ちます。RILL_VIRTUAL_TIME=1 を付けると、プログラムは独自の時計で動きます。何かが実行されている間は時間が進まず、すべての strand が待機していて、待たれているもののうち最も近いものがタイマーであれば、時計はそこまで一気に進みます。

$ time ./turnstile
real 3.18
$ RILL_VIRTUAL_TIME=1 time ./turnstile
real 0.00

出力は同一で、時間は 3.18 秒からゼロになります。そのうえ仮想時間では実行のトレースが毎回同じになるので、モデルの振る舞いをテストできるようになります。RILL_TRACE=1 はマシンのイベントを1行ずつ書き出し、rill msc はそれらの行を、SDL ツールがかつて描いていたメッセージシーケンスチャートに変換します。

マシンにできることはもっとたくさんあります。後でマシンにシグナルを送るタイマー（set(Retry, 200)）、シグナルを後の状態のために取っておくこと（save）、優先度（priority）、ネストした状態、そしてマシン間の経路がコンパイル時にチェックされる system ブロック図などです。これらは別の記事のテーマにします。

スケール：100万個の strand

strand がどれほど安価なのか、自分で計測してみたくなりました。下のプログラムは N 個の strand を起動し、それぞれを「停止」チャネルで停止させます。そして1回の close でそれら全員を起こし、各 strand がメールボックスに報告します。

fn parked(stop, done) =
  r = select
    recv(stop) -> 1
  send(done, r)

fn count(done, k, acc) = if k == 0 then acc else count(done, k - 1, acc + recv(done))

fn main() =
  n = str_to_int(unwrap_or(nth(args(), 1), "1000"))
  stop = channel()
  done = mailbox()
  t0 = now_nanos()
  for _ in 0..n
    spawn parked(stop, done)
  yield()
  sleep_ms(1)
  t1 = now_nanos()
  close(stop)
  woke = count(done, n, 0)
  t2 = now_nanos()
  println(int_to_str(n) + " strand: başlat+park " + int_to_str((t1 - t0) / 1000000) + " ms, close ile uyandır " + int_to_str((t2 - t1) / 1000000) + " ms, uyanan " + int_to_str(woke))

1000 strand: başlat+park 1 ms, close ile uyandır 0 ms, uyanan 1000
10000 strand: başlat+park 5 ms, close ile uyandır 1 ms, uyanan 10000
100000 strand: başlat+park 22 ms, close ile uyandır 11 ms, uyanan 100000
1000000 strand: başlat+park 162 ms, close ile uyandır 124 ms, uyanan 1000000

100万個の strand が 162 ms で起動して停止し、1回の close ブロードキャストで 124 ms のうちに全員が起きます。時間もメモリも線形にスケールします。メモリについて正直に言うと、このプログラムでは strand 1つあたり約 1.5 KB になります。リポジトリの「停止中の strand ≈ 300 バイト」より多いのは、ここではすべての strand が select の中で待ち、起きたあとに上限のないメールボックスへ書き込むからです。100万件のメッセージを抱えたそのメールボックスもメモリに含まれています。これらは1回限りの計測で、正確な数値よりも桁の大きさを示すことが目的です。

複数のコア

デフォルトでは、すべての strand は OS スレッド1本の上で動きます。そのおかげで参照カウントに非アトミックな、つまり高速な操作を使えます。本当の並列性は、ビルド時と実行時の2回、明示的に要求します。

rill build main.rill -o main --parallel   # reference counts become safe across threads
RILL_THREADS=8 ./main                     # run on eight workers

ワーカーは、まだ開始していない strand を互いに奪い合います（work stealing）。すでに開始した strand は自分のワーカーに留まります。スタックフレームがそのワーカーのアドレスにあるからです。並列性が役立つのは、strand がそれぞれ独立に計算する場合です。ある strand が次の strand に値を渡していくランデブーチェーン（上のふるいやリングのようなもの）には重ねられる部分がなく、ワーカー1つで動かすのが最速です。

共有はコンパイル時にチェックされる

2つの strand の間で食い違いが起こりうる唯一の種類の値はストレージ、つまり Buf、Map、StrBuf、あるいはそれらを保持する何かです。一方が書き込んでいる間に他方が読み書きすれば、それは競合です。--parallel ビルドは、プログラムを実行する前に strand が何を共有しているかをチェックします。

fn fill(b: Buf(Int), v) = for i in 0..#b
  b[i] = v

fn main() =
  b = buf_int(8)
  spawn fill(b, 1)
  spawn fill(b, 2)
  println(int_to_str(b[0]))

$ rill check share.rill --parallel
share.rill: type error at line 6: `b` is written by this strand and still used here after
the `spawn`: two strands writing one piece of storage is a race. Hand it over (do not use
it after the spawn), or send each strand its own, or, where the writes are kept apart on
purpose, say so: `apart(b)`

規則は単純です。

1. strand に渡された値（spawn にキャプチャされたもの、またはチャネルで送られたもの）で、渡した側がその後使わないものは、その strand のものになります。何をしても構いません。
2. その後も使われる値は共有されています。読むための共有は問題ありません。共有して書き込むことは拒否され、エラーは値と行を名指しします。
3. 関数値は不透明です。それが何を保持し何に書き込むかは型に表れません。プログラムが知っていることを明示しない限り、strand 間での共有は拒否されます。

プログラムが知っていることは apart(x) で表明します。「この値は意図的に共有されていて、書き込みは互いに分離されている」という意味です。たとえば、各 strand がバッファの自分の範囲に書き込んでいる場合です。コンパイラーはその主張を検証できませんが、主張が書かれていることは確認します。ワーカー1つのビルドでは、strand が交互に動くのは待機する場所だけなので、同じプログラムでもそこではチェックされません。

実例：このブログのサーバー

このページはまさにこれらのツールで配信されています。Rill の HTTP サーバーは接続ごとに strand を開き、静かな接続の strand は停止していて何も保持しません。データベース接続はバッファ付きチャネルの中に置かれたプールです。リクエストは接続を1つ取り、使って、戻します。接続を見つけられなかったリクエストは待つので、サーバーはタダでバックプレッシャーを得られます。

fn with_conn(pool, ask) =
  fd = recv(pool)
  res = ask(fd)
  send(pool, fd)
  res

記事が読まれると、閲覧記録は別の strand でデータベースに書き込まれます。ページはデータベースがディスクに書き込むのを待ちません。

fn spawn_view(app: App, slug, referrer) =
  spawn with_conn(app.db, \fd -> record_view(fd, slug, referrer))
  ()

ここでは、spawn が式全体を新しい strand で実行することがまさに望みどおりの動作です。プールから接続を取り、書き込み、接続を戻す処理がすべて、リクエストに応答する strand の外で行われます。

まとめ

• channel() はハンドシェイクです。send が戻ったことは recv が完了したことの証明です。channel(n) は速度差を吸収してバックプレッシャーを与えます。mailbox() は決して待ちません。
• close はブロードキャストで、バッファ内のものを失わず、2回行っても構いません。recv_opt は終わりのあるチャネルのためのものです。規則を覚えておきましょう：送る側が close する。
• select は複数のチャネルを待ちます。default はブロックしないポーリング、timeout(ms) は strand を消費しない締め切りです。
• wait_signal + close はグレースフルシャットダウンの2つの半分です。本物のデッドロックは黙っていません。
• machine は strand を SDL スタイルのステートマシンとして書きます。rill diagram がそれを図にし、仮想時間がテスト可能にします。
• --parallel は本当の並列性を与え、共有ストレージへの書き込みをコンパイル時に拒否します。意図的な例外は apart で表明します。

strand はとても安価なので、設計上の問いは「スレッドをいくつ開けるか？」ではなく「この仕事の自然な形は何か？」に変わります。接続は1つの strand、段は1つの strand、素数は1つの strand です。

次の記事では、このブログを支えるデータベース OxiDB について書きます。質問や試してみたことを X や LinkedIn で聞かせてもらえると嬉しいです。