Linked Lists, Stacks, and Queues Questions

Requirements clarification:

• Fixed capacity N, multiple producers and consumers, low dequeue latency, clients must see backpressure when full, support batching for GPU/TPU workers and integrate with autoscaler.

High-level API:

• enqueue(request, timeout=None) -> bool/RejectReason
• try_enqueue(request) -> bool
• dequeue_batch(max_batch_size, max_wait_ms) -> List[Request]
• size(), capacity(), drain()

Queue internals:

• Circular buffer of pointers to requests with head/tail indices and current size counter (atomic).
• One mutex for short critical sections (enqueue/dequeue index updates) + two condition variables: not_full and not_empty.
• Producers: on enqueue, acquire mutex, if size==capacity:
  • If timeout==0 return REJECT (immediate backpressure)
  • Else wait on not_full with timeout -> return failure on timeout (explicit backpressure).
  • On success, insert at tail, increment size (atomic or protected), notify not_empty.
• Consumers (model workers): call dequeue_batch(max_batch_size, max_wait_ms).
  • If size==0 wait on not_empty up to max_wait_ms.
  • When woken, take up to min(size, max_batch_size) items in one mutex-protected splice (move pointers out of circular buffer to local vector), decrement size atomically, notify not_full once (or notify_all) after batch removal.
  • Doing batch removal under short lock ensures low dequeue latency and contiguous memory for batching.

Why this locking strategy:

• Condition variables + short mutexes give deterministic latency and simple correctness for bounded queue with backpressure. Lock-free variants (e.g., MPMC ring with CAS) can reduce latency but are complex when supporting batch atomic dequeue and timed waits; edge cases and memory reclamation make correctness harder. Our hybrid (atomic counters + short locks) yields low contention and simple batching semantics.

Performance optimizations:

• Use a fixed-size array to avoid allocations.
• Use cache-line padding for head/tail/size to avoid false sharing.
• Use notify_one for not_empty when a single consumer will likely handle batch; use notify_all on not_full when multiple producers blocked.
• Pre-allocate request wrappers; use pointer swapping to avoid copying payloads.

Backpressure semantics:

• Immediate reject on try_enqueue; blocking enqueue with timeout for clients that can wait; expose async enqueue via client SDK that returns 429 HTTP or connection backpressure when full.

Autoscaling integration:

• Expose queue metrics: current size, enqueue rate, dequeue rate, avg wait time, number of rejected enqueues.
• Scale-up triggers: sustained queue length > high-watermark (e.g., 70% capacity) or increasing enqueue/dequeue gap, or rising avg wait time.
• Scale-down: sustained low utilization and empty queue plus worker idle duration.
• When scaling up, new workers start and call dequeue_batch; when scaling down, drain signals: mark worker as draining, finish in-flight batches, then exit. To avoid thundering herd on scale-up, stagger worker startup and use exponential ramp.

Failure and correctness considerations:

• Ensure durable or replicated queue if requests cannot be lost (optional persistent write-ahead log).
• Handle spurious wakeups, respect timeouts, and ensure no lost notifications.
• Test with stress tests: N producers, M consumers, varying request sizes and network latencies.

This design balances low dequeue latency (batch atomic removal), clear backpressure semantics, and operational observability for autoscaling while keeping correctness simple with condition variables.

Requirements & constraints:

• Per-key, sliding windows of variable size (user can query arbitrary window lengths), high throughput (millions/sec), approximate quantiles with tunable error, exact min/max, support out-of-order events with watermarks, low-latency updates for ML feature pipelines.

High-level idea:

• Use two techniques combined:
  1. Exact min/max via per-key monotonic deque (O(1) amortized update, exact).
  2. Approximate quantiles via mergeable sketches (KLL preferred; t-digest as alternative for heavy-tail focus). To support arbitrary window sizes and high throughput, maintain sketches in a time-partitioned structure (fixed-size subwindows) so variable windows are answered by merging a small number of sketches.

Architecture & components:

• Ingest → Stream router (shard by key) → Per-shard operator:
  • Time-bucketed state: ring buffer of N subwindows (e.g., 1s buckets for 1h retention → 3600 buckets)
  • For each bucket and key:
    • monotonic deque for max/min (stores (value, timestamp))
    • sketch (KLL) for quantiles
  • Eviction thread rotates buckets; old buckets persisted/cleared.
• Query/Feature API: given key & window [t1,t2], compute exact min/max by scanning deques (or using bucket-level extrema), compute quantiles by merging sketches for all buckets overlapping window.

Monotonic deque details (exact max/min):

• Maintain decreasing deque: on arrival, pop from tail while tail.value <= new.value, append (value, ts). Evict head while head.ts < window_start.
• O(1) per element amortized, exact.

Approximate sketch details:

• Use KLL because:
  • Strong worst-case error guarantees (ε)
  • Small memory for given ε
  • Efficient merges (mergeable summaries → add sketches for adjacent buckets)
• Implement one KLL per key per bucket. To answer query, merge O(#buckets_over_window) KLLs (merge cost proportional to sketch size).
• If extreme tails are most important, use t-digest instead or use hybrid: KLL for central quantiles, t-digest for tails.

Windowing strategy to support variable sizes & high throughput:

• Fixed-size sub-buckets (power-of-two or configurable) so any variable window decomposes into ~O(log T) or O(window_size / bucket_size) merges. Use hierarchical buckets (like segment tree/windows-of-two) to reduce merges: pre-aggregate sketches in levels (1s,2s,4s...) enabling log(window) merges.
• For strict low-latency ingestion, only update current bucket; background compaction merges some buckets into higher-level buckets.

Parallelization & state management:

• Shard by key across worker nodes to distribute CPU and memory.
• Use a local RocksDB-backed state store for large cardinality; keep hottest keys in memory.
• Checkpoint operator state (deques + sketches) for fault tolerance.
• For cross-worker aggregation (when keys skew), use consistent hashing with hot-key replicates or a dedicated hot-key partitioning.

Memory / accuracy / CPU trade-offs:

• Sketch size k controls memory vs error: KLL k ~ O(1/ε) items. Doubling k halves error roughly; increases merge/cpu cost.
• Bucket granularity: smaller buckets → more buckets to merge for wide windows (higher query CPU) but smaller per-bucket memory and better time-precision for evictions; larger buckets → fewer merges, more memory per bucket, worse freshness bounds.
• Hierarchical aggregation reduces merge time at cost of additional storage for aggregated sketches.
• Monotonic deque memory equals number of candidate extrema; adversarial inputs can push it to O(bucket_size) but in practice small for many distributions.
• Latency vs accuracy: tighter error needs bigger sketches and more CPU for merges; for low-latency feature lookups, pre-merge most-likely windows (common sizes), or serve slightly stale precomputed features.

Accuracy and ML implications:

• Quantile error should be bounded and monitored; produce confidence intervals as features (e.g., quantile ± ε) or use deterministic features (median-of-means) if model sensitivity demands exactness.
• For downstream training/serving parity, record whether features are exact or approximate and set consistent sketch parameters between offline training and online serving.

Operational considerations:

• Metrics: sketch memory, merge latency, per-key state size, error estimates (calibration via offline replay).
• Backpressure: bound outstanding merges; if overloaded, sacrifice freshness (serve last precomputed window) or degrade sketch k dynamically.
• Hot-key handling: pre-aggregate hot-key sketches to avoid repeated merges; use separate fast path.

Example: monotonic deque (simplified)

from collections import deque

class MonotonicMax:
    def __init__(self):
        self.d = deque()  # stores (value, ts)

    def push(self, value, ts):
        while self.d and self.d[-1][0] <= value:
            self.d.pop()
        self.d.append((value, ts))

    def evict_older_than(self, tmin):
        while self.d and self.d[0][1] < tmin:
            self.d.popleft()

    def max(self):
        return self.d[0][0] if self.d else None

Query-time steps:

• Determine buckets overlapping [t1,t2]
• For max/min: evict with t1 and use bucket-level stored extrema or inspect deque head
• For quantiles: merge KLL sketches of those buckets and extract quantile

Alternatives & trade-offs:

• Use reservoir sampling for small-memory approximate quantiles (but no mergeability).
• Use t-digest if tails matter more; KLL if theoretical error bounds and mergeability prioritized.
• Use sliding-window exact quantiles impractical at scale — approximate required.

Summary:

• Combine monotonic deque for exact min/max with per-time-bucket mergeable KLL sketches for quantiles.
• Partition time to support variable windows by merging O(log T) or O(#buckets) sketches; shard by key for throughput.
• Tune sketch size, bucket granularity, and hierarchical aggregation to balance memory, accuracy, and latency for production ML pipelines.

Approach: Use a lock-free tail-append using atomic CAS on the list tail pointer and the predecessor's next field. We keep a single atomic tail to make appends mostly constant-time; concurrent threads attempt to link a new node at tail.next then swing tail to the new node using CAS.

# Pseudocode (conceptual) using CAS primitives
class Node:
    def __init__(self, value):
        self.value = value
        self.next = atomic_ptr(None)

tail = atomic_ptr(sentinel)   # sentinel is head initially

def append(value):
    new = Node(value)
    while True:
        last = tail.load()            # read current tail (atomic)
        next = last.next.load()
        if next is None:
            # try to link new after last
            if CAS(&last.next, None, new):
                # linked successfully; try to advance tail (best-effort)
                CAS(&tail, last, new)
                return
            # if CAS failed some other thread linked; retry
        else:
            # tail is lagging; help advance tail
            CAS(&tail, last, next)

Key concepts:

• Fast path: link at last.next with CAS; then best-effort advance tail.
• Helping: if another thread already appended, advance tail to keep progress.
• This is Michael & Scott non-blocking queue append pattern adapted to singly-linked list.

ABA problem:

• ABA occurs when a pointer A is read, changed to B by another thread, then back to A; a CAS that checks A appears to succeed though state mutated. Here it can affect CAS on tail or next if nodes are reclaimed and reallocated with same address. Use version tags (pointer + counter) or double-wide CAS (pointer, tag) to detect changes.

Memory-reclamation strategies:

• Hazard pointers: each thread publishes pointers it might dereference; nodes are only freed when no hazard pointer references them. Simple, low-latency, per-thread overhead.
• Epoch-based reclamation (EBR/RCU-like): advance global epoch; nodes retired in epoch N freed when all threads moved past N. Scales well for many threads, lower per-access cost but may delay reclamation.
• Choose hazard pointers if bounded memory is critical and reclamation must be prompt; choose EBR for throughput at scale.

Complexity & edge cases:

• Amortized O(1) per append; progress is lock-free (some thread always makes progress).
• Edge cases: node reuse causing ABA (use tags), CAS starvation (fairness not guaranteed), safe memory free must be ensured before reuse.
• Testing: stress tests under high concurrency, use ASAN/thread sanitizer, validate with model ingestion workloads.

To remove a cycle in a singly linked list in O(1) space, use Floyd’s Tortoise and Hare to detect the cycle, locate the cycle start, then find the node just before the cycle start and set its next to None. This preserves node identities and works whether the cycle starts at head or in the middle.

class ListNode:
    def __init__(self, val=0, next=None):
        self.val = val
        self.next = next

def remove_cycle(head):
    if not head or not head.next:
        return head

    # 1) Detect cycle using Floyd's algorithm
    slow = head
    fast = head
    has_cycle = False
    while fast and fast.next:
        slow = slow.next
        fast = fast.next.next
        if slow is fast:
            has_cycle = True
            break
    if not has_cycle:
        return head  # no cycle

    # 2) Find start of cycle
    # Move one pointer to head; advance both one step at a time;
    # they meet at cycle start.
    slow = head
    while slow is not fast:
        slow = slow.next
        fast = fast.next
    cycle_start = slow

    # 3) Find node before cycle_start (tail of cycle) and unlink it
    ptr = cycle_start
    while ptr.next is not cycle_start:
        ptr = ptr.next
    ptr.next = None

    return head

Key points:

• Detection: O(n) time average, O(1) space.
• Locating start: after meeting, reset one pointer to head; both move at 1x to meet at cycle entry.
• Removing: iterate from cycle_start until node whose next points to cycle_start and set next to None.
  Time complexity: O(n). Space complexity: O(1).

Edge cases:

• Empty list or single node without cycle.
• Cycle that begins at head (works because cycle_start == head; we find tail and break).
• Multiple nodes pointing into cycle (we only remove the cycle edge, preserving nodes).

A monotonic queue (deque) maintains elements in monotonically decreasing order so the front is always the current maximum for the window. For sliding-window maximums you:

• Push new element indices at the back, popping any smaller values (they can't be max while the new element is in window).
• Pop the front if it falls outside the window range.
• The front index corresponds to the window maximum.

Pseudocode:

def sliding_max(nums, k):
    from collections import deque
    dq = deque()   # store indices, nums[dq] decreasing
    result = []
    for i, x in enumerate(nums):
        while dq and nums[dq[-1]] < x:
            dq.pop()
        dq.append(i)
        if dq[0] <= i - k:
            dq.popleft()
        if i >= k - 1:
            result.append(nums[dq[0]])
    return result

Example:
Input [1,3,-1,-3,5,3,6,7], k=3 -> output [3,3,5,5,6,7].

Complexity: O(n) time (each index enters/exits deque once), O(k) space.

ML feature-engineering uses monotonic queues to compute streaming rolling-window features (max, min, range) efficiently for large time-series, enabling low-latency online feature extraction, memory-efficient pipelines, and real-time model scoring.