Heaps and Priority Queues Questions

Situation: You're seeing intermittent high-latency spikes from a production service that computes running medians with two heaps. Use this checklist to diagnose root causes and short-circuit mitigation.

1. Baseline & reproduce

• Capture timestamped traces/metrics (latency p50/p95/p99), logs, and request IDs around spikes.
• Enable CPU/heap profiling for the spike window (e.g., perf, async-profiler, py-spy, go pprof).

2. CPU saturation

• Check system and process CPU (top, sar, Cloud metrics). Correlate CPU% with spike times.
• Profile hot code paths to see if heap operations (push/pop, rebalancing) spike CPU.
• Mitigation: offload heavy preprocessing, increase concurrency, use optimized heap library or native code.

3. Garbage collection

• Collect GC logs (JVM -Xlog:gc*, Python tracemalloc/GC, Go GC stats). Look for full GC/stop-the-world pauses coincident with latency.
• Mitigation: tune GC settings, reduce allocation churn (reuse objects/buffers), increase heap size or use incremental GC.

4. Memory fragmentation and allocator behavior

• Inspect RSS vs. heap usage (pmap/memory profiler). Look for high RSS but low used heap.
• Check allocator (glibc malloc, jemalloc, tcmalloc) stats if available.
• Mitigation: switch allocator, enable memory pooling, reduce many small allocations.

5. Heap balancing inefficiencies

• Verify invariants: sizes differ by at most 1, max(left) <= min(right).
• Instrument counts and operation latencies for push/pop and rebalance operations.
• Look for pathological input patterns (sorted streams) causing frequent rebalances; mitigate with batch inserts or median-of-medians pivoting.

6. Lazy deletion accumulation

• If deletions are lazy (mark-and-clean), measure pending-delete queue size and cleanup cost.
• Ensure cleanup is amortized (background sweeper) and bounded per request.
• Mitigation: convert to eager clean on low-cost paths, throttle lazy queue, or use indexed heaps for O(log n) deletes.

7. Contention and synchronization

• Profile locks and GC pauses (jstack, go trace, pythread dump). Look for mutexes around heap ops causing stalls.
• Check thread pool saturation, request queueing, and blocking I/O.
• Mitigation: minimize critical sections, use lock-free or partitioned heaps (sharding by key), increase pool size.

8. End-to-end checks

• Correlate spikes with external resources (DB, network, model scoring). Ensure median code isn't blocked waiting for IO.
• Run load tests with synthetic patterns that mimic production to reproduce.

9. Short-term mitigations

• Rate-limit, shed noncritical traffic, serve cached medians during incident.
• Roll back recent changes if correlation exists.

10. Long-term fixes & monitoring

• Add metrics: per-operation latency, heap sizes, pending deletions, lock contention, GC pause durations.
• Automated alerts on p99 latency, GC pauses, and pending-delete growth.
• Postmortem: root cause, fix, tests, and runbook.

This checklist helps you triage quickly and decide whether to tune GC/allocator, fix algorithmic inefficiency, reduce contention, or apply operational mitigations.

Approach summary

• Two viable families: multiple heap pairs (exact or near-exact per-quantile) vs a single streaming sketch (t-digest / KLL) that reports many quantiles from one compact synopsis. Choice depends on required accuracy, throughput, memory, and read latency.

Design with heaps (per-quantile pairs)

• For each quantile q (0.1, 0.5, 0.9) maintain two heaps:
  • max-heap L_q for the smallest floor(q*N) items
  • min-heap R_q for the rest
• On each incoming x:
  • For each q, insert x to L_q or R_q based on comparison with top(s), then rebalance sizes so |L_q| = floor(q*N) (or within ±1 for streaming).
• Reads: the q-th percentile is top(L_q) — O(1).
• Update cost: O(k * log s) per element where k=number of quantiles (3) and s ≈ sizes of heaps (~qN). Memory: O(k * N) if you store all elements (or O(k * target-buffer-size) if you evict/compact). This is effectively exact (modulo how you manage N changes).

Trade-offs for heap approach

• Pros: constant-time reads, exact or controllable exactness per quantile, simple logic.
• Cons: linear memory blow-up if you retain full data per quantile; update cost scales with k. Not practical for high-throughput or very large N unless you cap stored window.

Single sketch (t-digest / KLL)

• Maintain one digest (t-digest for floating distributions; KLL has provable error bounds).
• On each x: update digest in O(log m) or amortized O(1) depending on implementation; digest compresses periodically keeping size m bounded (e.g., hundreds).
• Query multiple quantiles from the same sketch in O(m) or O(log m) each; memory O(m), independent of total N.
• Accuracy: configurable with compression parameter; typically excellent for tails with t-digest and provable error for KLL.

Trade-offs for sketch approach

• Pros: very low, bounded memory; cheap updates; single structure supplies many quantiles; suitable for high-throughput production.
• Cons: approximate (but controllable), read latency slightly higher than a simple heap top but still low; worst-case accuracy depends on distribution and compression tuning.

Recommendation

• If you need exact or near-exact percentiles and only a few quantiles with modest throughput/retention, use per-quantile heap pairs (fast reads).
• For production ML pipelines with high-throughput streams, many quantiles, bounded memory, and acceptable approximation, use a single well-tuned sketch (KLL for provable bounds, t-digest if tail accuracy matters). Sketches generally give best memory/latency/throughput trade-off.

Requirements:

• Functional: maintain running median per feature stream (thousands of independent streams).
• Non-functional: 5 ms median read SLA per stream, total ingest 100k inserts/sec, high availability, crash recovery.

High-level architecture:

• Ingest API (HTTP/gRPC) → Ingest router (Kafka or Kinesis topics keyed by stream_id) → Stream processors (stateful workers) → Read API (low-latency cache / query tier) → Persistent storage (WAL + periodic snapshots) → Orchestrator (service registry + autoscaling).

Sharding strategy:

• Hash-based partitioning on stream_id into N partitions. Each partition assigned to one stateful worker (process/container) to guarantee single-writer semantics per stream and avoid cross-node coordination. Choose N >> number of workers to allow elastic rebalancing (consistent hashing + sticky assignments).

State management per partition:

• Maintain two heaps: max-heap for lower half, min-heap for upper half, plus counts. Keep per-stream metadata in an in-memory map keyed by stream_id. For thousands of streams per worker this fits memory.

Persistence & recovery:

• Append-only WAL: on every insert, append compact record (stream_id, value, timestamp) to partitioned log (local SSD with async replicate to durable object store like S3 or cloud block store). Also emit to Kafka which retains history.
• Snapshotting: periodic in-memory snapshot of heaps per stream (e.g., every 1k ops or 5s) flushed to durable store. Snapshots are compact: serialize heap arrays and counts.
• Recovery: on startup rehydrate latest snapshot, then replay WAL entries after snapshot offset. Use sequence numbers to avoid duplication.
• Fast reads during recovery: route reads for a recovering partition to replicas; keep a read-replica cache (Redis) updated asynchronously for sub-5ms reads.

Availability & latency:

• Keep hot-path reads served from in-memory state inside worker; expose pinned replica via gRPC with connection pooling. For <5ms median read, co-locate read API with workers and use local caches; use thread-safe lock-free heaps or per-stream locks to avoid contention.

Scaling and trade-offs:

• For higher ingest, increase partitions and workers; snapshot frequency vs durability/IO trade-off; WAL provides durability at cost of write IO—batch WAL writes to amortize.
• Alternative: use CRDT-based distributed quantile sketches (t-digest) if exact median not required — saves memory and speeds recovery.

Monitoring & ops:

• Track per-stream latency, heap sizes, WAL lag, snapshot intervals. Add automatic partition rebalancer and controlled draining for code deploys.

To maintain a running median in O(log n) insert and O(1) retrieval, use two heaps: a max-heap for the lower half and a min-heap for the upper half. In Python use heapq (min-heap); store negatives for max-heap.

import heapq

class RunningMedian:
    def __init__(self):
        # max_heap stores the lower half as negatives to simulate a max-heap
        self.max_heap = []  # max-heap (as negatives)
        # min_heap stores the upper half as usual min-heap
        self.min_heap = []  # min-heap

    def insert(self, num: float):
        """
        Insert num in O(log n).
        - If num <= max of lower half, push to max_heap; else to min_heap.
        - Rebalance so sizes differ by at most 1.
        Duplicates are handled naturally by heap insertion.
        """
        if not self.max_heap or num <= -self.max_heap[0]:
            heapq.heappush(self.max_heap, -num)
        else:
            heapq.heappush(self.min_heap, num)

        # Rebalance heaps to maintain size invariant
        if len(self.max_heap) > len(self.min_heap) + 1:
            heapq.heappush(self.min_heap, -heapq.heappop(self.max_heap))
        elif len(self.min_heap) > len(self.max_heap) + 1:
            heapq.heappush(self.max_heap, -heapq.heappop(self.min_heap))

    def get_median(self) -> float:
        """
        Return median in O(1).
        - If counts equal: average of two heap tops.
        - If unequal: top of the larger heap.
        """
        if not self.max_heap and not self.min_heap:
            raise IndexError("no elements")
        if len(self.max_heap) == len(self.min_heap):
            return (-self.max_heap[0] + self.min_heap[0]) / 2.0
        elif len(self.max_heap) > len(self.min_heap):
            return -self.max_heap[0]
        else:
            return self.min_heap[0]

Key points:

• Time: insert O(log n), get_median O(1). Space: O(n).
• Handles even counts by averaging two middle values; odd counts by returning middle.
• Duplicates work correctly because heaps allow repeated values.

Approach:

• Generate random integer streams of varying lengths (e.g., [10^2, 10^3, 10^4, 10^5]).
• For each stream, maintain two implementations:
  1. two-heap running median (max-heap for lower, min-heap for upper)
  2. ground-truth: sorted container (bisect insertion into list)
• At each insertion compare medians; log mismatches and timings.

Example implementation (core loop):

import heapq, bisect, random, time

def running_median_two_heaps(stream):
    lo, hi = [], []  # lo: max-heap (store negatives), hi: min-heap
    medians = []
    for x in stream:
        if not lo or x <= -lo[0]:
            heapq.heappush(lo, -x)
        else:
            heapq.heappush(hi, x)
        # rebalance
        if len(lo) > len(hi) + 1:
            heapq.heappush(hi, -heapq.heappop(lo))
        elif len(hi) > len(lo):
            heapq.heappush(lo, -heapq.heappop(hi))
        # median
        if len(lo) == len(hi):
            med = (-lo[0] + hi[0]) / 2
        else:
            med = -lo[0]
        medians.append(med)
    return medians

def running_median_sorted(stream):
    s, medians = [], []
    for x in stream:
        bisect.insort(s, x)
        n = len(s)
        med = (s[n//2] if n%2==1 else (s[n//2-1]+s[n//2])/2)
        medians.append(med)
    return medians

Correctness measurement:

• For each stream and RNG seed, assert elementwise equality (or float tolerance).
• Run many seeds (fuzz-testing) and random distributions (uniform, normal, skewed, duplicates).
• Record and report number of failures and example counterexamples.

Performance measurement:

• Time-to-compute using time.perf_counter for each method per stream length; average over runs.
• Memory: track peak process memory (psutil) or approximate sizes (len(list)+heap sizes).
• Plot median runtime vs stream length; compute throughput (elements/sec).

Reporting:

• Table/plots: correctness pass rate, avg latency per insertion, total runtime, memory.
• Discuss trade-offs: two-heaps O(log n) per insert vs sorted-list O(n) per insert; expect two-heaps to scale linearly overall while sorted becomes quadratic.

Edge cases & best practices:

• Test empty stream, all-equal elements, monotonic streams.
• Use deterministic seeds and CI job to catch regressions.