c - 具有不同功能的线程之间的互斥锁

Question

问题:类似于我的其他问题之一 Other Question我正在尝试用 C 语言创建一个程序，该程序允许我使用可变数量的线程搜索 10 个文本文件以找到最大的素数。它还应该有一个管理器线程，允许读取工作线程的最大素数(而不是修改它)。管理器线程还发布所有工作线程找到的最大质数，以便工作线程可以读取并使用它。工作线程必须将其本地最大素数发布到全局数组 (privateLargestPrime)，并且在执行此操作之前，必须锁定它，以便管理器线程在工作线程更新它之前不会读取它。

奇怪的部分:当我单步执行程序时，当工作线程想要调用锁时，它会将线程切换到管理器，管理器调用锁并被授予锁，然后它继续循环，使工作线程挨饿。我不确定这是怎么回事。如果我能对这个问题有任何见解，我将不胜感激。

/*
 * The Reason for Worker Initialization + Manager Initialization is that we need both types of threads to exist at the same time
 * so I just combined them into one loop, although I believe that they could have been created seperatly.
 * Basically just call pthread_Join at the end
 */

    #include <stdio.h>
    #include <stdlib.h>
    #include <math.h>
    #include <pthread.h>
    #include <time.h>
    #include <string.h>
    #include <fileTest.h>

    clock_t Start, End;
    double elapsed = 0;
    pthread_cond_t managerVar;
    pthread_mutex_t mutex;
    unsigned int globalLargestPrime = 0;
    int numThreads = 1;//Number of Threads
    int LINES_PER_THREAD;
    pthread_cond_t *WorkerConditionaVar;
    pthread_cond_t *ManagerConditionaVar;
    unsigned int *privateLocalLargest;//will need to be changed
    int *statusArray;


    FILE *fileOut;
    typedef enum{
      FREE,
      IN_USE
    }lrgstPrm;
    lrgstPrm monLargestPrime;//create enum
    lrgstPrm workerLargestPrime;//create enum

    typedef enum{
      Finished,
      Not_Finished
    }Status;
    Status is_Finished;

    typedef struct threadFields{
      int id;
      int StartPos;//gets seek for worker thread
      int EndPos;
    }tField;



    int ChkPrim(unsigned int n){
      unsigned int i;
      unsigned int root = sqrt(n);
      for(i=2; i<root; i++){
            if(n % i == 0)
              return 0;

         }
      //printf("%d \n", isPrime);
      return 1;
    }

    void *Worker(void *threadStruct){//Create Threads
      struct threadFields *info = threadStruct;
      int index;
      int id = info->id;
      unsigned int currentNum = 0;
      int Seek = info->StartPos;
      unsigned int localLargestPrime = 0;
      char *buffer = malloc(50);
      int isPrime = 0;

        while(Seek<info->EndPos){
        for(index = 0; index < 1000; index++){//Loop 1000 times
        fseek(fileOut,Seek*sizeof(char)*20, SEEK_SET);
        fgets(buffer,20,fileOut);
        Seek++;
        currentNum = atoi(buffer);
        if(currentNum>localLargestPrime && currentNum > 0){
          isPrime = ChkPrim(currentNum);
          if( isPrime == 1)
            localLargestPrime = currentNum;
        }
      }

        //while(monLargestPrime == IN_USE)
        //pthread_cond_wait(&monitor[id], &mutex);//wait untill mutex is unlocked
        //monLargestPrime = IN_USE;
        //Critical Zone
        //printf("Entering Critical Zone My ID: %d\n",id);

        /*Should Lock the Private Largest Prime from any other thread using it*/
        if(pthread_mutex_lock(&mutex) != 0)//Lock
          printf("Failed To Lock");
        while(workerLargestPrime == IN_USE)//Wait untill Workers largest prime is free
          pthread_cond_wait(ManagerConditionaVar, &mutex);
        workerLargestPrime = IN_USE;//Local Largest is in use
        privateLocalLargest[id] = localLargestPrime;//Assign Local Largest to each workers Shared Variable
        workerLargestPrime = FREE;
        pthread_cond_signal(ManagerConditionaVar);//Signal to any waiting thread that wants to touch(read) this workers privateLocalLargest
        pthread_mutex_unlock(&mutex);



        /*
        pthread_mutex_lock(&mutex);
       while(workerLargestPrime == FREE){
         workerLargestPrime = IN_USE;
         //pthread_cond_wait(&managerVar,&mutex);
        */
        if(localLargestPrime < globalLargestPrime)
          localLargestPrime = globalLargestPrime;
        /*
       workerLargestPrime = FREE;
       pthread_mutex_unlock(&mutex);

       // for(index = 0; index < numThreads; index++)
         // if(index != id)
           // pthread_cond_signal(&monitor[id]);//signal all threads that mutex is unlocked
        //monLargestPrime = FREE;
        //printf("Exiting Critical Zone My ID: %d\n",id);
    */
        //pthread_mutex_unlock(&mutex);
       }//End of While
    statusArray[id] = 1;
    void *i = 0;
    return i;
    }

    void *manager(){
        int index, MlocalLargestPrime;


        while(is_Finished==Not_Finished){
        /*Should Lock the Private Largest Prime from any other thread using it*/
            if(pthread_mutex_lock(&mutex) != 0)//Lock
              printf("Failed To Lock");
            while(workerLargestPrime == IN_USE)//Wait untill Workers largest prime is free
              pthread_cond_wait(ManagerConditionaVar, &mutex);
            workerLargestPrime = IN_USE;//Local Largest is in use
            //Critical Zone
            for(index =0; index < numThreads; index++)
                  if(privateLocalLargest[index] > MlocalLargestPrime)
                    MlocalLargestPrime = privateLocalLargest[index];
            //Critical Zone
            workerLargestPrime = FREE;
            pthread_cond_signal(ManagerConditionaVar);//Signal to any waiting thread that wants to touch(read) this workers privateLocalLargest
            pthread_mutex_unlock(&mutex);
    /*
       pthread_mutex_lock(&mutex);
       while(workerLargestPrime == FREE){
         workerLargestPrime = IN_USE;
         globalLargestPrime = MlocalLargestPrime;
         workerLargestPrime = FREE;
         pthread_cond_signal(&managerVar);

       }
       pthread_mutex_unlock(&mutex);
    */
       /*check if workers have finished*/
    for(index = 0; index < numThreads; index++)
      if(statusArray[index] == 0)
        is_Finished = Not_Finished;
    }
        void *i = 0;
        return i;
       }

    int main(){
      //setFile();
      LINES_PER_THREAD = (getLineNum()/numThreads);
      fileOut = fopen("TextFiles/dataBin.txt", "rb");
      Start = clock();
      //pthread_t managerThread;
      pthread_t threads[numThreads];
      pthread_cond_t monitor[numThreads];
      pthread_cond_t managerCon;
      WorkerConditionaVar = monitor;//Global Pointer points to the array created in main
      ManagerConditionaVar = &managerCon;
      unsigned int WorkerSharedVar[numThreads];
      privateLocalLargest = WorkerSharedVar;
      pthread_mutex_init(&mutex, NULL);
      int finishedArr[numThreads];
      statusArray = finishedArr;
      is_Finished = Not_Finished;
      int index;


          /*Worker Initialization + Manager Initialization*/
          pthread_cond_init(&managerCon,NULL);
          /*Worker Thread Struct Initalization*/
          tField *threadFields[numThreads];//sets number of thread structs
          rewind(fileOut);
          for(index = 0; index < numThreads; index++){//run through threads; inizilize the Struct for workers
              pthread_cond_init(&monitor[index], NULL);//Initialize all the conditional variables
              threadFields[index] = malloc(sizeof(tField));
              threadFields[index]->id = index;
              threadFields[index]->StartPos = index*LINES_PER_THREAD;// Get Position for start of block
              threadFields[index]->EndPos = (index+1)*LINES_PER_THREAD-1;// Get Position for end of block
        }
          /*Worker Thread Struct Initalization*/

          for(index = 0; index<numThreads+1; index++)
            if(index == numThreads)//Last Thread is Manager Thread
              pthread_create(&threads[index],NULL,manager,NULL);//Create Manager
            else//Worker Threads
              pthread_create(&threads[index],NULL,Worker,(void *)threadFields[index]);//Pass struct to each worker

          for(index = 0; index<numThreads+1; index++)
            pthread_join(threads[index], NULL);
          /*Worker Initialization + Manager Initialization*/


      /*Destroy the mutexes & conditional signals*/
          for(index = 0; index < numThreads; index++){
             pthread_cond_destroy(&WorkerConditionaVar[index]);
             }
          pthread_cond_destroy(&managerCon);

      pthread_mutex_destroy(&mutex);
    End = clock();
    elapsed = ((double) (End - Start)) / CLOCKS_PER_SEC;

    printf("This is the Time %f\n", elapsed);
    printf("This is the Largest Prime Number: %u", globalLargestPrime);
    return 0;
    }



  [1]: https://stackoverflow.com/questions/13672456/slightly-complicated-thread-synchronization

还有另一个C源代码，我只使用了一种方法，它是给我10个文本文件的行数，我也会发布它(但不是必需的):

/*
 * fileTest.c
 *
 *  Created on: Dec 8, 2012
 *      Author: kevin
 *
 *  count number of lines
 *  divide by number of threads
 *  get the positions to hand to each thread
 *  to get positions, one needs to get the number of lines per thread,
 *  add number of lines to each: Seek*sizeof(char)*10, SEEK_SET.
 *  and hand out these positions to each thread
 *
 */

#include <stdio.h>
#include <stdlib.h>
#include <math.h>
#include <pthread.h>
#include <time.h>
#include <string.h>

  FILE *filesIn[10], *fileOut;
  int Seek;

void createText(){
  FILE *fOUT = fopen("data9.txt", "w");
  int i;
  srand(time(NULL));
  for(i=0; i<10000; i++)
  fprintf(fOUT, "%d\n",rand()%9000);
  fclose(fOUT);
}

void setFile(){
  int index;
  Seek = 0;
  char *buffer = malloc(50);
  filesIn[0] = fopen("TextFiles/primes1.txt", "r");//read Text
  filesIn[1] = fopen("TextFiles/primes2.txt", "r");//read Text
  filesIn[2] = fopen("TextFiles/primes3.txt", "r");//read Text
  filesIn[3] = fopen("TextFiles/primes4.txt", "r");//read Text
  filesIn[4] = fopen("TextFiles/primes5.txt", "r");//read Text
  filesIn[5] = fopen("TextFiles/primes6.txt", "r");//read Text
  filesIn[6] = fopen("TextFiles/primes7.txt", "r");//read Text
  filesIn[7] = fopen("TextFiles/primes8.txt", "r");//read Text
  filesIn[8] = fopen("TextFiles/primes9.txt", "r");//read Text
  filesIn[9] = fopen("TextFiles/primes10.txt", "r");//read Text
  fileOut = fopen("TextFiles/dataBin.txt", "wb");//write in bin

for(index = 0; index < 10; index++)//Run through 10 files
  while(!feof(filesIn[index])){

    fscanf(filesIn[index],"%s", buffer);//take line from input
    fseek(fileOut,Seek*sizeof(char)*20, SEEK_SET);
    fputs(buffer,fileOut);//Print line to output file
    Seek++;
  }
  fclose(filesIn[0]);
  fclose(filesIn[1]);
  fclose(filesIn[2]);
  fclose(filesIn[3]);
  fclose(filesIn[4]);
  fclose(filesIn[5]);
  fclose(filesIn[6]);
  fclose(filesIn[7]);
  fclose(filesIn[8]);
  fclose(filesIn[9]);
  fclose(fileOut);
}

void getFile(){
  int Seek = 0;
  int currentSeek = 0;
  int currentNum = 0;
  int localLargestPrime = 0;
  char *buffer = malloc(50);
  fileOut = fopen("TextFiles/dataBin.txt", "rb");
  rewind(fileOut);
  while(!feof(fileOut)){
  fseek(fileOut,Seek*sizeof(char)*20, SEEK_SET);
  fgets(buffer,10,fileOut);
  Seek++;
  currentNum = atoi(buffer);
  if(currentNum>localLargestPrime)
        if(ChkPrim(currentNum) == 1){
          localLargestPrime = currentNum;
          currentSeek = Seek*sizeof(char)*20;
          printf("the current seek is: %d\n", currentSeek);
        }
  }
  printf("This is the largest Prime: %d\n", localLargestPrime);
}

int getLineNum(){
  Seek = 0;
  int index;
  char c;
  filesIn[0] = fopen("TextFiles/primes1.txt", "r");//read Text
  filesIn[1] = fopen("TextFiles/primes2.txt", "r");//read Text
  filesIn[2] = fopen("TextFiles/primes3.txt", "r");//read Text
  filesIn[3] = fopen("TextFiles/primes4.txt", "r");//read Text
  filesIn[4] = fopen("TextFiles/primes5.txt", "r");//read Text
  filesIn[5] = fopen("TextFiles/primes6.txt", "r");//read Text
  filesIn[6] = fopen("TextFiles/primes7.txt", "r");//read Text
  filesIn[7] = fopen("TextFiles/primes8.txt", "r");//read Text
  filesIn[8] = fopen("TextFiles/primes9.txt", "r");//read Text
  filesIn[9] = fopen("TextFiles/primes10.txt", "r");//read Text
  for(index = 0; index < 10; index++)
    while((c = fgetc(filesIn[index])) != EOF)
      if(c == '\n')
        Seek++;

  return Seek;
}

enter link description here

Answer 1

您似乎过度同步访问 globalLargestPrime。但是，与其尝试解决这个问题，可能有更好的方法将每个线程的值传递给管理器 - 只需让线程函数将它找到的值返回为 unsigned int 转换为 void *。然后，管理器只需等待每个线程的 pthread_join() 完成即可收集这些值。

类似下面的伪代码:

void *Worker(void *threadStruct)
{
    unsigned int largest_prime;

    // do whatever you need to do to find the largest prime in the set of numbers
    //  this thread has to deal with
    //
    // Note that nothing here should require synchronization, since the data should be
    //  completely independent of other threads

    return (void*) largest_prime;
}

 void *manager()
 {
    unsigned int largest_prime = 0;

    // do whatever to spin up the threads and keep track of them in a
    //  pthread_t[] array...

    // now wait for the threads to finish up and keep deal with the value
    //  each thread has found:
    for each (pthread* p in the pthread_t[]) { // remember - pseudo code
        void* result = 0;

        // get the result that thread found
        pthread_join( p, &result);
        unsigned int thread_prime = (unsigned int) result;

        if (largest_prime < thread_prime) {
            largest_prime = thread_prime;
        }
    }

    printf("largest prime: %u\n", largest_prime);
 }

现在所有同步麻烦都由 pthread_join() 处理。