Неблокирующий параллельный алгоритм поиска максимума и его индекса
Существует ли неблокирующий параллельный алгоритм поиска максимального элемента массива а также его индекса? Есть такая реализация,
AtomicInteger atomicMin = new AtomicInteger(99999);
AtomicInteger atomicMax = new AtomicInteger(-99999);
AtomicInteger atomicMinIndex = new AtomicInteger();
AtomicInteger atomicMaxIndex = new AtomicInteger();
AtomicInteger atomicIndex = new AtomicInteger(0);
IntStream.range(0, arr.length).parallel().forEach(i -> {
int oldMin;
int newMin;
int oldMax;
int newMax;
int oldMinIndex;
int newMinIndex;
int oldMaxIndex;
int newMaxIndex;
do {
oldMin = atomicMin.get();
oldMax = atomicMax.get();
oldMinIndex = atomicMinIndex.get();
oldMaxIndex = atomicMaxIndex.get();
newMin = arr[i] < oldMin ? arr[i] : oldMin;
newMax = arr[i] > oldMax ? arr[i] : oldMax;
newMinIndex = arr[i] < oldMin ? i : oldMinIndex;
newMaxIndex = arr[i] > oldMax ? i : oldMaxIndex;
} while (!atomicMin.compareAndSet(oldMin, newMin) ||
!atomicMax.compareAndSet(oldMax, newMax) ||
!atomicMinIndex.compareAndSet(oldMinIndex, newMinIndex) ||
!atomicMaxIndex.compareAndSet(oldMaxIndex, newMaxIndex));
});
но смущает вот этот участок кода
while (!atomicMin.compareAndSet(oldMin, newMin) ||
!atomicMax.compareAndSet(oldMax, newMax) ||
!atomicMinIndex.compareAndSet(oldMinIndex, newMinIndex) ||
!atomicMaxIndex.compareAndSet(oldMaxIndex, newMaxIndex))
тут по сути операция обновления максимума и его индекса не атомарна, то есть другой поток может записать другое значение индекса элемента, в результате максимальный элемент будет найден правильно, а его индекс - нет.
Подскажите, это корректный алгоритм и как в этом можно формально убедиться?
Upd. Переписал вот так, по идее теперь сомнений быть не должно
public static AtomicSearchResult getMax(final Array array, final Comparator<Integer> comparator) {
if (array.size() < 1) {
throw new IllegalArgumentException("Array must contain at least one element");
}
AtomicSearchResult searchResult = new AtomicSearchResult(array.get(0), 0);
IntStream.range(0, array.size()).parallel().forEach((i) -> {
int currentElement = array.get(i);
AtomicSearchResult.SearchResult currentSearchResultValue;
do {
currentSearchResultValue = searchResult.get();
} while (comparator.compare(currentElement, currentSearchResultValue.getValue()) > 0 && !searchResult.compareAndSet(currentSearchResultValue, currentElement, i));
});
return searchResult;
}
public final static class AtomicSearchResult {
private final AtomicLong max; // stores index and value as 64 bit long in such format - <32 bit of max value><32 bit of corresponding index>. See method 'convertToLong'
public AtomicSearchResult(int maxValue, int index) {
max = new AtomicLong(convertToLong(maxValue, index));
}
public SearchResult get() {
long currentValue = max.get();
return new SearchResult(getMaxValue(currentValue), getElementIndex(currentValue));
}
private int getMaxValue(long value) {
return (int) (value >>> 32);
}
private int getElementIndex(long value) {
return (int) (value & 0xFFFFFFFFL);
}
public boolean compareAndSet(SearchResult expected, int actualMaxValue, int actualMaxIndex) {
return compareAndSet(expected.getValue(), expected.getIndex(), actualMaxValue, actualMaxIndex);
}
public boolean compareAndSet(int expectedMaxValue, int expectedMaxIndex, int actualMaxValue, int actualMaxIndex) {
long expectedLong = convertToLong(expectedMaxValue, expectedMaxIndex);
long actualLong = convertToLong(actualMaxValue, actualMaxIndex);
return max.compareAndSet(expectedLong, actualLong);
}
private long convertToLong(int value, int index) {
return ((long) value << 32) | index;
}
@Data
public final static class SearchResult {
private final int value;
private final int index;
}
}
Что думаете?
Ответы (1 шт):
Автор решения: user7860670
→ Ссылка
Делаете N потоков, каждый из которых ищет индекс максимального элемента на своем участке массива, затем выбираете максимальный из найденных N максимумов - это и будет максимальный элемент во всем массиве. Любители buzzwords и bullshit bingo наверное сразу бы вспомнили про map-reduce, а на деле все элементарно, для потоков даже синхронизация не требуется, надо только подождать, пока они все окончат работу:
public class t_Main
{
private static final class
t_MaxFinder implements Runnable
{
private final int m_begin_index;
private final int m_end_index;
private final int m_thread_index;
private final int [] m_items;
private final int [] m_max_indexes;
public t_MaxFinder
(
final int begin_index
, final int items_count
, final int [] items
, final int [] max_indexes
, final int thread_index
)
{
m_begin_index = begin_index;
m_end_index = begin_index + items_count;
m_thread_index = thread_index;
m_items = items;
m_max_indexes = max_indexes;
return;
}
@Override
public void run()
{
int max_index = m_begin_index;
int max_value = m_items[m_begin_index];
for (int index = m_begin_index; index < m_end_index; ++index)
{
final int value = m_items[index];
if (max_value < value)
{
max_index = index;
max_value = value;
}
}
m_max_indexes[m_thread_index] = max_index;
System.out.printf
(
"thread %d (%d - %d, %d) max %d at %d\n"
, m_thread_index, m_begin_index, m_end_index
, m_end_index - m_begin_index, max_value, max_index
);
return;
}
}
private static int
Find_MaxIndex
(
final int [] items
, final int threads_count
) throws InterruptedException
{
final int [] max_indexes = new int[threads_count];
{
java.util.concurrent.ExecutorService pool = java.util.concurrent.Executors.newFixedThreadPool(threads_count);
{
int items_count = items.length / threads_count;
int begin_index = 0;
for (int thread_index = 0; thread_index < threads_count; ++thread_index)
{
if (thread_index == (threads_count - 1))
{
items_count += items.length % threads_count;
}
pool.execute(new t_MaxFinder(begin_index, items_count, items, max_indexes, thread_index));
begin_index += items_count;
}
}
pool.shutdown();
pool.awaitTermination(Long.MAX_VALUE, java.util.concurrent.TimeUnit.SECONDS);
}
int max_index = max_indexes[0];
int max_value = items[max_index];
for (int max_index_index = 1; max_index_index < max_indexes.length; ++max_index_index)
{
final int index = max_indexes[max_index_index];
final int value = items[index];
if (max_value < value)
{
max_index = index;
max_value = value;
}
}
return max_index;
}
public static void main
(
String [] args
) throws InterruptedException
{
final long seed = 42;
final int threads_count = 3;
final int items_count = 20;
final int item_max = 100;
final int [] items = new int[items_count];
{
final java.util.Random rd = new java.util.Random(seed);
for (int item_index = 0; item_index < items.length; ++item_index)
{
items[item_index] = rd.nextInt(item_max);
System.out.println(items[item_index]);
}
}
final int max_index = Find_MaxIndex(items, threads_count);
System.out.printf("result max %d at %d", items[max_index], max_index);
}
}