Programming in D - ファイバー

ファイバー

ファイバーは、単一のスレッドが複数のタスクを実行可能にする実行スレッドである。並列処理や並行処理で一般的に使用される通常のスレッドと比べて、ファイバー間の切り替えはより効率的だ。ファイバーは、コルーチンや グリーンスレッドと類似している。

ファイバーは、1つのスレッドにつき複数のコールスタックを可能にする。そのため、ファイバーを完全に理解し、その利点を理解するためには、まずスレッドのコールスタックを理解する必要がある。

コールスタック

関数のパラメータ、非静的ローカル変数、戻り値、一時式、および実行中に必要となる追加情報は、その関数のローカル状態を構成する。関数のローカル状態は、その関数が呼び出されるたびに、実行時に自動的に割り当てられ、初期化される。

関数呼び出しのローカル状態用に割り当てられる記憶領域は、フレーム(またはスタックフレーム)と呼ばれる。スレッドの実行中に、関数が他の関数を呼び出すと、それらのフレームは概念的には互いに積み重ねられて、フレームのスタックを形成する。現在アクティブな関数呼び出しのフレームのスタックが、そのスレッドのコールスタックとなる。

例えば、次のプログラムのメインスレッドがbar()関数の実行を開始したとき、main()がfoo()を呼び出し、foo()がbar()を呼び出すため、3つのレベルのアクティブな関数呼び出しがある。

void main() {
    int a;
    int b;

    int c = foo(a, b);
}

int foo(int x, int y) {
    bar(x + y);
    return 42;
}

void bar(int param) {
    string[] arr;
    // ...
}

bar()の実行中、コールスタックは、現在アクティブな関数呼び出しのローカル状態を格納する3つのフレームで構成される。

関数呼び出しが深くなるにつれて、
コールスタックは上方向に成長する。            ▲  ▲
                                 │  │
コールスタックの先頭    → ┌──────────────┐
                           │ int param    │ ← barのフレーム
                           │ string[] arr │
                           ├──────────────┤
                           │ int x        │
                           │ int y        │ ← fooのフレーム
                           │ return value │
                           ├──────────────┤
                           │ int a        │
                           │ int b        │ ← mainのフレーム
                           │ int c        │
コールスタックの最下部  → └──────────────┘

関数が他の関数を呼び出すと関数呼び出しの層が深くなり、関数が戻ると層が浅くなるため、呼び出しスタックのサイズはそれに応じて増減する。例えば、bar()が戻ると、そのフレームは不要になり、そのスペースは後で別の関数呼び出しに使用される。

                           ┌──────────────┐
                           │ int param    │
                           │ string[] arr │
コールスタックの先頭    → ├──────────────┤
                           │ int a        │
                           │ int b        │ ← fooのフレーム
                           │ return value │
                           ├──────────────┤
                           │ int a        │
                           │ int b        │ ← mainのフレーム
                           │ int c        │
コールスタックの最下部  → └──────────────┘

これまで作成したすべてのプログラムで、コールスタックの利点を活用してきた。コールスタックの利点は、再帰関数で特に明らかだ。

再帰

再帰とは、関数が直接または間接的に自分自身を呼び出す状況のことである。再帰は、分割統治法に分類されるような特定の種類のアルゴリズムを大幅に簡略化する。

スライスの要素の合計を計算する次の関数を考えてみよう。この関数は、受け取ったスライスより1要素短いスライスを引数として、自分自身を再帰的に呼び出すことでこのタスクを実行する。再帰は、スライスが空になるまで続く。現在の結果は、2番目の引数として次の再帰ステップに引き継がれる。

import std.array;

int sum(int[] arr, int currentSum = 0) {
    if (arr.empty) {
        /* 追加する要素はない。結果は、これまでに
         * 計算されたものだ。 */
        return currentSum;
    }

    /* 最初の要素を現在の合計に追加し、
     * 残りの要素でselfを呼び出す。 */
    return sum(arr[1..$], currentSum + arr.front);
}

void main() {
    assert(sum([1, 2, 3]) == 6);
}

注釈:上記のコードは、説明のためのものである。実際には、範囲の要素の合計は、浮動小数点型に対してより正確な計算を行う特別なアルゴリズムを使用するstd.algorithm.sum

上記の[1, 2, 3]の最初の引数として空のスライスが指定されて、最終的にsum()が呼び出されると、呼び出しスタックの関連部分は次のフレームで構成される。各パラメータの値は、==記号の後に示されている。フレームの内容は下から上に向かって読むことを忘れないで。

              ┌─────────────────────────┐
              │ arr        == []        │ ← sumの最終呼び出し
              │ currentSum == 6         │
              ├─────────────────────────┤
              │ arr        == [3]       │ ← sumの3回目の呼び出し
              │ currentSum == 3         │
              ├─────────────────────────┤
              │ arr        == [2, 3]    │ ← sumの2回目の呼び出し
              │ currentSum == 1         │
              ├─────────────────────────┤
              │ arr        == [1, 2, 3] │ ← sumの最初の呼び出し
              │ currentSum == 0         │
              ├─────────────────────────┤
              │            ...          │ ← mainのフレーム
              └─────────────────────────┘

注釈:実際には、再帰関数がそれ自体の呼び出しの結果を直接返す場合、コンパイラは"末尾呼び出し最適化"と呼ばれる手法を使用して、再帰呼び出しごとに個別のフレームを削除する。

マルチスレッドプログラムでは、各スレッドが独自のタスクを実行するため、各スレッドは独自の実行状態を維持するための独自のコールスタックを取得する。

ファイバーの力は、ファイバーはスレッドではないにもかかわらず、独自のコールスタックを取得し、実質的にスレッドごとに複数のコールスタックを可能にするという事実に基づいている。1つのコールスタックが1つのタスクの実行状態を維持するため、複数のコールスタックにより、スレッドは複数のタスクを処理できる。

使用

以下は、ファイバーの一般的な操作である。これらの例については、後で説明する。

ファイバーは、引数を受け取らず、何も返さない呼び出し可能エンティティ (関数ポインタ、デリゲートなど) から実行を開始する。例えば、次の関数はファイバー関数として使用できる。
```
void fiberFunction() {
    // ...
}
```
D
IDEで開く
ファイバーは、呼び出し可能なエンティティを引数として、core.thread.Fiberクラスのオブジェクトとして作成できる。
```
import core.thread;

// ...

    auto fiber = new Fiber(&fiberFunction);
```
D
IDEで開く
あるいは、Fiberのサブクラスを定義し、ファイバー関数をスーパークラスのコンストラクタに渡すこともできる。次の例では、ファイバー関数はメンバー関数である。
```
class MyFiber : Fiber {
    this() {
        super(&run);
    }

    void run() {
        // ...
    }
}

// ...

    auto fiber = new MyFiber();
```
D
IDEで開く
ファイバーは、そのcall()メンバー関数によって開始および再開される。
```
fiber.call();
```
D
IDEで開く
スレッドとは異なり、ファイバーの実行中は呼び出し元が一時停止される。
ファイバーは、Fiber.yield()によって自身を一時停止する(呼び出し元に実行を譲渡する)。
```
void fiberFunction() {
    // ...

        Fiber.yield();

    // ...
}
```
D
IDEで開く
ファイバーが実行を譲渡すると、呼び出し元の実行が再開される。
ファイバーの実行状態は、その.stateプロパティによって決まる。
```
if (fiber.state == Fiber.State.TERM) {
    // ...
}
```
D
IDEで開く
Fiber.Stateは、次の値を持つ列挙型である。
- HOLD: ファイバーは一時停止中で、開始または再開可能である。
- EXEC: ファイバーは現在実行中である。
- TERM: ファイバーが終了した。再利用するには、reset()にする必要がある。

範囲実装におけるファイバー

ほとんどすべての範囲は、反復の状態を記憶するために何らかの情報を格納する必要がある。これは、popFront()が次に呼び出されたときに何をすべきかを判断するために必要だ。Rangesおよびそれ以降の章で見たほとんどの範囲の例は、そのタスクを実行するために何らかの状態を格納していた。

例えば、先ほど定義したFibonacciSeriesは、系列の次の次の数を計算するために2つのメンバー変数を保持していた。

struct FibonacciSeries {
    int current = 0;
    int next = 1;

    enum empty = false;

    int front() const {
        return current;
    }

    void popFront() {
        const nextNext = current + next;
        current = next;
        next = nextNext;
    }
}

FibonacciSeriesのような一部の範囲では反復状態を維持することは単純だが、再帰的なデータ構造(例えば二分木)では意外に難しい。これが意外なのは、このようなデータ構造では、再帰的に実装した場合、同じアルゴリズムが単純だからだ。

例えば、insert()およびprint()の次の再帰的実装は、変数を一切定義しておらず、ツリーに含まれる要素の数に依存しない。再帰呼び出しは強調表示されている。(insert()は、insertOrSet()を通じて間接的に再帰的であることに注意。)

import std.stdio;
import std.string;
import std.conv;
import std.random;
import std.range;
import std.algorithm;

/* バイナリツリーのノードを表す。この型は、
 * 以下の構造体Treeの実装で使用されるため、
 * 直接使用してはならない。 */
struct Node {
    int element;
    Node * left;     // 左サブツリー
    Node * right;    // 右サブツリー

    void insert(int element) {
        if (element < this.element) {
            /* 小さい要素は左サブツリーに入る。 */
            insertOrSet(left, element);

        } else if (element > this.element) {
            /* 大きい要素は右サブツリーに入る。 */
            insertOrSet(right, element);

        } else {
            throw new Exception(format("%s already exists",
                                       element));
        }
    }

    void print() const {
        /* まず、左サブツリーの要素を表示する */
        if (left) {
            left.print();
            write(' ');
        }

        /* 次にこの要素を表示する */
        write(element);

        /* 最後に右サブツリーの要素を表示する */
        if (right) {
            write(' ');
            right.print();
        }
    }
}

/* 指定されたサブツリーに要素を挿入し、必要に応じて
 * そのノードを初期化する。 */
void insertOrSet(ref Node * node, int element) {
    if (!node) {
        /* このサブツリーの最初の要素。 */
        node = new Node(element);

    } else {
        node.insert(element);
    }
}

/* これは実際のツリー表現。'root'が'null'である場合、空の
 * ツリーを許可する。 */
struct Tree {
    Node * root;

    /* 要素をこのツリーに挿入する。 */
    void insert(int element) {
        insertOrSet(root, element);
    }

    /* 要素をソート順に表示する。 */
    void print() const {
        if (root) {
            root.print();
        }
    }
}

/* '10 * n'個の数字からランダムに選んだ
 * 'n'個の数字でツリーを埋める。 */
Tree makeRandomTree(size_t n) {
    auto numbers = iota((n * 10).to!int)
                   .randomSample(n, Random(unpredictableSeed))
                   .array;

    randomShuffle(numbers);

    /* それらの数字でツリーを埋める。 */
    auto tree = Tree();
    numbers.each!(e => tree.insert(e));

    return tree;
}

void main() {
    auto tree = makeRandomTree(10);
    tree.print();
}

注釈:上記のプログラムは、Phobosの以下の機能を使用している。

std.range.iota 指定された値の範囲の要素を遅延生成する。(デフォルトでは、最初の要素は.initの値になる。)例えば、iota(10)は、0から9までのint要素の範囲だ。
std.algorithm.each std.algorithm.mapと類似している。map()は各要素に対して新しい結果を生成するが、each()は各要素に対して副作用を生成する。さらに、map()は遅延評価であり、each()は即時評価である。
std.random.randomSample 指定された範囲から要素のランダムなサンプルを選択し、その順序を変更しない。
std.random.randomShuffle 範囲内の要素をランダムにシャッフルする。

ほとんどのコンテナと同様に、このツリーにも、その要素を既存のrangeアルゴリズムで使用できるように、rangeインターフェースを提供したいと思うだろう。これは、opSlice()メンバー関数を定義することで実現できる。

struct Tree {
// ...

    /* ツリー内の要素に、ソートされた順で
     * アクセスする。 */
    struct InOrderRange {
        ... What should the implementation be? ...
    }

    InOrderRange opSlice() const {
        return InOrderRange(root);
    }
}

上記のprint()メンバー関数は、基本的に、ソートされた順にすべての要素を訪問するという同じタスクを実現しているが、ツリーに対してInputRangeを実装するのは簡単ではない。ここではInOrderRangeの実装は試みませんが、ツリーイテレータを実装するか、少なくとも研究することをお勧めする。(一部の実装では、ツリーノードに、各ノードの親を指す追加のNode*が必要になる。

print()のような再帰的な木アルゴリズムが単純なのは、コールスタックが自動的に管理されるためだ。コールスタックには、現在の要素が何かだけでなく、プログラムの実行がどのようにその要素に到達したか(例えば、実行が左ノードと右ノードのどちらを追跡したか)に関する情報も暗黙的に含まれている。

例えば、left.print()の再帰呼び出しが左サブツリーの要素を出力して戻った場合、現在のprint()呼び出しのローカル状態は、スペース文字を出力するタイミングが来たことをすでに意味している。

void print() const {
    if (left) {
        left.print();
        write(' ');   // ← コールスタックはこれが次であることを意味する
    }

    // ...
}

ファイバーは、コールスタックを使用する方が明示的に状態を維持するよりもはるかに簡単な、同様のケースで有用だ。

ファイバーの利点は、フィボナッチ数列を生成するような単純なタスクでは明らかではないが、簡素化のため、ファイバーの実装における一般的なファイバー操作を説明する。後で木範囲の実装を実装する。

import core.thread;

/* これは、各要素を生成し、
 * その後、その要素に'ref'パラメータを設定するファイバー関数だ。 */
void fibonacciSeries(ref int current) {                 // (1)
    current = 0;    // 'current'はパラメータであることに注意
    int next = 1;

    while (true) {
        Fiber.yield();                                  // (2)
        /* 次のcall()はこの地点から続行される */ // (3)

        const nextNext = current + next;
        current = next;
        next = nextNext;
    }
}

void main() {
    int current;                                        // (1)
                         // (4)
    Fiber fiber = new Fiber(() => fibonacciSeries(current));

    foreach (_; 0 .. 10) {
        fiber.call();                                   // (5)

        import std.stdio;
        writef("%s ", current);
    }
}

上記のファイバー関数は、intへの参照を受け取る。このパラメータを使用して、現在の要素を呼び出し元に伝える。(このパラメータは、refではなくoutとして指定することもできる)。
現在の要素が使用可能になると、ファイバーはFiber.yield()を呼び出して自身を一時停止する。
その後、call()が呼び出されると、ファイバーの最後のFiber.yield()呼び出しの直後に関数が再開される。(最初のcall()が関数を起動する。)
ファイバー関数はパラメータを受け取らないため、fibonacciSeries()はファイバー関数として直接使用することはできない。その代わりに、パラメータのないデリゲートをアダプタとして使用し、Fiberコンストラクタに渡す。
呼び出し元は、call()メンバー関数によってファイバーを開始および再開する。

その結果、main()はcurrentを通じてシリーズの要素を受け取り、それらを出力する:

`std.concurrency.Generator`ファイバーを範囲として表現するため

ファイバーを使用してフィボナッチ数列を生成する方法は実現したが、その実装には以下の欠点がある:

上記の解決策は範囲インターフェースを提供しないため、既存の範囲アルゴリズムと互換性がない。
refパラメータをmutateして要素を表示する方法は、要素を呼び出し元のコンテキストにコピーする設計に比べて望ましくない。
ファイバーをそのメンバー関数を使って明示的に構築して使用すると、他の設計に比べて、より低レベルの実装の詳細が露見してしまう。

std.concurrency.Generatorクラスは、これらの問題すべてに対処している。以下のfibonacciSeries()が単純な関数として記述されていることに注目しよう。唯一の違いは、returnによって単一の要素を返す代わりに、yield()によって複数の要素(この例では無限の要素)を利用可能にする点だ。

また、今回は、上記で使用したFiber.yieldメンバー関数ではなく、std.concurrency.yield関数を使用していることに注意。

import std.stdio;
import std.range;
import std.concurrency;

/* このエイリアスは、std.range.Generatorとの名前衝突を
 * 解決するために使用される。 */
alias FiberRange = std.concurrency.Generator;

void fibonacciSeries() {
    int current = 0;
    int next = 1;

    while (true) {
        yield(current);

        const nextNext = current + next;
        current = next;
        next = nextNext;
    }
}

void main() {
    auto series = new FiberRange!int(&fibonacciSeries);
    writefln("%(%s %)", series.take(10));
}

その結果、ファイバー関数によって生成された要素は、InputRangeとして便利に使用できる。

Generatorを使用すると、ツリーの要素をInputRangeとして簡単に表現することもできる。さらに、ツリーにInputRangeインターフェースが定義されていると、print()メンバー関数は不要になるため、削除することができる。特に、byNode()が、再帰関数nextNode()のアダプタとして実装されている点に注意。

import std.concurrency;

alias FiberRange = std.concurrency.Generator;

struct Node {
// ...

    /* 注釈: print()メンバー関数は、
     * もう必要ないため削除された。 */

    auto opSlice() const {
        return byNode(&this);
    }
}

/* これは、ソートされた順で次のツリーノードを
 * 返すファイバー関数である。 */
void nextNode(const(Node) * node) {
    if (!node) {
        /* このnode以下には要素はない */
        return;
    }

    nextNode(node.left);    // まず、左側の要素
    yield(node);            // 次に、この要素
    nextNode(node.right);   // 最後に、右側の要素
}

/* ツリーのノードへのInputRangeを返す。 */
auto byNode(const(Node) * node) {
    return new FiberRange!(const(Node)*)(
        () => nextNode(node));
}

// ...

struct Tree {
// ...

    /* 注釈: print()メンバー関数は、
     * もう必要ないため削除された。 */

    auto opSlice() const {
        /* nodeから要素への変換。 */
        return byNode(this).map!(n => n.element);
    }
}

/* ツリーのnodeへのInputRangeを返す。
 * ツリーに要素がない場合(つまり、
 * 'root'が'null'の場合)、返される範囲は空になる。 */
auto byNode(const(Tree) tree) {
    if (tree.root) {
        return byNode(tree.root);

    } else {
        alias RangeType = typeof(return);
        return new RangeType(() {});    // ← 空の範囲
    }
}

Treeオブジェクトは[]でスライスでき、結果はInputRangeとして使用できる:

writefln("%(%s %)", tree[]);

非同期入出力におけるファイバー

ファイバーのコールスタックは、非同期入出力タスクを簡素化することもできる。

例として、ユーザーがサーバーに接続し、名前、メールアドレス、年齢の順で入力してサービスにサインオンするシステムを考えてみよう。これは、ウェブサイトのサインオンユーザーフローに似ている。例をシンプルにするため、実際のウェブサービスを実装する代わりに、コマンドラインに入力されたデータを使用してユーザーの操作をシミュレートしよう。入力データが強調表示された、以下のシンプルなサインオンプロトコルを使用しよう。

hi: ユーザーが接続し、フローIDが生成される
id data:idに対応するフローのユーザーが、次に期待されるデータを入力する。例えば、フロー42で期待されるデータが名前である場合、Aliceのコマンドは42 Aliceとなる。
bye: プログラムが終了する

例えば、シミュレーションプログラムへの入力が強調表示された、AliceとBobのやり取りは次のようになる。各ユーザーが接続し、名前、E メールアドレス、年齢を入力する。

> こんにちは                     ← アリスが接続
フロー0が開始された。
> 0 アリス
> 0 alice@example.com
> 0 20
フロー0が完了した。    ← アリスが終了
ユーザー'Alice'を追加した。
> こんにちは                     ← Bobが接続した
フロー1が開始された。
> 1 ボブ
> 1 bob@example.com
> 1 30
フロー1が完了した。    ← Bobが終了した
ユーザー'Bob'を追加した。
> さようなら
さようなら。
ユーザー:
  User("Alice", "alice@example.com", 20)
  User("Bob", "bob@example.com", 30)

このプログラムは、hiコマンドをループで待機し、接続しているユーザーの入力データを受け取る関数を呼び出すように設計することができる。

if (input == "hi") {
    signOnNewUser();    // ← 警告: ブロックするデザイン
}

プログラムにマルチタスクのサポートがない場合、このような設計はブロック型とみなされ、現在のユーザーがサインオンフローを完了するまで他のすべてのユーザーがブロックされる。ユーザーがデータを提供するのに数分かかる場合、この設計は使用頻度の低いサービスでも応答性に影響を与える。

このサービスを非ブロッキングにする設計は複数あり、複数のユーザーが同時にサインオンできるようになる:

タスクを明示的に維持する:メインスレッドは、ユーザーサインオンごとに1つのスレッドを生成し、メッセージによってそのスレッドに入力データを渡すことができる。この方法は機能するが、スレッドの同期が必要になり、ファイバーよりも速度が低下する可能性がある。(この潜在的なパフォーマンスの低下の理由については、以下の協調型マルチタスクのセクションで説明する。)
状態を維持する:プログラムは複数のサインオンを受け入れ、各サインオンの状態を明示的に記憶することができる。例えば、アリスがこれまでに名前だけを入力した場合、彼女の状態は、次の入力データが彼女のEメールアドレスであることを示す必要がある。

あるいは、ファイバーに基づく設計では、サインオンフローごとに1つのファイバーを使用することができる。これにより、プロトコルに完全に一致した、直線的なフローを実装することができる。まず名前、次にEメールアドレス、最後に年齢という順になる。例えば、以下のrun()では、サインオンフローの状態を記憶するために特別な処理は必要ない。次回call()が呼び出されると、一時停止していた最後のFiber.yield()呼び出しの直後に実行が再開される。次に実行される行は、コールスタックによって暗黙的に決定される。

以前の例とは異なり、次のプログラムはFiberのサブクラスを使用している。

import std.stdio;
import std.string;
import std.format;
import std.exception;
import std.conv;
import std.array;
import core.thread;

struct User {
    string name;
    string email;
    uint age;
}

/* このFiberサブクラスは、ユーザーの
 * サインオンフローを表している。 */
class SignOnFlow : Fiber {
    /* このフローで最後に読み込まれたデータ。 */
    string inputData_;

    /* ユーザーオブジェクトを構築するための情報。 */
    string name;
    string email;
    uint age;

    this() {
        /* 'run'メンバー関数を
         * ファイバーの開始点として設定する。 */
        super(&run);
    }

    void run() {
        /* 最初の入力は名前。 */
        name = inputData_;
        Fiber.yield();

        /* 2番目の入力はメールアドレス。 */
        email = inputData_;
        Fiber.yield();

        /* 最後の入力は年齢。 */
        age = inputData_.to!uint;

        /* この時点で、ユーザーを構築するためのすべての情報を
         * 収集した。ここで'Fiber.yield()'ではなく
         * 'return'を実行する。その結果、このファイバーの状態は
         * Fiber.State.TERMになる。 */
    }

    /* このプロパティ関数は、呼び出し元からデータを受け取る
     * ためのもの。 */
    void inputData(string data) {
        inputData_ = data;
    }

    /* このプロパティ関数は、ユーザーを構築し、
     * それを呼び出し元に返すためのもの。 */
    User user() const {
        return User(name, email, age);
    }
}

/* 特定のフローの入力から読み込まれたデータを表す。 */
struct FlowData {
    size_t id;
    string data;
}

/* フローに関連するデータを解析する。 */
FlowData parseFlowData(string line) {
    size_t id;
    string data;

    const items = line.formattedRead!" %s %s"(id, data);
    enforce(items == 2, format("Bad input '%s'.", line));

    return FlowData(id, data);
}

void main() {
    User[] users;
    SignOnFlow[] flows;

    bool done = false;

    while (!done) {
        write("> ");
        string line = readln.strip;

        switch (line) {
        case "hi":
            /* 新しい接続のフローを開始する。 */
            flows ~= new SignOnFlow();

            writefln("Flow %s started.", flows.length - 1);
            break;

        case "bye":
            /* プログラムを終了する。 */
            done = true;
            break;

        default:
            /* 入力データをフローデータとして使用しようとする。 */
            try {
                auto user = handleFlowData(line, flows);

                if (!user.name.empty) {
                    users ~= user;
                    writefln("Added user '%s'.", user.name);
                }

            } catch (Exception exc) {
                writefln("Error: %s", exc.msg);
            }
            break;
        }
    }

    writeln("Goodbye.");
    writefln("Users:\n%(  %s\n%)", users);
}

/* 入力の所有ファイバーを識別し、その入力データを設定し、
 * そのファイバーを再開する。フローが完了している場合は、
 * 有効なフィールドを持つユーザーを返す。 */
User handleFlowData(string line, SignOnFlow[] flows) {
    const input = parseFlowData(line);
    const id = input.id;

    enforce(id < flows.length, format("Invalid id: %s.", id));

    auto flow = flows[id];

    enforce(flow.state == Fiber.State.HOLD,
            format("Flow %s is not runnable.", id));

    /* フローデータを設定する。 */
    flow.inputData = input.data;

    /* フローを再開する。 */
    flow.call();

    User user;

    if (flow.state == Fiber.State.TERM) {
        writefln("Flow %s has completed.", id);

        /* 戻り値を新しく作成されたユーザーに設定する。 */
        user = flow.user;

        /* TODO: このファイバーの'flows'配列内のエントリは、
         * 将来新しいフローで再利用可能。ただし、
         * まず'flow.reset()'でリセットする必要がある。 */
    }

    return user;
}

main() 入力から行を読み取り、それらを解析し、処理される適切なフローにフローデータをディスパッチする。各フローのコールスタックは、フローの状態を自動的に維持する。完全なユーザー情報が利用可能になると、新しいユーザーがシステムに追加される。

上記のプログラムを実行すると、ユーザーが個々のサインオンフローを完了するのにどれほど時間がかかっても、システムは常に新しいユーザーの接続を受け入れることがわかる。例として、Aliceの操作を強調表示する。

> こんにちは                     ← アリスが接続
フロー0が開始された。
> 0 アリス
> こんにちは                     ← ボブが接続
フロー1が開始された。
> こんにちは                     ← シンディが接続
フロー2が開始された。
> 0 alice@example.com
> 1 ボブ
> 2 シンディ
> 2 cindy@example.com
> 2 40                   ← シンディが終了
フロー2が完了した。
ユーザー'Cindy'を追加した。
> 1 bob@example.com
> 1 30                   ← ボブが終了
フロー1が完了した。
ユーザー'Bob'を追加した。
> 0 20                   ← アリスが終了
フロー0が完了した。
ユーザー'Alice'を追加した。
> さようなら
さようなら。
ユーザー:
  User("Cindy", "cindy@example.com", 40)
  User("Bob", "bob@example.com", 30)
  User("Alice", "alice@example.com", 20)

Alice、Bob、Cindyがその順で接続しても、サインオンフローを完了するペースは異なる。その結果、users配列はフローが完了した順に埋まっていく。

このプログラムでファイバーを使用するメリットの1つは、SignOnFlow.run()がユーザーの入力速度に関係なく簡単に記述できることだ。さらに、他のサインオンフローが進行中でもユーザーがブロックされることはない。

vibe.dのような多くの非同期入出力フレームワークは、ファイバーに基づく同様の設計を採用している。

例外とファイバー

例外の章では、"下位レベルの関数からスローされた例外オブジェクトは、1レベルずつ上位の関数に転送される"ことを説明した。また、キャッチされなかった例外は"main()関数を最終的に終了させる"ことも説明した。その章ではコールスタックについては触れていなかったが、例外メカニズムのこの動作もコールスタックによって実現されている。

この章の最初の例を続けて、bar()内で例外がスローされた場合、まずbar()のフレームがコールスタックから削除され、次にfoo()のフレームが削除され、最後にmain()のフレームが削除される。関数が終了し、そのフレームがコールスタックから削除されると、ローカル変数のデストラクタが実行され、最終的な操作が行われる。例外がスローされたために関数を終了し、ローカル変数のデストラクタを実行するプロセスは、スタック巻き戻しと呼ばれる。

ファイバーは独自のスタックを持っているため、ファイバーの実行中にスローされた例外は、そのファイバーの呼び出し元ではなく、そのファイバーの呼び出しスタックをアンワインドする。例外がキャッチされなかった場合、ファイバー関数は終了し、ファイバーの状態はFiber.State.TERMになる。

これは場合によっては望ましい動作だが、ファイバーは、実行状態を失うことなく、呼び出し元にエラー状態を通知する必要がある場合もある。Fiber.yieldAndThrowを使用すると、ファイバーは、呼び出し元のコンテキストで例外を即座にスローして、実行を中断することができる。

使用例として、サインオンプログラムに無効な年齢データを渡す場合を考えてみよう。

> こんにちは
フロー0が開始された。
> 0 アリス
> 0 alice@example.com
> 0 こんにちは                       ← ユーザーが無効な年齢を入力した
エラー: 文字列型からuint型に変換する際に予期しない'h'が検出された
> 0 20                          ← エラーを修正しようとした
エラー: フロー0は実行できない。  ← しかし、フローは終了した

ファイバーを終了してサインオンフロー全体を失う代わりに、ファイバーは変換エラーをキャッチし、yieldAndThrow()を使用して呼び出し元に通知することができる。これは、プログラムでファイバーが年齢データを変換する次の行を次のように置き換えることで実現できる:

age = inputData_.to!uint;

この行を、無条件ループ内のtry-catch文で囲むだけで、uintに変換できるデータがあるまでファイバーを存続させることができる。

	while (true) {
        try {
            age = inputData_.to!uint;
            break;  // ← 変換成功; ループはそのままに

        } catch (ConvException exc) {
            Fiber.yieldAndThrow(exc);
        }
    }

この場合、ファイバーはデータが有効になるまで無条件ループ内に留まる:

> こんにちは
フロー0が開始された。
> 0 アリス
> 0 alice@example.com
> 0 こんにちは                       ← ユーザーが無効な年齢を入力した
エラー: 文字列型からuint型に変換する際に予期しない'h'が検出された
> 0 world                       ← enters invalid age again
エラー: 文字列型からuint型に変換する際に予期しない'w'が検出された
> 0 20                          ← ついに有効なデータを入力した
フロー0が完了した。
ユーザー'Alice'が追加された。
> さようなら
さようなら。
ユーザー:
  User("Alice", "alice@example.com", 20)

出力からわかるように、今回はサインオンフローが失われることはなく、ユーザーはシステムに追加される。

協調型マルチタスク

オペレーティングシステムのスレッドは、オペレーティングシステムによって不確定なタイミングで一時停止(サスペンド)され、再開されるのとは異なり、ファイバーは明示的に自分自身を一時停止し、呼び出し元によって明示的に再開される。この違いにより、オペレーティングシステムが提供するマルチタスクは"プリエンプティブマルチタスク"と呼ばれ、ファイバーが提供するマルチタスクは"協調マルチタスク"と呼ばれる。

プリエンプティブマルチタスクでは、オペレーティングシステムは、スレッドの実行を開始または再開する際に、そのスレッドに一定量の時間を割り当てる。その時間が経過すると、そのスレッドは一時停止され、別のスレッドがその代わりに再開される。あるスレッドから別のスレッドに移動することを、コンテキストスイッチと呼ぶ。コンテキストスイッチには比較的長い時間がかかり、その時間はスレッドによる実際の作業に費やしたほうがいいだろう。

システムは通常、多数のthreadで忙しく動作しているため、コンテキストの切り替えは避けられないだけでなく、実際には望ましいことでもある。しかし、スレッドは、割り当てられた時間をすべて使い切る前に、自発的に一時停止する必要がある場合がある。これは、スレッドが別のスレッドやデバイスからの情報を必要とする場合に発生する。スレッドが一時停止すると、オペレーティングシステムは別のスレッドに切り替えるために再び時間を費やす必要がある。その結果、実際の作業に使用できたはずの時間が、コンテキストの切り替えに費やされてしまう。

ファイバーを使用すると、呼び出し元とファイバーは、同じスレッドの一部として実行される。(これが、呼び出し元とファイバーが同時に実行できない理由だ。) この利点として、呼び出し元とファイバー間のコンテキストの切り替えによるオーバーヘッドがない。(ただし、通常の関数呼び出しのオーバーヘッドと同程度のわずかなオーバーヘッドは発生する。)

協調型マルチタスクのもう1つの利点は、呼び出し元とファイバーが交換するデータがCPUのデータキャッシュにある可能性が高くなることだ。CPUキャッシュにあるデータは、システムメモリから読み戻す必要のあるデータよりも数百倍も高速にアクセスできるため、ファイバーのパフォーマンスがさらに向上する。

さらに、呼び出し元とファイバーは同時に実行されないため、レースコンディションが発生する可能性がなく、データの同期化も不要になる。ただし、ファイバーが意図したタイミング(例えば、データが実際に準備できたとき)に確実にイールドすることを、プログラマは確認する必要がある。例えば、以下のfunc()呼び出しは、sharedDataの値が2倍になる前に、間接的にであってもFiber.yield()呼び出しを実行してはならない。

void fiberFunction() {
    // ...

        func();           // ← 早すぎるyieldは不可
        sharedData *= 2;
        Fiber.yield();    // ← yieldする意図のあるポイント

    // ...
}

ファイバーの明らかな欠点の一つは、呼び出し元とそのファイバーにCPUの1つのコアしか使用されないことだ。CPUの他のコアはアイドル状態になり、リソースが無駄になる可能性がある。M:Nスレッドモデル(ハイブリッドスレッド)のような、他のコアも使用する異なる設計を採用することも可能だ。異なるスレッドモデルを調査し、比較することをお勧めする。

要約

コールスタックは、ローカル状態の効率的な割り当てを可能にし、特に再帰的なアルゴリズムを簡素化する。
ファイバーは、スレッドごとにデフォルトの単一のコールスタックではなく、複数のコールスタックを可能にする。
ファイバーとその呼び出し元は同じスレッド上で実行される(つまり、同時に実行されない)。
ファイバーは呼び出し元に制御を譲ることで自身を一時停止し、呼び出し元はファイバーを再呼び出しすることで実行を再開する。
GeneratorファイバーをInputRangeとして表現する。
ファイバーは、コールスタックに依存するアルゴリズムを簡素化する。
ファイバーは、非同期の入出力操作を簡略化する。
ファイバーは、プリエンプティブマルチタスクとはトレードオフが異なる協調型マルチタスクを提供する。