Luna Flow
日本語

backend/native API

パッケージ Luna-Flow/luna_thread/backend/native は @native としてインポートされ、C の外部関数インターフェースによるバックエンドです。C ランタイムの整数カーネルを MoonBit の配列に対して実行し、ワークフローを C スケジューラに投入し、プランを型付きのネイティブリクエストに変換します。native ターゲットでのみビルドでき、ネイティブスタブから C ランタイムをリンクします。

このパッケージの列挙型はパッケージの外では読み取り専用です。コンストラクタでパターンマッチはできます。リクエストレコードは *_from_plan 関数で作ります。

バックエンド情報

backend_target

Native を返します。

pub fn backend_target() -> @shared.BackendTarget

default_plan_kind

Map を返します。

pub fn default_plan_kind() -> @plan.PlanKind

package_name

"backend/native" を返します。

pub fn package_name() -> String

supports_openmp

true を返します。

pub fn supports_openmp() -> Bool

この値は定数です。moon によるビルドは OpenMP なしで C スタブをコンパイルするため、そのビルドではカーネルは呼び出し元のスレッドで実行されます。ネイティブバックエンドの設計を参照してください。

カーネル

すべてのカーネルは長さ nn の FixedArray[Int]、ワーカー数 ww、チャンクサイズ cc を受け取り、終わるまでブロックし、入力を変更しません。次の条件が必要です。

n>0,0<w≤n,0<c≤n,w≥⌈nc⌉n > 0, \qquad 0 < w \le n, \qquad 0 < c \le n, \qquad w \ge \left\lceil \frac{n}{c} \right\rceil

満たさない場合は InvalidArgument で失敗します。core API にある同名のファサード関数は、これらを省略可能な引数付きでラップしています。

execute_map_i32

すべての要素を 2 倍にした新しい配列を返します。ある 2xi2 x_i が 32 ビットに収まらない場合は Overflow を返します。

pub fn execute_map_i32(FixedArray[Int], Int, Int) -> Result[FixedArray[Int], NativeRequestError]

execute_reduce_sum_i32

要素の総和を返します。失敗した場合は常に 0 を返します。

pub fn execute_reduce_sum_i32(FixedArray[Int], Int, Int) -> Int

カーネルは各チャンクを左から順に足し、次にチャンクの和を左から順に足します。これらの部分和のどれかがオーバーフローすると失敗します。返された 0 以外の値は正確な和です。0 は和である場合も失敗である場合もあります。

execute_scan_sum_i32

包含的な累積和を返します。

pub fn execute_scan_sum_i32(FixedArray[Int], Int, Int) -> Result[FixedArray[Int], NativeRequestError]

部分和がオーバーフローすると Overflow で失敗します。

test "kernels" {
  let input : FixedArray[Int] = [3, 1, 4, 1, 5, 9]
  debug_inspect(
    @native.execute_scan_sum_i32(input, 3, 2),
    content="Ok(<FixedArray: [3, 4, 8, 9, 14, 23]>)",
  )
  assert_eq(@native.execute_reduce_sum_i32(input, 3, 2), 23)
  debug_inspect(@native.execute_map_i32(input, 2, 2), content="Err(InvalidArgument)")
}

ワークフロー

workflow_is_supported

@workflow.validate が問題を見つけず、ワークフローのポリシーがネイティブバックエンドを指定しているとき true を返します。

pub fn workflow_is_supported(@workflow.Workflow) -> Bool

submit_workflow

@workflow.submit(workflow, backend=Native) を返します。これは検証済みの投入記録で、ワークフローは実行しません。

pub fn submit_workflow(@workflow.Workflow) -> @workflow.Submission

WorkflowRuntimeState

C ランタイムにおけるワークフローの状態です。

pub enum WorkflowRuntimeState {
  Submitted
  Running
  Completed
  Failed
  Rejected
} derive(Eq, @debug.Debug)

ワーカーが最初のノードを取り出すまで、状態は Submitted のままです。ランタイムが Rejected を設定することはなく、拒否された投入では代わりに空のハンドルが返ります。

WorkflowResult

ワークフローのスナップショットです。状態、ランタイムのステータスコード、完了したノード数、失敗したノードの id（失敗がなければ -1）からなります。

pub struct WorkflowResult {
  state : WorkflowRuntimeState
  status : Int
  completed_nodes : Int
  failed_node_id : Int
} derive(Eq, @debug.Debug)

status はアーキテクチャガイドに挙げたコードのいずれかです。

NativeWorkflowHandle と WorkflowHandle

C ランタイムで実行中のワークフローを指す不透明なポインタと、その MoonBit ラッパーです。

#external
pub type NativeWorkflowHandle

pub struct WorkflowHandle {
  raw : NativeWorkflowHandle
}

submit_workflow_async

ワークフローを平坦な整数配列にコピーし、ポリシーの worker_count 個のスレッドで C スケジューラ上で開始します。

pub fn submit_workflow_async(@workflow.Workflow) -> Result[WorkflowHandle, NativeRequestError]

ワークフローは MoonBit では検証されません。C ランタイムが検査し、グラフを拒否した場合は空のハンドルを返します。この関数は常に Ok を返します。計算ノードはスケジュールされますが、そのプランは実行されません。

poll_workflow

待たずにスナップショットを返します。

pub fn poll_workflow(WorkflowHandle) -> WorkflowResult

wait_workflow

ワークフローが Completed か Failed になるまで待ち、スナップショットを返します。

pub fn wait_workflow(WorkflowHandle) -> WorkflowResult

空のハンドルに対しては、状態 Submitted、ステータス 7 ですぐに返ります。ワークフローがデッドロックすると戻りません。

drop_workflow

ランタイムのスレッドを停止して合流させ、ワークフローを解放します。ハンドルごとに一度だけ呼んでください。空のハンドルに対しては何もしません。

pub fn drop_workflow(WorkflowHandle) -> Unit

型付きのネイティブリクエスト

NativeValueType と NativeReductionKernel

C ランタイムが実装している値の型とリダクションカーネルです。

pub enum NativeValueType {
  I32
  I64
} derive(Eq, @debug.Debug)

pub enum NativeReductionKernel {
  Sum
  Min
  Max
} derive(Eq, @debug.Debug)

NativeRequestError

プランをネイティブリクエストにできなかった理由、またはカーネルが失敗した理由です。

pub enum NativeRequestError {
  UnsupportedBackend
  UnsupportedMode
  UnsupportedValueType
  UnsupportedReductionKernel
  InvalidArgument
  Overflow
} derive(Eq, @debug.Debug)

NativeMapRequest、NativeReduceRequest、NativeScanRequest

C のリクエスト構造体に対応する型付きのレコードです。

pub struct NativeMapRequest {
  input : @shared.NativeBuffer
  output : @shared.NativeBuffer
  element_count : Int
  value_type : NativeValueType
  worker_count : Int
  chunk_size : Int
} derive(Eq, @debug.Debug)

pub struct NativeReduceRequest {
  input : @shared.NativeBuffer
  output : @shared.NativeBuffer
  element_count : Int
  value_type : NativeValueType
  reduction_kernel : NativeReductionKernel
  worker_count : Int
  chunk_size : Int
} derive(Eq, @debug.Debug)

pub struct NativeScanRequest {
  input : @shared.NativeBuffer
  output : @shared.NativeBuffer
  element_count : Int
  value_type : NativeValueType
  reduction_kernel : NativeReductionKernel
  worker_count : Int
  chunk_size : Int
} derive(Eq, @debug.Debug)

プランから作ったリクエストの input と output のバッファは NativeBuffer::new(0, 0) です。プランが表すのはデータの形であってデータそのものではありません。このモジュールにはこれらのレコードを C に送る関数はありません。将来の実行器がバッファを埋めるための、検証済みの記述です。

map_request_from_plan

Map プランを NativeMapRequest に変換します。

pub fn map_request_from_plan(@plan.Plan) -> Result[NativeMapRequest, NativeRequestError]

プランが Map プランでなければ UnsupportedMode で失敗します。そうでなければ @plan.validate の最初の問題から対応づけたエラーで失敗します。値の型、リダクションカーネル、バックエンド、モードの問題は同名のエラーに、MissingReductionKernel は UnsupportedReductionKernel に、残りの問題は InvalidArgument に対応します。

reduce_request_from_plan

Reduce または MapReduce プランを NativeReduceRequest に変換します。エラーの規則は同じです。

pub fn reduce_request_from_plan(@plan.Plan) -> Result[NativeReduceRequest, NativeRequestError]

scan_request_from_plan

Scan プランを、カーネルが Sum の NativeScanRequest に変換します。エラーの規則は同じです。

pub fn scan_request_from_plan(@plan.Plan) -> Result[NativeScanRequest, NativeRequestError]

map_request_is_valid、reduce_request_is_valid、scan_request_is_valid

C の検証器も検査するリクエストのフィールドを検査します。要素数、ワーカー数、チャンクサイズが正であること、値の型がサポートされていること、スキャンではカーネルが Sum であることです。

pub fn map_request_is_valid(NativeMapRequest) -> Bool
pub fn reduce_request_is_valid(NativeReduceRequest) -> Bool
pub fn scan_request_is_valid(NativeScanRequest) -> Bool

map_request_value_type、reduce_request_value_type、scan_request_value_type、reduce_request_kernel、scan_request_kernel

リクエストの値の型（I32 は 0、I64 は 1）とカーネル（Sum は 0、Min は 1、Max は 2）の C のコードを返します。

pub fn map_request_value_type(NativeMapRequest) -> Int
pub fn reduce_request_value_type(NativeReduceRequest) -> Int
pub fn scan_request_value_type(NativeScanRequest) -> Int
pub fn reduce_request_kernel(NativeReduceRequest) -> Int
pub fn scan_request_kernel(NativeScanRequest) -> Int
test "requests from plans" {
  let policy = @shared.make_execution_policy(worker_count=2, chunk_size=4).unwrap()
  let plan = @plan.map_reduce("sum", @plan.i64_type(), 8, @plan.max_reduction(), policy~)
  guard @native.reduce_request_from_plan(plan) is Ok(request) else {
    fail("expected a request")
  }
  assert_true(@native.reduce_request_is_valid(request))
  assert_eq(@native.reduce_request_value_type(request), 1)
  assert_eq(@native.reduce_request_kernel(request), 2)
  let scan = @plan.scan("prefix", @plan.i32_type(), 8, policy~)
  debug_inspect(@native.map_request_from_plan(scan), content="Err(UnsupportedMode)")
}

生の外部関数

これらの extern "C" 宣言は公開されており、型付きの関数がラップしていないカーネル、たとえば Int64 のカーネルや最小値・最大値のリダクションを呼び出せます。配列は呼び出しの間だけ借用されます。マップとスキャンはステータスコードを返して output に書き込みます。output には少なくとも length 個の要素が必要です。リダクションは結果を返し、失敗時は 0 を返します。引数は入力、その長さ、ワーカー数、チャンクサイズの順です。

ffi_execute_map_i32 と ffi_execute_map_i64

すべての要素を 2 倍にして output に書き込みます。

pub fn ffi_execute_map_i32(FixedArray[Int], Int, Int, Int, FixedArray[Int]) -> Int
pub fn ffi_execute_map_i64(FixedArray[Int64], Int, Int, Int, FixedArray[Int64]) -> Int

ffi_execute_reduce_sum_i32、ffi_execute_reduce_sum_i64、ffi_execute_reduce_min_i32、ffi_execute_reduce_min_i64、ffi_execute_reduce_max_i32、ffi_execute_reduce_max_i64

入力の総和、最小値、または最大値を返します。

pub fn ffi_execute_reduce_sum_i32(FixedArray[Int], Int, Int, Int) -> Int
pub fn ffi_execute_reduce_sum_i64(FixedArray[Int64], Int, Int, Int) -> Int64
pub fn ffi_execute_reduce_min_i32(FixedArray[Int], Int, Int, Int) -> Int
pub fn ffi_execute_reduce_min_i64(FixedArray[Int64], Int, Int, Int) -> Int64
pub fn ffi_execute_reduce_max_i32(FixedArray[Int], Int, Int, Int) -> Int
pub fn ffi_execute_reduce_max_i64(FixedArray[Int64], Int, Int, Int) -> Int64

最小値と最大値はオーバーフローしません。

ffi_execute_scan_sum_i32 と ffi_execute_scan_sum_i64

包含的な累積和を output に書き込みます。

pub fn ffi_execute_scan_sum_i32(FixedArray[Int], Int, Int, Int, FixedArray[Int]) -> Int
pub fn ffi_execute_scan_sum_i64(FixedArray[Int64], Int, Int, Int, FixedArray[Int64]) -> Int

ffi_submit_workflow_async

平坦な配列からワークフローを開始します。引数はワーカー数、ケイパビリティの id・種類コード・個数、ノードの id・種類コード・ケイパビリティ id（なければ -1）・個数、エッジの始点・終点・種類コード・個数です。ランタイムがグラフを拒否するとヌルハンドルを返します。

pub fn ffi_submit_workflow_async(Int, FixedArray[Int], FixedArray[Int], Int, FixedArray[Int], FixedArray[Int], FixedArray[Int], Int, FixedArray[Int], FixedArray[Int], FixedArray[Int], Int) -> NativeWorkflowHandle

ffi_workflow_poll_state、ffi_workflow_poll_status、ffi_workflow_poll_completed_nodes、ffi_workflow_poll_failed_node_id、ffi_workflow_wait_status、ffi_workflow_destroy

ワークフローのスナップショットのフィールドを 1 つ読む、待って最終ステータスを返す、またはワークフローを破棄します。

pub fn ffi_workflow_poll_state(NativeWorkflowHandle) -> Int
pub fn ffi_workflow_poll_status(NativeWorkflowHandle) -> Int
pub fn ffi_workflow_poll_completed_nodes(NativeWorkflowHandle) -> Int
pub fn ffi_workflow_poll_failed_node_id(NativeWorkflowHandle) -> Int
pub fn ffi_workflow_wait_status(NativeWorkflowHandle) -> Int
pub fn ffi_workflow_destroy(NativeWorkflowHandle) -> Unit

ヌルハンドルに対しては、ポーリング関数は 0 を、ffi_workflow_wait_status は 7 を返し、ffi_workflow_destroy は何もしません。

test "raw kernels" {
  let input : FixedArray[Int64] = [5L, -2L, 7L, 0L]
  assert_eq(@native.ffi_execute_reduce_min_i64(input, 4, 2, 2), -2L)
  assert_eq(@native.ffi_execute_reduce_max_i64(input, 4, 2, 2), 7L)
  let output : FixedArray[Int64] = FixedArray::make(4, 0L)
  assert_eq(@native.ffi_execute_scan_sum_i64(input, 4, 2, 2, output), 0)
  debug_inspect(output, content="<FixedArray: [5, 3, 10, 10]>")
}

等価性

T::equal

構造的な等価性で、NativeMapRequest、NativeReduceRequest、NativeScanRequest、NativeReductionKernel、NativeRequestError、NativeValueType、WorkflowResult、WorkflowRuntimeState のメソッドとして昇格されています。== と != を使ってください。WorkflowHandle には等価性がありません。

pub fn WorkflowResult::equal(Self, Self) -> Bool