Ошибка сегментации при использовании pthreads недетерминированным образом

Question

Проблема в том, что когда я запускаю приведенный ниже код на одном ядре, иногда он работает правильно, а иногда я получаю ошибку сегментации. Вероятно, эта проблема будет возникать чаще на многоядерной машине. Мне нужно знать, где этот недетерминизм присутствует в моей программе, и как я могу его устранить. Спасибо.

int numThreads = 4;

class Evaluator;

struct E {
    Evaluator* evaluator;
    int id;
};

class Evaluator {
public:
    pthread_t * threads;
    sem_t* fork_sync;
    sem_t* join_sync;
    int tin;
    pthread_mutex_t tin_mut;

    double * d;
    int sz;
    int cursor;
    pthread_mutex_t c_mut;

    Evaluator(sem_t* fs, sem_t* js) {
        fork_sync = fs;
        join_sync = js;
        threads = new pthread_t[numThreads];
        tin = 0;
        pthread_mutex_init(&tin_mut,NULL);
        for(int i=0 ;i<numThreads; i++) {
            E arg;
            arg.evaluator = this;
            arg.id = i;
            pthread_create(&threads[i],NULL,(void* (*) (void*) )func,(void*)&arg);
        }


        //dummy init
        sz = 20;
        d = new double[sz];
        for(int i=0; i<sz ; i++) d[i] = .5 + i;
        cursor = 0;
        pthread_mutex_init(&c_mut,NULL);
    }

    static void func(E* e) {        
        Evaluator* eval = e -> evaluator;
        eval -> go(e -> id);
    }

    void reset() {
        cursor = 0;
    }

    void go(int id) {
        while(1) {
            sem_wait(fork_sync);

            pthread_mutex_lock(&tin_mut);
            ++tin;
            pthread_mutex_unlock(&tin_mut);

            while(1) {
                int idx;
                pthread_mutex_lock(&c_mut);
                idx = cursor++;
                pthread_mutex_unlock(&c_mut);
                if(idx >= sz ) break;
                // do the evaluation
                cout << "evaluating  index " << idx << " using thread " << id << endl;
            }

            int remain;
            pthread_mutex_lock(&tin_mut);
            remain = --tin;
            pthread_mutex_unlock(&tin_mut);
            if(remain == 0) sem_post(join_sync);
        }
    }


};

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

    sem_t fork_sync;
    sem_t join_sync;

    sem_init(&fork_sync,0,0);
    sem_init(&join_sync,0,0);

    Evaluator e(&fork_sync,&join_sync);


    //evaluating t times
    int t = 3;
    for(int i=0; i<t; i++) {
        cout << "---------- evaluation number :" << i << endl;
        e.reset();
        for(int j=0; j<numThreads; j++) sem_post(&fork_sync);
        sem_wait(&join_sync);
        cout << endl;
    }

    return 0;
}

Answer 1

Вот быстрое исправление второй ошибки. Это гарантирует, что рабочий поток завершает работу до запуска другой итерации.

--- a/misc/so/sem_wait/q.cpp
+++ b/misc/so/sem_wait/q.cpp
@@ -83,7 +83,7 @@ public:
             pthread_mutex_lock(&tin_mut);
             remain = --tin;
             pthread_mutex_unlock(&tin_mut);
-            if(remain == 0) sem_post(join_sync);
+            sem_post(join_sync);
         }
     }

@@ -107,9 +107,11 @@ int main(int argc, char *argv[]) {
         cout << "---------- evaluation number :" << i << endl;
         e.reset();
         for(int j=0; j<numThreads; j++) sem_post(&fork_sync);
-        sem_wait(&join_sync);
+        for(int j=0; j<numThreads; j++) sem_wait(&join_sync);
         cout << endl;
     }

+    cout << "exit" << endl;
+
     return 0;
 }

Answer 2

arg находится в стеке. Вы берете его адрес и передаете этот адрес другому потоку. Состояние гонки (значение в стеке может быть перезаписано до того, как вновь созданный поток его прочитает).

E arg;
arg.evaluator = this;
arg.id = i;
pthread_create(&threads[i],NULL,(void* (*) (void*) )func,(void*)&arg);

Решение:

E* arg = new E();
arg->evaluator = this;
arg->id = i;
pthread_create(&threads[i],NULL,(void* (*) (void*) )func,(void*)arg);

И не забудьте delete e в func,

Answer 3

Навид, не могли бы вы привести пример, который выйдет из коробки в следующий раз?

не так уж и сложно добавить следующие строки вверху на вашем примере

#include <pthread.h>
#include <semaphore.h>
#include <iostream>

using namespace std;

// compile with: g++ -g -pthread main.cpp -o main -lrt -lpthread

Когда я запускаю программу в отладчике, она действительно иногда падает в строке sem_wait() (а иногда не падает!)

void go(int id) {
    while(1) {
        sem_wait(fork_sync); //  <--- seems to crash here
        ...

Answer 4

Адрес вашего объекта будет поврежден. Это вызвано размещением элемента args в стеке. Когда потоки запускаются, он может содержать или не содержать допустимые значения.
Это поддерживает ответ Vokuhila-Oliba, поскольку fork_sync - это первый раз, когда поток пытается получить доступ к памяти объектов.
редактировать
У меня работает код со следующими изменениями (20 тестов без сбоев)

 
for(int i=0 ;i<numThreads; i++) {
                            E* arg = new E;
                            arg->evaluator = this;
                            arg->id = i;
                            pthread_create(&threads[i],NULL,func,arg);
                        }

static void* func(void* e) {
        Evaluator* eval = reinterpret_cast<E*>(e) -> evaluator;
        eval -> go(reinterpret_cast<E*>(e) -> id);
        delete(e);
        return NULL;
    }

Answer 5

Вам удалось воспроизвести проблему в отладчике?

Если вы скомпилировали оптимизированный код; Хотя проблема, скорее всего, заключается в вашем коде, возможно, стоит попробовать отключить оптимизацию компилятора (-O0)