java-topology/defects/storm/unit/StormWorkerStateTaskLookupTest.java

package unit;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * storm-0002: WorkerState refreshConnections — ArrayList.contains() O(n²) vs HashSet O(n).
 *
 * Standalone unit test — no JUnit required.
 * Compile: javac -d . StormWorkerStateTaskLookupTest.java
 * Run:     java unit.StormWorkerStateTaskLookupTest
 */
public class StormWorkerStateTaskLookupTest {

    static long slowOps;
    static long fastOps;

    /**
     * Slow: mirrors WorkerState.refreshConnections() — localTaskIds is ArrayList.
     * For each of m tasks in assignment, calls localTaskIds.contains() = O(n).
     * Total: O(m * n). With m ≈ n (all tasks outbound) → O(n²).
     */
    static int slowRefresh(List<Integer> localTaskIds, Set<Integer> outboundTasks,
                           List<Integer> allTasks) {
        slowOps = 0;
        int neededConnections = 0;
        for (Integer task : allTasks) {
            slowOps++;
            if (outboundTasks.contains(task)) {         // Set — O(1)
                slowOps++;
                boolean isLocal = false;
                for (Integer local : localTaskIds) {    // ArrayList.contains() scan — O(n)
                    slowOps++;
                    if (local.equals(task)) { isLocal = true; break; }
                }
                if (!isLocal) neededConnections++;
            }
        }
        return neededConnections;
    }

    /**
     * Fast: patched — localTaskIdSet is HashSet<Integer>.
     * contains() is O(1); total loop is O(m).
     */
    static int fastRefresh(HashSet<Integer> localTaskIdSet, Set<Integer> outboundTasks,
                           List<Integer> allTasks) {
        fastOps = 0;
        int neededConnections = 0;
        for (Integer task : allTasks) {
            fastOps++;
            if (outboundTasks.contains(task)) {         // Set — O(1)
                fastOps++;
                if (!localTaskIdSet.contains(task)) {   // HashSet — O(1)
                    fastOps++;
                    neededConnections++;
                }
            }
        }
        return neededConnections;
    }

    static void run(int n, int expectedNx) {
        // n local tasks, 2n total tasks in assignment (half remote)
        List<Integer> localTaskIds = new ArrayList<>();
        HashSet<Integer> localTaskIdSet = new HashSet<>();
        for (int i = 0; i < n; i++) {
            localTaskIds.add(i);
            localTaskIdSet.add(i);
        }

        List<Integer> allTasks = new ArrayList<>();
        Set<Integer> outboundTasks = new HashSet<>();
        for (int i = 0; i < 2 * n; i++) {
            allTasks.add(i);
            outboundTasks.add(i);   // all tasks are outbound
        }

        int slowResult = slowRefresh(localTaskIds, outboundTasks, allTasks);
        int fastResult = fastRefresh(localTaskIdSet, outboundTasks, allTasks);

        boolean resultsMatch = (slowResult == fastResult);
        boolean quadraticWorse = slowOps > fastOps * expectedNx;

        System.out.printf("n=%-4d  slow=%6d  fast=%4d  ratio=%5.1fx  match=%b  PASS=%b%n",
            n, slowOps, fastOps, (double) slowOps / fastOps, resultsMatch,
            resultsMatch && quadraticWorse);

        if (!resultsMatch || !quadraticWorse) {
            throw new AssertionError(
                "FAIL n=" + n + " slowResult=" + slowResult + " fastResult=" + fastResult +
                " slowOps=" + slowOps + " fastOps=" + fastOps +
                " needed ratio>" + expectedNx);
        }
    }

    public static void main(String[] args) {
        System.out.println("=== storm-0002: WorkerState task lookup O(n^2) vs O(n) ===");
        run(100,  10);
        run(500,  30);
        run(1000, 60);
        System.out.println("3/3 PASS");
    }
}