C++实现进程间的同步原语：利用操作系统信号量（Semaphore）与互斥体（Mutex）

C++ 进程间同步：信号量与互斥体

大家好，今天我们来深入探讨 C++ 中进程间同步的两种重要原语：信号量（Semaphore）和互斥体（Mutex）。在多进程并发编程中，这两个工具至关重要，它们能够确保多个进程在访问共享资源时的正确性和一致性，避免出现数据竞争和其他并发问题。

1. 进程间同步的必要性

在单进程环境下，线程间的同步机制已经可以解决大部分并发问题。但是，当程序涉及到多个独立的进程协同工作时，线程间的同步机制就无法直接应用。这是因为：

• 地址空间隔离: 每个进程都有自己独立的地址空间，线程共享同一个地址空间，因此线程间同步可以直接通过共享内存来实现。而进程间不能直接访问彼此的内存。
• 资源竞争: 多个进程可能需要访问同一个文件、网络端口、硬件设备或其他系统资源，如果没有适当的同步机制，就会导致资源竞争。

因此，我们需要进程间同步机制，允许进程以可控的方式共享资源和进行通信。

2. 信号量（Semaphore）

信号量是一种计数器，用于控制多个进程对共享资源的访问。它维护一个整数值，表示可用资源的数量。进程可以通过 wait (或 acquire, P操作) 操作来减少信号量的值，表示请求一个资源。如果信号量的值为零，则进程会被阻塞，直到有其他进程释放资源（增加信号量的值）。进程可以通过 signal (或 release, V操作) 操作来增加信号量的值，表示释放一个资源。

2.1 信号量的类型

• 二进制信号量（Binary Semaphore）： 信号量的值只能是 0 或 1，类似于互斥锁。它通常用于实现互斥访问。
• 计数信号量（Counting Semaphore）： 信号量的值可以是任意非负整数，表示可用资源的数量。它通常用于控制对有限数量资源的并发访问。

2.2 C++ 中信号量的实现

C++ 标准库本身并没有提供直接的信号量实现，但我们可以利用操作系统提供的 API 来创建和使用信号量。不同的操作系统提供了不同的 API，这里我们以 POSIX 信号量为例。

POSIX 信号量

POSIX 信号量提供了两组 API：

• 命名信号量： 通过文件名来标识信号量，可以在不相关的进程间使用。
• 匿名信号量： 只能在相关的进程（例如，通过 fork 创建的父子进程）间使用。

我们主要讨论命名信号量，因为它们更适合一般情况下的进程间同步。

2.2.1 命名信号量的创建和销毁

#include <semaphore.h>
#include <fcntl.h>
#include <unistd.h>
#include <iostream>
#include <errno.h>

const char* semaphore_name = "/my_semaphore";

sem_t* create_semaphore(int initial_value) {
    sem_t* semaphore = sem_open(semaphore_name, O_CREAT | O_EXCL, 0666, initial_value);
    if (semaphore == SEM_FAILED) {
        if (errno == EEXIST) {
            // Semaphore already exists, try to open it
            semaphore = sem_open(semaphore_name, 0);
            if (semaphore == SEM_FAILED) {
                std::cerr << "Error opening existing semaphore: " << strerror(errno) << std::endl;
                return SEM_FAILED;
            }
        } else {
            std::cerr << "Error creating semaphore: " << strerror(errno) << std::endl;
            return SEM_FAILED;
        }
    }
    return semaphore;
}

void destroy_semaphore(sem_t* semaphore) {
    if (sem_close(semaphore) == -1) {
        std::cerr << "Error closing semaphore: " << strerror(errno) << std::endl;
    }

    // Unlink the semaphore to remove it from the system
    if (sem_unlink(semaphore_name) == -1) {
        std::cerr << "Error unlinking semaphore: " << strerror(errno) << std::endl;
    }
}

int main() {
    sem_t* semaphore = create_semaphore(1); // Create a binary semaphore with initial value 1

    if (semaphore == SEM_FAILED) {
        return 1;
    }

    // ... use the semaphore ...

    destroy_semaphore(semaphore); // Destroy the semaphore
    return 0;
}

• sem_open(semaphore_name, O_CREAT | O_EXCL, 0666, initial_value): 创建或打开一个命名信号量。
  • semaphore_name: 信号量的名称，必须以 / 开头。
  • O_CREAT: 如果信号量不存在，则创建它。
  • O_EXCL: 如果 O_CREAT 被指定，并且信号量已经存在，则 sem_open 会失败。
  • 0666: 设置信号量的权限（类似于文件权限）。
  • initial_value: 信号量的初始值。
• sem_close(semaphore): 关闭信号量。
• sem_unlink(semaphore_name): 从系统中删除信号量。 只有所有打开信号量的进程关闭了信号量，sem_unlink 才会真正删除信号量。

2.2.2 信号量的 Wait 和 Signal 操作

#include <semaphore.h>
#include <iostream>
#include <errno.h>

// ... (create_semaphore and destroy_semaphore functions from the previous example)

void wait_semaphore(sem_t* semaphore) {
    if (sem_wait(semaphore) == -1) {
        std::cerr << "Error waiting on semaphore: " << strerror(errno) << std::endl;
    }
}

void signal_semaphore(sem_t* semaphore) {
    if (sem_post(semaphore) == -1) {
        std::cerr << "Error posting to semaphore: " << strerror(errno) << std::endl;
    }
}

int main() {
    sem_t* semaphore = create_semaphore(1);

    if (semaphore == SEM_FAILED) {
        return 1;
    }

    // Critical section - only one process can access this at a time
    wait_semaphore(semaphore);
    std::cout << "Process " << getpid() << " is in the critical section." << std::endl;
    sleep(2); // Simulate some work
    std::cout << "Process " << getpid() << " is leaving the critical section." << std::endl;
    signal_semaphore(semaphore);

    destroy_semaphore(semaphore);
    return 0;
}

• sem_wait(semaphore): 减少信号量的值。如果信号量的值为零，则调用进程会被阻塞，直到信号量的值大于零。
• sem_post(semaphore): 增加信号量的值。如果有进程因为调用 sem_wait 而被阻塞，则其中一个进程会被唤醒。

2.3 信号量的应用场景

• 互斥访问： 使用二进制信号量可以实现互斥访问共享资源。
• 资源计数： 使用计数信号量可以控制对有限数量资源的并发访问。例如，限制同时访问数据库的连接数。
• 进程同步： 信号量可以用于协调多个进程的执行顺序。例如，一个进程产生数据，另一个进程消费数据。

3. 互斥体（Mutex）

互斥体（Mutex，Mutual Exclusion）是一种锁机制，用于保护共享资源，确保在任何时候只有一个进程可以访问该资源。 与信号量不同，互斥体的目的是提供独占访问，而不是控制资源的数量。

3.1 C++ 中互斥体的实现

与信号量类似，C++ 标准库并没有提供直接的进程间互斥体实现。我们需要使用操作系统提供的 API。 同样，我们以 POSIX 互斥体为例。

POSIX 互斥体

POSIX 互斥体也提供了两组 API：

• 进程共享互斥体： 可以被多个进程共享。
• 非进程共享互斥体： 只能在同一个进程的线程间使用。

我们需要使用进程共享互斥体来实现进程间同步。

3.1.1 进程共享互斥体的创建和销毁

#include <pthread.h>
#include <iostream>
#include <unistd.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <errno.h>

typedef struct {
    pthread_mutex_t mutex;
} shared_memory_t;

const char* shared_memory_name = "/my_shared_memory";

shared_memory_t* create_shared_memory() {
    int fd = shm_open(shared_memory_name, O_CREAT | O_RDWR, 0666);
    if (fd == -1) {
        std::cerr << "Error creating shared memory: " << strerror(errno) << std::endl;
        return nullptr;
    }

    if (ftruncate(fd, sizeof(shared_memory_t)) == -1) {
        std::cerr << "Error truncating shared memory: " << strerror(errno) << std::endl;
        close(fd);
        shm_unlink(shared_memory_name);
        return nullptr;
    }

    shared_memory_t* shared_data = (shared_memory_t*)mmap(NULL, sizeof(shared_memory_t), PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
    if (shared_data == MAP_FAILED) {
        std::cerr << "Error mapping shared memory: " << strerror(errno) << std::endl;
        close(fd);
        shm_unlink(shared_memory_name);
        return nullptr;
    }

    close(fd);

    pthread_mutexattr_t mutex_attr;
    pthread_mutexattr_init(&mutex_attr);
    pthread_mutexattr_setpshared(&mutex_attr, PTHREAD_PROCESS_SHARED); // Enable process sharing

    if (pthread_mutex_init(&shared_data->mutex, &mutex_attr) != 0) {
        std::cerr << "Error initializing mutex: " << strerror(errno) << std::endl;
        munmap(shared_data, sizeof(shared_memory_t));
        shm_unlink(shared_memory_name);
        return nullptr;
    }

    pthread_mutexattr_destroy(&mutex_attr);

    return shared_data;
}

void destroy_shared_memory(shared_memory_t* shared_data) {
    if (pthread_mutex_destroy(&shared_data->mutex) != 0) {
        std::cerr << "Error destroying mutex: " << strerror(errno) << std::endl;
    }

    if (munmap(shared_data, sizeof(shared_memory_t)) == -1) {
        std::cerr << "Error unmapping shared memory: " << strerror(errno) << std::endl;
    }

    if (shm_unlink(shared_memory_name) == -1) {
        std::cerr << "Error unlinking shared memory: " << strerror(errno) << std::endl;
    }
}

int main() {
    shared_memory_t* shared_data = create_shared_memory();

    if (shared_data == nullptr) {
        return 1;
    }

    // ... use the mutex ...

    destroy_shared_memory(shared_data);
    return 0;
}

• shm_open(shared_memory_name, O_CREAT | O_RDWR, 0666): 创建或打开一个共享内存对象。
  • shared_memory_name: 共享内存对象的名称，必须以 / 开头。
  • O_CREAT: 如果共享内存对象不存在，则创建它。
  • O_RDWR: 以读写方式打开共享内存对象。
  • 0666: 设置共享内存对象的权限。
• ftruncate(fd, sizeof(shared_memory_t)): 设置共享内存对象的大小。
• mmap(NULL, sizeof(shared_memory_t), PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0): 将共享内存对象映射到进程的地址空间。
  • PROT_READ | PROT_WRITE: 设置映射的内存区域的权限为可读可写。
  • MAP_SHARED: 指定映射是共享的，这意味着对映射的内存区域的修改对其他进程可见。
• pthread_mutexattr_t mutex_attr: 互斥锁属性。
• pthread_mutexattr_init(&mutex_attr): 初始化互斥锁属性。
• pthread_mutexattr_setpshared(&mutex_attr, PTHREAD_PROCESS_SHARED): 设置互斥锁为进程共享。
• pthread_mutex_init(&shared_data->mutex, &mutex_attr): 初始化互斥锁。
• pthread_mutexattr_destroy(&mutex_attr): 销毁互斥锁属性。
• munmap(shared_data, sizeof(shared_memory_t)): 取消映射共享内存对象。
• shm_unlink(shared_memory_name): 删除共享内存对象。

3.1.2 互斥体的 Lock 和 Unlock 操作

#include <pthread.h>
#include <iostream>
#include <unistd.h>
#include <errno.h>

// ... (create_shared_memory and destroy_shared_memory functions from the previous example)

void lock_mutex(shared_memory_t* shared_data) {
    if (pthread_mutex_lock(&shared_data->mutex) != 0) {
        std::cerr << "Error locking mutex: " << strerror(errno) << std::endl;
    }
}

void unlock_mutex(shared_memory_t* shared_data) {
    if (pthread_mutex_unlock(&shared_data->mutex) != 0) {
        std::cerr << "Error unlocking mutex: " << strerror(errno) << std::endl;
    }
}

int main() {
    shared_memory_t* shared_data = create_shared_memory();

    if (shared_data == nullptr) {
        return 1;
    }

    // Critical section - only one process can access this at a time
    lock_mutex(shared_data);
    std::cout << "Process " << getpid() << " is in the critical section." << std::endl;
    sleep(2); // Simulate some work
    std::cout << "Process " << getpid() << " is leaving the critical section." << std::endl;
    unlock_mutex(shared_data);

    destroy_shared_memory(shared_data);
    return 0;
}

• pthread_mutex_lock(&shared_data->mutex): 尝试获取互斥锁。如果互斥锁已经被其他进程持有，则调用进程会被阻塞，直到互斥锁被释放。
• pthread_mutex_unlock(&shared_data->mutex): 释放互斥锁。

3.2 互斥体的应用场景

• 互斥访问： 互斥体的主要用途是保护共享资源，确保在任何时候只有一个进程可以访问该资源。例如，保护对共享文件的写入操作。

4. 信号量 vs. 互斥体

虽然信号量和互斥体都可以用于实现互斥访问，但它们之间存在一些重要的区别：

特性	信号量	互斥体
所有权	无所有权概念，任何进程都可以 signal	互斥体有所有权，只有持有互斥体的进程可以 unlock
用途	用于控制对资源的并发访问，以及进程同步	用于提供对共享资源的独占访问
值	可以是任意非负整数	只能是 0 或 1
复杂性	更灵活，但使用不当容易出错	更简单，更安全

选择哪种原语？

• 如果只需要提供对共享资源的独占访问，那么互斥体是更简单、更安全的选择。
• 如果需要控制对资源的并发访问数量，或者需要实现进程间的复杂同步，那么信号量是更合适的选择。

5. 死锁（Deadlock）

在使用信号量和互斥体时，需要特别注意死锁问题。死锁是指两个或多个进程无限期地等待对方释放资源，导致所有进程都无法继续执行。

5.1 死锁的产生条件

死锁的产生通常需要满足以下四个条件：

• 互斥条件： 资源必须以独占方式访问。
• 占有且等待条件： 进程已经占有一个资源，但同时又请求另一个资源。
• 不可剥夺条件： 进程已经获得的资源不能被强制剥夺。
• 循环等待条件： 存在一个进程链，每个进程都在等待链中下一个进程所占有的资源。

5.2 避免死锁的方法

• 避免循环等待： 对所有资源进行排序，并要求进程按照固定的顺序请求资源。
• 超时机制： 为获取资源设置超时时间，如果超过超时时间仍未获取到资源，则放弃请求。
• 死锁检测与恢复： 定期检测系统是否存在死锁，如果检测到死锁，则采取措施恢复系统。例如，杀死一个或多个进程。

6. 总结与建议

信号量和互斥体是 C++ 中实现进程间同步的重要原语。信号量更灵活，可以用于控制资源的并发访问数量和实现进程间的复杂同步。互斥体更简单，更安全，用于提供对共享资源的独占访问。在使用这些原语时，需要特别注意死锁问题，并采取措施避免死锁的发生。

如何更好地使用这些原语？

• 理解底层原理： 深入理解信号量和互斥体的底层实现机制，可以帮助你更好地使用它们。
• 选择合适的原语： 根据实际需求选择合适的原语。
• 避免死锁： 采取措施避免死锁的发生。
• 进行充分的测试： 对并发程序进行充分的测试，以确保其正确性和可靠性。

掌握信号量和互斥体的使用，能够让你编写出更加健壮和高效的并发程序。 希望今天的分享对大家有所帮助，谢谢！

更多IT精英技术系列讲座，到智猿学院