package unit;

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

/**
 * hadoop-0002: HeartbeatManager.heartbeatCheck() — ArrayList.contains() O(n)
 * called inside nested loops over datanodes and their storage infos.
 * Total complexity: O(D * S * K) where D=datanodes, S=storages/node, K=dead nodes.
 *
 * Standalone unit test — no JUnit required.
 * Compile: javac -d . HadoopHeartbeatManagerTest.java
 * Run:     java -ea unit.HadoopHeartbeatManagerTest
 */
public class HadoopHeartbeatManagerTest {

    static long slowOps;
    static long fastOps;

    /**
     * Simulates the defective heartbeat check loop.
     * deadDatanodes is ArrayList — contains() is O(K) per call.
     */
    static int slowHeartbeatCheck(int numDatanodes, int storagesPerNode, int numDead) {
        slowOps = 0;
        List<Integer> deadDatanodes = new ArrayList<>();
        // Pre-populate dead set (first numDead datanodes are "dead")
        for (int i = 0; i < numDead; i++) deadDatanodes.add(i);

        int failedStoragesCount = 0;
        for (int d = 0; d < numDatanodes; d++) {         // outer: all datanodes
            slowOps++;
            for (int s = 0; s < storagesPerNode; s++) {  // inner: all storages
                slowOps++;
                // Simulate areBlocksOnFailedStorage() — true for storage 0 of every node
                boolean failedStorage = (s == 0);
                if (failedStorage) {
                    for (Integer dead : deadDatanodes) {  // ArrayList.contains() scan
                        slowOps++;
                        if (dead.equals(d)) break;
                    }
                    if (!deadDatanodes.contains(d)) {
                        failedStoragesCount++;
                    }
                }
            }
        }
        return failedStoragesCount;
    }

    /**
     * Patched: deadDatanodes is HashSet — contains() is O(1).
     */
    static int fastHeartbeatCheck(int numDatanodes, int storagesPerNode, int numDead) {
        fastOps = 0;
        Set<Integer> deadDatanodes = new HashSet<>();
        for (int i = 0; i < numDead; i++) deadDatanodes.add(i);

        int failedStoragesCount = 0;
        for (int d = 0; d < numDatanodes; d++) {         // outer: all datanodes
            fastOps++;
            for (int s = 0; s < storagesPerNode; s++) {  // inner: all storages
                fastOps++;
                boolean failedStorage = (s == 0);
                if (failedStorage) {
                    fastOps++;  // O(1) HashSet.contains()
                    if (!deadDatanodes.contains(d)) {
                        failedStoragesCount++;
                    }
                }
            }
        }
        return failedStoragesCount;
    }

    static void run(int D, int S, int K, int expectedRatio) {
        int slowResult = slowHeartbeatCheck(D, S, K);
        int fastResult = fastHeartbeatCheck(D, S, K);

        boolean resultsMatch = (slowResult == fastResult);
        boolean quadraticWorse = slowOps > fastOps * expectedRatio;
        boolean pass = resultsMatch && quadraticWorse;

        System.out.printf("D=%-4d S=%-3d K=%-4d  slow=%8d  fast=%6d  ratio=%6.1fx  match=%b  PASS=%b%n",
            D, S, K, slowOps, fastOps, (double) slowOps / fastOps, resultsMatch, pass);

        if (!pass) {
            throw new AssertionError(
                "FAIL D=" + D + " S=" + S + " K=" + K +
                " resultsMatch=" + resultsMatch +
                " slowOps=" + slowOps + " fastOps=" + fastOps +
                " needed ratio>" + expectedRatio);
        }
    }

    public static void main(String[] args) {
        System.out.println("=== hadoop-0002: HeartbeatManager deadDatanodes.contains() O(D*S*K) vs O(D*S) ===");
        run(100,  5,  10, 2);
        run(500,  10, 30, 3);
        run(1000, 10, 50, 4);
        System.out.println("3/3 PASS");
    }
}