无原子RMW操作的跨机器Spinlock实现：问题问询与替代方案

我正在使用一款不支持原子Read-Modify-Write（RMW）操作的光纤反射内存卡（RFM2g型号），需要仅借助该卡提供的共享内存与中断，实现跨多进程（可能运行于不同计算机）的spinlock。我已基于内存操作实现了一款spinlock，简化逻辑的代码如下：

#include <chrono>
#include <iostream>
#include <random>
#include <thread>


void spin_lock(volatile void *lock_area, const uint32_t lock_id) {
    uint32_t lock = static_cast<volatile uint32_t *>(lock_area)[0];

spin:
    while (lock != 0 && lock != lock_id) {
        lock = static_cast<volatile uint32_t *>(lock_area)[0];
    }

    static_cast<volatile uint32_t *>(lock_area)[0] = lock_id;

    auto start = std::chrono::high_resolution_clock::now();
    while (std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::high_resolution_clock::now() - start).
           count() < 10) {
        lock = static_cast<volatile uint32_t *>(lock_area)[0];
    }

    lock = static_cast<volatile uint32_t *>(lock_area)[0];

    if (lock != lock_id) {
        goto spin;
    }
}

void spin_unlock(volatile void *lock_area) {
    static_cast<volatile uint32_t *>(lock_area)[0] = 0;
}

uint32_t generate_lock_id(const uint32_t node_id) {
    static std::random_device random_device{};
    static std::uniform_int_distribution<uint16_t> distribution(0, 32768);
    const auto process_id = distribution(random_device);
    return (static_cast<uint32_t>(node_id) << 16) + process_id;
}

void add_data(volatile void *lock_area, volatile void *data_area, const uint32_t lock_id) {
    spin_lock(lock_area, lock_id);
    const auto count = static_cast<volatile uint32_t *>(data_area)[0];
    static_cast<volatile uint32_t *>(data_area)[0] = count + 1;
    spin_unlock(lock_area);
}


int main(int argc, char **argv) {
    const auto start = std::chrono::high_resolution_clock::now();

    uint16_t node_id = 1;
    volatile auto lock_area = malloc(4);
    volatile auto data_area = malloc(4);
    memset(data_area, 0, 4);
    memset(lock_area, 0, 4);

    auto thread1 = std::thread([lock_area, data_area, node_id]() {
        const auto lock_id = generate_lock_id(node_id);
        for (int i = 0; i < 200000; i++) {
            add_data(lock_area, data_area, lock_id);
        }
    });

    auto thread2 = std::thread([lock_area, data_area, node_id]() {
        const auto lock_id = generate_lock_id(node_id);
        for (int i = 0; i < 200000; i++) {
            add_data(lock_area, data_area, lock_id);
        }
    });

    thread1.join();
    thread2.join();

    const auto end = std::chrono::high_resolution_clock::now();
    const auto nanoseconds = std::chrono::duration_cast<std::chrono::nanoseconds>(end - start).count();
    const auto seconds = static_cast<double>(nanoseconds) / 1000000000.0;

    const auto count = static_cast<volatile uint32_t *>(data_area)[0];

    std::cout << "Time usage: " << seconds << "\nCount: " << count << std::endl;

    free(lock_area);
    free(data_area);

    return 0;
}

技术问题 #

1. 潜在缺陷：除generate_lock_id()中可能出现的process_id重复问题外，该实现是否存在其他问题？例如“等待时间（代码中为10ns）超过内存同步最大延迟则方法可行”这一假设是否成立？
2. 替代实现：是否存在其他适用于跨机器场景的纯内存型spinlock实现方案（无需CPU原子操作）？

关键约束 #

• 程序可能运行于不同计算机（CPU原子RMW操作不可用）
• 仅可使用RFM2g卡提供的共享内存与中断

回答 #

一、现有实现的潜在缺陷 #

除锁ID重复问题外，还存在以下核心问题：

1. 固定等待时间的不可靠性：RFM2g的内存同步延迟受节点负载、光纤链路状态影响，并非固定值。如果实际同步延迟超过10ns，后续的锁校验会失效，多个进程可能同时认为自己持有锁，引发竞态条件。
2. 无冲突时的冗余开销：当锁本身为0或属于当前进程时，仍会执行10ns等待，无端增加锁获取的时间成本，降低并发性能。
3. 解锁操作的竞态风险：spin_unlock直接将锁设为0，无任何同步或校验逻辑。若解锁时恰好有其他进程写入锁ID，会覆盖合法的锁状态，导致持有锁的进程被错误剥夺权限。
4. 忙等待的资源浪费：持续轮询共享内存会占用大量总线带宽，增加RFM2g卡的负载，反而可能加剧同步延迟，在跨机器高并发场景下问题更突出。

二、替代实现方案 #

针对跨机器、无原子RMW的场景，有两种成熟的纯内存型spinlock方案：

1. Lamport面包店算法

   • 核心逻辑：每个进程获取锁前先“取号”，等待所有号码更小的进程完成操作，仅依赖内存读写，无需原子RMW。
   • 适配RFM2g的要点：
     • 在共享内存中维护两个数组：ticket[]（存储各进程的取号）和choosing[]（标记进程是否正在取号）
     • 进程取号时，先设choosing[pid] = 1，读取所有进程的ticket值，取最大值加1作为自己的ticket，再设choosing[pid] = 0
     • 等待阶段轮询所有进程，确认无进程正在取号且自己的ticket是当前最小值
   • 优势：无死锁、公平性好；劣势：高并发下轮询开销大，需为每个进程分配固定索引。

2. 结合RFM2g中断的自旋锁优化

   • 核心逻辑：利用RFM2g的中断机制替代持续轮询，进程获取锁失败时注册中断回调，等待持有锁的进程解锁时触发中断唤醒自己。
   • 适配RFM2g的要点：
     • 在共享内存中维护锁状态和等待进程的中断通知列表
     • 进程获取锁失败时，写入自身中断信息到列表，进入休眠或低功耗轮询状态
     • 解锁进程释放锁后，遍历通知列表触发对应节点的中断，唤醒等待进程
   • 优势：大幅降低总线和卡的负载，适合高并发跨机器场景；劣势：需对接RFM2g的中断API，实现复杂度稍高。

内容的提问来源于stack exchange，提问作者MikuSoft