pthreads：独占+并发函数（反向信号量）

Question

我有代码可以锁定（某个库的）每个函数，并且我想对其进行优化。给定函数A和B，我不介意任何A与任何其他A同时运行，或任何B与任何其他B同时运行，但没有A可以同时运行任何B 正在运行，反之亦然。线程数是动态的，出于我无法控制的原因，我不得不对互斥锁和条件变量使用静态分配（即PTHREAD_MUTEX_INITIALIZER）。

我有一种预感，最有效的方法是两个条件变量。使用pthread_mutex_trylock 允许一个函数（即A）并行运行，而另一个必须序列化。还有 *trylock 静态初始化实际上是未定义的行为...

编辑：

也许是这样的？我不确定这是否：

• 可以更简单。毕竟，互斥锁是使用信号量实现的，但需要四个互斥锁和两个条件变量来实现基本上是逆信号量的东西。
• 涵盖所有比赛条件。
• “公平”（超出默认优先级和调度）吗？

static int countA = 0;
static int countB = 0;
static pthread_mutex_t lockCountA = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t lockCountB = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t lockA = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t lockB = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t condA = PTHREAD_COND_INITIALIZER;
static pthread_cond_t condB = PTHREAD_COND_INITIALIZER;

// for B(), just s/B/A/g
static void A(void) {
    pthread_mutex_lock(&lockB);
    while(countB)
        pthread_cond_wait(&condB, &lockB);
    pthread_mutex_lock(&lockCountA);
    countA += 1;
    pthread_mutex_unlock(&lockCountA)
    doA();
    pthread_mutex_lock(&lockCountA)
    countA -= 1;
    if countA == 0:
        pthread_cond_signal(&condA);
    mutex_unlock(&lockCountA)
    mutex_unlock(&lockB);
}

Answer 1

您的解决方案存在竞争条件。考虑countA 和countB 都为零，并且两个线程同时调用A() 和B() 的情况。第一个线程锁定lockB，第二个线程锁定lockA，两者都看到他们检查的计数为零，然后继续增加各自的计数并继续错误。

您的解决方案中的另一个问题是它使用pthread_cond_signal()，它不一定会唤醒一个以上的等待线程，因此如果您有 10 个线程等待进入 B() 但只有一个线程在运行 A()，当后一个线程只完成一个B()线程保证继续，其他9个可能无限期等待。

它也不允许一个以上的线程同时运行doA()，因为lockB 被该调用保留。

要解决第一个问题，您可以使用一个同时保护countA 和countB 的互斥锁（因为我们必须检查的共享状态涉及这两个变量的组合）。同时，你也可以只使用一个条件变量：等待条件变量的线程要么都在等待进入A()，要么都在等待进入B()，但两者的混合是不可能的。解决第二个问题只需要pthread_cond_broadcast()。这导致更简单：

static int countA = 0;
static int countB = 0;
static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

void A(void)
{
    pthread_mutex_lock(&lock);
    while (countB)
        pthread_cond_wait(&cond, &lock);
    countA++;
    pthread_mutex_unlock(&lock);

    do_A();

    pthread_mutex_lock(&lock);
    countA--;
    if (!countA)
        pthread_cond_broadcast(&cond);
    pthread_mutex_unlock(&lock);
}

这个解决方案是正确的，但不是“公平的”——如果有一个连续的线程流在执行A()（这样一个新的线程进入并在前一个线程完成并减少它之前增加countA）然后等待执行B() 的线程将永远等待。这在您的特定用途中可能不是问题 - 例如，如果您知道执行A() 的任何线程最终都会执行B()，那么饥饿情况必须最终解决。

可以通过在有线程排队进入B() 时阻止新条目进入A() 来改进系统以防止这种饥饿：

static int countA = 0;
static int countB = 0;
static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
static int queuedA = 0;
static int queuedB = 0;
static pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER;

void A(void)
{
    pthread_mutex_lock(&lock);
    while (queuedB)
        pthread_cond_wait(&queue_cond, &lock);
    while (countB)
    {
        queuedA++;
        pthread_cond_wait(&cond, &lock);
        queuedA--;
    }
    if (!queuedA)
        pthread_cond_broadcast(&queue_cond);
    countA++;
    pthread_mutex_unlock(&lock);

    do_A();

    pthread_mutex_lock(&lock);
    countA--;
    if (!countA)
        pthread_cond_broadcast(&cond);
    pthread_mutex_unlock(&lock);
}

这可以防止饥饿，因为：

1. 当B() 中有任何线程在等待cond 时，queuedB 始终为非零；
2. 虽然queuedB 不为零，但没有线程可以增加countA，因此countA 必须最终达到零并允许等待cond 的线程继续。
3. 当countA 为零时，没有线程可以增加queuedB，因此queuedB 必须最终达到零并允许等待queue_cond 的线程继续进行。