Measure dirty scheduler contention and retain I/O inference

.github/workflows/ci.yml

@@ -42,9 +42,9 @@ jobs:
       - run: mix deps.get --check-locked
       - run: mix format --check-formatted
       - run: cargo +1.93.1 fmt --manifest-path native/ortex/Cargo.toml --check
-      - name: Check provider feature mappings
+      - name: Check optional native feature mappings
         if: matrix.api == 17 && matrix.runtime == '1.24.4'
-        run: cargo +1.93.1 check --locked --manifest-path native/ortex/Cargo.toml --features coreml,cuda,tensorrt,directml,acl,onednn,rocm
+        run: cargo +1.93.1 check --locked --manifest-path native/ortex/Cargo.toml --features coreml,cuda,tensorrt,directml,acl,onednn,rocm,benchmark-dirty-cpu
       - name: Select API floor
         env:
           API: ${{ matrix.api }}

bench/schedulers.exs

@@ -0,0 +1,138 @@
+# Compare with four normal, four dirty CPU, and four dirty I/O schedulers.
+# Build/run each mode in fresh VMs; see docs/schedulers.md.
+mode = Ortex.Native.inference_scheduler()
+expected = if System.get_env("ORTEX_BENCHMARK_DIRTY_CPU") == "1", do: :dirty_cpu, else: :dirty_io
+if mode != expected, do: raise("stale NIF: force a native rebuild before measuring")
+options = [intra_threads: 1, intra_op_spinning: false]
+path = Path.expand("../test/fixtures/xgboost.onnx", __DIR__)
+{%{rows: rows, expected: scores}, _} = Code.eval_file(Path.rootname(path) <> "_cases.ex")
+model = Ortex.load(path, options)
+input = Nx.tensor(rows, type: :f32) |> Nx.backend_transfer(Ortex.Backend)
+
+run = fn ->
+  {output} = Ortex.run(model, input)
+  actual = Nx.to_flat_list(output)
+
+  unless Enum.all?(Enum.zip(actual, scores), fn {a, b} -> abs(a - b) < 1.0e-6 end),
+    do: raise("score mismatch")
+end
+
+for _ <- 1..200, do: run.()
+{serial_us, timings} = :timer.tc(fn -> for _ <- 1..2000, do: elem(:timer.tc(run), 0) end)
+
+{concurrent_us, _} =
+  :timer.tc(fn ->
+    1..2000
+    |> Task.async_stream(fn _ -> run.() end, max_concurrency: 8)
+    |> Enum.each(fn {:ok, _} -> :ok end)
+  end)
+
+percentile = fn values, fraction ->
+  values |> Enum.sort() |> Enum.at(min(length(values) - 1, trunc(length(values) * fraction)))
+end
+
+# Native.from_binary always uses dirty CPU. Its latency reveals whether queued
+# inference mutex waiters can monopolize that pool and delay unrelated work.
+binary = :binary.copy(<<0>>, 2_097_152)
+
+probe = fn ->
+  case Ortex.Native.from_binary(binary, {524_288}, {:f, 32}) do
+    {:error, error} -> raise error
+    _resource -> :ok
+  end
+end
+
+for _ <- 1..5, do: probe.()
+copy_baseline = for _ <- 1..10, do: elem(:timer.tc(probe), 0)
+loop = Ortex.load(Path.expand("../test/fixtures/cancellable_loop.onnx", __DIR__), options)
+iterations = 20_000
+
+refs =
+  for value <- [iterations, 0] do
+    %Nx.Tensor{data: %Ortex.Backend{ref: reference}} =
+      Nx.tensor(value, type: :s64) |> Nx.backend_transfer(Ortex.Backend)
+
+    reference
+  end
+
+pool_width =
+  case mode do
+    :dirty_cpu -> :erlang.system_info(:dirty_cpu_schedulers_online)
+    :dirty_io -> :erlang.system_info(:dirty_io_schedulers)
+  end
+
+workers = max(pool_width * 4, 16)
+owner = self()
+
+wait_claimed = fn recurse, controls, deadline ->
+  if Enum.count(controls, &Ortex.RunControl.started?/1) < pool_width do
+    if System.monotonic_time(:millisecond) >= deadline,
+      do: raise("inference pool did not become occupied")
+
+    Process.sleep(1)
+    recurse.(recurse, controls, deadline)
+  end
+end
+
+contended =
+  for _ <- 1..10 do
+    controls = for _ <- 1..workers, do: Ortex.RunControl.new(loop)
+
+    tasks =
+      Enum.map(controls, fn control ->
+        Task.async(fn ->
+          send(owner, {:ready, self()})
+
+          receive do
+            :go -> :ok
+          end
+
+          Ortex.Native.run_controlled(loop.reference, refs, control.reference)
+        end)
+      end)
+
+    for task <- tasks do
+      pid = task.pid
+
+      receive do
+        {:ready, ^pid} -> :ok
+      after
+        5000 -> raise("worker did not become ready")
+      end
+    end
+
+    for task <- tasks, do: send(task.pid, :go)
+    wait_claimed.(wait_claimed, controls, System.monotonic_time(:millisecond) + 10_000)
+    {latency, _} = :timer.tc(probe)
+
+    for task <- tasks do
+      [{:tensor, {reference, [], :s, 64}}] = Task.await(task, 30_000)
+
+      unless Ortex.Native.to_binary(reference, 64, 1) == <<iterations::signed-native-64>>,
+        do: raise("loop result mismatch")
+    end
+
+    latency
+  end
+
+IO.inspect(
+  %{
+    mode: mode,
+    schedulers: :erlang.system_info(:schedulers_online),
+    dirty_cpu: :erlang.system_info(:dirty_cpu_schedulers_online),
+    dirty_io: :erlang.system_info(:dirty_io_schedulers),
+    tree_serial_calls_per_second: 2_000_000_000 / serial_us,
+    tree_concurrent_calls_per_second: 2_000_000_000 / concurrent_us,
+    tree_p50_us: percentile.(timings, 0.5),
+    tree_p99_us: percentile.(timings, 0.99),
+    copy_baseline_p50_us: percentile.(copy_baseline, 0.5),
+    contended_copy_p50_us: percentile.(contended, 0.5),
+    contended_copy_p99_us: percentile.(contended, 0.99),
+    contended_copy_samples_us: contended,
+    queued_calls_per_round: workers,
+    loop_iterations: iterations,
+    note: "Diagnostic on this host; one contended CPU session, no accelerator or placement claim."
+  },
+  limit: :infinity,
+  charlists: :as_lists
+)

config/config.exs

@@ -8,3 +8,8 @@ config :ortex,
 
 # CPU-only by default. Accelerator features require an explicit source build and
 # a matching ONNX Runtime library; never infer GPU support from the host OS.
+
+# Test-only source-build diagnostic, separate from distributed precompiled NIFs.
+if config_env() == :test and System.get_env("ORTEX_BENCHMARK_DIRTY_CPU") == "1" do
+  config :ortex, Ortex.Native, features: ["load-dynamic", "api-17", "benchmark-dirty-cpu"]
+end

docs/integration.md

@@ -36,6 +36,7 @@ writes to another repository. The broad original PR is superseded by the stack.
 | 26 | Runtime-independent ONNX graph summaries |
 | 27 | One-attempt native cancellation controls |
 | 28 | Managed deadlines and caller-exit cancellation |
+| 29 | Dirty-scheduler measurement and policy |
 
 Each PR targets the preceding branch. Merge or retarget in order; the integration
 branch is for validation rather than a replacement broad PR. CI on each branch
@@ -130,7 +131,6 @@ with separate focused PRs for:
 
 - Per-run profiling toggles require API 25 and remain deferred beyond the API
   17/24 runtime matrix; session profiling is available now.
-- Dirty-scheduler classification under representative concurrent inference.
 
 GPU/NN primitives remain a later or measurement-dependent stage: shared weights
 and pooling, AOT compilation, I/O binding, device allocation, and adapters. Keep

docs/schedulers.md

@@ -0,0 +1,51 @@
+# Inference scheduling decision
+
+Keep dirty I/O scheduling for inference. A session run can wait for the per-model
+mutex or the runtime's worker threads. Scheduling those waits on dirty CPU can
+occupy that entire pool and delay unrelated CPU NIFs, including input conversion
+and control preparation. Control creation uses dirty CPU; cancellation itself is
+a short normal NIF that only sets the runtime flag.
+
+A local Apple silicon experiment used four normal, four dirty CPU, and four dirty
+I/O schedulers. It verified the loaded NIF's actual mode, checked golden tree-model
+scores, and compared a 2 MiB dirty-CPU input-copy probe while 16 native calls
+contended for one finite counting-Loop session. Ten rounds used 20,000 iterations
+per call, with one probe after each inference pool became occupied.
+
+| Measurement | Dirty I/O inference | Dirty CPU inference |
+| --- | --- | --- |
+| Tree serial calls/second | 141,233 | 153,527 |
+| Tree concurrent calls/second (8 callers) | 86,062 | 86,749 |
+| Tree p50 / p99, microseconds | 4 / 29 | 5 / 16 |
+| Contended copy p50, microseconds | 52 | 71,779 |
+| Contended copy p99, microseconds | 97 | 77,888 |
+
+Tree throughput is similar in this short local sample. The contended probe shows
+that moving mutex waits to dirty CPU can delay other native CPU work. This is a
+workload-specific result, not a claim about every graph/provider. Heavy independent
+CPU sessions, neural-network workloads, or accelerators need separate measurement
+before changing policy. Avoiding native mutex waits through a different queueing
+architecture would also change this tradeoff.
+
+## Reproduce
+
+Use a source checkout with a supported runtime library in `ORT_DYLIB_PATH`:
+
+```sh
+export MIX_ENV=test ORTEX_BUILD=true
+export ERL_FLAGS='+S 4:4 +SDcpu 4:4 +SDio 4'
+ORTEX_BENCHMARK_DIRTY_CPU=0 mix compile --force --warnings-as-errors
+ORTEX_BENCHMARK_DIRTY_CPU=0 mix run bench/schedulers.exs
+ORTEX_BENCHMARK_DIRTY_CPU=1 mix compile --force --warnings-as-errors
+ORTEX_BENCHMARK_DIRTY_CPU=1 mix run bench/schedulers.exs
+# Restore the default native artifact before ordinary use:
+ORTEX_BENCHMARK_DIRTY_CPU=0 mix compile --force --warnings-as-errors
+```
+
+Each build/run uses a fresh VM. The script reads the native scheduling mode and
+rejects a stale artifact. The test-only config flag enables the diagnostic Cargo
+feature `benchmark-dirty-cpu`; normal and precompiled builds never enable it.
+This mode is for the experiment, not a supported deployment setting. It shares
+the CPU pool with control preparation, which is another reason to avoid that
+policy for contended sessions. CI compile-checks the alternate feature together
+with all exposed provider mappings; normal runtime tests use the default policy.

lib/ortex/native.ex

@@ -40,6 +40,7 @@ defmodule Ortex.Native do
   def new_run_control(_model, _options), do: :erlang.nif_error(:nif_not_loaded)
   def cancel_run(_control), do: :erlang.nif_error(:nif_not_loaded)
   def run_started(_control), do: :erlang.nif_error(:nif_not_loaded)
+  def inference_scheduler, do: :erlang.nif_error(:nif_not_loaded)
   def run_controlled(_model, _inputs, _control), do: :erlang.nif_error(:nif_not_loaded)
   def run_named_controlled(_model, _inputs, _control), do: :erlang.nif_error(:nif_not_loaded)
   def from_binary(_bin, _shape, _type), do: :erlang.nif_error(:nif_not_loaded)

native/ortex/Cargo.toml

@@ -18,6 +18,8 @@ half = "2.2.1"
 
 [features]
 default = ["load-dynamic", "api-17"]
+# Diagnostic-only alternate schedule for bench/schedulers.exs; never in prebuilts.
+benchmark-dirty-cpu = []
 load-dynamic = ["ort/load-dynamic"]
 download-binaries = ["ort/download-binaries", "ort/copy-dylibs", "ort/tls-rustls"]
 tls-rustls = ["ort/tls-rustls"]

native/ortex/src/constants.rs

@@ -9,6 +9,7 @@ pub mod ortex_atoms {
         s, u, f, bf, c,
         // Execution provider atoms
         cpu, cuda, tensorrt, acl, dnnl,
-        onednn, coreml, directml, rocm
+        onednn, coreml, directml, rocm,
+        dirty_cpu, dirty_io
     }
 }

native/ortex/src/lib.rs

@@ -76,23 +76,35 @@ fn show_session(
     Ok(model::show(model))
 }
 
-#[rustler::nif(schedule = "DirtyIo")]
+#[cfg_attr(feature = "benchmark-dirty-cpu", rustler::nif(schedule = "DirtyCpu"))]
+#[cfg_attr(
+    not(feature = "benchmark-dirty-cpu"),
+    rustler::nif(schedule = "DirtyIo")
+)]
 fn run(
     model: ResourceArc<model::OrtexModel>,
     inputs: Vec<ResourceArc<OrtexTensor>>,
 ) -> NifResult<Vec<(ResourceArc<OrtexTensor>, Vec<usize>, Atom, usize)>> {
     model::run(model, inputs).map_err(|e| rustler::Error::Term(Box::new(e.to_string())))
 }
 
-#[rustler::nif(schedule = "DirtyIo")]
+#[cfg_attr(feature = "benchmark-dirty-cpu", rustler::nif(schedule = "DirtyCpu"))]
+#[cfg_attr(
+    not(feature = "benchmark-dirty-cpu"),
+    rustler::nif(schedule = "DirtyIo")
+)]
 fn run_named(
     model: ResourceArc<model::OrtexModel>,
     inputs: Vec<(String, ResourceArc<OrtexTensor>)>,
 ) -> NifResult<model::NamedOutputs> {
     model::run_named(model, inputs).map_err(|e| rustler::Error::Term(Box::new(e.to_string())))
 }
 
-#[rustler::nif(schedule = "DirtyIo")]
+#[cfg_attr(feature = "benchmark-dirty-cpu", rustler::nif(schedule = "DirtyCpu"))]
+#[cfg_attr(
+    not(feature = "benchmark-dirty-cpu"),
+    rustler::nif(schedule = "DirtyIo")
+)]
 fn run_with_options(
     model: ResourceArc<model::OrtexModel>,
     inputs: Vec<ResourceArc<OrtexTensor>>,
@@ -102,7 +114,11 @@ fn run_with_options(
         .map_err(|e| rustler::Error::Term(Box::new(e.to_string())))
 }
 
-#[rustler::nif(schedule = "DirtyIo")]
+#[cfg_attr(feature = "benchmark-dirty-cpu", rustler::nif(schedule = "DirtyCpu"))]
+#[cfg_attr(
+    not(feature = "benchmark-dirty-cpu"),
+    rustler::nif(schedule = "DirtyIo")
+)]
 fn run_named_with_options(
     model: ResourceArc<model::OrtexModel>,
     inputs: Vec<(String, ResourceArc<OrtexTensor>)>,
@@ -112,7 +128,11 @@ fn run_named_with_options(
         .map_err(|e| rustler::Error::Term(Box::new(e.to_string())))
 }
 
-#[rustler::nif(schedule = "DirtyIo")]
+#[cfg_attr(feature = "benchmark-dirty-cpu", rustler::nif(schedule = "DirtyCpu"))]
+#[cfg_attr(
+    not(feature = "benchmark-dirty-cpu"),
+    rustler::nif(schedule = "DirtyIo")
+)]
 fn run_values<'a>(
     env: Env<'a>,
     model: ResourceArc<model::OrtexModel>,
@@ -126,7 +146,11 @@ fn run_values<'a>(
         .map_err(|e| rustler::Error::Term(Box::new(e.to_string())))
 }
 
-#[rustler::nif(schedule = "DirtyIo")]
+#[cfg_attr(feature = "benchmark-dirty-cpu", rustler::nif(schedule = "DirtyCpu"))]
+#[cfg_attr(
+    not(feature = "benchmark-dirty-cpu"),
+    rustler::nif(schedule = "DirtyIo")
+)]
 fn run_named_values<'a>(
     env: Env<'a>,
     model: ResourceArc<model::OrtexModel>,
@@ -254,7 +278,11 @@ fn run_started(control: ResourceArc<run_control::RunControl>) -> bool {
     control.started()
 }
 
-#[rustler::nif(schedule = "DirtyIo")]
+#[cfg_attr(feature = "benchmark-dirty-cpu", rustler::nif(schedule = "DirtyCpu"))]
+#[cfg_attr(
+    not(feature = "benchmark-dirty-cpu"),
+    rustler::nif(schedule = "DirtyIo")
+)]
 fn run_controlled<'a>(
     env: Env<'a>,
     model: ResourceArc<model::OrtexModel>,
@@ -266,7 +294,11 @@ fn run_controlled<'a>(
         .map_err(|e| rustler::Error::Term(Box::new(e.to_string())))
 }
 
-#[rustler::nif(schedule = "DirtyIo")]
+#[cfg_attr(feature = "benchmark-dirty-cpu", rustler::nif(schedule = "DirtyCpu"))]
+#[cfg_attr(
+    not(feature = "benchmark-dirty-cpu"),
+    rustler::nif(schedule = "DirtyIo")
+)]
 fn run_named_controlled<'a>(
     env: Env<'a>,
     model: ResourceArc<model::OrtexModel>,
@@ -286,3 +318,12 @@ fn environment_with_options(
         .map(ResourceArc::new)
         .map_err(|e| rustler::Error::Term(Box::new(e)))
 }
+
+#[rustler::nif]
+fn inference_scheduler() -> Atom {
+    if cfg!(feature = "benchmark-dirty-cpu") {
+        constants::ortex_atoms::dirty_cpu()
+    } else {
+        constants::ortex_atoms::dirty_io()
+    }
+}