Netflix AtomicCounter：并发增减与等待归零

题意与假设

这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。

• 训练版计数器不得为负，decrement(0) 抛 ValueError；wait_zero 等当前值变为零，若调用时已为零立即 True。
• timeout=None 无限等待，有限 timeout 用 monotonic deadline；超时返回 False。题目未明确返回类型。
• 计数器可重用：从零再次 increment 后，新的 wait_zero 会等待下一次归零。调用方若要‘所有已提交任务完成’的生命周期保证，必须另定任务提交关闭边界。

解题路线

1. 一个 Condition 保护 value。increment、decrement、get 都在同一锁内进行；decrement 从 1 到 0 时 notify_all 唤醒所有等零线程。
2. wait_zero 在 `while value!=0` 中等待，醒来后重查谓词。任何线程都可能在通知后先拿锁并 increment，waiter 不可只因为收到通知就直接返回 True。
3. 有限 timeout 用调用开始时的 deadline 计算剩余时间，避免虚假唤醒或多次通知让总等待无界。若要 wait for a specific generation，应再加 generation/epoch，不能把瞬时零视为稳定结束。

手动走一遍

训练例初值 2，两名 worker 各完成后 decrement：第一次变 1 不通知归零，第二次变 0 通知所有 waiter。若另一提交线程在 waiter 抢到锁前 increment 至 1，waiter 会继续等待，这符合‘返回时观察到零’的契约。

Python 参考实现

下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。

代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。

from threading import Condition
from time import monotonic


class AtomicCounter:
    def __init__(self, initial=0):
        if initial < 0:
            raise ValueError("initial count must be nonnegative")
        self._value = initial
        self._condition = Condition()

    def increment(self):
        with self._condition:
            self._value += 1
            return self._value

    def decrement(self):
        with self._condition:
            if self._value == 0:
                raise ValueError("counter cannot become negative")
            self._value -= 1
            if self._value == 0:
                self._condition.notify_all()
            return self._value

    def get(self):
        with self._condition:
            return self._value

    def wait_zero(self, timeout=None):
        if timeout is not None and timeout < 0:
            raise ValueError("negative timeout")
        deadline = None if timeout is None else monotonic() + timeout
        with self._condition:
            while self._value != 0:
                remaining = None if deadline is None else deadline - monotonic()
                if remaining is not None and remaining <= 0:
                    return False
                self._condition.wait(remaining)
            return True

复杂度

每次增、减、读在锁内 O(1)；等待空间 O(1) 每个等待线程的系统开销。达到零时 notify_all 唤醒 W 名等待者有 O(W) 调度成本；高争用下单锁限制吞吐。

易错点与追问

容易失分的边界

• 只用普通 int 加减，在多线程中假定复合操作原子。
• 降到零未通知等待者，导致永久阻塞。
• 用 if 代替 while，虚假唤醒后错误返回。
• 通知一名 waiter 却有多名等待者。
• 任务还可能继续提交却把一次归零当作整个任务批次已结束。

继续深挖

• 如果要 await 某一批任务完成，应如何加 epoch 或闭合提交？
• 高频读少量写时锁与 atomic primitive 如何取舍？
• 负值是否允许？若允许 wait_zero 的语义如何变化？
• 进程间共享计数如何持久化并处理 worker 崩溃？