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的互斥锁(因为我们必须检查的共享状态涉及这两个变量的组合(。 同时，你也可以只使用一个条件变量：等待条件变量的线程必须全部等待进入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()，那么饥饿情况最终必须解决。

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

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. queuedB始终为非零，而B()中有任何线程在等待cond;
2. 虽然queuedB不为零，但任何线程都不能递增countA，因此countA最终必须达到零并允许等待cond的线程继续。
3. 当countA为零时，任何线程都不能递增queuedB，因此queuedB最终必须达到零并允许等待queue_cond的线程继续。