core チュートリアル

このチュートリアルでは、luna_thread のファサードを使って MoonBit から並列の整数カーネルを実行する方法、同じ作業をライブラリが検証するプランとして記述する方法、ワークフローグラフを作って検査する方法を示します。すべてネイティブターゲットで動きます。

クイックスタート

モジュールを追加します。

moon add Luna-Flow/luna_thread@0.1.0

ネイティブターゲット向けにビルドするパッケージからファサードをインポートします。

import {
  "Luna-Flow/luna_thread",
}

supported_targets = "native"

最大 3 要素のチャンクをそれぞれ受け持つ 2 つのワーカーで、配列の和を求めます。

test "quick start" {
  let input : FixedArray[Int] = [1, 2, 3, 4, 5, 6]
  let total = @luna_thread.execute_reduce_sum_i32(
    input,
    worker_count=2,
    chunk_size=3,
  )
  inspect(total, content="21")
}

moon test --target native の報告：

Total tests: 1, passed: 1, failed: 0.

日常的なタスク

ワーカー数とチャンクサイズを選ぶ

カーネルは $n$ 個の要素を、最大 $c$ 要素の $\lceil n / c \rceil$ 個のチャンクに分け、チャンクごとに 1 つのワーカーを必要とします。そこでまず $c$ を選び、次に $w = \lceil n / c \rceil$ とします。どちらも $n$ 以下でなければなりません。

fn chunked_sum(input : FixedArray[Int], chunk_size : Int) -> Int {
  let n = input.length()
  let workers = (n + chunk_size - 1) / chunk_size
  @luna_thread.execute_reduce_sum_i32(input, worker_count=workers, chunk_size~)
}

test "choose the parameters" {
  let input : FixedArray[Int] = [10, 20, 30, 40, 50, 60, 70]
  inspect(chunked_sum(input, 3), content="280")
  inspect(chunked_sum(input, 7), content="280")
}

要素を 2 倍にしてオーバーフローを扱う

execute_map_i32 はすべての要素を 2 倍にし、オーバーフローは回り込ませずにエラーとして報告します。

test "map with overflow" {
  let small : FixedArray[Int] = [1, -2, 3, 4]
  debug_inspect(
    @luna_thread.execute_map_i32(small, worker_count=2, chunk_size=2),
    content="Ok(<FixedArray: [2, -4, 6, 8]>)",
  )
  let large : FixedArray[Int] = [1, 2147483647]
  debug_inspect(
    @luna_thread.execute_map_i32(large, worker_count=2, chunk_size=1),
    content="Err(Overflow)",
  )
}

累積和を計算する

execute_scan_sum_i32 は途中までの合計を返します。

test "prefix sums" {
  let deposits : FixedArray[Int] = [100, -30, 50, 20, -10]
  match @luna_thread.execute_scan_sum_i32(deposits, worker_count=2, chunk_size=3) {
    Ok(balance) => debug_inspect(balance, content="<FixedArray: [100, 70, 120, 140, 130]>")
    Err(error) => fail("scan failed: \{Repr(error)}")
  }
}

作業をプランとして記述し検証する

プランは何を実行すべきかを記録するだけで、実行はしません。検証により v1 のランタイムがそれをサポートするかがわかります。

test "plans" {
  let policy = @luna_thread.make_policy(worker_count=2, chunk_size=3).unwrap()
  let plan = @luna_thread.reduce(
    "total",
    @luna_thread.i32_type(),
    6,
    @luna_thread.sum_reduction(),
    policy~,
  )
  assert_true(@luna_thread.is_ready(plan))
  let doubles = @luna_thread.map("halve", @luna_thread.f64_type(), 6, policy~)
  debug_inspect(@luna_thread.validate(doubles), content="[UnsupportedValueType]")
}

ワークフローを作って検査する

ワークフローはタスクのグラフです。この例はスポーンし、上のプランを計算し、ジョインします。順序はエッジで固定します。

test "workflow" {
  let policy = @luna_thread.make_policy(worker_count=2, chunk_size=3).unwrap()
  let plan = @luna_thread.reduce(
    "total",
    @luna_thread.i32_type(),
    6,
    @luna_thread.sum_reduction(),
    policy~,
  )
  let graph = @luna_thread.workflow("sum-job", policy~)
    .add_node(@luna_thread.spawn_task(1, "spawn"))
    .add_node(@luna_thread.compute_task(2, "sum", plan))
    .add_node(@luna_thread.join_task(3, "join"))
    .add_edge(@workflow.Edge::new(1, 2, @workflow.control_dependency()))
    .add_edge(@workflow.Edge::new(2, 3, @workflow.control_dependency()))
  assert_true(@luna_thread.workflow_is_ready(graph))
  let submission = @luna_thread.submit_workflow(graph)
  inspect(submission.accepted(), content="true")
}

Edge を使うには、インポートに "Luna-Flow/luna_thread/workflow" も必要です。submit_workflow は検証して投入を記録するだけで、グラフは実行しません。

さらに進んで

ワークフローの同期プロトコルを本物のスレッドで実行するには、submit_workflow_async で投入し、wait_workflow と drop_workflow を呼びます。C のブリッジに既知の不具合があるため非同期の経路は実験的です。使う前に core API の警告を読んでください。

ほかの要素型やカーネルについては、backend/native が Int64 や最小値・最大値のリダクションを含む生の C の入口を公開しています。ネイティブバックエンドのチュートリアルを参照してください。プランとグラフをより細かく制御するには plan と workflow のパッケージを直接使います。plan チュートリアルと workflow チュートリアルではプランとグラフをフィールドごとに組み立てます。作業を記述するだけのパッケージはこの 2 つをインポートして、すべてのターゲットでビルドできます。

よくある落とし穴

• 既定値の worker_count=1 と chunk_size=1 が使えるのは要素が 1 つの入力だけです。それより長い入力では、カーネルは Err(InvalidArgument) を返し、execute_reduce_sum_i32 は 0 を返します。
• execute_reduce_sum_i32 は、和が 0 のときも失敗したときも 0 を返します。区別が必要なら入力と引数を自分で確認するか、スキャンを使って最後の要素を取ってください。
• 部分和が検査されるので、Int の和がオーバーフローするかどうかはチャンク分けに依存することがあります。[-1, 0, 2147483647, 1] は chunk_size=4 では和が求まりますが、chunk_size=2 では失敗します。
• is_ready は「チャンクごとに 1 つのワーカー」という条件を検査しないので、準備のできたプランでもカーネルに拒否されることがあります。
• javascript_policy() は v1 では中断し、make_policy は JavaScript バックエンドと非同期モードを拒否します。
• Workflow::add_node などのビルダーはワークフローをその場で変更します。
• ファサードはネイティブターゲットでのみビルドできます。それをインポートするパッケージには supported_targets = "native" が必要か、ネイティブだけでビルドする必要があります。

次のステップ

core API にはファサードのすべての関数とその正確な意味が載っています。core の設計ではファサードが既定値付きの自由関数の集まりである理由を、ネイティブバックエンドの設計ではカーネルとワークフロースケジューラがスレッドとメモリをどう使うかを説明しています。パッケージと C ランタイムの組み合わせ方はアーキテクチャガイドにあります。