c - 适当的条件变量用法

Question

我想确保我了解条件变量的工作原理，因此我将使用编写的程序来问我的问题。

在我的程序中，我有一个“生产者”线程(一个)和“ worker 线程”(几个，让我们假设 3 )。

生产者线程“处理”一个FIFO链接列表，这意味着，它只是在列表的开头(由全局指针指向)检查是否有一个项目(在我的程序中请求称为Req类型)。在我的程序中称为前端的指针)，如果是这样，则将其分配给全局请求元素(称为globalReq)。

辅助线程循环运行，通过将全局请求变量提取到自己的本地变量中来等待处理请求(每个线程都是“私有(private)”的，因为每个线程都有一个独立堆栈-如果我输入错误，请更正我)，然后处理该请求。

为此，我将互斥锁与条件变量一起使用。

一个重要的注意事项是，一旦一个请求存在(目前让我们假设只有一个存在)，它就不重要，哪个工作线程将“照顾”它(假设它们都是“自由的”- sleep )在条件变量上)。

在提取请求并将其分配给全局请求之后，生产者线程调用pthread_cond_signal-据我所知，它至少会解除阻塞一个“被阻塞”的线程->因此它可以解除阻塞，例如2个线程。

所以我的问题是关于我当前的代码(如下):

1)如何确保只有一个线程(来自工作线程)可以处理请求。
我是否需要像所有通用的“生产者使用者”实现中一样添加“while check loop”？

2)如何通过pthread_cond_broadcast解除阻塞的线程(或者如果pthread_cond_signal解除阻塞多个线程)，互斥对象的内容，可能我还没有掌握...

(每个)工作线程的代码是:

void *worker(void *arg)
{

       while(1)
       {
         printf("\n BEFORE LOCKING sthread_mutex  with thread: %d \n", syscall(SYS_gettid));
                 pthread_mutex_lock(&sthread_mutex);
         printf("\n AFTER UNLOCKING sthread_mutex  with thread: %d \n", syscall(SYS_gettid));

         printf("\n BEFORE WAITING ON cond_var with thread: %d \n", syscall(SYS_gettid));           
                 pthread_cond_wait(&cond_var,&sthread_mutex); //wait on condition variable
         printf("\n AFTER WAITING ON cond_var with thread: %d \n", syscall(SYS_gettid));            

                 printf("\n got signal for thread: %d \n",syscall(SYS_gettid));

         // extract the current request into g local variable 
             // within the "private stack" of this thread   
                 Req localReq = globalReq;  
                 pthread_mutex_unlock(&sthread_mutex);
         printf("\n AFTER UNLOCKING sthread_mutex with thread: %d \n", syscall(SYS_gettid));

            // perform the desired task
                   task(localReq);
           printf("\n BEFORE calling sem_post with thread: %d \n", syscall(SYS_gettid));
           sem_post(&sem);

       } // end while (1)

} // end of worker thread function

生产者线程的代码是:

void *producer(void *arg)
{
       while(1)
       {

              if(front != NULL)  // queue not empty  
              {
              // go to sleep if all "worker threads" are occuipied 
              // or decrement number of free "worker threads" threads by 1
              printf(" schedualer thread BEFORE calling sem_wait on sem \n");
                          sem_wait(&sem);

              // lock the sthread mutex in order to "synchronize" with the
              // "worker threads"...
              printf(" schedualer thread BEFORE locking sthread_mutex \n");
                          pthread_mutex_lock(&sthread_mutex);
              printf(" schedualer thread AFTER locking sthread_mutex \n");


                          globalReq = extract_element(); // this is the global request variable 

              // notify the worker threads that an "arriving request" needed to 
              // be taking care of
              printf(" schedualer thread BEFORE calling signal on cond_var \n");
                          // pthread_cond_signal(&cond_var);
              pthread_cond_broadcast(&cond_var);
              printf(" schedualer thread AFTER calling signal on cond_var \n");

              // unlock the smutex
              printf(" schedualer thread BEFORE UNLOCKING sthread_mutex \n");
                          pthread_mutex_unlock(&sthread_mutex);
              printf(" schedualer thread AFTER UNLOCKING sthread_mutex \n");

               }  // queue not empty  

               else
                    continue;


        }  // end while (1)

 } // end of producer

另一个问题:

生产者线程在 全局信号量上调用 sem_wait(它以工作线程的数量开始初始化，在本例中为 3 )，以指示自身当前有多少个工作线程正在处理请求，并且，为了完成此“机制”，工作线程一旦完成处理它们“中奖”的请求(当争用条件变量时)，就调用 sem_post指示“另一个工作线程可用”。

3)这是实现这种“发信号通知多少个可用工作线程”的正确方法吗？

4)在生产者和//*段落中提到的工作线程之间共享的 全局变量通过 全局变量“传递”请求有什么优点和缺点？传递它是一种明智的方法，还是最好“仅”创建一个“新请求变量”(使用malloc在堆上)，该变量将“专用”于每个工作线程和请求(并在内部将其释放)每个工作线程处理完请求后)？

5)随意用这段代码可能会引起您的其他注意(好或坏)来告诉我。

编辑:

大家好

一些其他问题:

除了生产者和工作线程之外，还有另一个线程，称为监听器，它的唯一任务是将到达的请求插入到链表(FIFO队列)中，所以这实际上不是生产者的任务之前提到过。

所以我的新问题是:

8)同样，有了有关程序的其他信息，我与信号量组成的“信号机制”是否有效？

9)由生产者和监听器线程管理的链表有两个全局指针front和rear分别指向链表的头和尾(列表的头是要处理的第一个请求)。

下面是监听器线程执行的插入功能和生产者线程执行的“提取”功能的实现。

为了在“队列”(链接列表)上同步这两个线程，我使用了一个在它们之间共享的互斥锁，称为qmutex。

我的问题是，关于下面的2个代码，在每个功能中“放置”互斥锁的“最佳”位置在哪里？

谢谢分配，

伙计

插入功能:

void insertion(void *toInsert)
{

    struct getInfo *req = (struct getInfo *)toInsert;
        newNode = (N*)malloc(sizeof(N));
        newNode->req = req;
        newNode->next = NULL;

    // WHERE SHULD I LOCK (AND UNLOCK) THE  QUEUE MUTEX ????????????????????????
        if(front == NULL)
    {
        front = newNode;
        printf("empty list - insert as head \n");     
    }

        else
    {
        rear->next = newNode;
        printf(" NOT AN EMPTY list - insert as last node \n");
    }

        rear = newNode;
}  // end of insertion

提取功能:

Req extract_element()
{

        if(front == NULL)
            printf("\n empty queue \n");
        else
        {
                Req ret;
                tmpExtract = front;
            ret.socketNum = tmpExtract->req->socketNum;
            ret.type = tmpExtract->req->type;
                printf("\n extracting node with sockNum: %d \n",ret.socketNum);
                front = front->next;
                free(tmpExtract);
                return(ret);
        }
} // end of extract_element

Answer 1

首先，不是直接回答您的问题，而是对执行此操作的典型方法的说明:

您有一种队列或列表，您可以在其中添加工作数据。每当添加一组工作数据时，都首先锁定互斥锁，添加数据，向条件变量发出信号，然后解锁互斥锁。

然后，您的工作线程将锁定互斥锁，并在队列为空时循环等待条件。发送信号后，一个或多个工作人员将醒来，但是(一次)只有一个工作人员将捕获互斥体。互斥锁处于锁定状态时，“优胜者”将检查队列中是否有东西，将其提取出来，解锁互斥锁，然后进行必要的工作。解锁互斥锁后，其他线程也可能会唤醒(如果情况已广播，也会唤醒)，并且将从队列中提取下一个工作，或者在队列为空时返回等待状态。

在代码中，它看起来像这样:

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

#define WORKER_COUNT 3

pthread_mutex_t mutex;
pthread_cond_t cond;
pthread_t workers[WORKER_COUNT];

static int queueSize = 0;

static void *workerFunc(void *arg)
{
    printf("Starting worker %d\n", (int)arg);
    while(1) {
        pthread_mutex_lock(&mutex);
        while(queueSize < 1) {
            pthread_cond_wait(&cond, &mutex);
        }
        printf("Worker %d woke up, processing queue #%d\n", (int)arg, queueSize);
        //Extract work from queue
        --queueSize;
        pthread_mutex_unlock(&mutex);
        //Do work
        sleep(1);
    }
}

int main()
{
    int i;

    pthread_mutex_init(&mutex, 0);
    pthread_cond_init(&cond, 0);

    for(i=0; i<WORKER_COUNT; ++i) {
        pthread_create(&(workers[i]), 0, workerFunc, (void*)(i+1));
    }

    sleep(1);
    pthread_mutex_lock(&mutex);
    //Add work to queue
    queueSize = 5;
    pthread_cond_broadcast(&cond);
    pthread_mutex_unlock(&mutex);

    sleep(10);

    return 0;
}

(我省去了在线程之后进行清理的工作，而将工作人员编号传递给线程既快速又肮脏，但是在这种情况下可以工作)。

在这里，工作人员将被 pthread_cond_broadcast()唤醒，并且只要队列中有东西就将运行(直到 queueSize返回0-想象还有一个实际的队列)，然后再返回等待。

回到问题:

1:互斥量和保护变量(此处为 queueSize)会处理此问题。您还需要保护变量，因为您的线程也可能由于其他原因而被唤醒(所谓的虚假唤醒，请参见 http://linux.die.net/man/3/pthread_cond_wait)。

2:如果调用 pthread_mutex_lock()，则唤醒的线程会像其他任何线程一样在互斥体上竞争。

3:我不确定为什么您需要将可用的工作线程数量发回生产者？

4:生产者和使用者都需要访问队列-但仍然可以通过各种方式将其与函数(或类(如果使用的是C++))一起封装。

5:我希望以上足够了吗？

6: pthread_cond_wait()的问题是它可能会虚假唤醒。也就是说，即使您没有发出信号，它也可能会唤醒。因此，您需要一个保护变量(在我的代码示例中 while()周围的 pthread_cond_wait()循环)，以确保在 pthread_cond_wait()返回时确实有唤醒的理由。然后，使用与条件使用相同的互斥量来保护保护变量(以及需要提取的任何工作数据)，然后可以确定每个工作只有一个线程可以完成。

7:我不会让生产者去 sleep ，而是让它添加可以提取到工作队列的任何数据。如果队列已满，则应进入休眠状态，否则应继续添加内容。

8:使用您的Listener线程，我看不到您为什么甚至需要Producer线程。为什么不让 worker 自己叫 extract_element()？

9:您需要保护对列表变量的所有访问。也就是说，在 insertion()中，在首次访问 front之前锁定互斥锁，在最后一次访问 rear之后将其解锁。 extract_element()中的内容相同-尽管您需要重写此函数以在队列为空时也具有有效的返回值。