竞态条件

概念定义

竞态条件是一类缺陷：程序结果取决于访问共享状态的多个操作之间非确定性的交织。两个线程可能都读出计数器、各自加一、再写回，结果丢掉一次自增。该缺陷与时序相关：它可能通过上千次运行，却在负载下失败。它与数据竞争不同——后者是对内存的非同步并发访问，属于未定义行为——竞态关乎逻辑正确性，即使使用了原子安全的原语也可能出现。

为什么重要

竞态是最难发现的缺陷类型，因为它时断时续。测试循环能过、压力测试能过，然后在真实并发下数值开始漂移。它们出现在计数器、缓存、惰性初始化、先检查后执行模式，以及无法简单加锁的分布式状态中。若不修复，会造成悄无声息的数据损坏、多计费或重复提交。

如何预防

让整个读-改-写保持原子，而不是只让其中一步原子：使用比较并交换、覆盖所有访问的互斥锁，或由单线程 actor 独占状态。优先使用不可变数据与消息传递，而非共享可变状态。当协调是分布式的，使用幂等键与服务端原子操作，并通过对交织情形进行推理来证明修复，而不是跑测试碰运气。

代码示例

use std::sync::mpsc;
use std::thread;

fn main() {
    let (sender, receiver) = mpsc::channel::<u32>();

    let producers: Vec<_> = (0..3)
        .map(|id| {
            let sender = sender.clone();
            thread::spawn(move || {
                for item in 1..=3 {
                    sender.send(id * 10 + item).expect("send failed");
                }
            })
        })
        .collect();
    drop(sender);

    let consumer = thread::spawn(move || {
        let mut total = 0u32;
        while let Ok(value) = receiver.recv() {
            total += value;
        }
        total
    });

    for handle in producers {
        handle.join().expect("producer panicked");
    }
    let total = consumer.join().expect("consumer panicked");
    println!("total = {total}");
}

from __future__ import annotations

import asyncio
from collections.abc import Awaitable, Iterable
from typing import TypeVar

T = TypeVar("T")


async def first_success(awaitables: Iterable[Awaitable[T]], *, timeout: float) -> T:
    tasks = [asyncio.create_task(awaitable) for awaitable in awaitables]
    if not tasks:
        raise ValueError("at least one awaitable is required")

    loop = asyncio.get_running_loop()
    deadline = loop.time() + timeout
    pending = set(tasks)
    failures: list[BaseException] = []
    try:
        while pending:
            remaining = deadline - loop.time()
            if remaining <= 0:
                raise TimeoutError("no replica succeeded before the deadline")
            done, pending = await asyncio.wait(
                pending,
                timeout=remaining,
                return_when=asyncio.FIRST_COMPLETED,
            )
            if not done:
                raise TimeoutError("no replica succeeded before the deadline")
            for task in done:
                error = task.exception()
                if error is None:
                    return task.result()
                failures.append(error)
        messages = "; ".join(str(error) for error in failures)
        raise RuntimeError(f"all replicas failed: {messages}")
    finally:
        for task in pending:
            task.cancel()
        if pending:
            await asyncio.gather(*pending, return_exceptions=True)


async def fetch(name: str, delay: float, *, fail: bool = False) -> str:
    await asyncio.sleep(delay)
    if fail:
        raise ConnectionError(f"{name} unavailable")
    return name


async def main() -> None:
    result = await first_success(
        [fetch("slow", 0.20), fetch("fast", 0.05), fetch("broken", 0.01, fail=True)],
        timeout=1.0,
    )
    print(f"winner: {result}")
    assert result == "fast"

    cancelled = asyncio.Event()

    async def cancellable() -> str:
        try:
            await asyncio.sleep(30)
        except asyncio.CancelledError:
            cancelled.set()
            raise
        return "never"

    assert await first_success(
        [fetch("quick", 0.01), cancellable()],
        timeout=1.0,
    ) == "quick"
    await asyncio.sleep(0)
    assert cancelled.is_set()

    try:
        await first_success(
            [fetch("a", 0.01, fail=True), fetch("b", 0.02, fail=True)],
            timeout=1.0,
        )
    except RuntimeError as exc:
        print(f"all failed: {exc}")


if __name__ == "__main__":
    asyncio.run(main())