使用 pthread 进行简单的流水线操作

Question

最近我开始研究 pthreads 并尝试使用 pthreads 实现软件流水线。为此，我自己编写了一个玩具程序，其中一个类似的程序将成为我主要项目的一部分。

所以在这个程序中，主线程创建整数类型的输入和输出缓冲区，然后创建一个主线程 并将这些缓冲区传递给 主线程。 主线程依次创建两个工作线程。

从 ma​​in 传递到 主线程 的 输入 和 输出缓冲区 的大小nxk（例如 5x10 的 int 大小）。 主线程 对大小为 k（即 10）的块迭代 n（即 5）次。 主线程 中有一个循环运行了 k（此处为 5）次。在 k 的每次迭代中，主线程 对大小为 n 的输入数据的一部分进行一些操作，并将其放在 common 主线程和工作线程之间共享的缓冲区。 主线程然后向工作线程发出信号，表明数据已被放置在公共缓冲区中。

如果公共缓冲区准备好，两个工作线程等待来自主线程的信号。 公共缓冲区上的操作在工作线程之间分成了一半。这意味着一个工作线程将在前半部分工作，而另一个工作线程将在公共缓冲区的下半部分工作。 一旦 工作线程 从 主线程 获得信号，每个 工作线程 对其一半的数据进行一些操作并复制它到输出缓冲区。然后工作线程通过设置标志值通知主线程他们在公共缓冲区上的操作已经完成。为工作线程创建了一组标志。 主线程继续检查是否设置了所有标志，这基本上意味着所有工作线程都完成了对公共缓冲区的操作，所以主线程可以将下一个数据块安全地放入公共缓冲区以供工作线程使用。

因此，主线程和工作线程之间基本上以流水线方式进行通信。最后，我在 主线程 中打印 输出缓冲区。但我根本没有得到任何输出。我在几乎所有步骤上都复制粘贴了带有完整 cmets 的代码。

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <sys/types.h>
#include <sys/time.h>
#include <semaphore.h>
#include <unistd.h>
#include <stdbool.h>
#include <string.h>

#define MthNum 1 //Number of Master threads
#define WthNum 2 //Number of Worker threads
#define times 5 // Number of times the iteration (n in the explanation)
#define elNum 10 //Chunk size during each iteration (k in the explanation)

pthread_mutex_t mutex; // mutex variable declaration
pthread_cond_t cond_var; //conditional variarble declaration
bool completion_flag = true; //This global flag indicates the completion of the worker thread. Turned false once all operation ends
                             //marking the completion
int *commonBuff; //common buffer between master and worker threads
int *commFlags; //array of flags that are turned to 1 by each worker threads. So worker thread i turns commFlags[i] to 1
                // the master thread turns commFlags[i] = 0 for i =0 to (WthNum - 1)
int *commFlags_s;
int counter; // This counter used my master thread to count if all the commFlags[i] that shows
             //all the threads finished their work on the common buffer
// static pthread_barrier_t barrier;
// Arguments structure passed to master thread
typedef struct{
    int *input; // input buffer
    int *output;// output buffer
}master_args;

// Arguments structure passed to worker thread
typedef struct{
    int threadId;
    int *outBuff;
}worker_args;

void* worker_func(void *arguments);
void *master_func(void *);

int main(int argc,char*argv[]){

    int *ipData,*opData;
    int i,j;

    // allocation of input buffer and initializing to 0
    ipData = (int *)malloc(times*elNum*sizeof(int));
    memset(ipData,0,times*elNum*sizeof(int));

    // allocation of output buffer and initializing to 0
    opData = (int *)malloc(times*elNum*sizeof(int));
    memset(opData,0,times*elNum*sizeof(int));

    pthread_t thread[MthNum];
    master_args* args[MthNum];


    //creating the single master thread and passing the arguments
    for( i=0;i<MthNum;i++){
        args[i] = (master_args *)malloc(sizeof(master_args));
        args[i]->input= ipData;
        args[i]->output= opData;
        pthread_create(&thread[i],NULL,master_func,(void *)args[i]);
    }

    //joining the master thred
    for(i=0;i<MthNum;i++){
        pthread_join(thread[i],NULL);
    }

    //printing the output buffer values
    for(j =0;j<times;j++ ){
        for(i =0;i<elNum;i++){
            printf("%d\t",opData[i+j*times]);
        }
      printf("\n");
    }

    return 0;
}

//This is the master thread function
void *master_func(void *arguments){

    //copying the arguments pointer to local variables
    master_args* localMasterArgs = (master_args *)arguments;
    int *indataArgs = localMasterArgs->input; //input buffer
    int *outdataArgs = localMasterArgs->output; //output buffer

    //worker thread declaration
    pthread_t Workers[WthNum];
    //worker thread arguments declaration
    worker_args* wArguments[WthNum];
    int i,j;

    pthread_mutex_init(&mutex, NULL);
    pthread_cond_init (&cond_var, NULL);
    counter =0;

    commonBuff = (int *)malloc(elNum*sizeof(int));

    commFlags = (int *)malloc(WthNum*sizeof(int));
    memset(commFlags,0,WthNum*sizeof(int) );
    commFlags_s= (int *)malloc(WthNum*sizeof(int));
    memset(commFlags_s,0,WthNum*sizeof(int) );

    for(i =0;i<WthNum;i++){

        wArguments[i] = (worker_args* )malloc(sizeof(worker_args));
        wArguments[i]->threadId = i;
        wArguments[i]->outBuff = outdataArgs;

        pthread_create(&Workers[i],NULL,worker_func,(void *)wArguments[i]);
    }

    for (i = 0; i < times; i++) {
        for (j = 0; j < elNum; j++)
            indataArgs[i + j * elNum] = i + j;

        while (counter != 0) {
            counter = 0;

            pthread_mutex_lock(&mutex);
            for (j = 0; j < WthNum; j++) {
                counter += commFlags_s[j];
            }
            pthread_mutex_unlock(&mutex);

        }
        pthread_mutex_lock(&mutex);
        memcpy(commonBuff, &indataArgs[i * elNum], sizeof(int));
        pthread_mutex_unlock(&mutex);
        counter = 1;
        while (counter != 0) {
            counter = 0;

            pthread_mutex_lock(&mutex);
            for (j = 0; j < WthNum; j++) {
                counter += commFlags[j];
            }
            pthread_mutex_unlock(&mutex);


        }
        // printf("master broad cast\n");
        pthread_mutex_lock(&mutex);
        pthread_cond_broadcast(&cond_var);
         //releasing the lock
        pthread_mutex_unlock(&mutex);

    }

    pthread_mutex_lock(&mutex);
     completion_flag = false;
    pthread_mutex_unlock(&mutex);

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

    pthread_mutex_destroy(&mutex);
    pthread_cond_destroy(&cond_var);

    return NULL;
}


void* worker_func(void *arguments){

    worker_args* localArgs = (worker_args*)arguments;

    //copying the thread ID and the output buffer
    int tid = localArgs->threadId;
    int *localopBuffer = localArgs->outBuff;
    int i,j;
    bool local_completion_flag=false;

    while(local_completion_flag){

        pthread_mutex_lock(&mutex);
        commFlags[tid] =0;
        commFlags_s[tid] =1;
        pthread_cond_wait(&cond_var,&mutex);
        commFlags_s[tid] =0;
        commFlags[tid] =1;
        if (tid == 0) {
            for (i = 0; i < (elNum / 2); i++) {
                localopBuffer[i] = commonBuff[i] * 5;
            }
        } else { // Thread ID 1 operating on the other half of the common buffer data and placing on the
                 // output buffer
            for (i = 0; i < (elNum / 2); i++) {
                localopBuffer[elNum / 2 + i] = commonBuff[elNum / 2 + i] * 10;
            }
        }
         local_completion_flag=completion_flag;
        pthread_mutex_unlock(&mutex);//releasing the lock

    }

    return NULL;
}

但我不知道我在实施中哪里做错了，因为从逻辑上讲它似乎是正确的。但是我的实现肯定有问题。我花了很长时间尝试不同的方法来修复它，但没有任何效果。很抱歉这篇长篇文章，但我无法确定我可能做错的部分，所以我无法简明扼要。因此，如果有人可以查看问题和实施并建议需要进行哪些更改才能按预期运行它，那将非常有帮助。感谢您的帮助和帮助。

Answer 1

这段代码有几个错误。

1. 您可以从修复工作线程的创建开始：

   wArguments[i] = (worker_args* )malloc(sizeof(worker_args));
   wArguments[i]->threadId = i;
   wArguments[i]->outBuff = outdataArgs;
   
   pthread_create(&Workers[i],NULL,worker_func, (void *)wArguments);
   

您正在初始化 worker_args 结构但不正确 - 将指针传递给数组 (void *)wArguments 而不是指向您刚刚初始化的数组元素的指针。

pthread_create(&Workers[i],NULL,worker_func, (void *)wArguments[i]);
//                                                             ^^^

1. 在启动使用它的值的线程之前初始化计数器：

   void *master_func(void *arguments)
   {
   /* (...) */
   pthread_mutex_init(&mutex, NULL);
   pthread_cond_init (&cond_var, NULL);
   counter = WthNum;
   

2. 在启动主线程时，您错误地将指针传递给指针：

   pthread_create(&thread[i],NULL,master_func,(void *)&args[i]);
   

请将其更改为：

pthread_create(&thread[i],NULL,master_func,(void *) args[i]);

4. 对counter 变量的所有访问（与任何其他共享内存一样）必须在线程之间同步。

Answer 2

我认为你应该像这样使用基于信号量的生产者-消费者模型

https://jlmedina123.wordpress.com/2014/04/08/255/