ArrayList.removeFirst() выдает исключение ArrayIndexOutOfBoundsException, но было синхронизировано ⇐ JAVA

[Anonymous]

Наличие этого пула потоков:

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

package xxx;

import java.util.ArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
* A thread pool that hold an amount of work.  During execution the amount of
* work may grow.
*
* @param  The work
*/
public class DepleatingFiFoThreadPool {

public static final Logger LOG = Logger.getLogger(DepleatingFiFoThreadPool.class.getCanonicalName());

public record Socket(A r, String threadPostfix) {
}

private final Thread[] threadsRunning;
private final ArrayList unstartedRunnables = new ArrayList();
private int nextFreeSocketIndex = 0;
private final Consumer errorHandler;
private final Object lock = new Object();
private boolean waitingForLock = false;

private final String prefix;
private final Consumer invoker;

/**
* Create the threadpool.
*
* @param threadsRunningMax The maximum of Threads running at the same time,
*                          must be higher than 0

* @param errorHandler The handler in case of problems, never
*

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

null

* @param prefix The prefix for the pool, required for debugging
* purposes, never

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

null

* @param invoker The invoker who encapsule the work, never
*

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

null

*/
public DepleatingFiFoThreadPool(final int threadsRunningMax, final Consumer errorHandler,
final String prefix, final Consumer invoker) {
this.errorHandler = errorHandler;
this.prefix = prefix;
this.invoker = invoker;
this.threadsRunning = new Thread[threadsRunningMax];
}

/**
* Add and start a thread.
*
* Work after returning from {@link #executeUntilDeplated(long)} will not be
* executed.
*
* @param notRunningThread The not running work that is designated to be done
* @param threadPostfix The postfix of the thread, never

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

null

*/
public void addAndStartThread(final A notRunningThread, final String threadPostfix) {
var mustCallLater = false;
if (nextFreeSocketIndex < threadsRunning.length) {
synchronized (this) {
if (nextFreeSocketIndex < threadsRunning.length) {
start(notRunningThread, threadPostfix);
} else {
mustCallLater = true;
}
}
} else {
mustCallLater = true;
}
if (mustCallLater) {
unstartedRunnables.add(new Socket(notRunningThread, threadPostfix));
}
}

private void finished() {
synchronized (this) {
threadsRunning[--nextFreeSocketIndex] = null;
if (nextFreeSocketIndex == 0) {
synchronized (lock) {
if (waitingForLock) {
lock.notify();
}
}
} else {
Socket e = null;
if (!unstartedRunnables.isEmpty()) {
synchronized (unstartedRunnables) {
if (!unstartedRunnables.isEmpty()) {
e = unstartedRunnables.removeFirst();
}
}
}
if (e != null) {
start(e.r, e.threadPostfix);
}
}
}
}

private void start(final A notRunningThread, final String name) {
var socket = nextFreeSocketIndex++;
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
invoker.accept(notRunningThread);
} catch (Throwable e) { // fault barrier!
try {
errorHandler.accept(e);
} catch (Throwable ta) {
ta.addSuppressed(e);
LOG.log(Level.SEVERE, ta.getMessage(), ta);
}
}
finished();
}
}, prefix + name);
threadsRunning[socket] = t;
t.start();
}

/**
* Execute the threads and every subsequent thread that is added during the
* progress.
*
* @param timeoutMs The timeout in miliseconds, should be an positive value
* @return

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

false

if the timeout has reached,

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

true

* otherwise
* @throws InterruptedException If the thread has been interrupted (maybe
* shutdown of application)
*/
public boolean executeUntilDeplated(final long timeoutMs) throws InterruptedException {
AtomicBoolean resultHolder = new AtomicBoolean(true);
if (nextFreeSocketIndex > 0) {
Thread timeout = new Thread(() -> {
try {
Thread.sleep(timeoutMs);
resultHolder.set(false);
synchronized (lock) {
lock.notify();
}
} catch (InterruptedException e) {
LOG.log(Level.FINE, "Not a bug, timeout not reached, watchdog not required.", e);
}
});
timeout.start();
if (nextFreeSocketIndex > 0) {
waitingForLock = true;
synchronized (lock) {
try {
lock.wait(timeoutMs);
} finally {
timeout.interrupt();
}
}
}
}
return resultHolder.get();
}
}

Иногда возникает исключение с такой трассировкой стека:

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

Exception in thread "NN/SQL-caller" java.lang.ArrayIndexOutOfBoundsException: arraycopy: last source index 50 out of bounds for object array[49]
at java.base/java.lang.System.arraycopy(Native Method)
at java.base/java.util.ArrayList.fastRemove(ArrayList.java:724)
at java.base/java.util.ArrayList.removeFirst(ArrayList.java:573)
at xxx.DepleatingFiFoThreadPool.finished(DepleatingFiFoThreadPool.java:74)
at xxx.DepleatingFiFoThreadPool$1.run(DepleatingFiFoThreadPool.java:100)
at java.base/java.lang.Thread.run(Thread.java:1570)

Я не могу отладить это, поскольку оно появляется только раз в три-пять недель.
Я использовал этот тест, но не смог найти проблема:

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

public void testMassiveQueuedFunction() throws InterruptedException {
var dtp = new DepleatingFiFoThreadPool(100, x -> System.out.println(x), "testQueueFunction",
Runnable::run);
var l = new CountDownLatch(0);
Runnable wait = () -> {
try {
Thread.sleep(2);
l.countDown();
} catch (InterruptedException e) {
}
};
for (int i = 0; i < 9999; i++) {
dtp.addAndStartThread(wait, i + "");
}
long start = System.currentTimeMillis();
dtp.executeUntilDeplated(3000);
long took = System.currentTimeMillis() - start;
assert l.getCount() == 0 : "Await a count of 0 but has: " + l.getCount();
assert took < 2900 : "Should execute faster but took: " + took;
}

Может ли кто-нибудь предложить лучший модульный тест?

Подробнее здесь: https://stackoverflow.com/questions/790 ... ynchronize