Skip to content

Commit be9c495

Browse files
springfall2008claudegithub-actions[bot]
authored
perf(kernel): persistent worker pool instead of threads per batch (6% faster) (#4546)
* docs: design for a persistent scratch and thread pool in the kernel Batching made planning 35% faster but threading on top of it bought 1.6%. A census showed why: 25,091 batch calls with a median of 6 jobs, against which a fresh thread pool and a fresh 70KB scratch are built per call. Staged so the low-risk half (per-thread scratch, which also helps the serial default) is measured before the concurrency half. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * perf(kernel): give each thread one scratch instead of one per call PARKED - measured as a no-op, do not merge on its own. Replaces the three per-call PkScratch constructions with a thread_local accessor, so a thread grows its ~70KB of buffers once rather than once per pk_run_batch call (~25,000 of them per plan). Measured on the serial benchmark, interleaved, best of 3: 26.291s before, 26.344s after - no difference against ~1.8% run-to-run spread. The hypothesis was wrong: the kernel's comment about the allocator being the bottleneck was recorded under four threads contending, and one thread repeatedly allocating and freeing the same-sized block never leaves the allocator's thread cache. Kept on this branch because the persistent-worker design needs per-worker scratch anyway, and the two tests are worth having either way: one pins that the buffers stop being rebuilt per call (it fails before this change), the other that a reused scratch cannot leak between contexts of different n_steps (verified by mutation - relaxing the assign to skip shrinking makes it fail). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * perf(kernel): keep a persistent worker pool instead of threads per batch pk_run_batch built and joined a fresh std::vector<std::thread> on every call. Measured over one benchmark plan that was 201,545 thread creations costing 1.5s against 3.1s of simulation, which is why threading was worth only 1.6% - the creation was cancelling what the parallelism won. Workers now park on a condition variable for the life of the process, and each has its own flag and condvar so a batch wakes exactly the lanes it needs. Broadcasting instead cost most of the win: a six-job batch woke all fifteen workers, ~250,000 wasted wakeups per plan, and measured 1.3% against 3.7% for targeted wakes. The calling thread takes lane 0 itself, so a six-job batch wakes five. Also gives each thread one scratch instead of one per call. That is a no-op on its own (26.291s vs 26.344s, inside noise - the allocator never left its thread cache) but the parked workers need it. 20-scenario benchmark, best of 3: 26.33s serial, 24.71s at 6 threads - 6.1%, on top of the batching work. All 20 scenarios byte-identical at every thread count, in both the normal regime and one with each job made eight times dearer. threads: auto stays at the core count. A cap looked right on a fast machine, where the curve peaks below it, but in the kernel-heavy regime the curve stops turning over - capping at 4 would cost 10.7% there against 1.3% for no cap at all. resolve_batch_threads carries the numbers. No ABI or parity bump: results are unchanged, so an older binary still loads and runs correctly. All six shipped binaries rebuilt. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * test(kernel): skip the pool tests when the binary predates pk_run_batch The batch and pool tests call run_prediction_kernel_batch, which returns None when the loaded binary has no pk_run_batch. Four of them treated that as a failure rather than a skip, so running against an older binary reported errors instead of degrading the way the product does. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * chore: update prediction kernel binaries for all platforms * fix(kernel): include <system_error> explicitly The pool catches std::system_error when the system refuses a thread. libc++ pulls that declaration in through <thread>, libstdc++ does not, so the file built on macOS and failed on GCC with a parse error at the catch clause. Caught by CI, which builds natively with g++; neither the local clang build nor the zig cross-build sees it, since both use libc++. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * chore: update prediction kernel binaries for all platforms --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
1 parent 9546ffe commit be9c495

12 files changed

Lines changed: 901 additions & 22 deletions

‎.cspell/custom-dictionary-workspace.txt‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ armhf
2525
armv
2626
ASHP
2727
asyncio
28+
atfork
2829
authoriser
2930
autodocstring
3031
autoflake
@@ -380,6 +381,7 @@ preseeded
380381
prevs
381382
protobuf
382383
psum
384+
pthread
383385
pvbat
384386
pvenergytotal
385387
pvlib
@@ -537,10 +539,14 @@ venv
537539
Victron
538540
visibilitychange
539541
waf
542+
waitpid
540543
Wallbox
541544
weblink
542545
welink
543546
Werror
547+
WEXITSTATUS
548+
WIFEXITED
549+
WNOHANG
544550
workmode
545551
writeonly
546552
wrongsha

‎apps/predbat/plan.py‎

Lines changed: 21 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,26 @@
3535
PLAN_PASS_WINDOW_BUDGET = 8
3636

3737

38+
def resolve_batch_threads(threads, cpu_count_value):
39+
"""Map the threads setting onto how many kernel lanes one batch may use.
40+
41+
'auto' takes the core count and is deliberately not capped. On a fast machine the curve is very
42+
flat and peaks slightly below the core count - measured on the 20-scenario benchmark, best of 3:
43+
serial 26.33s, 4 threads 24.89s, 6 threads 24.71s, 8 threads 24.91s, 16 threads 25.04s - so a cap
44+
looks attractive. But re-running with each job made eight times dearer, which is how a machine
45+
where the kernel dominates behaves, the curve stops turning over entirely: 48.92s serial, 32.03s
46+
at 4, 29.98s at 6, 29.85s at 8, 28.94s at 16.
47+
48+
That makes the risk asymmetric. Capping at 4 costs 0.7% on the fast machine but 10.7% on the
49+
kernel-heavy one, while not capping costs 1.3% at worst. The worst case for a low cap is far
50+
worse than the worst case for none, so 'auto' is left alone and anyone who wants fewer lanes sets
51+
threads: explicitly.
52+
"""
53+
if threads == "auto":
54+
return max(cpu_count_value, 1)
55+
return max(int(threads), 1)
56+
57+
3858
def slots_around(target_slots, slot_lengths):
3959
"""
4060
Return a list of slot lengths around the target slots
@@ -1306,11 +1326,7 @@ def calculate_plan(self, recompute=True, debug_mode=False, publish=True):
13061326
# The kernel spreads one batched fan-out across threads with the GIL released for the whole
13071327
# call, so these are real cores - unlike a Python ThreadPool, which peaked at 1.15x on two
13081328
# threads and then degraded below serial (perf/threadpool-prototype).
1309-
threads = self.get_arg("threads", "auto")
1310-
if threads == "auto":
1311-
self.prediction.batch_threads = cpu_count()
1312-
else:
1313-
self.prediction.batch_threads = max(int(threads), 1)
1329+
self.prediction.batch_threads = resolve_batch_threads(self.get_arg("threads", "auto"), cpu_count())
13141330
self.log("Prediction batch using {} kernel thread(s)".format(self.prediction.batch_threads))
13151331
kernel_message, kernel_is_warning = kernel_status_summary(self.prediction)
13161332
self.log("{}Prediction kernel: {}".format("Warn: " if kernel_is_warning else "", kernel_message))

‎apps/predbat/prediction_kernel.cpp‎

Lines changed: 244 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616
// Build: bash apps/predbat/build_kernel.sh (g++/clang, C++17, no dependencies)
1717

1818
#include <algorithm>
19+
#include <atomic>
20+
#include <condition_variable>
1921
#include <cstdint>
2022
#include <limits>
2123
#include <cstdio>
@@ -25,9 +27,16 @@
2527
#include <map>
2628
#include <memory>
2729
#include <mutex>
30+
// std::system_error, thrown when the system refuses a thread. libc++ pulls this in through <thread>,
31+
// libstdc++ does not, so leaving it implicit builds on macOS and fails on GCC.
32+
#include <system_error>
2833
#include <thread>
2934
#include <vector>
3035

36+
#if defined(__unix__) || defined(__APPLE__)
37+
#include <pthread.h>
38+
#endif
39+
3140
// ABI 4: PkScenario::soc_out may be null, and pk_run_batch was added. The null is the reason this had
3241
// to move - Python now passes no SoC buffer for every cached run, and an ABI 3 binary writes to it
3342
// unconditionally, so loading one against this Python segfaults on the first prediction rather than
@@ -304,12 +313,47 @@ struct PkBatchResult {
304313
// each scenario would otherwise heap-allocate ~70KB, and 1200 of those per batch put every worker
305314
// thread in contention on the allocator. Measured 4 threads at 0.6x of serial before this, because
306315
// the simulation is cheap enough that malloc, not arithmetic, was the bottleneck.
316+
//
317+
// See thread_scratch() below: one of these now serves every batch a thread runs, not one per call.
318+
319+
// Counts PkScratch constructions for the life of the library, for pk_scratch_construct_count(). The
320+
// suite asserts this stops growing with the number of batches rather than trusting the reuse.
321+
std::atomic<int64_t> g_scratch_constructions{0};
322+
323+
// Counts worker threads started for the life of the library, for pk_worker_thread_count(). Thread
324+
// creation measured 7.5us each and a benchmark plan used to start 201,545 of them - 1.5s of a 26s
325+
// plan, which is what the pool exists to stop paying. The suite asserts this stops growing.
326+
std::atomic<int64_t> g_worker_threads_started{0};
327+
307328
struct PkScratch {
308329
std::vector<int32_t> charge_member, export_member;
309330
std::vector<int32_t> clipped_start, clipped_end;
310331
std::vector<double> clipped_limit;
332+
333+
PkScratch()
334+
{
335+
g_scratch_constructions.fetch_add(1, std::memory_order_relaxed);
336+
}
311337
};
312338

339+
// One scratch per thread, reused for that thread's lifetime.
340+
//
341+
// Every field is fully rewritten before it is read on each run - build_window_membership assigns
342+
// exactly n_steps entries and the clip helpers clear-then-push - which is the same invariant that
343+
// already let one scratch serve a whole stride within a batch. Extending it across calls means a
344+
// thread grows these buffers once instead of once per call: a benchmark plan issues ~25,000 batch
345+
// calls, and at a median of 6 jobs each the ~70KB allocation was the same order as the simulation it
346+
// wrapped.
347+
//
348+
// Thread-local rather than per-context because nothing here survives a simulation, so it never needs
349+
// to belong to a particular context. A thread that later runs a context with a different n_steps
350+
// simply has the vectors reassigned to the new length.
351+
static PkScratch &thread_scratch()
352+
{
353+
static thread_local PkScratch scratch;
354+
return scratch;
355+
}
356+
313357
// Deep-copied context storage so Python-side buffers can be freed after create
314358
struct ContextStore {
315359
std::vector<double> rate_import, rate_export, alert_keep;
@@ -1281,8 +1325,7 @@ int32_t pk_run(int64_t handle, const PkScenario *s, PkResult *out)
12811325
if (!c) {
12821326
return 1;
12831327
}
1284-
PkScratch scratch;
1285-
return pk_run_one(c, s, out, -1, -1, nullptr, nullptr, scratch); // c is the store
1328+
return pk_run_one(c, s, out, -1, -1, nullptr, nullptr, thread_scratch()); // c is the store
12861329
}
12871330

12881331
// Run n_jobs scenarios against one context in a single call.
@@ -1317,6 +1360,170 @@ static void run_batch_job(const ContextStore *c, const PkBatchJob &job, PkBatchR
13171360
out.status = pk_run_one(c, &scenario, &out.result, job.soc_range_start_step, job.soc_range_end_step, &out.soc_range_min, &out.soc_range_max, scratch);
13181361
}
13191362

1363+
1364+
// A pool of worker threads that outlives every batch and every context.
1365+
//
1366+
// pk_run_batch used to build and join a fresh std::vector<std::thread> per call. Measured over one
1367+
// benchmark plan that was 201,545 thread creations costing 1.5s - against 3.1s of simulation, so the
1368+
// creation cost was cancelling most of what the parallelism won and threading measured 1.6% overall.
1369+
// Parking workers on a condition variable replaces each creation with a wake.
1370+
//
1371+
// Workers park until `generation` changes, then take lane `index` of a stride partition. The calling
1372+
// thread takes lane 0 itself, so a six-job batch wakes five workers rather than six and the caller
1373+
// does useful work instead of blocking. Nobody holds the mutex while simulating; run_batch_job only
1374+
// reads the shared const context and writes its own result slot, which is what made the previous
1375+
// per-call threading safe and is unchanged here.
1376+
class PkThreadPool {
1377+
public:
1378+
// Start workers until at least `want` exist. Grows, never shrinks. Returns how many there are.
1379+
int32_t reserve_workers(int32_t want)
1380+
{
1381+
std::lock_guard<std::mutex> lock(mutex);
1382+
while (static_cast<int32_t>(workers.size()) < want) {
1383+
const int32_t lane = static_cast<int32_t>(workers.size()) + 1; // lane 0 is the caller
1384+
std::unique_ptr<Worker> worker(new Worker());
1385+
try {
1386+
worker->thread = std::thread(&PkThreadPool::worker_loop, this, lane, worker.get());
1387+
} catch (const std::system_error &) {
1388+
// Out of thread resources - keep what we have and let the caller use fewer lanes.
1389+
break;
1390+
}
1391+
workers.push_back(std::move(worker));
1392+
g_worker_threads_started.fetch_add(1, std::memory_order_relaxed);
1393+
}
1394+
return static_cast<int32_t>(workers.size());
1395+
}
1396+
1397+
// Run jobs[0..n_jobs) across `use` lanes, the calling thread taking lane 0. `use` must be at
1398+
// least 1 and no more than worker_count() + 1.
1399+
void run(const ContextStore *c, const PkBatchJob *jobs, int32_t n_jobs, PkBatchResult *results, int32_t use)
1400+
{
1401+
{
1402+
std::lock_guard<std::mutex> lock(mutex);
1403+
cur_ctx = c;
1404+
cur_jobs = jobs;
1405+
cur_results = results;
1406+
cur_n_jobs = n_jobs;
1407+
cur_use = use;
1408+
outstanding = use - 1;
1409+
// Each worker has its own flag and its own condition variable, so exactly the lanes this
1410+
// batch needs are woken. Broadcasting instead would wake every worker in the pool - with
1411+
// a median batch of six jobs against sixteen workers that was ~250,000 wakeups per plan
1412+
// spent entirely on threads that immediately went back to sleep.
1413+
for (int32_t lane = 1; lane < use; lane++) {
1414+
workers[lane - 1]->has_work = true;
1415+
}
1416+
}
1417+
for (int32_t lane = 1; lane < use; lane++) {
1418+
workers[lane - 1]->ready.notify_one();
1419+
}
1420+
1421+
PkScratch &scratch = thread_scratch();
1422+
for (int32_t i = 0; i < n_jobs; i += use) {
1423+
run_batch_job(c, jobs[i], results[i], scratch);
1424+
}
1425+
1426+
std::unique_lock<std::mutex> lock(mutex);
1427+
work_done.wait(lock, [this] { return outstanding == 0; });
1428+
}
1429+
1430+
private:
1431+
// One parked worker. The flag is written under the pool mutex before its condition variable is
1432+
// signalled, so a wake cannot be lost even if the worker has not re-parked yet.
1433+
struct Worker {
1434+
std::condition_variable ready;
1435+
bool has_work = false;
1436+
std::thread thread;
1437+
};
1438+
1439+
void worker_loop(int32_t lane, Worker *self)
1440+
{
1441+
std::unique_lock<std::mutex> lock(mutex);
1442+
for (;;) {
1443+
self->ready.wait(lock, [self] { return self->has_work; });
1444+
self->has_work = false;
1445+
const ContextStore *c = cur_ctx;
1446+
const PkBatchJob *jobs = cur_jobs;
1447+
PkBatchResult *results = cur_results;
1448+
const int32_t n_jobs = cur_n_jobs;
1449+
const int32_t use = cur_use;
1450+
lock.unlock();
1451+
1452+
PkScratch &scratch = thread_scratch();
1453+
for (int32_t i = lane; i < n_jobs; i += use) {
1454+
run_batch_job(c, jobs[i], results[i], scratch);
1455+
}
1456+
1457+
lock.lock();
1458+
if (--outstanding == 0) {
1459+
work_done.notify_one();
1460+
}
1461+
}
1462+
}
1463+
1464+
std::mutex mutex;
1465+
std::condition_variable work_done; // the caller waits here
1466+
const ContextStore *cur_ctx = nullptr;
1467+
const PkBatchJob *cur_jobs = nullptr;
1468+
PkBatchResult *cur_results = nullptr;
1469+
int32_t cur_n_jobs = 0;
1470+
int32_t cur_use = 1;
1471+
int32_t outstanding = 0;
1472+
std::vector<std::unique_ptr<Worker>> workers;
1473+
};
1474+
1475+
// The pool is created on first threaded use and deliberately never destroyed: a static destructor
1476+
// running while CPython tears the process down risks a deadlock for no benefit, and the OS reclaims
1477+
// the threads at exit.
1478+
std::atomic<PkThreadPool *> g_pool{nullptr};
1479+
std::mutex g_pool_create_mutex;
1480+
// Held for a whole batch. Python can enter pk_run_batch from two threads at once because ctypes
1481+
// releases the GIL, and one shared pool cannot serve two batches concurrently - its published job
1482+
// state would race. Nothing in predbat does this today; serialising is the safe answer if it ever does.
1483+
std::mutex g_dispatch_mutex;
1484+
1485+
#if defined(__unix__) || defined(__APPLE__)
1486+
// Threads do not survive fork(), so a child would inherit a pool whose workers do not exist and hang
1487+
// waiting for them. The prepare/parent pair keep the dispatch mutex consistent across the fork, and
1488+
// the child drops the pool so its next batch builds a fresh one. Nothing here allocates or takes a
1489+
// new lock, which is what the child handler is allowed to do; the dead pool leaks in the child.
1490+
void pk_atfork_prepare()
1491+
{
1492+
g_dispatch_mutex.lock();
1493+
}
1494+
1495+
void pk_atfork_parent()
1496+
{
1497+
g_dispatch_mutex.unlock();
1498+
}
1499+
1500+
void pk_atfork_child()
1501+
{
1502+
g_dispatch_mutex.unlock();
1503+
g_pool.store(nullptr, std::memory_order_relaxed);
1504+
}
1505+
#endif
1506+
1507+
// Fetch the pool, creating it on first use.
1508+
static PkThreadPool *pool_singleton()
1509+
{
1510+
PkThreadPool *pool = g_pool.load(std::memory_order_acquire);
1511+
if (pool) {
1512+
return pool;
1513+
}
1514+
std::lock_guard<std::mutex> lock(g_pool_create_mutex);
1515+
pool = g_pool.load(std::memory_order_relaxed);
1516+
if (!pool) {
1517+
#if defined(__unix__) || defined(__APPLE__)
1518+
static std::once_flag atfork_once;
1519+
std::call_once(atfork_once, [] { pthread_atfork(pk_atfork_prepare, pk_atfork_parent, pk_atfork_child); });
1520+
#endif
1521+
pool = new PkThreadPool();
1522+
g_pool.store(pool, std::memory_order_release);
1523+
}
1524+
return pool;
1525+
}
1526+
13201527
// Run n_jobs scenarios against one context in a single call.
13211528
//
13221529
// The context lookup and the Python/C boundary are paid once for a whole fan-out rather than once
@@ -1338,30 +1545,50 @@ int32_t pk_run_batch(int64_t handle, const PkBatchJob *jobs, int32_t n_jobs, PkB
13381545
// Scenarios are independent - each reads the shared const context and writes only its own result
13391546
// slot - so they are split across threads by stride.
13401547
if (n_threads > 1 && n_jobs > 1) {
1341-
const int32_t use = n_threads < n_jobs ? n_threads : n_jobs;
1342-
std::vector<std::thread> pool;
1343-
pool.reserve(use);
1344-
for (int32_t t = 0; t < use; t++) {
1345-
pool.emplace_back([c, jobs, results, n_jobs, use, t]() {
1346-
PkScratch scratch;
1347-
for (int32_t i = t; i < n_jobs; i += use) {
1348-
run_batch_job(c, jobs[i], results[i], scratch);
1349-
}
1350-
});
1548+
std::lock_guard<std::mutex> dispatch(g_dispatch_mutex);
1549+
int32_t use = n_threads < n_jobs ? n_threads : n_jobs;
1550+
PkThreadPool *pool = pool_singleton();
1551+
// Clamp to the lanes that actually exist: reserve_workers stops early if the system refuses a
1552+
// thread, and a batch must never wait on a worker that was never started.
1553+
const int32_t lanes = pool->reserve_workers(use - 1) + 1;
1554+
if (use > lanes) {
1555+
use = lanes;
13511556
}
1352-
for (auto &th : pool) {
1353-
th.join();
1557+
if (use > 1) {
1558+
pool->run(c, jobs, n_jobs, results, use);
1559+
return 0;
13541560
}
1355-
return 0;
1561+
// No workers available - fall through and run the batch inline.
13561562
}
1357-
PkScratch scratch;
13581563
for (int32_t i = 0; i < n_jobs; i++) {
1359-
run_batch_job(c, jobs[i], results[i], scratch);
1564+
run_batch_job(c, jobs[i], results[i], thread_scratch());
13601565
}
13611566
return 0;
13621567
}
13631568

13641569

1570+
// Test hook: how many PkScratch instances have been constructed since the library was loaded.
1571+
//
1572+
// The scratch buffers are the only heap allocation a simulation makes, and the kernel's whole
1573+
// batching win rests on not paying them per call - so the suite asserts this counter stops growing
1574+
// with the number of batches rather than trusting the code to be reusing them.
1575+
int64_t pk_scratch_construct_count(void)
1576+
{
1577+
return g_scratch_constructions.load(std::memory_order_relaxed);
1578+
}
1579+
1580+
1581+
// Test hook: how many worker threads have been started since the library was loaded.
1582+
//
1583+
// Batches used to build and join a fresh set of threads on every call. The suite asserts this count
1584+
// stops growing once the pool is warm, because a thread creation costs several microseconds against
1585+
// a median batch of six jobs - it was cancelling the parallelism it was there to provide.
1586+
int64_t pk_worker_thread_count(void)
1587+
{
1588+
return g_worker_threads_started.load(std::memory_order_relaxed);
1589+
}
1590+
1591+
13651592
// Test hook: sweep SoC densely and confirm the precomputed bucket boundaries give exactly the same
13661593
// percent as the round_py path they replace. Returns the number of disagreements (0 = equivalent).
13671594
int32_t pk_verify_soc_percent_table(double soc_max, int32_t samples)
5.86 KB
Binary file not shown.
5.28 KB
Binary file not shown.
2.72 KB
Binary file not shown.
6.56 KB
Binary file not shown.
5.49 KB
Binary file not shown.
5.34 KB
Binary file not shown.

0 commit comments

Comments
 (0)