java-topology/defects/activemq/unit/ActiveMQTest.java

package unit;

import java.util.*;
import java.util.concurrent.*;

/**
 * ActiveMQTest — CWE-407 benchmark for activemq-0001
 *
 * Models Topic.addSubscription() O(N²) CopyOnWriteArrayList.contains()
 * duplicate check vs. O(1) parallel-Set guard.
 *
 * Real code (activemq-broker/.../region/Topic.java:151,167,293):
 *   synchronized (consumers) {
 *       if (!consumers.contains(sub)) {  // O(N) CopyOnWriteArrayList scan
 *           consumers.add(sub);
 *       }
 *   }
 *
 * Fix: parallel Set<Subscription> from ConcurrentHashMap.newKeySet() for O(1) add.
 */
public class ActiveMQTest {

    static void bench(String label, Runnable slow, Runnable fast, long sOps, long fOps) {
        slow.run(); fast.run();
        long t0 = System.nanoTime(); slow.run(); long sMs = (System.nanoTime() - t0) / 1_000_000;
        long t1 = System.nanoTime(); fast.run(); long fMs = (System.nanoTime() - t1) / 1_000_000;
        double speedup = fMs > 0 ? (double) sMs / fMs : 0;
        System.out.printf("  %-54s  slow:%4dms (%,d ops)  fast:%4dms (%,d ops)  speedup:%.0fx%n",
                label, sMs, sOps, fMs, fOps, speedup);
    }

    // ---------- slow: ArrayList contains scan (the defect) ----------

    /** Returns total membership-test comparisons performed */
    static long slowSubscribe(int N) {
        List<Integer> consumers = new ArrayList<>();
        long ops = 0;
        for (int sub = 0; sub < N; sub++) {
            // simulate contains(): walk every existing entry
            boolean found = false;
            for (int i = 0; i < consumers.size(); i++) {
                ops++;
                if (consumers.get(i).equals(sub)) { found = true; break; }
            }
            if (!found) consumers.add(sub);
        }
        return ops;
    }

    // ---------- fast: Set.add() O(1) guard (the fix) ----------

    /** Returns total membership-test operations (O(1) each) */
    static long fastSubscribe(int N) {
        Set<Integer> consumerSet = new HashSet<>(N * 2);
        List<Integer> consumers = new ArrayList<>();
        long ops = 0;
        for (int sub = 0; sub < N; sub++) {
            ops++; // O(1) set probe
            if (consumerSet.add(sub)) consumers.add(sub);
        }
        return ops;
    }

    public static void main(String[] args) {
        System.out.println("ActiveMQTest — activemq-0001: Topic.addSubscription() CopyOnWriteArrayList → Set guard");
        System.out.println();

        int[][] cases = {{100, 5000}, {500, 2000}, {1000, 1000}};
        System.out.println("  [Topic.addSubscription duplicate guard — consumers CopyOnWriteArrayList.contains()]");
        for (int[] c : cases) {
            int N = c[0], R = c[1];
            bench(
                String.format("N=%d subscribers, %d subscribe() calls", N, R),
                () -> { for (int i = 0; i < R; i++) slowSubscribe(N); },
                () -> { for (int i = 0; i < R; i++) fastSubscribe(N); },
                (long) N * (N - 1) / 2 * R,
                (long) N * R
            );
        }

        System.out.println();
        System.out.println("Defect : activemq-broker/.../region/Topic.java:151,167,293 — consumers.contains(sub) CopyOnWriteArrayList O(n)");
        System.out.println("Fix    : parallel ConcurrentHashMap.newKeySet() guard — O(1) add/contains");
        System.out.println("Ticket : activemq-0001-topic-consumers-copyonwrite-contains-quadratic.md");

        System.out.println();
        int pass = 0;
        long s0 = slowSubscribe(500), f0 = fastSubscribe(500);
        assert s0 > f0 * 50 : "activemq-0001 expected >50x; slow=" + s0 + " fast=" + f0; pass++;
        System.out.printf("%d/1 PASS — activemq-0001: CWE-407 in ActiveMQ Topic subscriber dedup%n", pass);
        System.out.printf("Hotpath: every new subscriber on a busy topic (fan-out bus workloads)%n");
    }
}