package unit;

import java.util.*;

/**
 * ceph-0001 — OSDMap::calc_pg_upmaps: std::find on underfull vector inside deviation_osd loop
 *
 * Models src/osd/OSDMap.cc:5981-5983:
 *
 *   for (auto& [deviation, osd] : deviation_osd) {
 *       if (std::find(underfull.begin(), underfull.end(), osd) ==
 *                     underfull.end())
 *           break;
 *       // ... try_drop_remap_underfull ...
 *   }
 *
 * deviation_osd = all OSDs sorted by fill deviation (size N_osds)
 * underfull     = OSDs below -max_deviation threshold (size U, up to N/2)
 *
 * This is inside while(max--) outer loop (up to 100 iterations).
 *
 * Defective: O(max_iters × N_osds × U) via std::find
 * Fixed:     O(max_iters × N_osds)     via unordered_set<int>
 *
 * Also models CrushWrapper::try_remap_rule:
 *   for (auto item : underfull) { std::find(orig.begin(), orig.end(), item) }
 *   → O(U × |orig|) per PG per level per iter
 *
 * No JUnit. Uses assert. Prints N/N PASS.
 *
 * Compile: javac -d . CephOSDMapUpmapAlgorithm.java
 * Run:     java -ea -cp . unit.CephOSDMapUpmapAlgorithm
 */
public class CephOSDMapUpmapAlgorithm {

    static int passed = 0;
    static int total  = 0;

    static void check(String desc, boolean cond) {
        total++;
        if (cond) {
            passed++;
            System.out.println("PASS: " + desc);
        } else {
            System.out.println("FAIL: " + desc);
            throw new AssertionError("FAIL: " + desc);
        }
    }

    // -----------------------------------------------------------------------
    // Slow: models deviation_osd scan with std::find on underfull vector
    // Returns total comparison operations performed across all iterations.
    // -----------------------------------------------------------------------
    static long calcPgUpmapsSlow(int nOsds, int nUnderfull, int maxIters) {
        long ops = 0;
        // Build deviation_osd: all OSDs sorted by deviation (just indices)
        List<Integer> deviationOsd = new ArrayList<>(nOsds);
        for (int i = 0; i < nOsds; i++) deviationOsd.add(i);

        // Build underfull vector: first nUnderfull OSDs
        List<Integer> underfull = new ArrayList<>(nUnderfull);
        for (int i = 0; i < nUnderfull; i++) underfull.add(i);

        for (int iter = 0; iter < maxIters; iter++) {
            // Simulate the scan loop at OSDMap.cc:5981
            for (int osd : deviationOsd) {
                // std::find(underfull.begin(), underfull.end(), osd) — O(U)
                boolean found = false;
                for (int u : underfull) {
                    ops++;
                    if (u == osd) { found = true; break; }
                }
                if (!found) break; // early break on first non-underfull OSD
                // ... try_drop_remap_underfull (modeled as O(1) here)
            }
        }
        return ops;
    }

    // -----------------------------------------------------------------------
    // Fast: unordered_set<int> for O(1) membership test
    // Returns total comparison operations performed across all iterations.
    // -----------------------------------------------------------------------
    static long calcPgUpmapsFast(int nOsds, int nUnderfull, int maxIters) {
        long ops = 0;
        List<Integer> deviationOsd = new ArrayList<>(nOsds);
        for (int i = 0; i < nOsds; i++) deviationOsd.add(i);

        // Build underfull vector: first nUnderfull OSDs
        List<Integer> underfullList = new ArrayList<>(nUnderfull);
        for (int i = 0; i < nUnderfull; i++) underfullList.add(i);

        for (int iter = 0; iter < maxIters; iter++) {
            // Build unordered_set once per iteration (after fill_overfull_underfull)
            Set<Integer> underfullSet = new HashSet<>(underfullList);
            ops += nUnderfull; // cost to build set

            for (int osd : deviationOsd) {
                ops++; // O(1) hash lookup
                if (!underfullSet.contains(osd)) break;
                // ... try_drop_remap_underfull
            }
        }
        return ops;
    }

    // -----------------------------------------------------------------------
    // Correctness: both identify the same underfull OSDs
    // -----------------------------------------------------------------------
    static List<Integer> findUnderfullSlow(List<Integer> deviationOsd, List<Integer> underfull) {
        List<Integer> result = new ArrayList<>();
        for (int osd : deviationOsd) {
            boolean found = false;
            for (int u : underfull) {
                if (u == osd) { found = true; break; }
            }
            if (!found) break;
            result.add(osd);
        }
        return result;
    }

    static List<Integer> findUnderfullFast(List<Integer> deviationOsd, List<Integer> underfull) {
        Set<Integer> underfullSet = new HashSet<>(underfull);
        List<Integer> result = new ArrayList<>();
        for (int osd : deviationOsd) {
            if (!underfullSet.contains(osd)) break;
            result.add(osd);
        }
        return result;
    }

    // -----------------------------------------------------------------------
    // CrushWrapper model: for(item in underfull) { std::find(orig, item) }
    // -----------------------------------------------------------------------
    static long tryRemapRuleSlow(List<Integer> underfull, List<Integer> orig) {
        long ops = 0;
        for (int item : underfull) {
            // std::find(orig.begin(), orig.end(), item)
            for (int o : orig) {
                ops++;
                if (o == item) break;
            }
        }
        return ops;
    }

    static long tryRemapRuleFast(List<Integer> underfull, List<Integer> orig) {
        long ops = 0;
        // Build orig_set once
        Set<Integer> origSet = new HashSet<>(orig);
        ops += orig.size();
        for (int item : underfull) {
            ops++; // O(1) hash lookup
            origSet.contains(item);
        }
        return ops;
    }

    public static void main(String[] args) {
        System.out.println("ceph-0001: OSDMap::calc_pg_upmaps underfull std::find O(N×U) → O(N)");
        System.out.println();

        // --- Correctness ---
        List<Integer> deviationOsd = Arrays.asList(0, 1, 2, 3, 4, 5, 6, 7);
        List<Integer> underfull    = Arrays.asList(0, 1, 2, 3); // first 4 are underfull

        List<Integer> slowResult = findUnderfullSlow(deviationOsd, underfull);
        List<Integer> fastResult = findUnderfullFast(deviationOsd, underfull);

        check("correctness: same underfull OSDs found", slowResult.equals(fastResult));
        check("correctness: 4 underfull OSDs", slowResult.size() == 4);
        check("correctness: OSD 0 in result",  slowResult.contains(0));
        check("correctness: OSD 3 in result",  slowResult.contains(3));
        check("correctness: OSD 4 not in result", !slowResult.contains(4));

        // --- Performance: calc_pg_upmaps deviation scan ---
        // Small cluster
        int nOsds = 100, nUnderfull = 50, maxIters = 100;
        long slowOps = calcPgUpmapsSlow(nOsds, nUnderfull, maxIters);
        long fastOps = calcPgUpmapsFast(nOsds, nUnderfull, maxIters);
        double ratio = (double) slowOps / fastOps;

        System.out.println("  deviation_osd scan: N=" + nOsds + " OSDs, U=" + nUnderfull + " underfull, iters=" + maxIters);
        System.out.println("  Slow ops: " + slowOps);
        System.out.println("  Fast ops: " + fastOps);
        System.out.printf ("  Ratio:    %.1fx%n", ratio);
        System.out.println();

        check("speedup >= 10x at N=100/U=50/iters=100", ratio >= 10.0);

        // Large cluster
        nOsds = 1000; nUnderfull = 500;
        long slowOps2 = calcPgUpmapsSlow(nOsds, nUnderfull, maxIters);
        long fastOps2 = calcPgUpmapsFast(nOsds, nUnderfull, maxIters);
        double ratio2 = (double) slowOps2 / fastOps2;

        System.out.println("  Scale: N=" + nOsds + " OSDs, U=" + nUnderfull + " underfull, iters=" + maxIters);
        System.out.println("  Slow ops: " + slowOps2);
        System.out.println("  Fast ops: " + fastOps2);
        System.out.printf ("  Ratio:    %.1fx%n", ratio2);
        System.out.println();

        check("scale: speedup >= 50x at N=1000/U=500/iters=100", ratio2 >= 50.0);

        // --- CrushWrapper: try_remap_rule ---
        // U=100 underfull, |orig|=100 OSD mapping
        List<Integer> underfullCrush = new ArrayList<>();
        List<Integer> origCrush      = new ArrayList<>();
        for (int i = 0; i < 100; i++) underfullCrush.add(i);
        for (int i = 50; i < 150; i++) origCrush.add(i); // partial overlap

        long slowCrush = tryRemapRuleSlow(underfullCrush, origCrush);
        long fastCrush = tryRemapRuleFast(underfullCrush, origCrush);
        double ratioCrush = (double) slowCrush / fastCrush;

        System.out.println("  CrushWrapper try_remap_rule: U=100, |orig|=100");
        System.out.println("  Slow ops: " + slowCrush);
        System.out.println("  Fast ops: " + fastCrush);
        System.out.printf ("  Ratio:    %.1fx%n", ratioCrush);
        System.out.println();

        check("crush: speedup >= 10x at U=100/orig=100", ratioCrush >= 10.0);

        // Correctness for crush model
        // Both should classify the same items as "in orig"
        Set<Integer> origSet = new HashSet<>(origCrush);
        int slowInOrig = 0, fastInOrig = 0;
        for (int item : underfullCrush) {
            if (origCrush.contains(item)) slowInOrig++;
            if (origSet.contains(item))   fastInOrig++;
        }
        check("crush correctness: same count in orig", slowInOrig == fastInOrig);

        System.out.println(passed + "/" + total + " PASS");
    }
}