Обновление максимального значения из нескольких потоков

Question

Есть ли способ обновить максимум из нескольких потоков с помощью атомных операций?

Иллюстративный пример:

std::vector<float> coord_max(128); 
#pragma omp parallel for 
for (int i = 0; i < limit; ++i) { 
    int j = get_coord(i); // can return any value in range [0,128) 
    float x = compute_value(j, i); 
    #pragma omp critical (coord_max_update) 
    coord_max[j] = std::max(coord_max[j], x); 
} 

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

Answer 1

Следуя предложению в комментарии, я нашел решение, которое не требует блокировки, и вместо этого использует функцию сравнения и обмена, найденную в std :: atomic/boost :: atomic. Я ограничен C++ 03, поэтому в этом случае я бы использовал boost :: atomic.

BOOST_STATIC_ASSERT(sizeof(int) == sizeof(float)); 
union FloatPun { float f; int i; }; 

std::vector< boost::atomic<int> > coord_max(128); 
#pragma omp parallel for 
for (int i = 0; i < limit; ++i) { 
    int j = get_coord(i); 
    FloatPun x, maxval; 
    x.f = compute_value(j, i); 

    maxval.i = coord_max[j].load(boost::memory_order_relaxed); 
    do { 
     if (maxval.f >= x.f) break; 
    } while (!coord_max[j].compare_exchange_weak(maxval.i, x.i, 
     boost::memory_order_relaxed)); 
} 

Существует несколько шаблонов, связанных с размещением значений float в ints, поскольку кажется, что атомарные поплавки не блокируются. Я не на 100% использую порядок памяти, но наименее ограничивающий уровень «расслабленный», похоже, в порядке, поскольку неатомная память не задействована.

Answer 2

Как насчет объявления, скажем, std::vector<std::mutex> (или boost::mutex) длины 128, а затем создания объекта блокировки с использованием элемента j?

Я имею в виду, что-то вроде:

std::vector<float> coord_max(128); 
std::vector<std::mutex> coord_mutex(128); 
#pragma omp parallel for 
for (int i = 0; i < limit; ++i) { 
    int j = get_coord(i); // can return any value in range [0,128) 
    float x = compute_value(j, i); 
    std::scoped_lock lock(coord_mutex[j]); 
    coord_max[j] = std::max(coord_max[j], x);  
} 

Или, как в Rahul Banerjee's suggestion #3:

std::vector<float> coord_max(128); 
const int parallelism = 16; 
std::vector<std::mutex> coord_mutex(parallelism); 
#pragma omp parallel for 
for (int i = 0; i < limit; ++i) { 
    int j = get_coord(i); // can return any value in range [0,128) 
    float x = compute_value(j, i); 
    std::scoped_lock lock(coord_mutex[j % parallelism]); 
    coord_max[j] = std::max(coord_max[j], x);  
} 

Answer 3

Не уверен в синтаксисе, но алгоритмически, у вас есть три варианта:

1. Заблокируйте весь вектор, чтобы гарантировать атомный доступ (это то, что вы сейчас делаете).

2. Заблокируйте отдельные элементы, чтобы каждый элемент мог обновляться независимо от других. Плюсы: максимальный параллелизм; Минусы: требуется множество замков!

3. Что-то в промежутке! Концептуально подумайте о разбиении своего вектора на 16 (или 32/64/...) «банки» следующим образом: bank0 состоит из векторных элементов 0, 16, 32, 48, 64, ... bank1 состоит из векторных элементов 1 , 17, 33, 49, 65, ... bank2 состоит из векторных элементов 2, 18, 34, 50, 66, ... ... Теперь, используя 16 явных блокировок, прежде чем вы получите доступ к элементу, вы можете имеют до 16-точечного параллелизма. Чтобы получить доступ к элементу n, приобретите блокировку (n% 16), завершите доступ, затем отпустите ту же блокировку.

Answer 4

Просто добавить свои два цента, прежде чем начать более мелкозернистых оптимизаций я хотел бы попробовать следующий подход, который исключает необходимость в omp critical:

std::vector<float> coord_max(128); 
float    fbuffer(0); 
#pragma omp parallel 
{ 
    std::vector<float> thread_local_buffer(128); 

    // Assume limit is a really big number 
    #pragma omp for  
    for (int ii = 0; ii < limit; ++ii) { 
    int jj = get_coord(ii); // can return any value in range [0,128) 
    float x = compute_value(jj,ii); 
    thread_local_buffer[jj] = std::max(thread_local_buffer[jj], x); 
    } 
    // At this point each thread has a partial local vector 
    // containing the maximum of the part of the problem 
    // it has explored 

    // Reduce the results 
    for(int ii = 0; ii < 128; ii++){ 
    // Find the max for position ii 
#pragma omp for schedule(static,1) reduction(max:fbuffer) 
    for(int jj = 0; jj < omp_get_thread_num(); jj++) { 
     fbuffer = thread_local_buffer[ii]; 
    } // Barrier implied here 
    // Write it in the vector at correct position 
#pragma omp single 
    { 
     coord_max[ii] = fbuffer; 
     fbuffer = 0; 
    } // Barrier implied here 

    } 
} 

Обратите внимание, что я не компилировать сниппет , поэтому я мог бы оставить внутри синтаксическую ошибку. Во всяком случае, я надеюсь, что я передал эту идею.