title: Simple pipelining using pthreads
body: <p>Very recently I have started working on pthreads and trying to implement software pipelining with pthreads. To do that I have a written a toy program myself, a similar of which would be a part of my main project. </p>
  
  <p>So in this program the <strong>main thread</strong> creates and <em>input</em> and <em>output</em> buffer of integer type and then creates a single <strong>master thread</strong> and passes those buffers to the <strong>master thread</strong>. The <strong>master thread</strong> in turn creates two <strong>worker threads</strong>. </p>
  
  <p>The <em>input</em> and the <em>output buffer</em> that is passed from the <strong>main</strong> to the <strong>master thread</strong> is of size <strong>n</strong>x<strong>k</strong> (e.g. 5x10 of size int). The <strong>master thread</strong> iterates over a chunk of size <strong>k</strong> (i.e. 10) for <strong>n</strong> (i.e. 5) number of times. 
  There is a loop running in the <strong>master thread</strong> for <strong>k</strong> (5 in here) number of times. In each iteration of <strong>k</strong> the <strong>master thread</strong> does some operation on a portion of input data of size <strong>n</strong> and place it in the <em>common buffer</em> shared between the <strong>master</strong> and the <strong>worker threads</strong>. The <strong>master thread</strong> then signals the worker threads that the data has been placed in the <em>common buffer</em>. </p>
  
  <p>The two <strong>worker threads</strong> waits for the signal from the <strong>master thread</strong> if the <em>common buffer</em> is ready. The operation on the <em>common buffer</em> is divided into half among the <strong>worker threads</strong>. Which means one <strong>worker thread</strong> would work on the first half and the other <strong>worker thread</strong> would work on the next half of the <em>common buffer</em>. 
  Once the <strong>worker threads</strong> gets the signal from the <strong>master thread</strong>, each of the <strong>worker thread</strong> does some operation on their half of the data and copy it to the <em>output buffer</em>. Then the <strong>worker threads</strong> informs the <strong>master thread</strong> that their operation is complete on the <em>common buffer</em> by setting flag values. An array of flags are created for  <strong>worker threads</strong>. The <strong>master thread</strong> keeps on checking if all the flags are set which basically means all the <strong>worker threads</strong> finished their operation on the <em>common buffer</em> and so <strong>master thread</strong> can place the next data chunk into the <em>common buffer</em> safely for <strong>worker thread's</strong> consumption.</p>
  
  <p>So essentially there is communication between the <strong>master</strong> and the <strong>worker threads</strong> in a pipelined fashion. In the very end I am printing the <em>output buffer</em> in the <strong>main thread</strong>. But I am getting no output at all. I have copy pasted my code with full comments on almost all steps.</p>
  
  <pre><code>#include &lt;stdio.h&gt;
  #include &lt;stdlib.h&gt;
  #include &lt;pthread.h&gt;
  #include &lt;sys/types.h&gt;
  #include &lt;sys/time.h&gt;
  #include &lt;semaphore.h&gt;
  #include &lt;unistd.h&gt;
  #include &lt;stdbool.h&gt;
  #include &lt;string.h&gt;
  
  #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&lt;MthNum;i++){
          args[i] = (master_args *)malloc(sizeof(master_args));
          args[i]-&gt;input= ipData;
          args[i]-&gt;output= opData;
          pthread_create(&amp;thread[i],NULL,master_func,(void *)args[i]);
      }
  
      //joining the master thred
      for(i=0;i&lt;MthNum;i++){
          pthread_join(thread[i],NULL);
      }
  
      //printing the output buffer values
      for(j =0;j&lt;times;j++ ){
          for(i =0;i&lt;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-&gt;input; //input buffer
      int *outdataArgs = localMasterArgs-&gt;output; //output buffer
  
      //worker thread declaration
      pthread_t Workers[WthNum];
      //worker thread arguments declaration
      worker_args* wArguments[WthNum];
      int i,j;
  
      pthread_mutex_init(&amp;mutex, NULL);
      pthread_cond_init (&amp;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&lt;WthNum;i++){
  
          wArguments[i] = (worker_args* )malloc(sizeof(worker_args));
          wArguments[i]-&gt;threadId = i;
          wArguments[i]-&gt;outBuff = outdataArgs;
  
          pthread_create(&amp;Workers[i],NULL,worker_func,(void *)wArguments[i]);
      }
  
      for (i = 0; i &lt; times; i++) {
          for (j = 0; j &lt; elNum; j++)
              indataArgs[i + j * elNum] = i + j;
  
          while (counter != 0) {
              counter = 0;
  
              pthread_mutex_lock(&amp;mutex);
              for (j = 0; j &lt; WthNum; j++) {
                  counter += commFlags_s[j];
              }
              pthread_mutex_unlock(&amp;mutex);
  
          }
          pthread_mutex_lock(&amp;mutex);
          memcpy(commonBuff, &amp;indataArgs[i * elNum], sizeof(int));
          pthread_mutex_unlock(&amp;mutex);
          counter = 1;
          while (counter != 0) {
              counter = 0;
  
              pthread_mutex_lock(&amp;mutex);
              for (j = 0; j &lt; WthNum; j++) {
                  counter += commFlags[j];
              }
              pthread_mutex_unlock(&amp;mutex);
  
  
          }
          // printf("master broad cast\n");
          pthread_mutex_lock(&amp;mutex);
          pthread_cond_broadcast(&amp;cond_var);
           //releasing the lock
          pthread_mutex_unlock(&amp;mutex);
  
      }
  
      pthread_mutex_lock(&amp;mutex);
       completion_flag = false;
      pthread_mutex_unlock(&amp;mutex);
  
      for (i = 0; i &lt; WthNum; i++) {
          pthread_join(Workers[i], NULL);
      }
  
      pthread_mutex_destroy(&amp;mutex);
      pthread_cond_destroy(&amp;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-&gt;threadId;
      int *localopBuffer = localArgs-&gt;outBuff;
      int i,j;
      bool local_completion_flag=false;
  
      while(local_completion_flag){
  
          pthread_mutex_lock(&amp;mutex);
          commFlags[tid] =0;
          commFlags_s[tid] =1;
          pthread_cond_wait(&amp;cond_var,&amp;mutex);
          commFlags_s[tid] =0;
          commFlags[tid] =1;
          if (tid == 0) {
              for (i = 0; i &lt; (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 &lt; (elNum / 2); i++) {
                  localopBuffer[elNum / 2 + i] = commonBuff[elNum / 2 + i] * 10;
              }
          }
           local_completion_flag=completion_flag;
          pthread_mutex_unlock(&amp;mutex);//releasing the lock
  
      }
  
      return NULL;
  }
  </code></pre>
  
  <p>But I have no idea where I have done wrong in my implementation since logically it seems to be correct. But definitely there is something wrong in my implementation. I have spent a long time trying different things to fix it but nothing worked. Sorry for this long post but I am unable to determine a section where I might have done wrong and so I couldn't concise the post. So if anybody could take a look into the problem and implementation and can suggest what changes needed to be done to run it as intended then that it would be really helpful. Thank you for your help and assistance.</p>