logos-messaging/nim-ffi
1
0
Fork 0
mirror of https://github.com/logos-messaging/nim-ffi.git synced 2026-08-05 14:33:13 +00:00
Code Issues Packages Projects Releases Wiki Activity
nim-ffi/tests/perf/cpp/perf_driver.cpp
NagyZoltanPeter a50df01e4b
feat(perf): C++ e2e perf test harness (tests/perf) 
Measures the full C++ -> Nim -> C++ foreign round trip across setups:
scalar family (2 x int64 + float64), payload family (seq[byte], size
sweep incl. large payloads), and event delivery — through sync, async
(bounded in-flight window) and event lanes, over a thread-count sweep.

All handlers compute the same O(1) parity predicate, so setups differ
purely in transport cost. Every reply is verified; a bad reply or lost
event exits non-zero.

`nimble perf_cpp_e2e` regenerates the cpp bindings, builds libperfbench
with -d:danger over the NIM_FFI_MM matrix (orc + refc) — deliberately
bypassing the debug-orc nim_ffi_lib.cmake template build — and runs the
Release driver. NIM_FFI_PERF_* env knobs control threads, volume,
iterations and payload sizes; each table row also emits a csv line for
diff-friendly capture.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-03 12:49:59 +02:00

462 lines
19 KiB
C++
Raw Blame History

// perf_driver — C++ -> Nim -> C++ e2e perf harness for nim-ffi.
//
// Drives the ACTUAL foreign path: this C++ process -> generated
// perfbench.hpp wrapper (CBOR encode) -> libperfbench C ABI -> FFI thread
// -> Nim handler, and back.
//
// Three lanes:
// sync C++ -> Nim -> C++ blocking round trip; per-call latency sampled
// async C++ -> Nim -> C++ `*Async` future turnaround, bounded in-flight
// window per thread; throughput only
// event C++ -> Nim -> C++ a sync trigger fires one on_perf_ping event
// per call; delivery latency is measured from a
// driver-side steady_clock stamp passed through
// the Nim provider verbatim
//
// Two payload families share ONE trivial O(1) parity predicate as the handler
// compute, so setups differ purely in TRANSPORT cost, never in handler work:
// scalar — 2 x int64 + 1 x float64 in, bool out
// payload — N-byte byte payload (N swept), predicate over (first byte,
// last byte, length); the bytes are never walked by the handler
//
// Correctness is enforced on every reply: a wrong or failed result aborts the
// run rather than being mistimed.
//
// Env knobs: NIM_FFI_PERF_THREADS ("1,2,4,8"), NIM_FFI_PERF_PER_THREAD (2000),
// NIM_FFI_PERF_ITERS (3, median reported), NIM_FFI_PERF_PAYLOAD_SIZES
// ("64,512,4096,65536"), NIM_FFI_PERF_EVENT_PAYLOAD (512),
// NIM_FFI_PERF_ASYNC_WINDOW (64 in-flight futures per thread).
#include <algorithm>
#include <atomic>
#include <chrono>
#include <cinttypes>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <deque>
#include <functional>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
#include "perfbench.hpp"
using sclock = std::chrono::steady_clock;
static int envInt(const char* name, int def) {
const char* v = std::getenv(name);
if (v == nullptr || *v == '\0') return def;
return std::atoi(v);
}
static std::vector<int> envIntList(const char* name, const char* def) {
const char* v = std::getenv(name);
std::string s = (v != nullptr && *v != '\0') ? v : def;
std::vector<int> out;
size_t pos = 0;
while (pos < s.size()) {
size_t comma = s.find(',', pos);
if (comma == std::string::npos) comma = s.size();
out.push_back(std::atoi(s.substr(pos, comma - pos).c_str()));
pos = comma + 1;
}
return out;
}
[[noreturn]] static void fatal(const char* what, const std::string& detail) {
std::fprintf(stderr, "FATAL: %s: %s\n", what, detail.c_str());
std::exit(2);
}
// ---------------------------------------------------------------------------
// Formatting — human-readable ns and table helpers.
// ---------------------------------------------------------------------------
static void fmtNs(char* buf, size_t cap, int64_t ns) {
if (ns >= 1'000'000'000) {
std::snprintf(buf, cap, "%.3f s", static_cast<double>(ns) / 1e9);
} else if (ns >= 1'000'000) {
std::snprintf(buf, cap, "%.3f ms", static_cast<double>(ns) / 1e6);
} else if (ns >= 1'000) {
std::snprintf(buf, cap, "%.1f µs", static_cast<double>(ns) / 1e3);
} else {
std::snprintf(buf, cap, "%" PRId64 " ns", ns);
}
}
static double median(std::vector<double> xs) {
std::sort(xs.begin(), xs.end());
const size_t n = xs.size();
return (n % 2 == 1) ? xs[n / 2] : (xs[n / 2 - 1] + xs[n / 2]) / 2.0;
}
static int64_t percentile(const std::vector<int64_t>& sorted, int pct) {
if (sorted.empty()) return 0;
const size_t i = (sorted.size() * static_cast<size_t>(pct)) / 100;
return sorted[std::min(i, sorted.size() - 1)];
}
// ---------------------------------------------------------------------------
// The one shared predicate — identical formula in the Nim provider
// (`parityPred` in perfbench.nim), so every reply is exactly predictable here.
// ---------------------------------------------------------------------------
static bool parityPred(int64_t a, int64_t b, double x) {
return ((a + b + static_cast<int64_t>(x)) & 1) == 0;
}
static std::vector<std::uint8_t> makePayload(int size) {
std::vector<std::uint8_t> p(static_cast<size_t>(size));
for (size_t i = 0; i < p.size(); ++i) p[i] = static_cast<std::uint8_t>(i & 0xFF);
return p;
}
static bool payloadPredOf(const std::vector<std::uint8_t>& p) {
if (p.empty()) return parityPred(0, 0, 0.0);
return parityPred(static_cast<int64_t>(p.front()), static_cast<int64_t>(p.back()),
static_cast<double>(p.size()));
}
// Scalar family inputs, derived per message index i.
static int64_t scalarA(int64_t i) { return i; }
static int64_t scalarB(int64_t i) { return 3 * i + 1; }
static double scalarX(int64_t i) { return 0.5 * static_cast<double>(i); }
static bool scalarPred(int64_t i) {
return parityPred(scalarA(i), scalarB(i), scalarX(i));
}
static std::unique_ptr<PerfbenchCtx> makeCtx() {
auto r = PerfbenchCtx::create(PerfbenchConfig{"perf"});
if (r.isErr()) fatal("PerfbenchCtx::create", r.error());
return r.take();
}
struct IterOut {
double msgPerSec = 0.0;
std::vector<int64_t> latNs; // empty for lanes without per-call samples
};
// ---------------------------------------------------------------------------
// sync lane — blocking round trips, per-call latency sampled.
// `call(ctx, i)` performs one verified round trip.
// ---------------------------------------------------------------------------
template <typename CallFn>
static IterOut runSync(int threads, int perThread, CallFn&& call) {
auto ctx = makeCtx();
std::atomic<bool> start{false};
std::vector<std::vector<int64_t>> perThreadLat(static_cast<size_t>(threads));
std::vector<std::thread> workers;
workers.reserve(static_cast<size_t>(threads));
for (int t = 0; t < threads; ++t) {
workers.emplace_back([&, t] {
auto& lat = perThreadLat[static_cast<size_t>(t)];
lat.reserve(static_cast<size_t>(perThread));
while (!start.load(std::memory_order_acquire)) {}
for (int64_t i = 0; i < perThread; ++i) {
const auto c0 = sclock::now();
call(*ctx, i);
const auto c1 = sclock::now();
lat.push_back(
std::chrono::duration_cast<std::chrono::nanoseconds>(c1 - c0)
.count());
}
});
}
const auto t0 = sclock::now();
start.store(true, std::memory_order_release);
for (auto& w : workers) w.join(); // blocking calls: join == all done
const auto t1 = sclock::now();
IterOut out;
for (auto& v : perThreadLat)
out.latNs.insert(out.latNs.end(), v.begin(), v.end());
const double sec = std::chrono::duration<double>(t1 - t0).count();
out.msgPerSec =
static_cast<double>(static_cast<uint64_t>(threads) *
static_cast<uint64_t>(perThread)) /
sec;
return out;
}
// ---------------------------------------------------------------------------
// async lane — `*Async` future turnaround with a bounded in-flight window.
// `issue(ctx, i)` returns the future; `expect(i)` the predicted predicate.
// ---------------------------------------------------------------------------
template <typename IssueFn, typename ExpectFn>
static IterOut runAsync(int threads, int perThread, int window, IssueFn&& issue,
ExpectFn&& expect) {
auto ctx = makeCtx();
std::atomic<bool> start{false};
std::atomic<uint64_t> bad{0};
std::vector<std::thread> workers;
workers.reserve(static_cast<size_t>(threads));
for (int t = 0; t < threads; ++t) {
workers.emplace_back([&] {
using Fut = decltype(issue(*ctx, int64_t{0}));
std::deque<std::pair<int64_t, Fut>> inflight;
auto drainOne = [&] {
auto [idx, fut] = std::move(inflight.front());
inflight.pop_front();
auto r = fut.get();
if (r.isErr() || r.value().ok != expect(idx))
bad.fetch_add(1, std::memory_order_relaxed);
};
while (!start.load(std::memory_order_acquire)) {}
for (int64_t i = 0; i < perThread; ++i) {
inflight.emplace_back(i, issue(*ctx, i));
if (inflight.size() >= static_cast<size_t>(window)) drainOne();
}
while (!inflight.empty()) drainOne();
});
}
const auto t0 = sclock::now();
start.store(true, std::memory_order_release);
for (auto& w : workers) w.join();
const auto t1 = sclock::now();
if (bad.load() != 0)
fatal("async correctness",
std::to_string(bad.load()) + " failed or mispredicted replies");
IterOut out;
const double sec = std::chrono::duration<double>(t1 - t0).count();
out.msgPerSec =
static_cast<double>(static_cast<uint64_t>(threads) *
static_cast<uint64_t>(perThread)) /
sec;
return out;
}
// ---------------------------------------------------------------------------
// event lane — each verified sync trigger fires one on_perf_ping event; the
// listener measures delivery latency from the driver-side stamp. The clock is
// std::chrono::steady_clock on both ends (the stamp is passed through Nim
// verbatim), so the delta is a meaningful per-event delivery latency.
// ---------------------------------------------------------------------------
static IterOut runEvent(int threads, int perThread, int payloadBytes) {
auto ctx = makeCtx();
const uint64_t total =
static_cast<uint64_t>(threads) * static_cast<uint64_t>(perThread);
std::vector<int64_t> latencies;
latencies.reserve(total);
std::mutex latMu;
std::atomic<uint64_t> delivered{0};
const auto handle = ctx->addOnPerfPingListener([&](const PerfPingEvent& evt) {
const int64_t nowNs = std::chrono::duration_cast<std::chrono::nanoseconds>(
sclock::now().time_since_epoch())
.count();
{
std::lock_guard<std::mutex> g(latMu);
latencies.push_back(nowNs - evt.stampNs);
}
delivered.fetch_add(1, std::memory_order_relaxed);
});
if (handle.id == 0) fatal("addOnPerfPingListener", "returned zero id");
std::atomic<bool> start{false};
std::vector<std::thread> workers;
workers.reserve(static_cast<size_t>(threads));
for (int t = 0; t < threads; ++t) {
workers.emplace_back([&] {
while (!start.load(std::memory_order_acquire)) {}
for (int64_t i = 0; i < perThread; ++i) {
const int64_t stampNs =
std::chrono::duration_cast<std::chrono::nanoseconds>(
sclock::now().time_since_epoch())
.count();
auto r = ctx->trigger_ping(
TriggerPingRequest{1, payloadBytes, stampNs});
if (r.isErr()) fatal("triggerPing", r.error());
if (r.value().emitted != 1)
fatal("triggerPing", "emitted != 1");
}
});
}
const auto t0 = sclock::now();
start.store(true, std::memory_order_release);
for (auto& w : workers) w.join();
// Drain: the run is over only when every event reached the listener.
const auto deadline = sclock::now() + std::chrono::seconds(60);
while (delivered.load(std::memory_order_acquire) < total) {
if (sclock::now() > deadline)
fatal("event drain", "timeout waiting for event delivery");
std::this_thread::sleep_for(std::chrono::microseconds(200));
}
const auto t1 = sclock::now();
ctx->removeEventListener(handle);
if (delivered.load() != total)
fatal("event correctness", "delivered != emitted");
IterOut out;
{
std::lock_guard<std::mutex> g(latMu);
out.latNs = std::move(latencies);
}
const double sec = std::chrono::duration<double>(t1 - t0).count();
out.msgPerSec = static_cast<double>(total) / sec;
return out;
}
// ---------------------------------------------------------------------------
// Sweep driver — one compact table per (lane, family) with a row per thread
// count: median msg/s over iters, latency percentiles pooled across iters.
// ---------------------------------------------------------------------------
struct Knobs {
std::vector<int> threadCounts;
int perThread = 0;
int iters = 0;
};
static void runScenario(const std::string& name, const std::string& csvTag,
const Knobs& k, int bytesPerMsg,
const std::function<IterOut(int threads)>& fn) {
std::printf("── %s — %d msgs/thread (median of %d) ──────\n", name.c_str(),
k.perThread, k.iters);
std::printf(" %-9s%-11s%-13s%-11s%-12s%-12s%s\n", "threads", "msgs", "msg/s",
"MB/s", "p50", "p99", "vs 1-thread");
double base = 0.0;
for (int threads : k.threadCounts) {
std::vector<double> rates;
std::vector<int64_t> lat;
for (int i = 0; i < k.iters; ++i) {
IterOut out = fn(threads);
rates.push_back(out.msgPerSec);
lat.insert(lat.end(), out.latNs.begin(), out.latNs.end());
}
const double med = median(rates);
if (base == 0.0) base = med;
std::sort(lat.begin(), lat.end());
const int64_t p50 = percentile(lat, 50);
const int64_t p99 = percentile(lat, 99);
char mbs[32], p50Buf[32], p99Buf[32];
if (bytesPerMsg > 0)
std::snprintf(mbs, sizeof(mbs), "%.2f",
med * static_cast<double>(bytesPerMsg) / 1e6);
else
std::snprintf(mbs, sizeof(mbs), "-");
if (lat.empty()) {
std::snprintf(p50Buf, sizeof(p50Buf), "-");
std::snprintf(p99Buf, sizeof(p99Buf), "-");
} else {
fmtNs(p50Buf, sizeof(p50Buf), p50);
fmtNs(p99Buf, sizeof(p99Buf), p99);
}
std::printf(" %-9d%-11llu%-13.0f%-11s%-12s%-12s%s\n", threads,
static_cast<unsigned long long>(
static_cast<uint64_t>(threads) *
static_cast<uint64_t>(k.perThread)),
med, mbs, p50Buf, p99Buf,
(std::to_string(med / base).substr(0, 4) + "x").c_str());
// CSV row: tag,threads,msg/s,p50_ns,p99_ns — diff-friendly capture.
std::printf("csv,perf_ffi,%s,%d,%.0f,%" PRId64 ",%" PRId64 "\n",
csvTag.c_str(), threads, med, p50, p99);
std::fflush(stdout);
}
std::printf("\n");
}
int main() {
Knobs k;
k.threadCounts = envIntList("NIM_FFI_PERF_THREADS", "1,2,4,8");
k.perThread = envInt("NIM_FFI_PERF_PER_THREAD", 2000);
k.iters = envInt("NIM_FFI_PERF_ITERS", 3);
const std::vector<int> payloadSizes =
envIntList("NIM_FFI_PERF_PAYLOAD_SIZES", "64,512,4096,65536");
const int eventPayload = envInt("NIM_FFI_PERF_EVENT_PAYLOAD", 512);
const int window = envInt("NIM_FFI_PERF_ASYNC_WINDOW", 64);
if (k.perThread < 1 || k.iters < 1 || window < 1 || k.threadCounts.empty() ||
payloadSizes.empty()) {
std::fprintf(stderr, "invalid NIM_FFI_PERF_* configuration\n");
return 2;
}
std::printf("# perfbench FFI e2e — shared O(1) parity predicate; setups "
"differ only in transport (scalars vs N-byte payload)\n\n");
// ── scalar family ────────────────────────────────────────────────────────
runScenario("sync C++ -> Nim -> C++ (blocking round trip) [scalar]",
"sync,scalar,0", k, 0, [&](int threads) {
return runSync(threads, k.perThread, [](PerfbenchCtx& ctx,
int64_t i) {
auto r = ctx.scalar_check(
ScalarCheckRequest{scalarA(i), scalarB(i), scalarX(i)});
if (r.isErr()) fatal("scalarCheck", r.error());
if (r.value().ok != scalarPred(i))
fatal("sync scalar correctness", "predicate mismatch");
});
});
runScenario("async C++ -> Nim -> C++ (future turnaround) [scalar]",
"async,scalar,0", k, 0, [&](int threads) {
return runAsync(
threads, k.perThread, window,
[](PerfbenchCtx& ctx, int64_t i) {
return ctx.scalar_checkAsync(ScalarCheckRequest{
scalarA(i), scalarB(i), scalarX(i)});
},
[](int64_t i) { return scalarPred(i); });
});
// ── payload family, size sweep ───────────────────────────────────────────
for (int size : payloadSizes) {
const auto payload = makePayload(size);
const bool expected = payloadPredOf(payload);
const std::string tag = "[" + std::to_string(size) + "B payload]";
runScenario("sync C++ -> Nim -> C++ (blocking round trip) " + tag,
"sync,payload," + std::to_string(size), k, size,
[&](int threads) {
return runSync(threads, k.perThread,
[&](PerfbenchCtx& ctx, int64_t) {
auto r = ctx.payload_check(
PayloadCheckRequest{payload});
if (r.isErr())
fatal("payloadCheck", r.error());
if (r.value().ok != expected)
fatal("sync payload correctness",
"predicate mismatch");
});
});
runScenario("async C++ -> Nim -> C++ (future turnaround) " + tag,
"async,payload," + std::to_string(size), k, size,
[&](int threads) {
return runAsync(
threads, k.perThread, window,
[&](PerfbenchCtx& ctx, int64_t) {
return ctx.payload_checkAsync(
PayloadCheckRequest{payload});
},
[&](int64_t) { return expected; });
});
}
// ── event lane ───────────────────────────────────────────────────────────
runScenario("event C++ -> Nim -> C++ (on_perf_ping delivery) [" +
std::to_string(eventPayload) + "B payload]",
"event,payload," + std::to_string(eventPayload), k, eventPayload,
[&](int threads) {
return runEvent(threads, k.perThread, eventPayload);
});
std::printf(" correctness: all counts and predicates matched expectations.\n");
return 0;
}
Reference in New Issue View Git Blame Copy Permalink