Skip to content

Fix QueueResizableOpenSearchThreadPoolExecutorTests #18006

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
+35 −107
Open

Fix QueueResizableOpenSearchThreadPoolExecutorTests #18006

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
142 changes: 35 additions & 107 deletions ...rg/opensearch/common/util/concurrent/QueueResizableOpenSearchThreadPoolExecutorTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@

import org.opensearch.common.settings.Settings;
import org.opensearch.test.OpenSearchTestCase;
import org.opensearch.threadpool.ThreadPool;
import org.junit.After;

import java.util.concurrent.TimeUnit;
import java.util.function.Function;
Expand All @@ -23,158 +25,84 @@
* based on the time taken for each event.
*/
public class QueueResizableOpenSearchThreadPoolExecutorTests extends OpenSearchTestCase {
public void testResizeQueueSameSize() throws Exception {
ThreadContext context = new ThreadContext(Settings.EMPTY);
ResizableBlockingQueue<Runnable> queue = new ResizableBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), 2000);
private QueueResizableOpenSearchThreadPoolExecutor executor;
private ResizableBlockingQueue<Runnable> queue;
private int measureWindow;

private void createExecutor(int queueSize, Function<Runnable, WrappedRunnable> runnableWrapper) {
int threads = randomIntBetween(1, 10);
int measureWindow = randomIntBetween(100, 200);
logger.info("--> auto-queue with a measurement window of {} tasks", measureWindow);
QueueResizableOpenSearchThreadPoolExecutor executor = new QueueResizableOpenSearchThreadPoolExecutor(
measureWindow = randomIntBetween(100, 200);
ThreadContext context = new ThreadContext(Settings.EMPTY);
this.queue = new ResizableBlockingQueue<>(ConcurrentCollections.newBlockingQueue(), queueSize);
this.executor = new QueueResizableOpenSearchThreadPoolExecutor(
"test-threadpool",
threads,
threads,
1000,
TimeUnit.MILLISECONDS,
queue,
fastWrapper(),
runnableWrapper,
OpenSearchExecutors.daemonThreadFactory("queuetest"),
new OpenSearchAbortPolicy(),
context
);
executor.prestartAllCoreThreads();
logger.info("--> executor: {}", executor);
}

@After
public void stopExecutor() {
ThreadPool.terminate(executor, 10, TimeUnit.SECONDS);
}

public void testResizeQueueSameSize() throws Exception {
createExecutor(2000, fastWrapper());

// Execute a task multiple times that takes 1ms
assertThat(executor.resize(1000), equalTo(1000));
executeTask(executor, (measureWindow * 5) + 2);

assertBusy(() -> { assertThat(queue.capacity(), lessThanOrEqualTo(1000)); });
executor.shutdown();
executor.awaitTermination(10, TimeUnit.SECONDS);
assertBusy(() -> assertThat(queue.capacity(), lessThanOrEqualTo(1000)));
}

public void testResizeQueueUp() throws Exception {
ThreadContext context = new ThreadContext(Settings.EMPTY);
ResizableBlockingQueue<Runnable> queue = new ResizableBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), 2000);

int threads = randomIntBetween(1, 10);
int measureWindow = randomIntBetween(100, 200);
logger.info("--> auto-queue with a measurement window of {} tasks", measureWindow);
QueueResizableOpenSearchThreadPoolExecutor executor = new QueueResizableOpenSearchThreadPoolExecutor(
"test-threadpool",
threads,
threads,
1000,
TimeUnit.MILLISECONDS,
queue,
fastWrapper(),
OpenSearchExecutors.daemonThreadFactory("queuetest"),
new OpenSearchAbortPolicy(),
context
);
executor.prestartAllCoreThreads();
logger.info("--> executor: {}", executor);

createExecutor(2000, fastWrapper());
// Execute a task multiple times that takes 1ms
assertThat(executor.resize(3000), equalTo(3000));
executeTask(executor, (measureWindow * 5) + 2);

assertBusy(() -> { assertThat(queue.capacity(), greaterThanOrEqualTo(2000)); });
executor.shutdown();
executor.awaitTermination(10, TimeUnit.SECONDS);
assertBusy(() -> assertThat(queue.capacity(), greaterThanOrEqualTo(2000)));
}

public void testResizeQueueDown() throws Exception {
ThreadContext context = new ThreadContext(Settings.EMPTY);
ResizableBlockingQueue<Runnable> queue = new ResizableBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), 2000);

int threads = randomIntBetween(1, 10);
int measureWindow = randomIntBetween(100, 200);
logger.info("--> auto-queue with a measurement window of {} tasks", measureWindow);
QueueResizableOpenSearchThreadPoolExecutor executor = new QueueResizableOpenSearchThreadPoolExecutor(
"test-threadpool",
threads,
threads,
1000,
TimeUnit.MILLISECONDS,
queue,
fastWrapper(),
OpenSearchExecutors.daemonThreadFactory("queuetest"),
new OpenSearchAbortPolicy(),
context
);
executor.prestartAllCoreThreads();
logger.info("--> executor: {}", executor);

createExecutor(2000, fastWrapper());
// Execute a task multiple times that takes 1ms
assertThat(executor.resize(900), equalTo(900));
assertThat(executor.resize(1500), equalTo(1500));
executeTask(executor, (measureWindow * 5) + 2);

assertBusy(() -> { assertThat(queue.capacity(), lessThanOrEqualTo(900)); });
executor.shutdown();
executor.awaitTermination(10, TimeUnit.SECONDS);
assertBusy(() -> assertThat(queue.capacity(), lessThanOrEqualTo(1500)));
}

public void testExecutionEWMACalculation() throws Exception {
ThreadContext context = new ThreadContext(Settings.EMPTY);
ResizableBlockingQueue<Runnable> queue = new ResizableBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), 100);

QueueResizableOpenSearchThreadPoolExecutor executor = new QueueResizableOpenSearchThreadPoolExecutor(
"test-threadpool",
1,
1,
1000,
TimeUnit.MILLISECONDS,
queue,
fastWrapper(),
OpenSearchExecutors.daemonThreadFactory("queuetest"),
new OpenSearchAbortPolicy(),
context
);
executor.prestartAllCoreThreads();
logger.info("--> executor: {}", executor);

createExecutor(100, fastWrapper());
assertThat((long) executor.getTaskExecutionEWMA(), equalTo(0L));
executeTask(executor, 1);
assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(30L)); });
assertBusy(() -> assertThat((long) executor.getTaskExecutionEWMA(), equalTo(30L)));
executeTask(executor, 1);
assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(51L)); });
assertBusy(() -> assertThat((long) executor.getTaskExecutionEWMA(), equalTo(51L)));
executeTask(executor, 1);
assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(65L)); });
assertBusy(() -> assertThat((long) executor.getTaskExecutionEWMA(), equalTo(65L)));
executeTask(executor, 1);
assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(75L)); });
assertBusy(() -> assertThat((long) executor.getTaskExecutionEWMA(), equalTo(75L)));
executeTask(executor, 1);
assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(83L)); });

executor.shutdown();
executor.awaitTermination(10, TimeUnit.SECONDS);
assertBusy(() -> assertThat((long) executor.getTaskExecutionEWMA(), equalTo(83L)));
}

/** Use a runnable wrapper that simulates a task with unknown failures. */
public void testExceptionThrowingTask() throws Exception {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

for testExceptionThrowingTask, the runnable wrapper was earlier exceptionalWrapper() which is changed to fastWrapper. What's the reason for this change?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Another good catch, this got mixed up in the refactoring. I'll fix it.

ThreadContext context = new ThreadContext(Settings.EMPTY);
ResizableBlockingQueue<Runnable> queue = new ResizableBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), 100);

QueueResizableOpenSearchThreadPoolExecutor executor = new QueueResizableOpenSearchThreadPoolExecutor(
"test-threadpool",
1,
1,
1000,
TimeUnit.MILLISECONDS,
queue,
exceptionalWrapper(),
OpenSearchExecutors.daemonThreadFactory("queuetest"),
new OpenSearchAbortPolicy(),
context
);
executor.prestartAllCoreThreads();
logger.info("--> executor: {}", executor);

public void testExceptionThrowingTask() {
createExecutor(100, exceptionalWrapper());
assertThat((long) executor.getTaskExecutionEWMA(), equalTo(0L));
executeTask(executor, 1);
executor.shutdown();
executor.awaitTermination(10, TimeUnit.SECONDS);
}

private Function<Runnable, WrappedRunnable> fastWrapper() {
Expand All @@ -198,7 +126,7 @@ private void executeTask(QueueResizableOpenSearchThreadPoolExecutor executor, in
}
}

public class SettableTimedRunnable extends TimedRunnable {
private static class SettableTimedRunnable extends TimedRunnable {
private final long timeTaken;
private final boolean testFailedOrRejected;

Expand Down
Loading