diff --git a/src/core/platform.c b/src/core/platform.c index bad99c91d..5251a6712 100644 --- a/src/core/platform.c +++ b/src/core/platform.c @@ -369,7 +369,7 @@ static uint64_t cache_sysfs_llc_bytes(void) { } #endif -uint64_t ray_cache_llc_bytes(void) { +static uint64_t cache_llc_probe(void) { static uint64_t cached = UINT64_MAX; if (cached != UINT64_MAX) return cached; uint64_t bytes = 0; @@ -642,7 +642,7 @@ uint32_t ray_physical_core_count(void) { /* Sum of every level-3 cache instance reported by the processor topology * (each SYSTEM_LOGICAL_PROCESSOR_INFORMATION cache record is one instance). * 0 when the query fails. */ -uint64_t ray_cache_llc_bytes(void) { +static uint64_t cache_llc_probe(void) { static uint64_t cached = UINT64_MAX; if (cached != UINT64_MAX) return cached; uint64_t bytes = 0; @@ -800,7 +800,7 @@ ray_err_t ray_thread_join(ray_thread_t t) { } uint32_t ray_thread_count(void) { return 1; } -uint64_t ray_cache_llc_bytes(void) { return 0; } +static uint64_t cache_llc_probe(void) { return 0; } /* Semaphore — counter-only. Single-threaded so wait never blocks (the * counter must already be positive when wait fires). */ @@ -818,3 +818,21 @@ void ray_sem_wait(ray_sem_t* s) { void ray_sem_signal(ray_sem_t* s) { (*s)++; } #endif /* RAY_OS_WASM */ + +#ifdef DEBUG +/* Test pin for the probed LLC size: routing that bounds replicated state by + * the cache (group dense slabs) otherwise picks a different strategy on + * every CI runner. 0 restores the platform probe. */ +static uint64_t g_llc_for_test = 0; + +void ray_cache_llc_set_for_test(uint64_t bytes) { + g_llc_for_test = bytes; +} +#endif + +uint64_t ray_cache_llc_bytes(void) { +#ifdef DEBUG + if (g_llc_for_test) return g_llc_for_test; +#endif + return cache_llc_probe(); +} diff --git a/src/core/platform.h b/src/core/platform.h index ea3633c56..e2d137856 100644 --- a/src/core/platform.h +++ b/src/core/platform.h @@ -180,6 +180,10 @@ uint32_t ray_physical_core_count(void); * 0 when the platform cannot report it. Bounds replicated per-task state * whose random-access working set must stay cache-resident to scale. */ uint64_t ray_cache_llc_bytes(void); +#ifdef DEBUG +/* Pin ray_cache_llc_bytes to `bytes` (0 = probe again). */ +void ray_cache_llc_set_for_test(uint64_t bytes); +#endif void ray_parallel_begin(void); void ray_parallel_end(void); diff --git a/test/test_agg_contract.c b/test/test_agg_contract.c index 2134755cf..3f61880c3 100644 --- a/test/test_agg_contract.c +++ b/test/test_agg_contract.c @@ -1734,7 +1734,13 @@ static test_result_t test_cancelled_group(void) { * 20-worker pool and a 100k-slot slab the raw replication (20 slabs) leaves * most caches, and the run must use at most floor(0.75 * LLC / slab) task * slabs (never fewer than the pool when everything fits). The result is - * identical either way. */ + * identical either way. + * + * The LLC is pinned to 32 MB (the size the route assumes when none is + * reported) for the routed query: ~10 slabs fit, so the bound bites but the + * run is not cache-starved. Unpinned, a small-cache runner (a macOS CI VM + * reports a few MB of L2) fits fewer than three slabs and the route rightly + * switches to partition ownership, failing the strategy assertion. */ static test_result_t test_dense_cache_bound(void) { ray_pool_destroy(); TEST_ASSERT_EQ_I(ray_pool_init_total(20), RAY_OK); @@ -1743,21 +1749,24 @@ static test_result_t test_dense_cache_bound(void) { "(set cb_t (table [k v] (list (as 'I32 (% (* cb_i 7919) 100000)) (% cb_i 13))))"); TEST_ASSERT_NOT_NULL(setup); TEST_ASSERT_FALSE(RAY_IS_ERR(setup)); ray_release(setup); agg_route_reset(); + ray_cache_llc_set_for_test(32ull << 20); ray_t* r = ray_eval_str("(select {from:cb_t by:k s:(sum v)})"); - TEST_ASSERT_NOT_NULL(r); TEST_ASSERT_FALSE(RAY_IS_ERR(r)); agg_route_stats_t stats = agg_route_stats(); + uint64_t llc = ray_cache_llc_bytes(); + ray_cache_llc_set_for_test(0); /* before any assert can return */ + TEST_ASSERT_NOT_NULL(r); TEST_ASSERT_FALSE(RAY_IS_ERR(r)); TEST_ASSERT_EQ_I(stats.routes[AGG_ROUTE_V2_DENSE], 1); TEST_ASSERT_EQ_I(stats.dense_strategy, AGG_DENSE_TASK_LOCAL); TEST_ASSERT_TRUE(stats.dense_tasks >= 2 && stats.dense_tasks <= 20); TEST_ASSERT_EQ_I(ray_table_nrows(r), 100000); - uint64_t llc = ray_cache_llc_bytes(); - if (llc > 0) { + { size_t block = agg_resolve(OP_SUM, RAY_I64)->state_size; double slots = (double)stats.dense_local_slots / stats.dense_tasks; double slab = slots * (block + sizeof(int64_t) + 1); double budget = (double)llc * 0.75; uint32_t cap = slab * 20 > budget ? (uint32_t)(budget / slab) : 20; if (cap < 2) cap = 2; + TEST_ASSERT_TRUE(cap >= 3 && cap < 20); /* the pin makes the bound bite */ TEST_ASSERT_EQ_I(stats.dense_tasks, cap); } /* The bounded run computes the same sums as the serial engine. */