Std :: atomic vs static переменная для синхронизации потоков

Anyway, I used a static variable and the threads do not always get the latest one. I was surprised to see one thread dominating everything

Да, несколько потоков, обращающихся к одному и тому же несинхронизированному значению, и хотя бы один из них записывает в него, является гонка данных, что является неопределенным поведением в соответствии со стандартом C++. Может произойти все, что угодно.

I am thinking of using std::atomic as my token. Would it work?

Да. Это предотвратит любую гонку данных по токену. Я не вижу другой прямой проблемы в вашем псевдокоде, так что с этой точки зрения это выглядит хорошо.

this feels like a bad design and I am not sure how to do it better. Any suggestions would be very much appreciated.

Весь дизайн выглядит несколько странно, но это зависит от вашей библиотеки потоков, если есть более простой способ выразить это. Например, с OpenMP вы можете сделать это за один проход (логика give_data и get_data слишком неясна, чтобы сделать это полным):

#pragma omp parallel
{
    int threadCount = omp_get_num_threads();
#pragma omp single
    for (int i = 0; i < threadCount; ++i)
        give_data(i);

#pragma omp ordered for ordered schedule(static)
    for (int i = 0; i < threadCount; ++i)
    {
        auto data = get_data();
        result = process_data(data);
#pragma omp ordered
        put_data(result); // this is a synchronized call 
    }
}

Директива ordered заставляет вызовы put_data выполняться точно в том же порядке (один за другим), как если бы цикл был последовательным, в то время как потоки все еще могут выполнять предшествующую обработку данных параллельно.

На самом деле с OpenMP все могло бы быть еще проще, если бы все, что вы хотели сделать В самом деле, - это сделать один большой цикл обработки данных параллельно с упорядоченной записью:

#pragma omp parallel for ordered schedule(static)
for (int i = 0; i < dataItemCount; ++i)
{
    auto data = get_data(i); // whatever this would entail
    auto result = process_data(data);
#pragma omp ordered
    put_data(result); // this is a synchronized call    
}

Не похоже, что вам нужно, чтобы распределение элементов данных было в порядке, но если вы действительно это сделаете, этот подход не будет работать так легко, потому что у вас может быть только один упорядоченный раздел на упорядоченный цикл.

Ответ Макса отличный. Если бы у меня была возможность использовать OpenMP, учитывая время, я бы это сделал. Однако это не так, поэтому я отправляю этот ответ на свой вопрос.

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

void give_threads_data(){
    vector<pair<data, promise<result>*> promises(threads.size());
    vector<future<result>> futures(threads.size());
    for(int i=0; i<threads.size(); i++){
        data d = get_data();
        threads[i].put_data(d, promises[i]);
        futures[i] = promises[i].get_future();
    }

    for(int i=0; i<futures.size(); i++){
        result = futures[i].get();
        // handle result
    }
}

Таким образом, я смог получить результаты так же, как отправляю их в потоки. Тело резьбы стало намного чище:

void thread_body(){
    while(true){
        pair<data, promise<result>*> item = queue.get(); // blocking call
        data d = item.first;
        promise<result>* promise = item.second;

        result r = process_data(d);
        promise->set_value(r);
    }
}

Нет люфта, и результаты отличные. В следующий раз, когда я буду использовать потоки, я рассмотрю OpenMP.