使用 Peterson 的 N 进程算法的信号量实现

Question

我需要对我的代码进行反馈以获取以下语句，我是否走在正确的道路上？

问题陈述：

一个。实现一个信号量类，它有一个私有 int 和三个公共方法：init、wait 和 signal。 wait 和 signal 方法的行为应该与信号量的预期一致，并且必须在其实现中使用 Peterson 的 N 进程算法。

b.编写一个程序，创建 5 个线程并发更新共享整数的值，并使用 a) 部分中创建的信号量类的对象来确保并发更新的正确性。

这是我的工作程序：

#include <iostream>
#include <pthread.h>

using namespace std;

pthread_mutex_t mid;                   //muted id
int shared=0;                           //global shared variable
class semaphore {
   int counter;
public:
   semaphore(){
   }
   void init(){
       counter=1;                     //initialise counter 1 to get first thread access
   }
   void wait(){
       pthread_mutex_lock(&mid);         //lock the mutex here
       while(1){
           if(counter>0){             //check for counter value
               counter--;             //decrement counter
               break;                   //break the loop
           }

       }
       pthread_mutex_unlock(&mid);       //unlock mutex here
   }
   void signal(){
       pthread_mutex_lock(&mid);       //lock the mutex here
           counter++;                   //increment counter
           pthread_mutex_unlock(&mid);   //unlock mutex here
       }

};
semaphore sm;
void* fun(void* id)
{
   sm.wait();                           //call semaphore wait
   shared++;                           //increment shared variable
   cout<<"Inside thread "<<shared<<endl;
   sm.signal();                       //call signal to semaphore
}


int main() {

   pthread_t id[5];                   //thread ids for 5 threads
   sm.init();
   int i;
   for(i=0;i<5;i++)                   //create 5 threads
   pthread_create(&id[i],NULL,fun,NULL);
   for(i=0;i<5;i++)
   pthread_join(id[i],NULL);           //join 5 threads to complete their task
   cout<<"Outside thread "<<shared<<endl;//final value of shared variable
   return 0;
}

Answer 1

您需要在等待循环中旋转时释放互斥锁。

测试恰好工作，因为线程很可能在任何上下文切换之前运行它们的函数开始完成，因此每个线程在下一个甚至开始之前完成。所以你对信号量没有争议。如果你这样做了，他们会被一名服务员卡住并拿着互斥锁旋转，从而阻止任何人访问柜台并因此释放微调器。

这是一个有效的示例（尽管它可能仍然存在导致它偶尔无法正确启动的初始化竞争）。它看起来更复杂，主要是因为它使用了 gcc 内置的原子操作。只要您有多个内核，就需要这些，因为每个内核都有自己的缓存。将计数器声明为“易失性”仅有助于编译器优化 - 对于有效的 SMP，缓存一致性需要跨处理器缓存失效，这意味着需要使用特殊的处理器指令。您可以尝试将它们替换为例如counter++ 和 counter--（对于“共享”也是如此）-并观察它在多核 CPU 上是如何不起作用的。 （有关 gcc 原子操作的更多详细信息，请参阅https://gcc.gnu.org/onlinedocs/gcc-4.8.2/gcc/_005f_005fatomic-Builtins.html）

#include <stdio.h>
#include <pthread.h>
#include <unistd.h>
#include <stdint.h>


class semaphore {
    pthread_mutex_t lock;
    int32_t counter;
public:
   semaphore() {
       init();

   }
   void init() {
       counter = 1;           //initialise counter 1 to get first access
   }

   void spinwait() {
       while (true) {
           // Spin, waiting until we see a positive counter
           while (__atomic_load_n(&counter, __ATOMIC_SEQ_CST) <= 0)
               ;

           pthread_mutex_lock(&lock);
           if (__atomic_load_n(&counter, __ATOMIC_SEQ_CST) <= 0) {
               // Someone else stole the count from under us or it was
               // a fluke - keep trying
               pthread_mutex_unlock(&lock);
               continue;
           }
           // It's ours
           __atomic_fetch_add(&counter, -1, __ATOMIC_SEQ_CST);
           pthread_mutex_unlock(&lock);
           return;
       }
   }

   void signal() {
       pthread_mutex_lock(&lock);   //lock the mutex here
       __atomic_fetch_add(&counter, 1, __ATOMIC_SEQ_CST);
       pthread_mutex_unlock(&lock); //unlock mutex here
   }

};

enum { 
    NUM_TEST_THREADS = 5,
    NUM_BANGS = 1000
 };

// Making semaphore sm volatile would be complicated, because the
// pthread_mutex library calls don't expect volatile arguments.

int shared = 0;       // Global shared variable
semaphore sm;         // Semaphore protecting shared variable

volatile int num_workers = 0;   // So we can wait until we have N threads


void* fun(void* id)
{
    usleep(100000);                   // 0.1s. Encourage context switch.


    const int worker = (intptr_t)id + 1;

    printf("Worker %d ready\n", worker);

    // Spin, waiting for all workers to be in a runnable state.  These printouts
    // could be out of order.
    ++num_workers;
    while (num_workers < NUM_TEST_THREADS)
        ;

    // Go!

    // Bang on the semaphore.  Odd workers increment, even decrement.
    if (worker & 1) {
        for (int n = 0; n < NUM_BANGS; ++n) {
            sm.spinwait();
            __atomic_fetch_add(&shared, 1, __ATOMIC_SEQ_CST);
            sm.signal();
        }
    } else {
        for (int n = 0; n < NUM_BANGS; ++n) {
            sm.spinwait();
            __atomic_fetch_add(&shared, -1, __ATOMIC_SEQ_CST);
            sm.signal();
        }
    }

    printf("Worker %d done\n", worker);

    return NULL;
}


int main() {

    pthread_t id[NUM_TEST_THREADS]; //thread ids

    // create test worker threads
    for(int i = 0; i < NUM_TEST_THREADS; i++)
        pthread_create(&id[i], NULL, fun, (void*)((intptr_t)(i)));

    // join threads to complete their task
    for(int i = 0; i < NUM_TEST_THREADS; i++)
        pthread_join(id[i], NULL);

    //final value of shared variable.  For an odd number of
    // workers this is the loop count, NUM_BANGS
    printf("Test done.  Final value: %d\n", shared);
    const int expected = (NUM_TEST_THREADS & 1) ? NUM_BANGS : 0;
    if (shared == expected) {
        puts("PASS");
    } else {
        printf("Value expected was: %d\nFAIL\n", expected);
    }

    return 0;
}