同步

February 10, 2025

预定义

本文的目标是对一些同步机制做一下高层次表述，附上一些抽象的实现。
如果从比较理论的角度来看待这些话题，一般就以资源为抽象视角，但这里的视角是物理内存。

我们首先希望考虑在裸机上编程（以 risc-v 为例），所以是不存在操作系统的。

因此这里定义一个类Linux的非常简单的模型：线程是调度的最小单位；进程只是共享内存的线程组。

调度简单来说，分为主动放弃CPU（暂时没有任务可以做了）和被动放弃CPU（如时间片耗尽）。
好的调度很复杂，还需要考虑不同的设计目标（比如实时/公平性），所以这里就从略了（包括比如锁唤醒等等调度场景也从略）。

目标

同步考虑的是一个内存一致性问题。

把任务的执行抽象为两种：

单核：任务可以被中断/继续（可以等价为并发）
多核：多个任务可以并行，并允许访问同一块内存

注：物理CPU有更复杂的机制，如：SIMD、超线程。这里只是简单探讨。

目标是任务完成前，避免任务需要的某些内存被意外修改了。（任务一旦开始，往往说任务“持有”资源）

原语

原子操作

#include <stdint.h>

// 原子加
uint32_t atomic_add(volatile uint32_t *addr, uint32_t value) {
  uint32_t result;
  asm volatile (
    "amoadd.w %[result], %[value], (%[addr])"
    : [result] "=r" (result)
    : [value] "r" (value), [addr] "r" (addr)
    : "memory"
  );
  return result;
}

// 原子交换
uint32_t atomic_swap(volatile uint32_t *addr, uint32_t value) {
    uint32_t result;
    asm volatile (
        "amoswap.w %[result], %[value], (%[addr])"
        : [result] "=r" (result)
        : [value] "r" (value), [addr] "r" (addr)
        : "memory"
    );
    return result;
}

// 原子比较并交换 (Compare and Swap - CAS)
uint32_t atomic_compare_and_swap(volatile uint32_t *addr, uint32_t expected, uint32_t new_value) {
    uint32_t result;
    asm volatile (
        "1:  lr.w  %[result], (%[addr])\n"
        "    bne   %[result], %[expected], 2f\n"
        "    sc.w  %[result], %[new_value], (%[addr])\n"
        "    bnez  %[result], 1b\n"
        "2:"
        : [result] "=&r" (result)
        : [new_value] "r" (new_value), [expected] "r" (expected), [addr] "r" (addr)
        : "memory"
    );
    return result; // 返回0表示成功，非0表示失败
}

互斥锁

#include <stdbool.h>

typedef struct {
  volatile uint32_t locked; // 0: unlocked, 1: locked
} mutex_t;

void mutex_init(mutex_t *mutex) {
  mutex->locked = 0;
}

void mutex_lock(mutex_t *mutex) {
  uint32_t expected = 0;
  uint32_t new_value = 1;

  // 自旋等待
  while (atomic_compare_and_swap(&mutex->locked, expected, new_value) != 0) {
    yield(); // 切换到其它进程/线程
  }
}

void mutex_unlock(mutex_t *mutex) {
  mutex->locked = 0; // 直接赋值，因为解锁时我们知道它是锁定的
  // 可选：加入内存屏障，确保解锁操作对其它核心可见
  asm volatile ("fence rw,w" ::: "memory");
}

信号量

typedef struct {
  volatile int32_t count;
} semaphore_t;

void semaphore_init(semaphore_t *sem, int32_t initial_count) {
  sem->count = initial_count;
}

void semaphore_wait(semaphore_t *sem) {
  while (1) {
    int32_t current_count = sem->count;
    if (current_count > 0) {
      if (atomic_compare_and_swap(&sem->count, current_count, current_count - 1) == 0) {
        break; // 获取信号量成功
      }
    } else {
      // 信号量为0，需要阻塞当前线程
      // ... (实现线程阻塞和加入等待队列的逻辑) ...
      // 例如：
      // block_current_thread(sem->wait_queue);
      // yield(); // 切换到其它线程
    }
  }
}

void semaphore_signal(semaphore_t *sem) {
  atomic_add(&sem->count, 1);
  // ... (唤醒等待队列中的一个线程) ...
  // 例如：
  // unblock_thread(sem->wait_queue);
  // yield(); // 考虑是否需要切换线程
}

条件变量

typedef struct {
  // ... (等待队列等数据结构) ...
} condvar_t;

void condvar_init(condvar_t *cv) {
  // ... (初始化等待队列) ...
}

void condvar_wait(condvar_t *cv, mutex_t *mutex) {
  // 1. 释放互斥锁
  mutex_unlock(mutex);

  // 2. 将当前线程加入条件变量的等待队列
  // ... (将当前线程加入 cv->wait_queue) ...
  // block_current_thread(cv->wait_queue);

  // 3. 调度到其它线程
  // yield();

  // 4. 重新获取互斥锁 (在线程被唤醒后)
  mutex_lock(mutex);
}

void condvar_signal(condvar_t *cv) {
  // 唤醒等待队列中的一个线程
  // ... (从 cv->wait_queue 中移除一个线程并唤醒) ...
  // unblock_thread(cv->wait_queue);
}

void condvar_broadcast(condvar_t *cv) {
  // 唤醒等待队列中的所有线程
  // ... (唤醒 cv->wait_queue 中的所有线程) ...
}

屏障

// 假设的线程结构体 (根据操作系统实现定义)
typedef struct {
    uint32_t thread_id;
    // ... 其它线程相关信息 ...
} thread_t;

typedef struct {
    volatile uint32_t threshold; // 需要等待的线程数量
    volatile uint32_t count;     // 当前已到达屏障的线程数量
    volatile uint32_t phase;     // 屏障的阶段 (用于唤醒线程)
    // 等待队列 (用于存放被阻塞的线程)
    // ... (需要实现一个线程等待队列) ...
} barrier_t;

void barrier_init(barrier_t *barrier, uint32_t threshold) {
    barrier->threshold = threshold;
    barrier->count = 0;
    barrier->phase = 0;
    // 初始化等待队列
    // ... (barrier->wait_queue = create_wait_queue();) ...
}

void barrier_wait(barrier_t *barrier) {
    uint32_t current_phase = barrier->phase;

    // 原子性地增加计数器
    uint32_t new_count = atomic_add(&barrier->count, 1) + 1; // atomic_add 返回的是旧值，所以要 + 1

    if (new_count == barrier->threshold) {
        // 最后一个到达的线程负责重置屏障并唤醒所有等待的线程

        // 重置计数器
        barrier->count = 0;

        // 增加屏障阶段 (用于唤醒等待的线程)
        atomic_add(&barrier->phase, 1);

        // 唤醒所有等待的线程
        // ... unblock_all_threads(barrier->wait_queue); ...
        // ... yield(); ... // 考虑是否需要切换线程
        // 实际情况可能需要遍历等待队列，逐个唤醒线程
    } else {
        // 其它线程需要等待

        // 将当前线程加入等待队列
        // ... block_current_thread(barrier->wait_queue); ...

        // 调度到其它线程
        // ... yield(); ...

        // 当线程被唤醒时，它会从这里继续执行
        // 确保线程是在正确的阶段被唤醒
        while (barrier->phase == current_phase) {
            // 自旋等待，直到屏障进入下一个阶段
            // 这种情况是为了防止假唤醒 (spurious wakeups)
            // 可以加入 yield() 或其它让步操作，避免过度占用CPU
            // 例如： asm volatile ("pause" ::: "memory");
        }
    }
}

读写锁

typedef struct {
  volatile uint32_t writer;   // 0: no writer, 1: writer present
  volatile uint32_t readers;  // Number of active readers
  // ... (等待队列，用于阻塞写线程和读线程) ...
} rwlock_t;

void rwlock_init(rwlock_t *rwlock) {
  rwlock->writer = 0;
  rwlock->readers = 0;
  // ... (初始化等待队列) ...
}

void rwlock_read_lock(rwlock_t *rwlock) {
  while (1) {
    // 检查是否有写线程正在写
    if (rwlock->writer == 0) {
      uint32_t current_readers = rwlock->readers;
      if (atomic_compare_and_swap(&rwlock->readers, current_readers, current_readers + 1) == 0) {
        // 获取读锁成功
        break;
      }
    } else {
      // 有写线程，需要阻塞当前读线程
      // ... (将当前线程加入读等待队列) ...
      // block_current_thread(rwlock->read_wait_queue);
      // yield();
    }
  }
}

void rwlock_read_unlock(rwlock_t *rwlock) {
  atomic_add(&rwlock->readers, -1);
  // ... (唤醒写等待队列中的一个线程，如果队列不为空) ...
  // if (rwlock->write_wait_queue is not empty) {
  //   unblock_thread(rwlock->write_wait_queue);
  //   yield(); // 考虑是否需要切换线程
  // }
}

void rwlock_write_lock(rwlock_t *rwlock) {
  while (1) {
    // 检查是否有读线程或写线程正在访问
    if (rwlock->writer == 0 && rwlock->readers == 0) {
      if (atomic_compare_and_swap(&rwlock->writer, 0, 1) == 0) {
        // 获取写锁成功
        break;
      }
    } else {
      // 有读线程或写线程，需要阻塞当前写线程
      // ... (将当前线程加入写等待队列) ...
      // block_current_thread(rwlock->write_wait_queue);
      // yield();
    }
  }
}

void rwlock_write_unlock(rwlock_t *rwlock) {
  rwlock->writer = 0;
  // ... (唤醒读等待队列中的所有线程，如果队列不为空) ...
  // unblock_all_threads(rwlock->read_wait_queue);
  // ... (唤醒写等待队列中的一个线程，如果队列不为空) ...
  // if (rwlock->write_wait_queue is not empty) {
  //   unblock_thread(rwlock->write_wait_queue);
  // }
  // yield(); // 考虑是否需要切换线程
}

自旋锁

typedef struct {
  volatile uint32_t locked; // 0: unlocked, 1: locked
} spinlock_t;

void spinlock_init(spinlock_t *lock) {
  lock->locked = 0;
}

void spinlock_lock(spinlock_t *lock) {
  uint32_t expected = 0;
  uint32_t new_value = 1;

  while (atomic_compare_and_swap(&lock->locked, expected, new_value) != 0) {
    // 自旋等待
    expected = 0; // 每次失败后重置expected
    // 可以考虑加入 yield() 或其它让步操作，避免过度占用CPU
    // 例如： asm volatile ("pause" ::: "memory"); // RISC-V 的 pause 指令
  }
}

void spinlock_unlock(spinlock_t *lock) {
  lock->locked = 0;
  asm volatile ("fence rw,w" ::: "memory"); // 确保解锁操作对其它核心可见
}

可重入锁

#include <stdint.h>
#include <stdbool.h>

// 假设的线程结构体 (需要根据你的操作系统实现定义)
typedef struct {
    uint32_t thread_id;
    // ... 其他线程相关信息 ...
} thread_t;

typedef struct {
    volatile uint32_t owner_thread_id; // 持有锁的线程 ID (0 表示未持有)
    volatile uint32_t recursion_count;   // 递归计数器
    // 等待队列 (用于存放被阻塞的线程)
    // ... (需要实现一个线程等待队列) ...
} reentrant_lock_t;

void reentrant_lock_init(reentrant_lock_t *lock) {
    lock->owner_thread_id = 0;
    lock->recursion_count = 0;
    // 初始化等待队列
    // ... lock->wait_queue = create_wait_queue(); ...
}

void reentrant_lock_lock(reentrant_lock_t *lock) {
    uint32_t current_thread_id = get_current_thread_id(); // 获取当前线程 ID

    // 循环尝试获取锁，直到成功
    while (1) {
        // 检查锁是否被持有
        if (lock->owner_thread_id == 0) {
            // 锁未被持有，尝试获取
            if (atomic_compare_and_swap(&lock->owner_thread_id, 0, current_thread_id) == 0) {
                // 获取锁成功
                lock->recursion_count = 1;
                return;
            }
            // 获取锁失败，重试
        } else if (lock->owner_thread_id == current_thread_id) {
            // 锁已被当前线程持有，增加递归计数器
            lock->recursion_count++;
            return;
        } else {
            // 锁已被其他线程持有，阻塞等待
            // ... block_current_thread(lock->wait_queue); ...
            // ... yield(); ...
        }
    }
}

void reentrant_lock_unlock(reentrant_lock_t *lock) {
    uint32_t current_thread_id = get_current_thread_id(); // 获取当前线程 ID

    // 检查当前线程是否持有锁
    if (lock->owner_thread_id != current_thread_id) {
        // 当前线程不持有锁，报错
        // ... error_handler("Current thread does not own the lock"); ...
        return;
    }

    // 减少递归计数器
    lock->recursion_count--;

    // 检查递归计数器是否为 0
    if (lock->recursion_count == 0) {
        // 释放锁
        lock->owner_thread_id = 0;

        // 唤醒等待线程
        // ... unblock_all_threads(lock->wait_queue); ...
        // ... yield(); ...
    }
}

概念

死锁

同时满足：互斥请求与保持不可剥夺循环等待

预防死锁： 破坏死锁产生的四个必要条件中的一个或多个。
- 破坏互斥条件： 尽量减少对独占资源的使用，例如使用可重入锁。但很多资源本身就是互斥的，所以这个条件很难破坏。
- 破坏请求与保持条件： 线程在请求资源时，要么一次性获得所有需要的资源，要么不请求任何资源。实现起来比较困难，因为线程可能无法预知需要哪些资源。
- 破坏不可剥夺条件： 当线程请求新的资源无法满足时，主动释放已持有的资源。这种方法可能会导致线程频繁释放和获取资源，增加开销。
- 破坏循环等待条件： 对所有资源进行排序，线程按照固定的顺序请求资源。需要对资源进行统一管理
避免死锁： 使用算法来动态地检测和避免死锁的发生，例如银行家算法。
- 银行家算法： 模拟操作系统分配资源的过程，在分配资源之前，判断分配后系统是否处于安全状态（即所有线程都能完成）。算法复杂度较高，需要维护资源分配信息。
检测死锁： 定期检测系统中是否存在死锁，如果发现死锁，则采取措施解除死锁。
- 资源分配图： 通过资源分配图来检测是否存在环路，如果存在环路，则说明存在死锁。适用于资源数量较少情况。
- 超时机制： 设置锁的超时时间，当线程获取锁的时间超过超时时间时，认为发生了死锁。可能会误判，因为线程可能只是因为某些原因阻塞了。
解除死锁： 可能会导致数据丢失或不一致。
- 终止线程： 终止一个或多个死锁中的线程，释放其占用的资源
- 资源剥夺： 强制剥夺一个或多个线程占用的资源，分配给其他线程

活锁

线程不断尝试获取资源，但由于其他线程也在不断尝试，导致所有线程都无法取得进展。

不合理的重试机制：线程在请求资源失败后，立即重试，而没有引入任何延迟或随机性，导致它们始终在竞争同一资源。
优先级反转：高优先级线程不断被低优先级线程抢占，导致高优先级线程无法获得资源，从而陷入活锁。

解决：

引入随机性：在重试之前引入随机的延迟，避免所有线程同时重试。
优先级调整：调整线程的优先级，让某些线程优先获得资源。
退避算法：使用退避算法，例如指数退避，逐渐增加重试的延迟时间。

饥饿

一个或多个线程长时间无法获得所需资源，导致无法执行。

解决：

公平锁：使用公平锁，保证线程按照请求资源的顺序获得资源。
优先级反转控制：解决高优先级线程被低优先级线程阻塞的问题。例如，当高优先级线程等待低优先级线程释放资源时，临时提升低优先级线程的优先级。
资源预留：为某些线程预留资源，保证它们能够及时获得所需资源。
避免长时间占用资源：线程在使用资源时，尽量避免长时间占用，及时释放资源。

乐观锁/悲观锁

特性	乐观锁	悲观锁
假设	数据很少被修改	数据总是会被修改
加锁时机	更新时检查数据是否被修改	读取时加锁，防止其他线程修改数据
冲突处理	重试 (自旋)	阻塞等待锁释放
适用场景	读多写少，冲突概率低	写多读少，冲突概率高
优点	并发性能高，锁开销小	保证数据一致性
缺点	存在 ABA 问题，自旋开销，冲突概率高	并发性能低，锁开销大

异步

同步机制之后有必要简单考虑一下异步机制

上述同步机制中，任务在运行时获取的是所需资源的原始内存。若改成获取资源内存的拷贝，即消息机制（或事件通信），则本文不称为同步。

因此我们可以再对同步加以定义：即同步时，任务需锁定（或者说持有）资源的原始内存。

异步机制总是需要需要额外的内存，这存在以下情况：

拷贝资源：拷贝资源的原始内存，内存开销较大，后续还需考虑写回原始内存。
传递消息：需要通过不同的标识符区分事件，至少需要额外线性内存空间。

同步机制虽然也需要额外内存，但可以发现，所需的都是常数级别内存，内存开销小。

	同步	异步
资源排布	场景越复杂就需要做越复杂的资源安排	资源安排较为简单
内存开销	内存开销小	内存开销做不到最小

至于时间开销，则根据不同场景和异步机制而不同，本文略。

分布式

如果我们的观察维度是任务，那么分布式（计算），即联机（而非单机）计算任务结果，也与同步有关。

联机情况下需要考虑的有单点故障（物理机器故障，如物理磁盘和内存损坏），网络故障（影响通信，如数据包丢失、延迟、乱序、重复），负载均衡，弹性扩容等。此外，由于分布式系统中的各个节点拥有独立的时钟，因此还存在时钟不同步的问题，这可能导致事件顺序判断错误。分布式同步需要处理这些额外的复杂性，常用的技术包括：

分布式锁: 用于协调多个节点对共享资源的访问。实现方式有基于 ZooKeeper, etcd, Redis 等。
分布式共识算法: 确保多个节点对某个值或操作顺序达成一致。常见的算法有 Paxos, Raft, Zab 等。
两阶段提交 (2PC) / 三阶段提交 (3PC): 用于实现分布式事务，保证多个节点上的操作要么全部成功，要么全部失败。
向量时钟 / Lamport 时间戳: 用于在分布式系统中确定事件的偏序关系，解决时钟不同步问题。
消息队列: 通过异步消息传递来解耦各个节点，提高系统的可用性和可伸缩性. 常见的消息队列有 RabbitMQ, Kafka, ActiveMQ 等。

分布式计算与单机上的编程必然是完全不同的。可以说，很多解决方案是在通过协调协议来模拟非分布式系统，对分布式本身的控制减少了。

CRDT

在分布式计算中，同时更新数据的多个副本（复制数据的并发更新）而不进行协调可能导致不一致和冲突。分布式的一些通用解决方案的开销可能比较大。

无冲突复制数据类型（CRDT）是一种特殊的数据结构，其数学特性保证了并发更新的无冲突合并，从而确保最终一致性，无需协调。基于此，乐观复制允许所有更新，并在稍后解决潜在的不一致。这种模式直接针对数据冲突本身，因此可以说对是否分布式都适用，前提是数据能通过此种结构表示。（解决同步的方法就是不同步？）

←Back to home

This work is licensed under CC BY-NC-SA 4.0