java-topology/defects/wasmtime/unit/WorkQueueLinearScanTest.java

package unit;

import java.util.*;

/**
 * wasmtime-0001: WorkQueue high_priority Vec<WorkItem> O(n) scan in async scheduler.
 *
 * Models the WorkQueue.promote_thread_work_item() hot path:
 *   slow: high_priority stored as flat Vec, promote scans all items O(n)
 *   fast: items keyed by thread ID in HashMap<ThreadId, Deque>, O(1) lookup
 *
 * Benchmark: T threads, each with K work items => N = T*K total items.
 * slow: each promote_thread_work_item() scans all N items => O(N) per call
 * fast: each promote_thread_work_item() looks up HashMap[tid] => O(K) per call
 * With T promote calls (one per thread): slow=O(T*N)=O(T^2*K), fast=O(T*K)
 * Speedup = T (e.g. T=50 => 50x).
 */
public class WorkQueueLinearScanTest {

    static final int THREAD_ITEMS_K = 1; // items per thread in queue

    // Synthetic work item: tagged with thread id
    static class WorkItem {
        enum Kind { RESUME_THREAD, GUEST_CALL, WORKER_FUNCTION }
        final Kind kind;
        final int threadId;
        WorkItem(Kind kind, int threadId) { this.kind = kind; this.threadId = threadId; }
    }

    /** SLOW: flat list — promote_thread scans all items */
    static class SlowWorkQueue {
        final List<WorkItem> high_priority = new ArrayList<>();

        void pushHighPriority(WorkItem item) {
            high_priority.add(item);
        }

        /** Returns op count (items inspected) to find item for targetThread */
        long promoteThread(int targetThread) {
            long ops = 0;
            for (WorkItem item : high_priority) {
                ops++;
                if ((item.kind == WorkItem.Kind.RESUME_THREAD ||
                     item.kind == WorkItem.Kind.GUEST_CALL) &&
                    item.threadId == targetThread) {
                    break;
                }
            }
            return ops;
        }
    }

    /** FAST: HashMap<threadId, Deque<WorkItem>> — O(1) promote_thread */
    static class FastWorkQueue {
        final Map<Integer, Deque<WorkItem>> byThread = new HashMap<>();
        final List<WorkItem> general = new ArrayList<>();

        void pushHighPriority(WorkItem item) {
            if (item.kind == WorkItem.Kind.RESUME_THREAD ||
                item.kind == WorkItem.Kind.GUEST_CALL) {
                byThread.computeIfAbsent(item.threadId, k -> new ArrayDeque<>()).add(item);
            } else {
                general.add(item);
            }
        }

        /** Returns op count (items inspected) to find item for targetThread */
        long promoteThread(int targetThread) {
            // O(1) map lookup + iterate over items for this thread only
            Deque<WorkItem> items = byThread.get(targetThread);
            if (items == null) return 1; // 1 op for map lookup miss
            long ops = 1; // map lookup
            for (WorkItem item : items) {
                ops++;
                break; // found first matching item
            }
            return ops;
        }
    }

    static long bench(boolean slow, int T, int K) {
        // Build: T threads, each with K work items
        SlowWorkQueue slowQ = slow ? new SlowWorkQueue() : null;
        FastWorkQueue fastQ = slow ? null : new FastWorkQueue();

        for (int tid = 0; tid < T; tid++) {
            for (int k = 0; k < K; k++) {
                WorkItem item = new WorkItem(WorkItem.Kind.RESUME_THREAD, tid);
                if (slow) slowQ.pushHighPriority(item);
                else      fastQ.pushHighPriority(item);
            }
        }

        // T promote calls, always targeting the LAST thread (worst case: its items
        // are at the end of the flat list after all other threads' items).
        // This gives slow=O(T*K) per call, fast=O(K) per call => T-fold speedup.
        int lastTid = T - 1;
        long totalOps = 0;
        for (int call = 0; call < T; call++) {
            if (slow) totalOps += slowQ.promoteThread(lastTid);
            else      totalOps += fastQ.promoteThread(lastTid);
        }
        return totalOps;
    }

    static void test(String name, int T, int K, int minSpeedup) {
        long sOps = bench(true,  T, K);
        long fOps = bench(false, T, K);

        double speedup = (double) sOps / fOps;
        boolean pass = sOps >= fOps * minSpeedup;

        System.out.printf("%-50s  slow=%,d  fast=%,d  speedup=%.1fx  %s%n",
            name, sOps, fOps, speedup, pass ? "PASS" : "FAIL");

        assert pass : String.format(
            "%s: expected speedup >=%dx, got %.1fx (slow=%d, fast=%d)",
            name, minSpeedup, speedup, sOps, fOps);
    }

    public static void main(String[] args) {
        System.out.println("wasmtime-0001: WorkQueue linear scan");
        System.out.println("=====================================");

        // T=10 threads, 1 item each: slow scans all 10 per promote => 10x over fast O(1)
        test("T=10  K=1   minSpeedup=5x",   10,  1,  5);
        // T=50 threads: slow scans 50 items per promote; fast O(1)
        test("T=50  K=1   minSpeedup=25x",  50,  1, 25);
        // T=100 threads: slow scans 100 items per promote
        test("T=100 K=1   minSpeedup=50x", 100,  1, 50);
        // T=20 threads, K=5 items each: slow scans 100 items per promote
        test("T=20  K=5   minSpeedup=10x",  20,  5, 10);
        // T=50 threads, K=2 items each: slow scans 100 items per promote
        test("T=50  K=2   minSpeedup=25x",  50,  2, 25);

        System.out.println("=====================================");
        System.out.println("ALL PASS");
    }
}