[Solution] OpenAI高频题 GPU Credits — 3 Parts 详解高频题
冷目灼灼的扁豆
最近在准备 OpenAI 的面试，整理了一道高频题 GPU Credits。题目是根据面经回忆重构的（非原题，练手用）。核心是一个带过期时间的额度账本，三个 part 层层加码，分享一下思路和代码。
## 题意
三个 API：
- `add_credit(credit_id, amount, timestamp, expiration)`：发放一笔额度，在闭区间 `[timestamp, timestamp+expiration]` 内有效。
- `subtract(amount, timestamp)`：从当前有效的额度里扣，按最早过期优先，可以跨多笔扣；永远不报错——扣超了变成 debt，后续 timestamp 更大的 grant 先还 debt。
- `get_balance(timestamp)`：把所有 `timestamp <= 查询时间` 的调用按 timestamp 顺序（不是调用顺序）重放一遍，`v = 剩余有效额度 − 未还 debt`；`v < 0` 返回 None，否则返回 v（v=0 时返回整数 0）。
约束：timestamp 互不相同；add + subtract ≤ 2e4，query ≤ 2e4。
## 核心洞察
这题本质是一个 event-sourced ledger：balance 不是存出来的，是按 timestamp 重放算出来的。所有 tricky 的地方都来自三件事：
1. grant 是闭区间有效（`t == expiration` 那一刻还能用）；
2. debt 不是"最后统一减一下"——grant 到达时先还 debt，这个顺序会影响过期 corner case；
3. query 看的是 timestamp 顺序，不是到达顺序。
## Part 1：到达有序
事件按 timestamp 到达。维护一个按 expiration 的 min-heap + 一个 debt 计数器，一遍过：add 先还 debt，有剩再入堆；subtract 先踢掉 `exp < t` 的过期 grant，再按最早过期扣，扣不完变 debt；query 踢掉过期，`sum(heap) − debt`。每个操作 O(log n)。
## Part 2：到达无序
调用顺序和 timestamp 顺序不一致。标准做法：append-only event log，每次 query 把 `ts <= T` 的事件过滤出来按 timestamp 排序重放。每次 query O(U log U)。简单但 query 多了贵——这就是 Part 3 的动机。
## Part 3：query 单调 + 事件只增
加两个前提：query 的 timestamp 单调不减；新事件的 timestamp 永远大于已回答的最大 query 时间（late event 直接拒绝）。
做法：unsorted pending buffer + cursor。每次 query 只把 `ts <= T` 的事件挑出来排序、重放一次、推进 cursor。每个事件只被排序和重放一次，总 O(U log U + Q)。
这也是常问的 follow-up（"GPU Credit II 怎么优化避免每次从 T0 replay"）的答案：checkpoint——账本状态推进到已回答的最大时间，之后只算增量。
## Follow-up 讨论
- undo 一次 subtract：得记住这次扣动了哪些 grant（各扣了多少）以及 debt 变化了多少，再逆操作恢复。注意 grant 可能已过期或被后续扣过，faithful 的 undo 需要版本化。
- timestamp 撞车：题设保证唯一；真撞了，约定同一 timestamp 先 add 后 subtract。
- late add_credit：记一个 max_answered，`ts <= max_answered` 的事件会改写已回答的答案，直接拒绝（或退化到 Part 2 重放）。
## 易错点
1. 闭区间：expire 条件是 `exp < t`，不是 `<=`。`t == expiration` 时 grant 依然有效。
2. debt 语义：grant 到达时先还 debt，不是查询时统一减。超扣之后来的 grant 先填坑，填完剩下的才算余额；剩下的过期了就是 0（不是 None）。
3. `get_balance` 只有 `v < 0` 才返回 None；`v == 0` 返回整数 0。
4. Part 2/3 一定要按 timestamp 重放，不能按到达顺序。
## 代码
```python
import heapq, itertools
class _Ledger:
"""核心重放引擎: 事件必须按 timestamp 顺序喂入"""
def __init__(self):
self._heap = [] # (expiration, seq, remaining) min-heap
self._seq = itertools.count()
self._debt = 0
def _expire(self, t):
# 闭区间: t == expiration 时 grant 依然有效, 只有 exp < t 才过期
while self._heap and self._heap[0][0] < t:
heapq.heappop(self._heap)
def add_credit(self, credit_id, amount, timestamp, expiration):
if self._debt: # 后到的 grant 先还 debt
repay = min(self._debt, amount)
self._debt -= repay
amount -= repay
if amount:
heapq.heappush(self._heap, (timestamp + expiration, next(self._seq), amount))
def subtract(self, amount, timestamp):
self._expire(timestamp)
while amount and self._heap: # 最早过期优先, 可跨 grant
exp, seq, rem = heapq.heappop(self._heap)
take = min(rem, amount)
rem, amount = rem - take, amount - take
if rem:
heapq.heappush(self._heap, (exp, seq, rem))
if amount:
self._debt += amount # 永不报错, 超扣变 debt
def get_balance(self, timestamp):
self._expire(timestamp)
v = sum(rem for _, _, rem in self._heap) - self._debt
return None if v < 0 else v # v==0 返回 0, 不是 None
class LedgerP1(_Ledger):
"""Part 1: 到达有序, 直接维护, 每次操作 O(log n)"""
class LedgerP2:
"""Part 2: 到达无序. append-only log, 每次 query 过滤+排序+重放"""
def __init__(self):
self._log = []
def add_credit(self, credit_id, amount, timestamp, expiration):
self._log.append((timestamp, 0, "add", (credit_id, amount, expiration)))
def subtract(self, amount, timestamp):
self._log.append((timestamp, 1, "sub", amount))
def get_balance(self, timestamp):
ledger = _Ledger()
evs = sorted((e for e in self._log if e[0] <= timestamp),
key=lambda e: (e[0], e[1]))
for ts, _, kind, p in evs:
if kind == "add":
ledger.add_credit(p[0], p[1], ts, p[2])
else:
ledger.subtract(p, ts)
return ledger.get_balance(timestamp)
class LedgerP3:
"""Part 3: query 单调不减 + 事件只增(晚于已回答的最大时间).
pending buffer + cursor: 每个事件只排序/重放一次, 总 O(U log U + Q)"""
def __init__(self):
self._pending = []
self._max_answered = float("-inf")
self._ledger = _Ledger()
def _push(self, ts, k, kind, p):
if ts <= self._max_answered:
raise ValueError("late event")
self._pending.append((ts, k, kind, p))
def add_credit(self, credit_id, amount, timestamp, expiration):
self._push(timestamp, 0, "add", (credit_id, amount, expiration))
def subtract(self, amount, timestamp):
self._push(timestamp, 1, "sub", amount)
def get_balance(self, timestamp):
assert timestamp >= self._max_answered
due = sorted([e for e in self._pending if e[0] <= timestamp],
key=lambda e: (e[0], e[1]))
self._pending = [e for e in self._pending if e[0] > timestamp]
for ts, _, kind, p in due:
if kind == "add":
self._ledger.add_credit(p[0], p[1], ts, p[2])
else:
self._ledger.subtract(p, ts)
self._max_answered = timestamp
return self._ledger.get_balance(timestamp)
```
三个 part 跑了 300 组随机对拍（和暴力重放逐个 query 对比），都一致。有问题欢迎讨论。
## 题意
三个 API：
- `add_credit(credit_id, amount, timestamp, expiration)`：发放一笔额度，在闭区间 `[timestamp, timestamp+expiration]` 内有效。
- `subtract(amount, timestamp)`：从当前有效的额度里扣，按最早过期优先，可以跨多笔扣；永远不报错——扣超了变成 debt，后续 timestamp 更大的 grant 先还 debt。
- `get_balance(timestamp)`：把所有 `timestamp <= 查询时间` 的调用按 timestamp 顺序（不是调用顺序）重放一遍，`v = 剩余有效额度 − 未还 debt`；`v < 0` 返回 None，否则返回 v（v=0 时返回整数 0）。
约束：timestamp 互不相同；add + subtract ≤ 2e4，query ≤ 2e4。
## 核心洞察
这题本质是一个 event-sourced ledger：balance 不是存出来的，是按 timestamp 重放算出来的。所有 tricky 的地方都来自三件事：
1. grant 是闭区间有效（`t == expiration` 那一刻还能用）；
2. debt 不是"最后统一减一下"——grant 到达时先还 debt，这个顺序会影响过期 corner case；
3. query 看的是 timestamp 顺序，不是到达顺序。
## Part 1：到达有序
事件按 timestamp 到达。维护一个按 expiration 的 min-heap + 一个 debt 计数器，一遍过：add 先还 debt，有剩再入堆；subtract 先踢掉 `exp < t` 的过期 grant，再按最早过期扣，扣不完变 debt；query 踢掉过期，`sum(heap) − debt`。每个操作 O(log n)。
## Part 2：到达无序
调用顺序和 timestamp 顺序不一致。标准做法：append-only event log，每次 query 把 `ts <= T` 的事件过滤出来按 timestamp 排序重放。每次 query O(U log U)。简单但 query 多了贵——这就是 Part 3 的动机。
## Part 3：query 单调 + 事件只增
加两个前提：query 的 timestamp 单调不减；新事件的 timestamp 永远大于已回答的最大 query 时间（late event 直接拒绝）。
做法：unsorted pending buffer + cursor。每次 query 只把 `ts <= T` 的事件挑出来排序、重放一次、推进 cursor。每个事件只被排序和重放一次，总 O(U log U + Q)。
这也是常问的 follow-up（"GPU Credit II 怎么优化避免每次从 T0 replay"）的答案：checkpoint——账本状态推进到已回答的最大时间，之后只算增量。
## Follow-up 讨论
- undo 一次 subtract：得记住这次扣动了哪些 grant（各扣了多少）以及 debt 变化了多少，再逆操作恢复。注意 grant 可能已过期或被后续扣过，faithful 的 undo 需要版本化。
- timestamp 撞车：题设保证唯一；真撞了，约定同一 timestamp 先 add 后 subtract。
- late add_credit：记一个 max_answered，`ts <= max_answered` 的事件会改写已回答的答案，直接拒绝（或退化到 Part 2 重放）。
## 易错点
1. 闭区间：expire 条件是 `exp < t`，不是 `<=`。`t == expiration` 时 grant 依然有效。
2. debt 语义：grant 到达时先还 debt，不是查询时统一减。超扣之后来的 grant 先填坑，填完剩下的才算余额；剩下的过期了就是 0（不是 None）。
3. `get_balance` 只有 `v < 0` 才返回 None；`v == 0` 返回整数 0。
4. Part 2/3 一定要按 timestamp 重放，不能按到达顺序。
## 代码
```python
import heapq, itertools
class _Ledger:
"""核心重放引擎: 事件必须按 timestamp 顺序喂入"""
def __init__(self):
self._heap = [] # (expiration, seq, remaining) min-heap
self._seq = itertools.count()
self._debt = 0
def _expire(self, t):
# 闭区间: t == expiration 时 grant 依然有效, 只有 exp < t 才过期
while self._heap and self._heap[0][0] < t:
heapq.heappop(self._heap)
def add_credit(self, credit_id, amount, timestamp, expiration):
if self._debt: # 后到的 grant 先还 debt
repay = min(self._debt, amount)
self._debt -= repay
amount -= repay
if amount:
heapq.heappush(self._heap, (timestamp + expiration, next(self._seq), amount))
def subtract(self, amount, timestamp):
self._expire(timestamp)
while amount and self._heap: # 最早过期优先, 可跨 grant
exp, seq, rem = heapq.heappop(self._heap)
take = min(rem, amount)
rem, amount = rem - take, amount - take
if rem:
heapq.heappush(self._heap, (exp, seq, rem))
if amount:
self._debt += amount # 永不报错, 超扣变 debt
def get_balance(self, timestamp):
self._expire(timestamp)
v = sum(rem for _, _, rem in self._heap) - self._debt
return None if v < 0 else v # v==0 返回 0, 不是 None
class LedgerP1(_Ledger):
"""Part 1: 到达有序, 直接维护, 每次操作 O(log n)"""
class LedgerP2:
"""Part 2: 到达无序. append-only log, 每次 query 过滤+排序+重放"""
def __init__(self):
self._log = []
def add_credit(self, credit_id, amount, timestamp, expiration):
self._log.append((timestamp, 0, "add", (credit_id, amount, expiration)))
def subtract(self, amount, timestamp):
self._log.append((timestamp, 1, "sub", amount))
def get_balance(self, timestamp):
ledger = _Ledger()
evs = sorted((e for e in self._log if e[0] <= timestamp),
key=lambda e: (e[0], e[1]))
for ts, _, kind, p in evs:
if kind == "add":
ledger.add_credit(p[0], p[1], ts, p[2])
else:
ledger.subtract(p, ts)
return ledger.get_balance(timestamp)
class LedgerP3:
"""Part 3: query 单调不减 + 事件只增(晚于已回答的最大时间).
pending buffer + cursor: 每个事件只排序/重放一次, 总 O(U log U + Q)"""
def __init__(self):
self._pending = []
self._max_answered = float("-inf")
self._ledger = _Ledger()
def _push(self, ts, k, kind, p):
if ts <= self._max_answered:
raise ValueError("late event")
self._pending.append((ts, k, kind, p))
def add_credit(self, credit_id, amount, timestamp, expiration):
self._push(timestamp, 0, "add", (credit_id, amount, expiration))
def subtract(self, amount, timestamp):
self._push(timestamp, 1, "sub", amount)
def get_balance(self, timestamp):
assert timestamp >= self._max_answered
due = sorted([e for e in self._pending if e[0] <= timestamp],
key=lambda e: (e[0], e[1]))
self._pending = [e for e in self._pending if e[0] > timestamp]
for ts, _, kind, p in due:
if kind == "add":
self._ledger.add_credit(p[0], p[1], ts, p[2])
else:
self._ledger.subtract(p, ts)
self._max_answered = timestamp
return self._ledger.get_balance(timestamp)
```
三个 part 跑了 300 组随机对拍（和暴力重放逐个 query 对比），都一致。有问题欢迎讨论。
已获得 24 大米
+2
共0条回复
✨ 您正在体验新版论坛UI
