Параллельная реализация циклического перебора не удалась ⇐ JAVA

[Anonymous]

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

Код: Выделить всё

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class LoadBalancer1 {

private final List instanceList = new ArrayList();
private static final int MAX_SIZE = 10;
private final LoadBalancerStrategy strategy;
private static final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
private static final ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
private static final ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();

public LoadBalancer1(LoadBalancerStrategy strategy) {
this.strategy = strategy;
}

public boolean register(String instance) {
writeLock.lock();
try {
if (size() == MAX_SIZE) {
return false;
}
if (instanceList.contains(instance)) {
return false;
}
instanceList.add(instance);
return true;
} finally {
writeLock.unlock();
}
}

public String get() {
readLock.lock();
try {
return this.strategy.get(this.instanceList);
}finally {
readLock.unlock();
}
}

public int size() {
readLock.lock();
try {
return this.instanceList.size();
} finally {
readLock.unlock();
}
}
}

interface LoadBalancerStrategy {
String get(List instances);
}

class RandomStrategy implements LoadBalancerStrategy {
@Override
public String get(List instances) {
if (instances.isEmpty()) {
return null;
}
return instances.get(ThreadLocalRandom.current().nextInt(instances.size()));
}
}

class RoundRobinStrategy implements LoadBalancerStrategy {

private final AtomicInteger currentIndex = new AtomicInteger(0);

@Override
public String get(List instances) {  // Synchronize access
if (instances.isEmpty()) {
return null;
}
int index = currentIndex.getAndUpdate(i -> (i + 1) % instances.size());
return instances.get(index);
}
}

Мой тест:

Код: Выделить всё

@RepeatedTest(100)
void should_keep_round_robin_pattern_in_concurrent_environment() throws InterruptedException, ExecutionException {
// given
loadBalancerRoundRobin.register("1");
loadBalancerRoundRobin.register("2");
loadBalancerRoundRobin.register("3");
Queue objects = new ArrayBlockingQueue(30);

int threads = Runtime.getRuntime().availableProcessors();
var executorService = Executors.newScheduledThreadPool(threads);
try {
for (int i = 0; i < 30; i++) {
Future submit = executorService.submit(() -> loadBalancerRoundRobin.get());
objects.add(submit);
}
} finally {
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.MINUTES);
}

for (int i = 0; i < objects.size() - 3; i += 3) {
assertEquals("1", objects.remove().get());
assertEquals("2", objects.remove().get());
assertEquals("3", objects.remove().get());
}
}

Для каждого теста я создаю новые объекты:

Код: Выделить всё

@BeforeEach
void setUp() {
randomStrategy = new RandomStrategy();
roundRobin = new RoundRobinStrategy();
loadBalancerRandomStrategy = new LoadBalancer1(randomStrategy);
loadBalancerRoundRobin = new LoadBalancer1(roundRobin);
}

У меня постепенно заканчиваются идеи о том, в чем проблема. Я считаю, что currentIndex изменяется небезопасным образом, не знаю, как найти причину. Или, может быть, есть проблема с методом size(), но, поскольку он заблокирован, я не знаю, что еще я могу сделать. Если бы кто-нибудь мог объяснить, что происходит, я был бы чрезвычайно благодарен.

Подробнее здесь: https://stackoverflow.com/questions/789 ... tion-fails