고정 크기 Min-Heap 기반 Top-K 선택

N=1,000,000  K=100:  실측 931회    공식 921회
N=5,000,000  K=100:  실측 1,077회  공식 1,081회

Ranking:
A가 B보다 좋다
=
A.latencyMicros > B.latencyMicros
또는 latency가 같고
A.requestId < B.requestId

Heap:
현재 Top-K만 저장
root
=
현재 Top-K 중 가장 나쁜 원소

새 candidate:
heap.size < K
-> 삽입
candidate > root
-> root 제거
-> candidate 삽입
candidate <= root
-> 폐기

복잡도:
Scan:
O(N log K)
Memory:
O(K)
최종 정렬:
O(K log K)

LatencySample
- requestId
- latencyMicros

가장 느린 요청 100개

requestId  latencyMicros
101        1,250
102           82
103        8,400
104          910
105        8,400
...

1순위:
latencyMicros가 클수록 우선
2순위:
latency가 같다면
requestId가 작을수록 우선

103 / 8400
105 / 8400
101 / 1250
104 / 910
102 / 82

103 / 8400
105 / 8400
101 / 1250

101 / 1250

106 / 500

500 < 1250
-> 즉시 폐기

107 / 2200

2200 > 1250
-> 101 제거
-> 107 삽입

#include <algorithm>
#include <concepts>
#include <cstddef>
#include <cstdint>
#include <queue>
#include <ranges>
#include <utility>
#include <vector>

class LatencySample final
{
public:
    constexpr LatencySample(
        std::uint64_t requestId,
        std::uint32_t latencyMicros) noexcept
        : requestId_(requestId),
          latencyMicros_(latencyMicros)
    {
    }

    [[nodiscard]]
    constexpr std::uint64_t getRequestId() const noexcept
    {
        return requestId_;
    }

    [[nodiscard]]
    constexpr std::uint32_t getLatencyMicros() const noexcept
    {
        return latencyMicros_;
    }

private:
    std::uint64_t requestId_;
    std::uint32_t latencyMicros_;
};

struct BetterLatencyRank final
{
    [[nodiscard]]
    constexpr bool operator()(
        const LatencySample& left,
        const LatencySample& right) const noexcept
    {
        if (left.getLatencyMicros()
            != right.getLatencyMicros())
        {
            return left.getLatencyMicros()
                > right.getLatencyMicros();
        }

        return left.getRequestId()
            < right.getRequestId();
    }
};

template<
    std::copy_constructible T,
    typename Better>
requires std::strict_weak_order<
    Better,
    const T&,
    const T&>
class BoundedTopK final
{
public:
    BoundedTopK(
        std::size_t limit,
        Better better)
        : limit_(limit),
          better_(std::move(better)),
          heap_(better_)
    {
    }

    void add(T candidate)
    {
        if (limit_ == 0)
        {
            return;
        }

        if (heap_.size() < limit_)
        {
            heap_.push(std::move(candidate));
            return;
        }

        replaceWorstIfBetter(
            std::move(candidate));
    }

    [[nodiscard]]
    std::vector<T> finish() &&
    {
        std::vector<T> result;
        result.reserve(heap_.size());

        while (!heap_.empty())
        {
            result.push_back(
                heap_.top());
            heap_.pop();
        }

        std::ranges::sort(
            result,
            better_);

        return result;
    }

private:
    std::size_t limit_;
    Better better_;
    std::priority_queue<
        T,
        std::vector<T>,
        Better> heap_;

    void replaceWorstIfBetter(
        T candidate)
    {
        if (!better_(
            candidate,
            heap_.top()))
        {
            return;
        }

        heap_.pop();
        heap_.push(
            std::move(candidate));
    }
};

BoundedTopK<
    LatencySample,
    BetterLatencyRank> topK{
        100,
        BetterLatencyRank{}};

for (const LatencySample& sample : samples)
{
    topK.add(sample);
}

std::vector<LatencySample> slowest =
    std::move(topK).finish();

from __future__ import annotations

from dataclasses import dataclass, field
from heapq import heapreplace, heappush


@dataclass(frozen=True, slots=True)
class LatencySample:
    request_id: int
    latency_micros: int


@dataclass(order=True, slots=True, frozen=True)
class _HeapItem:
    latency_micros: int
    inverted_request_id: int
    sample: LatencySample = field(
        compare=False
    )


class BoundedTopKLatency:
    def __init__(
        self,
        limit: int,
    ) -> None:
        if limit < 0:
            raise ValueError(
                "limit은 0 이상이어야 합니다."
            )

        self._limit = limit
        self._heap: list[_HeapItem] = []

    def add(
        self,
        sample: LatencySample,
    ) -> None:
        if self._limit == 0:
            return

        if len(self._heap) < self._limit:
            heappush(
                self._heap,
                _to_heap_item(sample),
            )
            return

        self._replace_if_better(
            sample
        )

    def finish(
        self,
    ) -> tuple[LatencySample, ...]:
        result = [
            item.sample
            for item in self._heap
        ]
        result.sort(
            key=lambda sample: (
                -sample.latency_micros,
                sample.request_id,
            )
        )
        return tuple(result)

    def _replace_if_better(
        self,
        sample: LatencySample,
    ) -> None:
        root = self._heap[0]
        root_request_id = -root.inverted_request_id

        if (
            sample.latency_micros < root.latency_micros
            or (
                sample.latency_micros
                == root.latency_micros
                and sample.request_id >= root_request_id
            )
        ):
            return

        heapreplace(
            self._heap,
            _to_heap_item(sample),
        )


def _to_heap_item(
    sample: LatencySample,
) -> _HeapItem:
    return _HeapItem(
        latency_micros=(
            sample.latency_micros
        ),
        inverted_request_id=(
            -sample.request_id
        ),
        sample=sample,
    )

top_k = BoundedTopKLatency(
    limit=100
)

for sample in samples:
    top_k.add(sample)

slowest = top_k.finish()

requestId가 작음
-> 더 좋음
requestId가 큼
-> 더 나쁨
-> heap root 쪽으로 가야 함

heap key:
(
    latency,
    -requestId
)

using System;
using System.Collections.Generic;

public readonly struct LatencySample
{
    private readonly ulong requestId;
    private readonly uint latencyMicros;

    public LatencySample(
        ulong requestId,
        uint latencyMicros)
    {
        this.requestId = requestId;
        this.latencyMicros = latencyMicros;
    }

    public ulong GetRequestId()
    {
        return this.requestId;
    }

    public uint GetLatencyMicros()
    {
        return this.latencyMicros;
    }
}

internal readonly struct RankKey
    : IComparable<RankKey>
{
    private readonly uint latencyMicros;
    private readonly ulong requestId;

    private RankKey(
        uint latencyMicros,
        ulong requestId)
    {
        this.latencyMicros = latencyMicros;
        this.requestId = requestId;
    }

    public static RankKey From(
        in LatencySample sample)
    {
        return new RankKey(
            sample.GetLatencyMicros(),
            sample.GetRequestId());
    }

    public int CompareTo(
        RankKey other)
    {
        int latencyOrder =
            this.latencyMicros.CompareTo(
                other.latencyMicros);

        if (latencyOrder != 0)
        {
            return latencyOrder;
        }

        return other.requestId.CompareTo(
            this.requestId);
    }
}

public sealed class BoundedTopKLatency
{
    private readonly int limit;
    private readonly PriorityQueue<
        LatencySample,
        RankKey> heap;

    public BoundedTopKLatency(
        int limit)
    {
        ArgumentOutOfRangeException.ThrowIfNegative(
            limit);

        this.limit = limit;
        this.heap =
            new PriorityQueue<
                LatencySample,
                RankKey>(
                Math.Max(1, limit));
    }

    public void Add(
        in LatencySample sample)
    {
        if (this.limit == 0)
        {
            return;
        }

        RankKey rank =
            RankKey.From(
                in sample);

        if (this.heap.Count < this.limit)
        {
            this.heap.Enqueue(
                sample,
                rank);
            return;
        }

        this.heap.EnqueueDequeue(
            sample,
            rank);
    }

    public LatencySample[] Finish()
    {
        LatencySample[] result =
            new LatencySample[
                this.heap.Count];

        for (
            int index = result.Length - 1;
            index >= 0;
            index -= 1)
        {
            result[index] =
                this.heap.Dequeue();
        }

        return result;
    }
}

BoundedTopKLatency topK =
    new(limit: 100);

foreach (LatencySample sample in samples)
{
    topK.Add(in sample);
}

LatencySample[] slowest =
    topK.Finish();

< 0
-> left가 더 나쁨
> 0
-> left가 더 좋음

export type LatencySample =
  Readonly<{
    requestId: bigint;
    latencyMicros: number;
  }>;

type Comparator<T> =
  (
    left: T,
    right: T,
  ) => number;

class MinHeap<T>
{
  readonly #values: T[] = [];
  readonly #compare: Comparator<T>;

  public constructor(
    compare: Comparator<T>,
  )
  {
    this.#compare = compare;
  }

  public getSize(): number
  {
    return this.#values.length;
  }

  public peek(): T | undefined
  {
    return this.#values[0];
  }

  public push(
    value: T,
  ): void
  {
    this.#values.push(value);
    this.#siftUp(
      this.#values.length - 1,
    );
  }

  public replaceRoot(
    value: T,
  ): T
  {
    const previous =
      this.#values[0];
    this.#values[0] = value;
    this.#siftDown(0);
    return previous;
  }

  public copyValues(): T[]
  {
    return [...this.#values];
  }

  #siftUp(
    startIndex: number,
  ): void
  {
    let index = startIndex;

    while (index > 0) {
      const parent =
        Math.floor(
          (index - 1) / 2,
        );

      if (!this.#isLess(
        index,
        parent,
      )) {
        return;
      }

      this.#swap(
        index,
        parent,
      );
      index = parent;
    }
  }

  #siftDown(
    startIndex: number,
  ): void
  {
    let index = startIndex;

    while (true) {
      const child =
        this.#findSmallerChild(
          index,
        );

      if (child === undefined
        || !this.#isLess(
          child,
          index,
        )) {
        return;
      }

      this.#swap(
        index,
        child,
      );
      index = child;
    }
  }

  #findSmallerChild(
    parent: number,
  ): number | undefined
  {
    const left =
      parent * 2 + 1;

    if (left >= this.#values.length) {
      return undefined;
    }

    const right = left + 1;

    if (right >= this.#values.length) {
      return left;
    }

    return this.#isLess(
      right,
      left,
    )
      ? right
      : left;
  }

  #isLess(
    left: number,
    right: number,
  ): boolean
  {
    return this.#compare(
      this.#values[left],
      this.#values[right],
    ) < 0;
  }

  #swap(
    left: number,
    right: number,
  ): void
  {
    [
      this.#values[left],
      this.#values[right],
    ] = [
      this.#values[right],
      this.#values[left],
    ];
  }
}

export class BoundedTopKLatency
{
  readonly #limit: number;
  readonly #heap =
    new MinHeap<LatencySample>(
      compareLatencyRank,
    );

  public constructor(
    limit: number,
  )
  {
    if (!Number.isSafeInteger(limit)
      || limit < 0) {
      throw new RangeError(
        "limit은 0 이상의 안전한 정수여야 합니다.",
      );
    }

    this.#limit = limit;
  }

  public add(
    sample: LatencySample,
  ): void
  {
    if (this.#limit === 0) {
      return;
    }

    if (this.#heap.getSize()
      < this.#limit) {
      this.#heap.push(sample);
      return;
    }

    this.#replaceIfBetter(
      sample,
    );
  }

  public finish():
    readonly LatencySample[]
  {
    const result =
      this.#heap.copyValues();

    result.sort(
      (left, right) =>
        -compareLatencyRank(
          left,
          right,
        ),
    );

    return result;
  }

  #replaceIfBetter(
    sample: LatencySample,
  ): void
  {
    const worst =
      this.#heap.peek();

    if (worst === undefined
      || compareLatencyRank(
        sample,
        worst,
      ) <= 0) {
      return;
    }

    this.#heap.replaceRoot(
      sample,
    );
  }
}

export function compareLatencyRank(
  left: LatencySample,
  right: LatencySample,
): number
{
  if (
    left.latencyMicros
    !== right.latencyMicros
  ) {
    return (
      left.latencyMicros
      - right.latencyMicros
    );
  }

  if (left.requestId
    < right.requestId) {
    return 1;
  }

  if (left.requestId
    > right.requestId) {
    return -1;
  }

  return 0;
}

const topK =
  new BoundedTopKLatency(
    100,
  );

for (const sample of samples) {
  topK.add(sample);
}

const slowest =
  topK.finish();

Network Event
-> Validation
-> Compact Ranking Record
-> Top-K Heap

export type LatencyEvent =
  Readonly<{
    requestId: bigint;
    durationNanos: bigint;
    requestBody?: Uint8Array;
    responseBody?: Uint8Array;
    headers:
      Readonly<Record<string, string>>;
  }>;

export type LatencySource =
  Readonly<{
    readChunkAsync: (
      signal?: AbortSignal,
    ) => Promise<
      readonly LatencyEvent[]
      | undefined
    >;
  }>;

export async function collectSlowestAsync(
  source: LatencySource,
  limit: number,
  signal?: AbortSignal,
): Promise<
  readonly LatencySample[]
>
{
  const topK =
    new BoundedTopKLatency(
      limit,
    );

  while (true) {
    const chunk =
      await source.readChunkAsync(
        signal,
      );

    if (chunk === undefined) {
      return topK.finish();
    }

    addChunk(
      topK,
      chunk,
    );
  }
}

function addChunk(
  topK: BoundedTopKLatency,
  events: readonly LatencyEvent[],
): void
{
  for (const event of events) {
    const sample =
      normalizeLatency(event);
    topK.add(sample);
  }
}

function normalizeLatency(
  event: LatencyEvent,
): LatencySample
{
  if (event.durationNanos < 0n) {
    throw new RangeError(
      "durationNanos는 음수일 수 없습니다.",
    );
  }

  const micros =
    event.durationNanos / 1_000n;

  if (micros > 0xffff_ffffn) {
    throw new RangeError(
      "latencyMicros 범위를 초과했습니다.",
    );
  }

  return {
    requestId:
      event.requestId,
    latencyMicros:
      Number(micros),
  };
}

LatencySource
= network/file I/O
= cancellation
= chunk lifetime

Boundary Validation
= ID 유효성
= 음수 duration 차단
= 단위 변환
= 정수 범위 검증

LatencySample
= Top-K에 필요한 최소 값만 소유

BoundedTopK
= ranking
= cutoff 유지
= O(K) 메모리

Caller
= I/O와 ranking orchestration

Reporting Layer
= 최종 K개에 대해서만
  상세 trace 재조회 가능

Top-K heap
-> requestBody 5 MiB
-> responseBody 20 MiB
-> header map
-> tracing graph

Heap:
requestId
latencyMicros

최종 Top-K 확정 후:
requestId
-> 필요하면 상세 trace 조회

1.
"더 좋은 값"의 ranking이 정확히 무엇인가
↓ 
2.
동점까지 포함해 total order가 있는가
↓ 
3.
Heap root가 현재 Top-K의
최악 원소인가
↓ 
4.
Heap 크기가 K를 절대 넘지 않는가
↓ 
5.
Candidate가 root보다 좋을 때만
교체하는가
↓ 
6.
Heap 내부 배열을
정렬된 결과라고 오해하지 않는가
↓ 
7.
최종 정렬은 K개에 대해서만 하는가
↓ 
8.
Heap이 무거운 원본 객체를
붙잡고 있지 않은가
↓ 
9.
Ranking에 사용되는 값이
삽입 후 변경되지 않는가
↓ 
10.
K와 N의 비율이
heap 선택을 실제로 정당화하는가

heap 내부 iteration
-> 최종 ranking 출력

N개 전체 정렬

K개만 최종 정렬

가장 큰 K개

min-heap

latency DESC
requestId ASC

primary score
secondary immutable ID

left.score > right.score

uint32 microseconds

finite
not NaN

heap.add(sample)
이후:
sample.latency = ...

immutable value

새 sample 계속 추가
-> 마지막에 Top-K

기존 item score 변경
기존 item 삭제
임의 item priority 갱신

indexed heap
balanced tree
ordered set
재계산

N = 1,000,000
K = 900,000

K << N
-> bounded heap 우선 검토
K ≈ N
-> sort / partition selection benchmark

running max

O(N)
memory O(1)

K = 100

K = 1000으로 바꾸자

source replay
더 큰 K로 처음부터 유지
raw archive 보존

request 42
request 42

Top-K
= ranking selection

Dedup
= identity uniqueness

O(K)

O(
  K × retainedObjectGraphSize
)

compact immutable summary

packed = (uint64(latencyMicros) << 32) | uint32(requestId)

Worker A ─┐
Worker B ─┼-> lock -> global heap
Worker C ─┘

각 partition에서 Local Top-K
-> Local Top-K들의 union에서
   다시 Top-K
-> Global Top-K

local heap
-> merge

x보다 좋은 원소가 최소 K개

TopK(
  전체 입력
)
=
TopK(
  각 partition의 TopK를 모두 합친 것
)

Partition A
Partition B

(score, uniqueId)

- Top-K largest인데 max-heap을 씀
- Heap root를 최고 원소라고 생각함
- Heap 배열 자체가 정렬됐다고 생각함
- latency 동점을 정의하지 않음
- container stability에 tie 처리를 맡김
- NaN이 들어가는 float comparator를 그대로 사용함
- 삽입 후 ranking field를 mutation함
- K=100인데 request body까지 heap에 보관함
- K가 N에 가까운데 무조건 heap을 사용함
- K=1인데 priority queue를 만듦
- 나중에 K를 늘릴 수 있다고 생각함
- Top-K heap이 duplicate도 제거한다고 생각함
- 여러 worker가 sample마다 global heap lock을 잡음

export function slowestRequestsBad(
  samples: LatencySample[],
  limit: number,
): LatencySample[]
{
  samples.sort(
    (left, right) =>
      right.latencyMicros
      - left.latencyMicros,
  );

  return samples.slice(
    0,
    limit,
  );
}

N = 50,000,000
K = 100

전체 N개를 정렬
O(N log N)

100개 후보 유지

samples.sort(...)

request 10 / 8400
request 20 / 8400

network
file
generator
database cursor

작은 N
-> sort가 더 단순

K ≈ N
-> sort/partition 검토

거대한 N
AND
작은 K
AND
streaming
-> bounded Top-K heap

full sort
O(N log N)

bounded heap
O(N log K) 상한
기대 O(N + K log K log(N/K))

nth_element
평균 O(N) / 평균 linear
일반적인 최악 복잡도는 표준이 규정하지 않음

모든 8천만 sample
-> 중앙 서버
-> Global Heap

Worker 1:
10,000,000
-> Top 100

Worker 2:
10,000,000
-> Top 100

...

Worker 8:
10,000,000
-> Top 100

Coordinator:
800개
-> 다시 Top 100

export function mergePartitionTopK(
  partitions:
    readonly (
      readonly LatencySample[]
    )[],
  limit: number,
): readonly LatencySample[]
{
  const global =
    new BoundedTopKLatency(
      limit,
    );

  for (const partition of partitions) {
    addPartition(
      global,
      partition,
    );
  }

  return global.finish();
}

function addPartition(
  global: BoundedTopKLatency,
  partition:
    readonly LatencySample[],
): void
{
  for (const sample of partition) {
    global.add(sample);
  }
}

export function selectPartitionTopK(
  samples:
    readonly LatencySample[],
  limit: number,
): readonly LatencySample[]
{
  const topK =
    new BoundedTopKLatency(
      limit,
    );

  for (const sample of samples) {
    topK.add(sample);
  }

  return topK.finish();
}

import assert from "node:assert/strict";
import test from "node:test";

test(
  "Bounded Top-K는 full sort 기준 모델과 같다",
  () =>
{
    const random =
      createDeterministicRandom(
        0x5eed,
      );

    for (
      let iteration = 0;
      iteration < 1_000;
      iteration += 1
    ) {
      const sampleCount =
        random() % 2_000;
      const limit =
        random() % 100;
      const samples =
        createSamples(
          sampleCount,
          random,
        );

      const actual =
        selectPartitionTopK(
          samples,
          limit,
        );
      const expected =
        referenceTopK(
          samples,
          limit,
        );

      assert.deepEqual(
        actual,
        expected,
      );
    }
  },
);

test(
  "Partition Top-K 병합은 전체 Top-K와 같다",
  () =>
{
    const random =
      createDeterministicRandom(
        1234,
      );
    const samples =
      createSamples(
        20_000,
        random,
      );
    const partitions =
      partition(
        samples,
        8,
      );

    const local =
      partitions.map(
        (values) =>
          selectPartitionTopK(
            values,
            100,
          ),
      );

    const merged =
      mergePartitionTopK(
        local,
        100,
      );

    const expected =
      referenceTopK(
        samples,
        100,
      );

    assert.deepEqual(
      merged,
      expected,
    );
  },
);

test(
  "Latency 동점은 requestId로 결정된다",
  () =>
{
    const samples:
      readonly LatencySample[] =
    [
      {
        requestId: 30n,
        latencyMicros: 5_000,
      },
      {
        requestId: 10n,
        latencyMicros: 5_000,
      },
      {
        requestId: 20n,
        latencyMicros: 5_000,
      },
    ];

    assert.deepEqual(
      selectPartitionTopK(
        samples,
        2,
      ),
      [
        {
          requestId: 10n,
          latencyMicros: 5_000,
        },
        {
          requestId: 20n,
          latencyMicros: 5_000,
        },
      ],
    );
  },
);

function referenceTopK(
  samples:
    readonly LatencySample[],
  limit: number,
): readonly LatencySample[]
{
  return [...samples]
    .sort(
      (left, right) =>
        -compareLatencyRank(
          left,
          right,
        ),
    )
    .slice(
      0,
      limit,
    );
}

function createSamples(
  count: number,
  random: () => number,
): readonly LatencySample[]
{
  return Array.from(
    { length: count },
    (_, index) => ({
      requestId:
        BigInt(index + 1),
      latencyMicros:
        random() % 100_000,
    }),
  );
}

function partition<T>(
  values: readonly T[],
  count: number,
): readonly (
  readonly T[]
)[]
{
  const result:
    T[][] =
      Array.from(
        { length: count },
        () => [],
      );

  for (
    let index = 0;
    index < values.length;
    index += 1
  ) {
    result[
      index % count
    ].push(
      values[index],
    );
  }

  return result;
}

function createDeterministicRandom(
  seed: number,
): () => number
{
  let state =
    seed >>> 0;

  return () =>
  {
    state = (
      Math.imul(
        state,
        1_664_525,
      )
      + 1_013_904_223
    ) >>> 0;

    return state;
  };
}

비교:
1. full sort
2. bounded min-heap
3. partition local Top-K + merge
4. 언어별 표준 Top-K helper

변수:
N
K
K/N
partition 수
tie 비율
element 크기
ranking 계산 비용
입력 source가 streaming인지 여부

관측:
전체 실행 시간
peak memory
allocation 수
retained bytes
comparison 수
merge 비용

requestId
latencyMicros

std::shared_ptr<FullRequest>

Element:
LatencySample

Priority:
RankKey

negative
-> 더 나쁨
positive
-> 더 좋음

bool better(left,right)

HeapItem의 자연 ordering

RankKey.CompareTo()

compare(left,right)
< 0  -> worse
> 0  -> better

Heap Root
=
현재 retained Top-K에서
가장 나쁜 원소

작은 summary object
vs
거대한 request object

Rank:
(latency DESC, requestId ASC)

Heap:
size <= K

Root:
현재 Top-K 중 worst

Candidate:
better(candidate, root)
-> replace

Final Output:
heap 내부 순서 사용 금지
-> K개만 정렬

Ownership:
compact immutable summary

Parallel:
local Top-K
-> global Top-K

Local K:
global K 이상

finish() 재호출:
C++ 파괴적, `&&` 한정
   lvalue 호출은 금지되지만 `std::move(...)`로 다시 호출할 수 있음
   첫 호출 뒤 두 번째 결과는 empty
C# 파괴적 (Dequeue로 비움)
Python·TypeScript 비파괴적 (복사)

상위 K개만 필요하다면
전체 N개를 정렬하지 마라.

현재 승자 K개만 유지하고,
그중 가장 약한 원소를 root에 둬라.

새 후보가 root보다 강할 때만 교체하라.

Top-K의 핵심은
"최고를 빨리 찾는 것"이 아니라
"탈락시킬 최저 승자를 항상 알고 있는 것"이다.