c - 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 (因为我们必须检查的共享状态涉及这两个变量的组合) 。同时，您也可以只使用一个条件变量:WAITING该条件变量的线程必须要么全部等待进入 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);
}

这可以防止饥饿，因为:

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