Расовые условия в потоках, несмотря на блокировку мьютекса - PullRequest
Новая
       33

Расовые условия в потоках, несмотря на блокировку мьютекса

2 голосов
/ 12 октября 2019

Этот код вычисляет сумму всех целых чисел в массиве, равномерно распределяя работу между несколькими потоками. Однако время от времени номера потоков, а также их локальные значения перепутаны. Я предполагаю, что это потому, что void * param и globalindex доступны для нескольких потоков одновременно. Что не имеет смысла, так это то, что это происходит, даже если я блокировал все глобальные переменные в этом коде. Как я могу это исправить? 

#include<string>
#include<iostream>
#include<fstream>
#include<unistd.h>
#include<pthread.h>
#include<stdlib.h>
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

using namespace std;
int y =0;
int sum=0;
int array[1000000];
int x=0;
int leftoverHandle = 0;
int globalindex = 0;  
int eachThreadHandles =0;

void* add(void* param){
    pthread_mutex_lock(&mutex);
    int localindexup = globalindex + eachThreadHandles;
    int localindexdown = globalindex;
    int localsum=0;
    long localparam = (long)param;

    if(y != leftoverHandle ){
            localindexup++;
            y++;

    }
    pthread_mutex_unlock(&mutex);


    while(localindexdown<localindexup){

            pthread_mutex_lock(&mutex);
            sum = sum+array[localindexdown];
            localsum = localsum+array[localindexdown];
            localindexdown++;
            pthread_mutex_unlock(&mutex);


    }




    pthread_mutex_lock(&mutex);
    globalindex = localindexdown;
    printf("Thread %ld", localparam);
    printf(": %d\n", localsum);
    pthread_mutex_unlock(&mutex);
}

int main(int argc, char ** argv){
    if(argc != 3){
            cout<<"Incorrect number of argument";
            exit(1);
    }
    string line;
    string f = argv[1];
    const char *filename = f.c_str();



    int maxthreads = atoi(argv[2]);

    FILE* inFile = fopen(filename,"r");
    int i=0;
    if(inFile == NULL){
    cout<<"fopen failed"<<endl;
    }

    fscanf(inFile, "%d",&i);

    while(!feof(inFile)){

    array[x]=i;
    x +=1;

    fscanf(inFile,"%d",&i);

    }

    fclose(inFile);

    pthread_t id[maxthreads];

    leftoverHandle = x%maxthreads;
    eachThreadHandles = (x - leftoverHandle)/maxthreads;


    for(long i=0; i< maxthreads;i++){
            long status = pthread_create(&id[i], NULL, add, (void*) i);
            if(status){
                    printf("Error creating thread! \n");
                    exit(0);
            }
    }

    for(long i=0; i<maxthreads;i++){
    pthread_join(id[i], NULL);
    }


    cout<<"Sum="<<sum<<endl;




return 0;
}

1 Ответ

1 голос
/ 12 октября 2019

Проблема в том, что вы не обновляете globalindex сразу после инициализации localindexup и localindexdown для каждого потока, т. Е. В первом критическом разделе.
В вашем коде есть три критических раздела.
Представьте, что thread0 запускает первый критический раздел, затем thread1 выгружает thread0 сразу после того, как thread0 снимает блокировку первого критического раздела. Но поскольку вы устанавливаете globalindex в localindexdown в третьем критическом разделе, а не в первом, thread1 по-прежнему будет видеть globalindex=0, точно так же как thread0, поэтому он будет пересчитывать ту же сумму, что и thread0. Вы должны поместить globalindex = localindexdown; в первый критический раздел.
На самом деле вообще нет необходимости в третьем критическом разделе: 

    pthread_mutex_lock(&mutex);
    int localindexup = globalindex + eachThreadHandles;
    int localindexdown = globalindex;
    int localsum=0;
    long localparam = (long)param;

    if(y != leftoverHandle ){
            localindexup++;
            y++;

    }
    globalindex = localindexdown; //<--- MOVED HERE
    pthread_mutex_unlock(&mutex);

И, забыв мой комментарий о цикле, я допустил ошибку:while(localindexdown<localindexup) можно безопасно прервать, потому что переменные не являются общими для потоков. Вы можете просто немного повысить производительность, сократив область мьютекса, включив в нее только общие данные: 

    while(localindexdown<localindexup)
    {
            pthread_mutex_lock(&mutex);
            sum = sum+array[localindexdown];
            pthread_mutex_unlock(&mutex);     //<--- MOVED HERE
            localsum = localsum+array[localindexdown];
            localindexdown++;
    }
...