基于futex实现的信号量生命周期问题及实践处理方式

在Linux系统中，futex系统调用常被用来实现信号量：通过FUTEX_WAIT原子性检查数值并进入睡眠，FUTEX_WAKE唤醒睡眠线程。但如果等待线程从sem_wait返回后立即销毁信号量，会引发内存访问风险——比如发布线程可能还在执行futex_wake，此时信号量内存已被释放，后续操作可能导致虚假唤醒或破坏其他数据结构。

问题复现（POSIX接口示例） #

#include <pthread.h>
#include <semaphore.h>

void* thread_fun(void* arg)
{
  sem_t* sem = (sem_t*)arg;
  sem_post(sem);
  return NULL;
}

int main()
{
    sem_t sem;
    sem_init(&sem, 0, 0); // 初始值为0

    pthread_t thread;
    pthread_create(&thread, NULL, thread_fun, &sem);

    sem_wait(&sem);
    sem_destroy(&sem); // 提前销毁，此时post线程可能还在执行sem_post的futex_wake

    pthread_join(thread, NULL);
}

问题根源（简化实现） #

void sem_post(sem_t* sem)
{
  atomic_fetch_add(&sem->data, +1); // A：增加信号量计数
  futex_wake(&sem->data); // B：唤醒等待线程
}

void sem_wait(sem_t* sem)
{
  int32_t previous = atomic_fetch_add(&sem->data, -1); // C：减少计数
  while(previous <= 0)
  {
    futex_wait(&sem->data, previous);
    previous = atomic_load(&sem->data);
    if(previous > 0 && atomic_compare_exchange_weak(&sem->data, &previous, previous - 1))
      break;
  }
}

可能的危险执行序列：

1. 发布线程执行A（计数+1）
2. 等待线程执行C（计数-1，此时计数回到0，退出循环返回）
3. 等待线程销毁信号量
4. 发布线程执行B（futex_wake访问已释放的内存）

实际应用中的解决方法 #

1. 调整线程同步顺序（最简单直接）

核心思路：确保所有操作信号量的线程完全执行完毕后，再销毁信号量。把pthread_join移到sem_destroy之前，保证发布线程的sem_post（包括futex_wake）全部执行完成：

int main()
{
    sem_t sem;
    sem_init(&sem, 0, 0);

    pthread_t thread;
    pthread_create(&thread, NULL, thread_fun, &sem);

    sem_wait(&sem);
    pthread_join(thread, NULL); // 先等待post线程执行完所有操作
    sem_destroy(&sem); // 再销毁信号量

    return 0;
}

这种方法适用于大多数场景，完全符合POSIX标准对sem_destroy的要求（只能在无线程等待且无线程会再使用该信号量时调用）。

2. 引用计数管理信号量生命周期（适用于动态分配的信号量）

如果信号量是动态分配的（比如用malloc创建），可以给信号量添加原子引用计数，确保只有当所有操作线程都完成后，才真正释放内存：

#include <stdatomic.h>
#include <stdlib.h>

typedef struct {
    atomic_int32_t data;       // 信号量计数
    atomic_uint32_t ref_count; // 引用计数
} sem_t;

// 初始化信号量
void sem_init(sem_t* sem, int pshared, unsigned int value) {
    atomic_store(&sem->data, value);
    atomic_store(&sem->ref_count, 1); // 初始引用计数为1
}

// 增加引用计数
static void sem_acquire_ref(sem_t* sem) {
    atomic_fetch_add(&sem->ref_count, 1);
}

// 减少引用计数，若为0则销毁内存
static void sem_release_ref(sem_t* sem) {
    if (atomic_fetch_sub(&sem->ref_count, 1) == 1) {
        free(sem); // 最后一个引用，真正释放内存
    }
}

void sem_post(sem_t* sem) {
    sem_acquire_ref(sem); // 操作前加引用
    atomic_fetch_add(&sem->data, +1);
    futex_wake(&sem->data);
    sem_release_ref(sem); // 操作完减引用
}

void sem_wait(sem_t* sem) {
    sem_acquire_ref(sem); // 操作前加引用
    int32_t previous = atomic_fetch_add(&sem->data, -1);
    while(previous <= 0)
    {
        futex_wait(&sem->data, previous);
        previous = atomic_load(&sem->data);
        if(previous > 0 && atomic_compare_exchange_weak(&sem->data, &previous, previous - 1))
          break;
    }
    sem_release_ref(sem); // 操作完减引用
}

// 外部调用的销毁接口
void sem_destroy(sem_t* sem) {
    sem_release_ref(sem); // 减少初始的引用计数
}

原理：每次调用sem_post或sem_wait时，先增加引用计数，确保信号量内存不会被提前释放；操作完成后减少计数。当sem_destroy被调用时，只是减少初始的引用计数，只有当所有操作线程都完成（引用计数归0），才真正释放内存，从根源上避免了futex_wake访问无效地址的问题。

3. 自定义实现时增加内存合法性检查（辅助手段）

如果是自己实现futex操作，可以在futex_wake前先检查内存是否还属于当前进程（比如通过mmap的区间判断），但这种方法依赖系统细节，可靠性不如前两种，仅作为额外的安全防护。

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