New posts

gatling-spring-boot/src/main/java/lab/gatling/DownstreamPool.java

@@ -0,0 +1,64 @@
+package lab.gatling;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+/**
+ * A stand-in for a slow dependency with limited capacity (a legacy pricing service, a
+ * connection-limited database). {@code workers} requests are served at once, each taking
+ * about {@code delayMs}; up to {@code queueSize} more wait; everything beyond is rejected.
+ *
+ * Capacity is therefore workers / delay: 8 / 0.050 s = 160 requests per second.
+ * Below that the service answers in ~50 ms. Above it the queue fills, latency climbs, and
+ * once the queue is full requests are rejected. That is the knee the load test looks for.
+ */
+@Component
+public class DownstreamPool {
+
+    private final ThreadPoolExecutor executor;
+    private final int delayMs;
+    private final AtomicLong rejected = new AtomicLong();
+
+    public DownstreamPool(@Value("${lab.downstream.workers:8}") int workers,
+                          @Value("${lab.downstream.queue:100}") int queueSize,
+                          @Value("${lab.downstream.delay-ms:50}") int delayMs) {
+        this.delayMs = delayMs;
+        this.executor = new ThreadPoolExecutor(workers, workers, 0, TimeUnit.SECONDS,
+                new ArrayBlockingQueue<>(queueSize), new ThreadPoolExecutor.AbortPolicy());
+    }
+
+    /** Runs the lookup on the bounded pool and waits for it. Throws {@link Saturated} when full. */
+    public double price(String symbol) throws InterruptedException {
+        Callable<Double> lookup = () -> {
+            // 0.9x - 1.1x of the nominal delay, derived from the symbol so runs are repeatable
+            long ms = delayMs - delayMs / 10 + Math.floorMod(symbol.hashCode(), Math.max(1, delayMs / 5 + 1));
+            Thread.sleep(ms);
+            return 10 + Math.floorMod(symbol.hashCode(), 49000) / 100.0;
+        };
+        try {
+            return executor.submit(lookup).get();
+        } catch (RejectedExecutionException e) {
+            rejected.incrementAndGet();
+            throw new Saturated();
+        } catch (ExecutionException e) {
+            throw new IllegalStateException(e.getCause());
+        }
+    }
+
+    public Stats stats() {
+        return new Stats(executor.getActiveCount(), executor.getQueue().size(),
+                executor.getCompletedTaskCount(), rejected.get());
+    }
+
+    public record Stats(int active, int queued, long completed, long rejected) { }
+
+    public static class Saturated extends RuntimeException { }
+}