使用 POSIX 信号量的可重用屏障实现

Question

需要一个创建 5 个 pthread 的解决方案。每个 pthread 执行一个函数，该函数涉及循环 10 次。在循环的每次迭代中，线程将 int 从 0 增加到 0.9*MAX_INT，然后打印迭代编号。确保 5 个线程中的每一个都在它们可以开始第 (i+1) 次迭代之前完成循环的第 i 次迭代（即所有线程在每次迭代结束时同步/会合）。我需要使用使用 POSIX 信号量实现的两阶段屏障来强制执行同步约束

我写了以下代码我正确吗？

#include <stdio.h>

#include <stdlib.h>

#include <pthread.h>

int thread_count;

void* MyThread(void* rank);

int main()

{

  long thread;

   pthread_t* thread_handles;

   thread_count = 5;

   thread_handles = malloc (thread_count*sizeof(pthread_t));

   for (thread = 0; thread < thread_count; thread++)

       pthread_create(&thread_handles[thread],NULL,MyThread,(void*) thread);

   for (thread = 0; thread < thread_count; thread++)

       pthread_join(thread_handles[thread], NULL);

   free(thread_handles);

   return 0;

}

void* Hello(void* rank)

{

    long my_rank = (long) rank;

    int a,i;

    a=0;

    for(i=0;i<10;i++)

    {

          int n = 5;
          int count = 0;

          pthread_mutex_t mutex = Semaphore(1)

          barrier = Semaphore(0)

          a = a + 0.9*MAX_INT;

          printf("this is %d iteration\n",i);

          mutex.wait()

          count = count + 1

          mutex.signal()

          if count == n: barrier.signal() # unblock ONE thread

          barrier.wait()

          barrier.signal()

   }

}

编辑：

#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
#include <time.h>
#include <semaphore.h>

typedef struct {
  int n;
  int count;
  sem_t mutex;
  sem_t turnstyle;
  sem_t turnstyle2;
} barrier_t;

void init_barrier(barrier_t *barrier, int n)
{
  barrier->n = n;
  barrier->count = 0;
  sem_init(&barrier->mutex, 0, 1);
  sem_init(&barrier->turnstyle, 0, 0);
  sem_init(&barrier->turnstyle2, 0, 0);
}

void phase1_barrier(barrier_t *barrier)
{
  sem_wait(&barrier->mutex);
  if (++barrier->count == barrier->n) {
    int i;
    for (i = 0; i < barrier->n; i++) {
      sem_post(&barrier->turnstyle);
    }
  }
  sem_post(&barrier->mutex);
  sem_wait(&barrier->turnstyle);
}

void phase2_barrier(barrier_t *barrier)
{
  sem_wait(&barrier->mutex);
  if (--barrier->count == 0) {
    int i;
    for (i = 0; i < barrier->n; i++) {
      sem_post(&barrier->turnstyle2);
    }
  }
  sem_post(&barrier->mutex);
  sem_wait(&barrier->turnstyle2);
}

void wait_barrier(barrier_t *barrier)
{
  phase1_barrier(barrier);
  phase2_barrier(barrier);
}

#define NUM_THREADS 5

void *myThread(void *);

int main(int argc, char **argv)
{
  pthread_t threads[NUM_THREADS];
  barrier_t barrier;
  int i;

  init_barrier(&barrier, NUM_THREADS);

  for (i = 0; i < NUM_THREADS; i++) {
    pthread_create(&threads[i], NULL, myThread, &barrier);
  }

  for (i = 0; i < NUM_THREADS, i++) {
    pthread_join(threads[i], NULL);
  }

  return 0;
}

void *myThread(void *arg)
{
      barrier_t *barrier = arg;
      int i,a;

        for(i=0;i<10;i++)

            {
                a = a + 0.9*MAX_INT;

                printf("this is %d iteration\n",i);
            }
  return NULL;
}

Answer 1

好的，如果我们检查Barrier“信号量小书”第 3.7.7 节中的对象，我们会看到我们需要 amutex和 2 个信号量，称为turnstile和turnstile2（互斥量可以是初始化为 1 的信号量）。

由于我们必须使用 POSIX 信号量、pthreads 和INT_MAX，我们首先包含必要的头文件：

#include <pthread.h>
#include <semaphore.h>
#include <limits.h>

书使Barrier对象成为对象；然而，在 C 中，我们并没有真正的对象，但是我们可以创建一个struct带有一些函数来操作它的对象：

typedef struct {
  int n;
  int count;
  sem_t mutex;
  sem_t turnstile;
  sem_t turnstile2;
} barrier_t;

我们可以创建一个函数来初始化屏障：

void init_barrier(barrier_t *barrier, int n)
{
  barrier->n = n;
  barrier->count = 0;
  sem_init(&barrier->mutex, 0, 1);
  sem_init(&barrier->turnstile, 0, 0);
  sem_init(&barrier->turnstile2, 0, 0);
}

并实现该phase1功能，如书中所述：

void phase1_barrier(barrier_t *barrier)
{
  sem_wait(&barrier->mutex);
  if (++barrier->count == barrier->n) {
    int i;
    for (i = 0; i < barrier->n; i++) {
      sem_post(&barrier->turnstile);
    }
  }
  sem_post(&barrier->mutex);
  sem_wait(&barrier->turnstile);
}

请注意，该sem_post函数仅发布一次，因此需要一个循环来发布turnstile n时间。

该phase2函数也以相同的方式直接遵循：

void phase2_barrier(barrier_t *barrier)
{
  sem_wait(&barrier->mutex);
  if (--barrier->count == 0) {
    int i;
    for (i = 0; i < barrier->n; i++) {
      sem_post(&barrier->turnstile2);
    }
  }
  sem_post(&barrier->mutex);
  sem_wait(&barrier->turnstile2);
}

最后，我们可以实现这个wait功能：

void wait_barrier(barrier_t *barrier)
{
  phase1_barrier(barrier);
  phase2_barrier(barrier);
}

现在，在您的main函数中，您可以分配和初始化一个屏障并将其传递给您的衍生线程：

#define NUM_THREADS 5

void *myThread(void *);

int main(int argc, char **argv)
{
  pthread_t threads[NUM_THREADS];
  barrier_t barrier;
  int i;

  init_barrier(&barrier, NUM_THREADS);

  for (i = 0; i < NUM_THREADS; i++) {
    pthread_create(&threads[i], NULL, myThread, &barrier);
  }

  for (i = 0; i < NUM_THREADS, i++) {
    pthread_join(threads[i], NULL);
  }

  return 0;
}

最后，实现线程：

void *myThread(void *arg)
{
  barrier_t *barrier = arg;
  int i;
  int a;

  for (i = 0; i < 10; i++) {
    for (a = 0; a < 0.9*INT_MAX; a++);
    printf("this is %d iteration\n", i);
    wait_barrier(barrier);
  }

  return NULL;
}