みなさま、こんにちは。サイオステクノロジー武井です。今宵は、Durable Functionsの複雑な動きを簡単なコードで説明しようと思います。

Durable Functionsの動きは複雑…

Durable Functionsの最も特徴的であり、最も複雑な動きの一つに「リプレイ」があり、この「リプレイ」という機構を理解することが、Durable Functionsを理解することに他ならないと思っております。

「リプレイ」とは、アクティビティ関数がコールされるごとに、その実行結果や戻り値を外部のストレージ(Azure Table Storage)に保存して、オーケストレーター関数をまた最初から起動するという仕組みです。

この仕組のおかげで、仮にDurable Functionsを実行しているサーバーが故障して、その処理を他のサーバーが引き継いだとしても、きちんと故障して中断された処理から再開されます。これがDurable Functionsが「Durable」と言われる由縁かと思います。

このあたりの基本的な動きについては、以下のブログにて記載してありますので、ご覧いただけたらと思います。

多分わかりやすいDurable Functions 〜サーバーレスは次のステージへ〜【理論編】

今回は、このDurable Functionsの中身の動きを再現した、簡単なコードを交えて説明することで、Durable Functionsの複雑な動きをご説明したいと思います。

ジェネレーター関数について

リプレイの解説のところで「アクティビティ関数を実行したら、オーケストレーター関数は終了して、また初めからやり直す」みたいな解説ありますよね。これ、なんだかよくイメージ湧きにくいのですが、ジェネレーター関数っていう仕組みをつかっています。

ジェネレーター関数の特徴は、yieldと記載したところで処理を一時中断することができます。

百聞は一見にしかずなので、まずかんたんなコードをお見せしたいと思います。

function* generator() {
    console.log("hoge");
    yield 1; // (1)
    console.log("fuga");
    yield 2; // (2)
    console.log("piyo");
    yield 3; // (3)
}
  
fn = generator(); // (4)
  
fn.next(); // (5) {value: 1, done: false}
fn.next(); // (6) {value: 2, done: false}
fn.next(); // (7) {value: 3, done: false}
fn.next(); // (8) {value: undefined, done: true}

ジェネレーター関数はfunction*のようにアスタリスクをつけます。このコードがどんなふうに動くのかを順を追って説明します。

(4)にて、変数fnにジェネレータ関数generatorを代入する。
(5)にて、nextメソッドを実行する。このときgenerator関数が起動する。
generator関数内で(1)のyield 1まで実行され、この関数を抜ける。
(5)のfn.next()の戻り値として、{value: 1, done: false}が返ってくる。valueはyieldの引数で指定された値(この場合は1)、doneはこのジェネレーター関数が最後まで起動するとtrue、そうでない場合はfalseになる。
(6)にて、nextメソッドを実行する。このとき、再びgenerator関数が起動する。
先程処理を中断したyield 1の次の行(console.log(“fuga”);)から処理を再開する。
generator関数内で(2)のyield 2が実行され、この関数を抜ける。
(6)のfn.next()の戻り値として、{value: 2, done: false}が返ってくる。
(7)にて、nextメソッドを実行する。このとき、再びgenerator関数が起動する。
先程処理を中断したyield 2の次の行(console.log(“piyo”);)から処理を再開する。
generator関数内で(3)のyield 3が実行され、この関数を抜ける。
(7)のfn.next()の戻り値として、{value: 3, done: false}が返ってくる。
(8)にて、next()メソッドを実行すると、generator関数は最後まで起動したので、doneがtrueになる。

いかがでしょうか？つまり、ジェネレーター関数はnextメソッドで呼び出すたびにyiledが実行されて、ジェネレーター関数を抜けて、またnextメソッドで呼び出すと、中断したところから再開するという動きをしています。

簡単なコードでDurable Functionsの内部の動きを再現

実際はもっともっと複雑な動きをしていると思いますが、説明を簡単にするために、簡単なコードでDurable Functionsの内部の動きを再現してみました。

// Durable FUnctionsの実行状態を格納する変数を定義します。
const context = [];

// DurableFunctionが開始したことを表すeventTypeをcontextに追加します。
context.push(
    {
        "eventType": "DurableFunctionStarted"
    }
);

while (true) {
    // DurableFunctionを終了していいかを判定するためのフラグです。
    // これがtrueになると、このwhileループから抜けて、本プログラムが終了します。
    // つまりDurable Functionが終了します。
    let isDurableFunctionFinished = false;

    // 先述したようにisProcessedは、アクティビティ関数が実行済みor未実行かをチェックする処理が
    // 実行されたかそうでないかを表すフラグです(ややこしい、、、)。
    // オーケストレーター関数を実行する前に、このisProcessedをすべてfalseにします。
    for (value of context) {
        if (value.eventType == "TaskCompleted")
            value.isProcessed = false;
    }

    // オーケストレーター関数を表すジェネレーター関数を取得します。
    const fn = orchestrator(context);

    let result = null;
    while (true) {
        let next = null;
        // 一回目のリプレイか、2回目以降のリプレイかを判別します。
        if (result) {
            // 2回目以降のリプレイの場合、直前に実行されたアクティビティ関数に結果を渡します。
            next = fn.next(result);
        } else {
            // 1回目のリプレイの場合には何もわたしません。
            next = fn.next();
        }


        // yiled関数がすべて終わったら、つまりすべてのアクティビティ関数の実行が終わったら
        // isDurableFunctionFinishedをtrueにしてwhileループを抜けます。
        if (next.done) {
            isDurableFunctionFinished = true;
            break;
        }

        // アクティビティ関数の結果を取得します。
        result = next.value;

        // isPlayedはアクティビティ関数が実行されたかどうかを表すフラグです。
        // もし、リプレイでスキップされず、アクティビティ関数が実行されているのであれば、
        // isPlayedはtrueのはずですので、このwhileループを抜け出して、もう一度最初から
        // オーケストレーター関数を実行します。
        if (result.isPlayed) break;

    }

    if (isDurableFunctionFinished) break;

}

// オーケストレーター関数に相当するジェネレーター関数です。
function* orchestrator(context) {
    // 別途定義しているcallActivity関数の第1引数にcontext(実行状態を保存する変数)、
    // 第2引数にアクティビティ関数の実体を指定します。
    const result01 = yield callActivity(context,
        (context) => {
            return "1";
        }
    )

    console.log(result01);

    const result02 = yield callActivity(context,
        (context) => {
            return "2";
        }
    )

    console.log(result02);
}

// アクティビティ関数の実行状態のチェック、及び
// アクティビティ関数を実行します。
function callActivity(context, fn) {
    // 具体的な処理は記載していないが、ここで、
    // Azure Table Storageに現在のcontextの状態を保存します(と思われる)

    // Durable Functionsの実行状態を格納しているcontext変数をチェックして、
    // アクティビティ関数の実行状態を確認します。
    for (const status of context) {
        // アクティビティ関数のeventTypeがTaskCompleted(実行済み)、かつ
        // isProcessedがtrueかどうか(実行状態のチェックが完了しているかどうか)をチェックする。
        if (status.eventType == "TaskCompleted" && !status.isProcessed) {
            // もしアクティビティ関数がすでに実行済みだったら、
            // 実行済みをチェックしたかどうかを表すisProcessedをtrueにします。
            // また、ここではアクティビティ関数は実行していない(contextに保存された結果を返す)ので
            // アクティビティ関数を実行したかどうかを表すisPlayedはfalseとします。
            status.isProcessed = true;
            status.isPlayed = false;

            return status;
        }
    }

    // アクティビティ関数が一度も実行されていない場合は、実行します。
    const value = fn(context);

    // アクティビティ関数の実行状態及び結果を返すJSONを生成して、
    // contextに入れます。
    const result = {
        "eventType": "TaskCompleted", // アクティビティ関数が実行完了であるため、TaskCompletedとします
        "isProcessed": false, // アクティビティ関数の実行済みかどうかはチェックしていないので、falseとします。
        "isPlayed": true, // アクティビティ関数を実行したので、trueとします。
        "result": value // アクティビティ関数の実行結果を返します。
    };

    context.push(result);
    return result;
}

処理の概要

まずは、先のコードの大まかな構造についてご説明します。

このプログラムは大きく3つに別れます。①のメインの処理で一番最初に実行される部分、②のオーケストレーター関数に相当するジェネレーター関数、③のアクティビティ関数を実行する部分です。

①では、contextという変数を定義しています。これはアクティビティ関数の実行状態を記録する変数になり、該当のアクティビティ関数がまだ未実行なのか、実行済みなのかといった状態を管理します。そして、そのcontextの変数の状態を確認して、オーケストレーター関数を実行するかどうか(リプレイ)を決定します。

②はオーケストレーター関数に相当するジェネレーター関数になります。アクティビティ関数を呼び出します。

③は、アクティビティ関数を実行したり、アクティビティ関数の実行状態を変更したりする関数です。

詳細な動きについては、先程ご紹介したコード、及びコード内のコメントを参照頂ければと思います(´・ω・｀)