honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Ensure AsyncTask#isScheduled remain false after close (#45687) 

If a scheduled task of an AbstractAsyncTask starts after it was closed,
then isScheduledOrRunning can remain true forever although no task is
running or scheduled.

Closes #45576
This commit is contained in:
Nhat Nguyen 2019-08-17 13:47:09 -04:00
parent 6f2daa85e3
commit 6f5d944fbd
2 changed files with 35 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
server/src/main/java/org/elasticsearch/common/util/concurrent/AbstractAsyncTask.java
View File

@ -134,6 +134,9 @@ public abstract class AbstractAsyncTask implements Runnable, Closeable {
@Override
public final void run() {
synchronized (this) {
if (isClosed()) {
return;
}
cancellable = null;
isScheduledOrRunning = autoReschedule;
}

32
server/src/test/java/org/elasticsearch/common/util/concurrent/AbstractAsyncTaskTests.java
View File

@ -18,18 +18,23 @@
*/
package org.elasticsearch.common.util.concurrent;
import org.elasticsearch.common.Randomness;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.core.internal.io.IOUtils;
import org.elasticsearch.test.ESTestCase;
import org.elasticsearch.threadpool.TestThreadPool;
import org.elasticsearch.threadpool.ThreadPool;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
public class AbstractAsyncTaskTests extends ESTestCase {
@ -203,4 +208,31 @@ public class AbstractAsyncTaskTests extends ESTestCase {
assertFalse(task.isScheduled());
assertTrue(task.isClosed());
}
public void testIsScheduledRemainFalseAfterClose() throws Exception {
int numTasks = between(10, 50);
List<AbstractAsyncTask> tasks = new ArrayList<>(numTasks);
AtomicLong counter = new AtomicLong();
for (int i = 0; i < numTasks; i++) {
AbstractAsyncTask task = new AbstractAsyncTask(logger, threadPool, TimeValue.timeValueMillis(randomIntBetween(1, 2)), true) {
@Override
protected boolean mustReschedule() {
return counter.get() <= 1000;
}
@Override
protected void runInternal() {
counter.incrementAndGet();
}
};
task.rescheduleIfNecessary();
tasks.add(task);
}
Randomness.shuffle(tasks);
IOUtils.close(tasks);
Randomness.shuffle(tasks);
for (AbstractAsyncTask task : tasks) {
assertTrue(task.isClosed());
assertFalse(task.isScheduled());
}
}
}