如何用每个条目仅4位实现双向表线程安全且无全局锁？

场景描述 #

我在设计安全高效的锁机制时遇到了困难，以下是简化后的核心场景：

• 有两个表A和B，各含10个条目：其中9个是指向另一表的指针，剩余1个为NULL
• 程序初始化完成后，非NULL条目构成双射关系（即从A的某条目指向B，再从B的对应条目返回A时，会回到原位置，反之亦然），同时用全局变量跟踪两个表中NULL的位置
• 启动两个工作线程：
  • Thread A：随机选择A中的一个非NULL条目，与A中的NULL位置交换，同时调整B中对应的指针以维持双射关系
  • Thread B：对表B执行完全相同的操作

核心问题 #

如何在不锁定整个表的前提下实现该系统的线程安全？优先考虑利用每个指针的低4位来实现锁机制。

实际场景中表的规模远大于10，每个条目附带少量额外数据，且移动操作并非完全随机。

补充：三个难度版本的需求 #

我需要针对以下三个版本分别提供解决方案：

• 简易版：如果选中的条目无法移动，可随机选择其他条目重试
• 中等版：仅允许表A放弃重试；表B必须阻塞等待，直到选中的条目可以移动
• 困难版：两个表的操作均不允许放弃重试，必须阻塞等待直到操作可执行

补充说明2：无锁示例代码（x86-64/Linux） #

当前代码设计使用指针的低3位存储额外标记，若需要可升级为128位条目以使用低4位：

#include <pthread.h>
#include <stdint.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

// 清除64位值的低3位，获取原始指针
#define GET_POINTER(val) (uint64_t*) ((val) & ~0x7ULL)

// 将新指针的高61位与旧值的低3位拼接（要求新指针的低3位为0）
#define MODIFY_VALUE(new_ptr, old_val) ((uint64_t) (new_ptr)) ^ ((old_val) & 0x7ULL) 

// 全局变量声明
uint64_t A[10], B[10];
int index_of_NULL_value_A, index_of_NULL_value_B;

// 初始化表结构
void init_globals() {
    // 初始化A[0]和B[0]为NULL指针
    A[0] = (uint64_t) NULL; // (uint64_t) NULL == 0
    B[0] = (uint64_t) NULL;

    // 记录初始的NULL位置索引
    index_of_NULL_value_A = 0;
    index_of_NULL_value_B = 0;

    // 创建A到B的指针映射
    for (int i = 1; i < 10; ++i) {
        A[i] = (uint64_t) &B[i];
    }

    // 创建B到A的指针映射
    for (int i = 1; i < 10; ++i) {
        B[i] = (uint64_t) &A[i];
    }
}

// 验证单个表的完整性
void verify_integrity_of_table(uint64_t* table, int null_index) {
    for (int i = 0; i < 10; ++i) {
        if (i == null_index) {
            // 检查NULL位置是否正确
            if (table[i] != (uint64_t) NULL) {
                fprintf(stderr, "表完整性检查失败！位置%d未找到NULL\n", i);
                exit(1);
            }
        } else {
            // 检查双射关系是否成立
            if (&table[i] != GET_POINTER(*GET_POINTER(table[i]))) {
                fprintf(stderr, "表完整性检查失败！位置%d的链接无效\n", i);
                exit(1);
            }
        }
    }
}

// 验证整个系统的完整性
void verify_integrity() {
    verify_integrity_of_table(A, index_of_NULL_value_A);
    verify_integrity_of_table(B, index_of_NULL_value_B);
}

// 创建线程，失败则退出
typedef void *(*start_routine_t)(void *);
pthread_t pthread_create_or_exit(start_routine_t start_routine) {
    pthread_t thread_id;
    int result = pthread_create(&thread_id, NULL, start_routine, NULL);
    if (result != 0) {
        perror("创建线程失败！");
        exit(EXIT_FAILURE);
    }
    return thread_id;
}

// 执行一次随机交换操作
void do_a_random_swap(uint64_t* table, int* null_index_ptr) {
    // 获取当前NULL的位置
    int null_index = *null_index_ptr;

    // 随机选择一个非NULL的条目
    int target_idx = rand() % 10;
    while (target_idx == null_index) {
        target_idx = rand() % 10;
    }

    // 更新反向指针
    uint64_t* new_back_ptr = &table[null_index];
    uint64_t old_back_val = *(GET_POINTER(table[target_idx]));
    *GET_POINTER(table[target_idx]) = MODIFY_VALUE(new_back_ptr, old_back_val);

    // 交换NULL位置和目标条目
    table[null_index] = table[target_idx];
    table[target_idx] = (uint64_t) NULL;

    // 更新NULL位置的跟踪变量
    *null_index_ptr = target_idx;
}

// 线程A的执行函数：操作表A
void* fst_start_routine(void* arg) {
    (void) arg; // 忽略未使用的参数
    while (1) {
        if (time(NULL) % 2 == 0) {
            do_a_random_swap(A, &index_of_NULL_value_A);
        } else {
            usleep(100000); // 睡眠0.1秒
        }
    }
    return NULL;
}

// 线程B的执行函数：操作表B
void* snd_start_routine(void* arg) {
    (void) arg; // 忽略未使用的参数
    while (1) {
        if (time(NULL) % 2 == 0) {
            do_a_random_swap(B, &index_of_NULL_value_B);
        } else {
            usleep(100000); // 睡眠0.1秒
        }
    }
    return NULL;
}

// 完整性检查线程的执行函数
void* integrity_checker_start_routine(void* arg) {
    (void) arg; // 忽略未使用的参数
    for (;; usleep(100000)) {
        if (time(NULL) % 2 == 1) {
            verify_integrity();
        }
    }
    return NULL;
}

int main() {
    // 初始化随机数种子
    srand(time(NULL));

    // 初始化表和NULL位置跟踪变量
    init_globals();

    // 验证初始状态的完整性
    verify_integrity();

    // 创建工作线程和检查线程
    pthread_create_or_exit(fst_start_routine);
    pthread_create_or_exit(snd_start_routine);
    pthread_create_or_exit(integrity_checker_start_routine);

    // 主线程永久睡眠
    while (1) {
        sleep(1);
    }
    return 0;
}

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