使用 OpenMP 锁定例程实现计数信号量

Question

我正在尝试使用 OpenMP 的锁定结构来实现 counting semaphores，但是在尝试在关键区域内使用 omp_set_lock() 时遇到了问题（程序挂起）。

我正在使用一个简单的生产者-消费者程序来测试实现。这是我想出的：

#include <omp.h>
#include <stdio.h>

#define N 50

int semaphore_count = 0;
omp_lock_t semaphore_lock;

int A[N];

void semaphore_increment()
{
  int my_id = omp_get_thread_num();

  #pragma omp critical
  {
    printf("[%lf][%d] semaphore_count %d --> %d\n", omp_get_wtime(), my_id,
           semaphore_count, semaphore_count + 1);

    semaphore_count++;

    if(semaphore_count == 1) {
      // Semaphore was previously locked, so unlock it.
      printf("[%lf][%d] Releasing lock.\n", omp_get_wtime(), my_id);
      omp_unset_lock(&semaphore_lock);
    }
  }
}

void semaphore_decrement()
{
  int my_id = omp_get_thread_num();
  #pragma omp critical
  {
    printf("[%lf][%d] semaphore_count: %d\n", omp_get_wtime(), my_id,
           semaphore_count);

    if (semaphore_count == 0) {
      printf("[%lf][%d] Sleeping\n", omp_get_wtime(), my_id);
      omp_set_lock(&semaphore_lock);
    }
    else {
      printf("[%lf][%d] Working\n", omp_get_wtime(), my_id);
      // Creating a critical region here instead of in the beginning of
      // the function solves the problem.
      // #pragma omp critical
      // {
        semaphore_count--;
      // }
      if (semaphore_count == 0) {
        omp_set_lock(&semaphore_lock);
      }
    }
  }
}

void produce() {
  for (int i = 0; i < N; ++i) {
    A[i] = i;
    #pragma omp flush
    semaphore_increment();
  }
}

void consume() {
  int sum = 0;
  for (int i = 0; i < N; ++i) {
    semaphore_decrement();
    sum += A[i];
  }

  printf("Sum is: %d\n", sum);
}

int main() {

  omp_init_lock(&semaphore_lock);
  omp_set_lock(&semaphore_lock);

  #pragma omp parallel
  {
    #pragma omp single nowait
    produce();

    #pragma omp single nowait
    consume();
  }

  omp_destroy_lock(&semaphore_lock);
  return 0;
}

每次消费者线程进入睡眠状态时，此版本的程序都会挂起。如果我修改它以将关键区域减少到代码的较小部分（如程序中的注释所示），那么它就可以工作。

我不明白的是：为什么会这样？似乎只增加信号量的生产者线程停止运行，然后一切都挂起，但我不明白为什么。

Answer 1

这个答案是错误（见 cmets）。我把它留在这里作为不如何做的一个例子。

---

正如EOF 在 cmets 中所指出的，问题中显示的程序是错误的，因为它有一个关于更新 semaphore_count 变量的竞争条件，因为这可能同时发生在两个不同的关键部分。

我最终将函数 semaphore_increment 和 semaphore_decrement 替换为以下函数，它可以根据作为 operation 传入的值增加或减少信号量值。

void semaphore_update(int operation) {
  int my_id = omp_get_thread_num();
  int set_lock = 0;

  #pragma omp critical
  {
    if (operation == 0) { // Decrement operation
      if (semaphore_count == 0) {
        set_lock = 1;
      }
      else {
        semaphore_count--;
        if (semaphore_count == 0) {
          // Locking here won't actually lock the current thread, only set
          // the semaphore so that the *next* thread will be put to sleep.
          set_lock = 1;
        }
      }
    }
    else { // Increment operation
      semaphore_count++;
      if(semaphore_count == 1) {
        // Semaphore was previously locked, so unlock it.
        omp_unset_lock(&semaphore_lock);
      }
    }
  }

  // The actual call to omp_set_lock has to happen outside the critical region
  // otherwise any threads trying to unlock the semaphore won't be able to
  // get access to the critical region.
  if (set_lock) {
    omp_set_lock(&semaphore_lock);
  }
}