1919 * multiple threads executing Reserve→Commit lifecycles simultaneously.
2020 *
2121 * Lifecycle tests ramp from 8 → 16 → 32 concurrent threads. Reserve-only
22- * tests add a 200-client fan-out in two shapes: one shared budget (the
23- * contention case) and 200 independent leaf budgets (the sharded case).
22+ * tests ramp from 1 → 10 → 50 → 200 clients in two shapes: one shared
23+ * budget (the contention case) and independent leaf budgets (the sharded
24+ * case).
2425 *
2526 * Results are CI-environment sensitive — latency and throughput depend on
2627 * container resources, Redis container networking, and JVM warm-up.
2728 *
2829 * Run separately: mvn test -Pbenchmark
30+ * Select fan-out levels for isolated fresh-process trials with
31+ * -Dbenchmark.fanout.clients=1,10,50,200.
2932 */
3033@ DisplayName ("Concurrent Load Benchmarks" )
3134@ Tag ("benchmark" )
3437class CyclesProtocolConcurrentBenchmarkTest extends BaseIntegrationTest {
3538
3639 private static final int WARMUP_OPS = 50 ;
40+ private static final int MAX_WARMUP_CONCURRENCY = 50 ;
3741 private static final long MEASURE_DURATION_MS = 5_000 ;
38- private static final int FANOUT_CLIENTS = 200 ;
42+ private static final List <Integer > DEFAULT_FANOUT_CLIENT_LEVELS =
43+ List .of (1 , 10 , 50 , 200 );
44+ private static final List <Integer > FANOUT_CLIENT_LEVELS =
45+ configuredFanoutClientLevels ();
3946 private static final long FANOUT_ALLOCATION = 1_000_000_000_000L ;
4047 private static final long FANOUT_RESERVE_AMOUNT = 100L ;
4148 /** Max acceptable error rate (%) before failing the test */
@@ -111,16 +118,20 @@ void concurrentLifecycle_32threads() throws Exception {
111118
112119 @ Test
113120 @ Order (4 )
114- @ DisplayName ("Reserve fan-out at 200 clients on one shared budget" )
115- void concurrentReserve_200clients_sharedBudget () throws Exception {
116- runConcurrentReserveFanout (false );
121+ @ DisplayName ("Reserve fan-out from 1 to 200 clients on one shared budget" )
122+ void concurrentReserve_sharedBudget () throws Exception {
123+ for (int clients : FANOUT_CLIENT_LEVELS ) {
124+ runConcurrentReserveFanout (false , clients );
125+ }
117126 }
118127
119128 @ Test
120129 @ Order (5 )
121- @ DisplayName ("Reserve fan-out at 200 clients on independent leaf budgets" )
122- void concurrentReserve_200clients_independentBudgets () throws Exception {
123- runConcurrentReserveFanout (true );
130+ @ DisplayName ("Reserve fan-out from 1 to 200 clients on independent leaf budgets" )
131+ void concurrentReserve_independentBudgets () throws Exception {
132+ for (int clients : FANOUT_CLIENT_LEVELS ) {
133+ runConcurrentReserveFanout (true , clients );
134+ }
124135 }
125136
126137 private void runConcurrentLifecycle (int threadCount ) throws Exception {
@@ -227,30 +238,27 @@ private void runConcurrentLifecycle(int threadCount) throws Exception {
227238 }
228239 }
229240
230- private void runConcurrentReserveFanout (boolean isolated ) throws Exception {
231- prepareFanoutBudgets (isolated );
241+ private void runConcurrentReserveFanout (
242+ boolean isolated ,
243+ int clientCount ) throws Exception {
244+ prepareFanoutBudgets (isolated , clientCount );
232245
233- // Prime JIT, auth cache, HTTP connection management, and EVALSHA before
234- // resetting the measured ledger state.
235- for (int i = 0 ; i < WARMUP_OPS ; i ++) {
236- ResponseEntity <Map > response = post (
237- "/v1/reservations" ,
238- API_KEY_SECRET_A ,
239- fanoutReservationBody (isolated , i % FANOUT_CLIENTS ));
240- assertThat (response .getStatusCode ().value ()).isEqualTo (200 );
241- assertThat (response .getBody ()).containsKey ("reservation_id" );
242- }
243- prepareFanoutBudgets (isolated );
246+ // Prime JIT, auth cache, HTTP connection management, and EVALSHA
247+ // before resetting the ledger. The bounded ramp exercises every
248+ // logical client without turning 200 simultaneous cold connections
249+ // into a warmup-only transport failure.
250+ warmUpFanout (isolated , clientCount );
251+ prepareFanoutBudgets (isolated , clientCount );
244252
245- ExecutorService executor = Executors .newFixedThreadPool (FANOUT_CLIENTS );
253+ ExecutorService executor = Executors .newFixedThreadPool (clientCount );
246254 try {
247255 ConcurrentLinkedQueue <Long > timings = new ConcurrentLinkedQueue <>();
248256 AtomicInteger errorCount = new AtomicInteger ();
249- AtomicLongArray successesByClient = new AtomicLongArray (FANOUT_CLIENTS );
257+ AtomicLongArray successesByClient = new AtomicLongArray (clientCount );
250258 CountDownLatch startLatch = new CountDownLatch (1 );
251259 AtomicBoolean running = new AtomicBoolean (true );
252260
253- for (int client = 0 ; client < FANOUT_CLIENTS ; client ++) {
261+ for (int client = 0 ; client < clientCount ; client ++) {
254262 int clientIndex = client ;
255263 executor .submit (() -> {
256264 try {
@@ -307,13 +315,13 @@ private void runConcurrentReserveFanout(boolean isolated) throws Exception {
307315 : 0.0 ;
308316 double opsPerSec = totalOps / (MEASURE_DURATION_MS / 1000.0 );
309317 long ledgerMismatches = countFanoutLedgerMismatches (
310- isolated , successesByClient , totalOps );
318+ isolated , successesByClient , totalOps , clientCount );
311319
312320 FanoutResult result ;
313321 if (totalOps > 0 ) {
314322 result = new FanoutResult (
315323 isolated ? "isolated" : "shared" ,
316- FANOUT_CLIENTS ,
324+ clientCount ,
317325 totalOps ,
318326 opsPerSec ,
319327 p (sorted , 50 ),
@@ -327,7 +335,7 @@ private void runConcurrentReserveFanout(boolean isolated) throws Exception {
327335 } else {
328336 result = new FanoutResult (
329337 isolated ? "isolated" : "shared" ,
330- FANOUT_CLIENTS ,
338+ clientCount ,
331339 0 , 0 , 0 , 0 , 0 , 0 , 0 ,
332340 errors , errorRate , ledgerMismatches );
333341 }
@@ -346,27 +354,61 @@ private void runConcurrentReserveFanout(boolean isolated) throws Exception {
346354 result .errorRatePercent , result .ledgerMismatches );
347355
348356 assertThat (errorRate )
349- .as ("Reserve error rate for %s 200-client fan-out" , result .shape )
357+ .as ("Reserve error rate for %s %d-client fan-out" ,
358+ result .shape , result .clients )
350359 .isLessThan (MAX_ERROR_RATE_PERCENT );
351360 assertThat (totalOps )
352- .as ("Successful reserves for %s 200-client fan-out" , result .shape )
361+ .as ("Successful reserves for %s %d-client fan-out" ,
362+ result .shape , result .clients )
353363 .isGreaterThan (0 );
354364 assertThat (ledgerMismatches )
355- .as ("Ledger mismatches for %s 200-client fan-out" , result .shape )
365+ .as ("Ledger mismatches for %s %d-client fan-out" ,
366+ result .shape , result .clients )
356367 .isZero ();
357368 } finally {
358369 executor .shutdownNow ();
359370 }
360371 }
361372
362- private void prepareFanoutBudgets (boolean isolated ) {
373+ private void warmUpFanout (boolean isolated , int clientCount )
374+ throws Exception {
375+ int warmupOps = Math .max (WARMUP_OPS , clientCount );
376+ int warmupConcurrency =
377+ Math .min (clientCount , MAX_WARMUP_CONCURRENCY );
378+ ExecutorService executor =
379+ Executors .newFixedThreadPool (warmupConcurrency );
380+ try {
381+ List <Future <ResponseEntity <Map >>> responses =
382+ new ArrayList <>(warmupOps );
383+ for (int operation = 0 ; operation < warmupOps ; operation ++) {
384+ int clientIndex = operation % clientCount ;
385+ responses .add (executor .submit (() -> post (
386+ "/v1/reservations" ,
387+ API_KEY_SECRET_A ,
388+ fanoutReservationBody (isolated , clientIndex ))));
389+ }
390+ executor .shutdown ();
391+ for (Future <ResponseEntity <Map >> future : responses ) {
392+ ResponseEntity <Map > response = future .get (30 , TimeUnit .SECONDS );
393+ assertThat (response .getStatusCode ().value ()).isEqualTo (200 );
394+ assertThat (response .getBody ()).containsKey ("reservation_id" );
395+ }
396+ assertThat (executor .awaitTermination (30 , TimeUnit .SECONDS ))
397+ .as ("Fan-out warmup workers terminated" )
398+ .isTrue ();
399+ } finally {
400+ executor .shutdownNow ();
401+ }
402+ }
403+
404+ private void prepareFanoutBudgets (boolean isolated , int clientCount ) {
363405 try (var jedis = jedisPool .getResource ()) {
364406 if (!isolated ) {
365407 seedBudget (jedis , TENANT_A , "TOKENS" , FANOUT_ALLOCATION );
366408 return ;
367409 }
368410 jedis .del ("budget:tenant:" + TENANT_A + ":TOKENS" );
369- for (int client = 0 ; client < FANOUT_CLIENTS ; client ++) {
411+ for (int client = 0 ; client < clientCount ; client ++) {
370412 seedScopeBudget (
371413 jedis ,
372414 fanoutScope (client ),
@@ -391,7 +433,8 @@ private Map<String, Object> fanoutReservationBody(boolean isolated, int client)
391433 private long countFanoutLedgerMismatches (
392434 boolean isolated ,
393435 AtomicLongArray successesByClient ,
394- int totalOps ) {
436+ int totalOps ,
437+ int clientCount ) {
395438 try (var jedis = jedisPool .getResource ()) {
396439 if (!isolated ) {
397440 long actual = Long .parseLong (jedis .hget (
@@ -400,7 +443,7 @@ private long countFanoutLedgerMismatches(
400443 return actual == totalOps * FANOUT_RESERVE_AMOUNT ? 0 : 1 ;
401444 }
402445 long mismatches = 0 ;
403- for (int client = 0 ; client < FANOUT_CLIENTS ; client ++) {
446+ for (int client = 0 ; client < clientCount ; client ++) {
404447 long actual = Long .parseLong (jedis .hget (
405448 "budget:" + fanoutScope (client ) + ":TOKENS" ,
406449 "reserved" ));
@@ -422,6 +465,28 @@ private static String fanoutScope(int client) {
422465 return "tenant:" + TENANT_A + "/agent:" + fanoutAgent (client );
423466 }
424467
468+ private static List <Integer > configuredFanoutClientLevels () {
469+ String configured = System .getProperty ("benchmark.fanout.clients" );
470+ if (configured == null || configured .isBlank ()) {
471+ return DEFAULT_FANOUT_CLIENT_LEVELS ;
472+ }
473+ try {
474+ List <Integer > levels = Arrays .stream (configured .split ("," ))
475+ .map (String ::trim )
476+ .map (Integer ::parseInt )
477+ .toList ();
478+ if (levels .isEmpty () || levels .stream ().anyMatch (level -> level <= 0 )) {
479+ throw new IllegalArgumentException (
480+ "benchmark.fanout.clients values must be positive" );
481+ }
482+ return levels ;
483+ } catch (NumberFormatException e ) {
484+ throw new IllegalArgumentException (
485+ "benchmark.fanout.clients must be a comma-separated list of integers" ,
486+ e );
487+ }
488+ }
489+
425490 private static long p (long [] sorted , int percentile ) {
426491 return sorted [percentileIndex (sorted .length , percentile )];
427492 }
0 commit comments