You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2022/04/27 13:25:49 UTC
[iotdb] branch add_comment created (now 5027a0b0b2)
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a change to branch add_comment
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 5027a0b0b2 spotless
This branch includes the following new commits:
new 45dc2b1a69 add some comment and rename Driver related class
new 5027a0b0b2 spotless
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 02/02: spotless
Posted by qi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch add_comment
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5027a0b0b20cf3f426b154b843a54a4d6947a694
Author: qiaojialin <64...@qq.com>
AuthorDate: Wed Apr 27 21:25:32 2022 +0800
spotless
---
.../org/apache/iotdb/db/mpp/execution/Driver.java | 1 -
.../org/apache/iotdb/db/mpp/execution/IDriver.java | 18 ++++-----
.../db/mpp/schedule/AbstractDriverThread.java | 3 +-
.../iotdb/db/mpp/schedule/DriverScheduler.java | 14 ++-----
.../iotdb/db/mpp/schedule/IDriverScheduler.java | 3 +-
.../iotdb/db/mpp/schedule/ITaskScheduler.java | 15 +++-----
.../iotdb/db/mpp/schedule/task/DriverTask.java | 10 ++---
.../db/mpp/schedule/DefaultTaskSchedulerTest.java | 25 ++++--------
.../iotdb/db/mpp/schedule/DriverSchedulerTest.java | 12 ++----
.../DriverTaskTimeoutSentinelThreadTest.java | 45 ++++++++--------------
10 files changed, 50 insertions(+), 96 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/Driver.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/Driver.java
index 27cb1d361f..ae220e4911 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/Driver.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/Driver.java
@@ -47,7 +47,6 @@ import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static java.lang.Boolean.TRUE;
import static org.apache.iotdb.db.mpp.operator.Operator.NOT_BLOCKED;
-
public abstract class Driver implements IDriver {
protected static final Logger LOGGER = LoggerFactory.getLogger(Driver.class);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/IDriver.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/IDriver.java
index 4914760351..61e96dbe42 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/IDriver.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/IDriver.java
@@ -25,8 +25,8 @@ import com.google.common.util.concurrent.ListenableFuture;
import io.airlift.units.Duration;
/**
- * IDriver encapsulates some methods which are necessary for FragmentInstanceTaskExecutor to run a fragment
- * instance
+ * IDriver encapsulates some methods which are necessary for FragmentInstanceTaskExecutor to run a
+ * fragment instance
*/
public interface IDriver {
@@ -38,14 +38,14 @@ public interface IDriver {
boolean isFinished();
/**
- * run the IDriver for {@param duration} time slice, the time of this run is likely not
- * to be equal to {@param duration}, the actual run time should be calculated by the caller
+ * run the IDriver for {@param duration} time slice, the time of this run is likely not to be
+ * equal to {@param duration}, the actual run time should be calculated by the caller
*
* @param duration how long should this IDriver run
* @return the returned ListenableFuture<Void> is used to represent status of this processing if
- * isDone() return true, meaning that this IDriver is not blocked and is ready for
- * next processing. Otherwise, meaning that this IDriver is blocked and not ready for
- * next processing.
+ * isDone() return true, meaning that this IDriver is not blocked and is ready for next
+ * processing. Otherwise, meaning that this IDriver is blocked and not ready for next
+ * processing.
*/
ListenableFuture<Void> processFor(Duration duration);
@@ -66,8 +66,6 @@ public interface IDriver {
*/
void failed(Throwable t);
- /**
- * @return get SinkHandle of current IDriver
- */
+ /** @return get SinkHandle of current IDriver */
ISinkHandle getSinkHandle();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/AbstractDriverThread.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/AbstractDriverThread.java
index 13118452a8..c79304b0aa 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/AbstractDriverThread.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/AbstractDriverThread.java
@@ -62,8 +62,7 @@ public abstract class AbstractDriverThread extends Thread implements Closeable {
}
/** Processing a task. */
- protected abstract void execute(DriverTask task)
- throws InterruptedException, ExecutionException;
+ protected abstract void execute(DriverTask task) throws InterruptedException, ExecutionException;
@Override
public void close() throws IOException {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/DriverScheduler.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/DriverScheduler.java
index 1089c3410c..63718faa6b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/DriverScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/DriverScheduler.java
@@ -73,12 +73,9 @@ public class DriverScheduler implements IDriverScheduler, IService {
private DriverScheduler() {
this.readyQueue =
new L2PriorityQueue<>(
- MAX_CAPACITY,
- new DriverTask.SchedulePriorityComparator(),
- new DriverTask());
+ MAX_CAPACITY, new DriverTask.SchedulePriorityComparator(), new DriverTask());
this.timeoutQueue =
- new L1PriorityQueue<>(
- MAX_CAPACITY, new DriverTask.TimeoutComparator(), new DriverTask());
+ new L1PriorityQueue<>(MAX_CAPACITY, new DriverTask.TimeoutComparator(), new DriverTask());
this.queryMap = new ConcurrentHashMap<>();
this.blockedTasks = Collections.synchronizedSet(new HashSet<>());
this.scheduler = new Scheduler();
@@ -91,8 +88,7 @@ public class DriverScheduler implements IDriverScheduler, IService {
public void start() throws StartupException {
for (int i = 0; i < WORKER_THREAD_NUM; i++) {
AbstractDriverThread t =
- new DriverTaskThread(
- "Worker-Thread-" + i, workerGroups, readyQueue, scheduler);
+ new DriverTaskThread("Worker-Thread-" + i, workerGroups, readyQueue, scheduler);
threads.add(t);
t.start();
}
@@ -124,9 +120,7 @@ public class DriverScheduler implements IDriverScheduler, IService {
public void submitDrivers(QueryId queryId, List<IDriver> instances) {
List<DriverTask> tasks =
instances.stream()
- .map(
- v ->
- new DriverTask(v, QUERY_TIMEOUT_MS, DriverTaskStatus.READY))
+ .map(v -> new DriverTask(v, QUERY_TIMEOUT_MS, DriverTaskStatus.READY))
.collect(Collectors.toList());
queryMap
.computeIfAbsent(queryId, v -> Collections.synchronizedSet(new HashSet<>()))
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IDriverScheduler.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IDriverScheduler.java
index b92f663eb2..30dc6aa38c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IDriverScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IDriverScheduler.java
@@ -44,7 +44,8 @@ public interface IDriverScheduler {
void abortQuery(QueryId queryId);
/**
- * Abort all Drivers of the fragment instance. If the instance is not existed, nothing will happen.
+ * Abort all Drivers of the fragment instance. If the instance is not existed, nothing will
+ * happen.
*
* @param instanceId the id of the fragment instance to be aborted.
*/
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/ITaskScheduler.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/ITaskScheduler.java
index 7f1b973c01..641aba8c7f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/ITaskScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/ITaskScheduler.java
@@ -25,16 +25,14 @@ import org.apache.iotdb.db.mpp.schedule.task.DriverTaskStatus;
public interface ITaskScheduler {
/**
- * Switch a task from {@link DriverTaskStatus#BLOCKED} to {@link
- * DriverTaskStatus#READY}.
+ * Switch a task from {@link DriverTaskStatus#BLOCKED} to {@link DriverTaskStatus#READY}.
*
* @param task the task to be switched.
*/
void blockedToReady(DriverTask task);
/**
- * Switch a task from {@link DriverTaskStatus#READY} to {@link
- * DriverTaskStatus#RUNNING}.
+ * Switch a task from {@link DriverTaskStatus#READY} to {@link DriverTaskStatus#RUNNING}.
*
* @param task the task to be switched.
* @return true if it's switched to the target status successfully, otherwise false.
@@ -42,8 +40,7 @@ public interface ITaskScheduler {
boolean readyToRunning(DriverTask task);
/**
- * Switch a task from {@link DriverTaskStatus#RUNNING} to {@link
- * DriverTaskStatus#READY}.
+ * Switch a task from {@link DriverTaskStatus#RUNNING} to {@link DriverTaskStatus#READY}.
*
* @param task the task to be switched.
* @param context the execution context of last running.
@@ -51,8 +48,7 @@ public interface ITaskScheduler {
void runningToReady(DriverTask task, ExecutionContext context);
/**
- * Switch a task from {@link DriverTaskStatus#RUNNING} to {@link
- * DriverTaskStatus#BLOCKED}.
+ * Switch a task from {@link DriverTaskStatus#RUNNING} to {@link DriverTaskStatus#BLOCKED}.
*
* @param task the task to be switched.
* @param context the execution context of last running.
@@ -60,8 +56,7 @@ public interface ITaskScheduler {
void runningToBlocked(DriverTask task, ExecutionContext context);
/**
- * Switch a task from {@link DriverTaskStatus#RUNNING} to {@link
- * DriverTaskStatus#FINISHED}.
+ * Switch a task from {@link DriverTaskStatus#RUNNING} to {@link DriverTaskStatus#FINISHED}.
*
* @param task the task to be switched.
* @param context the execution context of last running.
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/DriverTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/DriverTask.java
index 4844a0de89..e008f8df1c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/DriverTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/DriverTask.java
@@ -23,8 +23,8 @@ import org.apache.iotdb.db.mpp.common.FragmentInstanceId;
import org.apache.iotdb.db.mpp.common.PlanFragmentId;
import org.apache.iotdb.db.mpp.common.QueryId;
import org.apache.iotdb.db.mpp.execution.IDriver;
-import org.apache.iotdb.db.mpp.schedule.ExecutionContext;
import org.apache.iotdb.db.mpp.schedule.DriverTaskThread;
+import org.apache.iotdb.db.mpp.schedule.ExecutionContext;
import org.apache.iotdb.db.mpp.schedule.queue.ID;
import org.apache.iotdb.db.mpp.schedule.queue.IDIndexedAccessible;
@@ -36,10 +36,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
-/**
- * the scheduling element of {@link DriverTaskThread}. It wraps a single
- * Driver.
- */
+/** the scheduling element of {@link DriverTaskThread}. It wraps a single Driver. */
public class DriverTask implements IDIndexedAccessible {
private DriverTaskID id;
@@ -84,8 +81,7 @@ public class DriverTask implements IDIndexedAccessible {
}
public boolean isEndState() {
- return status == DriverTaskStatus.ABORTED
- || status == DriverTaskStatus.FINISHED;
+ return status == DriverTaskStatus.ABORTED || status == DriverTaskStatus.FINISHED;
}
public IDriver getFragmentInstance() {
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DefaultTaskSchedulerTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DefaultTaskSchedulerTest.java
index 5689663133..213f742e69 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DefaultTaskSchedulerTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DefaultTaskSchedulerTest.java
@@ -80,8 +80,7 @@ public class DefaultTaskSchedulerTest {
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(testTask));
clear();
}
- DriverTask testTask =
- new DriverTask(mockDriver, 100L, DriverTaskStatus.BLOCKED);
+ DriverTask testTask = new DriverTask(mockDriver, 100L, DriverTaskStatus.BLOCKED);
manager.getBlockedTasks().add(testTask);
Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
@@ -130,8 +129,7 @@ public class DefaultTaskSchedulerTest {
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(testTask));
clear();
}
- DriverTask testTask =
- new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
+ DriverTask testTask = new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
manager.getQueryMap().put(queryId, taskSet);
@@ -178,8 +176,7 @@ public class DefaultTaskSchedulerTest {
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(testTask));
clear();
}
- DriverTask testTask =
- new DriverTask(mockDriver, 100L, DriverTaskStatus.RUNNING);
+ DriverTask testTask = new DriverTask(mockDriver, 100L, DriverTaskStatus.RUNNING);
Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
manager.getQueryMap().put(queryId, taskSet);
@@ -231,8 +228,7 @@ public class DefaultTaskSchedulerTest {
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(testTask));
clear();
}
- DriverTask testTask =
- new DriverTask(mockDriver, 100L, DriverTaskStatus.RUNNING);
+ DriverTask testTask = new DriverTask(mockDriver, 100L, DriverTaskStatus.RUNNING);
Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
manager.getQueryMap().put(queryId, taskSet);
@@ -284,8 +280,7 @@ public class DefaultTaskSchedulerTest {
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(testTask));
clear();
}
- DriverTask testTask =
- new DriverTask(mockDriver, 100L, DriverTaskStatus.RUNNING);
+ DriverTask testTask = new DriverTask(mockDriver, 100L, DriverTaskStatus.RUNNING);
Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
manager.getQueryMap().put(queryId, taskSet);
@@ -325,8 +320,7 @@ public class DefaultTaskSchedulerTest {
};
for (DriverTaskStatus status : invalidStates) {
DriverTask testTask1 = new DriverTask(mockDriver1, 100L, status);
- DriverTask testTask2 =
- new DriverTask(mockDriver2, 100L, DriverTaskStatus.BLOCKED);
+ DriverTask testTask2 = new DriverTask(mockDriver2, 100L, DriverTaskStatus.BLOCKED);
Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask1);
taskSet.add(testTask2);
@@ -354,9 +348,7 @@ public class DefaultTaskSchedulerTest {
}
DriverTaskStatus[] validStates =
new DriverTaskStatus[] {
- DriverTaskStatus.RUNNING,
- DriverTaskStatus.READY,
- DriverTaskStatus.BLOCKED,
+ DriverTaskStatus.RUNNING, DriverTaskStatus.READY, DriverTaskStatus.BLOCKED,
};
for (DriverTaskStatus status : validStates) {
Mockito.reset(mockDriver1);
@@ -366,8 +358,7 @@ public class DefaultTaskSchedulerTest {
DriverTask testTask1 = new DriverTask(mockDriver1, 100L, status);
- DriverTask testTask2 =
- new DriverTask(mockDriver2, 100L, DriverTaskStatus.BLOCKED);
+ DriverTask testTask2 = new DriverTask(mockDriver2, 100L, DriverTaskStatus.BLOCKED);
Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask1);
taskSet.add(testTask2);
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DriverSchedulerTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DriverSchedulerTest.java
index 07f3e67305..2dc325fb30 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DriverSchedulerTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DriverSchedulerTest.java
@@ -69,11 +69,9 @@ public class DriverSchedulerTest {
Assert.assertEquals(2, manager.getQueryMap().get(queryId).size());
Assert.assertEquals(2, manager.getTimeoutQueue().size());
Assert.assertEquals(2, manager.getReadyQueue().size());
- DriverTask task1 =
- manager.getTimeoutQueue().get(new DriverTaskID(instanceId1));
+ DriverTask task1 = manager.getTimeoutQueue().get(new DriverTaskID(instanceId1));
Assert.assertNotNull(task1);
- DriverTask task2 =
- manager.getTimeoutQueue().get(new DriverTaskID(instanceId2));
+ DriverTask task2 = manager.getTimeoutQueue().get(new DriverTaskID(instanceId2));
Assert.assertNotNull(task2);
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(task1));
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(task2));
@@ -91,8 +89,7 @@ public class DriverSchedulerTest {
Assert.assertEquals(3, manager.getQueryMap().get(queryId).size());
Assert.assertEquals(3, manager.getTimeoutQueue().size());
Assert.assertEquals(3, manager.getReadyQueue().size());
- DriverTask task3 =
- manager.getTimeoutQueue().get(new DriverTaskID(instanceId3));
+ DriverTask task3 = manager.getTimeoutQueue().get(new DriverTaskID(instanceId3));
Assert.assertNotNull(task3);
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(task3));
Assert.assertEquals(DriverTaskStatus.READY, task3.getStatus());
@@ -110,8 +107,7 @@ public class DriverSchedulerTest {
Assert.assertEquals(1, manager.getQueryMap().get(queryId2).size());
Assert.assertEquals(4, manager.getTimeoutQueue().size());
Assert.assertEquals(4, manager.getReadyQueue().size());
- DriverTask task4 =
- manager.getTimeoutQueue().get(new DriverTaskID(instanceId4));
+ DriverTask task4 = manager.getTimeoutQueue().get(new DriverTaskID(instanceId4));
Assert.assertNotNull(task4);
Assert.assertTrue(manager.getQueryMap().get(queryId2).contains(task4));
Assert.assertEquals(DriverTaskStatus.READY, task4.getStatus());
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DriverTaskTimeoutSentinelThreadTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DriverTaskTimeoutSentinelThreadTest.java
index 4ce669c1d2..9e6e7b3293 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DriverTaskTimeoutSentinelThreadTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DriverTaskTimeoutSentinelThreadTest.java
@@ -55,18 +55,15 @@ public class DriverTaskTimeoutSentinelThreadTest {
PlanFragmentId fragmentId = new PlanFragmentId(queryId, 0);
FragmentInstanceId instanceId = new FragmentInstanceId(fragmentId, "inst-0");
IndexedBlockingQueue<DriverTask> taskQueue =
- new L1PriorityQueue<>(
- 100, new DriverTask.TimeoutComparator(), new DriverTask());
+ new L1PriorityQueue<>(100, new DriverTask.TimeoutComparator(), new DriverTask());
IDriver mockDriver = Mockito.mock(IDriver.class);
Mockito.when(mockDriver.getInfo()).thenReturn(instanceId);
AbstractDriverThread executor =
- new DriverTaskThread(
- "0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
+ new DriverTaskThread("0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
// FINISHED status test
- DriverTask testTask =
- new DriverTask(mockDriver, 100L, DriverTaskStatus.FINISHED);
+ DriverTask testTask = new DriverTask(mockDriver, 100L, DriverTaskStatus.FINISHED);
executor.execute(testTask);
Assert.assertEquals(DriverTaskStatus.FINISHED, testTask.getStatus());
Mockito.verify(mockDriver, Mockito.never()).processFor(Mockito.any());
@@ -112,8 +109,7 @@ public class DriverTaskTimeoutSentinelThreadTest {
return true;
});
IndexedBlockingQueue<DriverTask> taskQueue =
- new L1PriorityQueue<>(
- 100, new DriverTask.TimeoutComparator(), new DriverTask());
+ new L1PriorityQueue<>(100, new DriverTask.TimeoutComparator(), new DriverTask());
// Mock the instance with a cancelled future
IDriver mockDriver = Mockito.mock(IDriver.class);
@@ -125,10 +121,8 @@ public class DriverTaskTimeoutSentinelThreadTest {
.thenReturn(Futures.immediateCancelledFuture());
AbstractDriverThread executor =
- new DriverTaskThread(
- "0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
- DriverTask testTask =
- new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
+ new DriverTaskThread("0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
+ DriverTask testTask = new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
executor.execute(testTask);
Mockito.verify(mockDriver, Mockito.times(1)).processFor(Mockito.any());
Assert.assertEquals(
@@ -154,8 +148,7 @@ public class DriverTaskTimeoutSentinelThreadTest {
return true;
});
IndexedBlockingQueue<DriverTask> taskQueue =
- new L1PriorityQueue<>(
- 100, new DriverTask.TimeoutComparator(), new DriverTask());
+ new L1PriorityQueue<>(100, new DriverTask.TimeoutComparator(), new DriverTask());
// Mock the instance with a cancelled future
IDriver mockDriver = Mockito.mock(IDriver.class);
@@ -166,10 +159,8 @@ public class DriverTaskTimeoutSentinelThreadTest {
Mockito.when(mockDriver.processFor(Mockito.any())).thenReturn(Futures.immediateVoidFuture());
Mockito.when(mockDriver.isFinished()).thenReturn(true);
AbstractDriverThread executor =
- new DriverTaskThread(
- "0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
- DriverTask testTask =
- new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
+ new DriverTaskThread("0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
+ DriverTask testTask = new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
executor.execute(testTask);
Mockito.verify(mockDriver, Mockito.times(1)).processFor(Mockito.any());
Assert.assertNull(testTask.getAbortCause());
@@ -194,8 +185,7 @@ public class DriverTaskTimeoutSentinelThreadTest {
return true;
});
IndexedBlockingQueue<DriverTask> taskQueue =
- new L1PriorityQueue<>(
- 100, new DriverTask.TimeoutComparator(), new DriverTask());
+ new L1PriorityQueue<>(100, new DriverTask.TimeoutComparator(), new DriverTask());
// Mock the instance with a blocked future
ListenableFuture<Void> mockFuture = Mockito.mock(ListenableFuture.class);
@@ -217,10 +207,8 @@ public class DriverTaskTimeoutSentinelThreadTest {
Mockito.when(mockDriver.processFor(Mockito.any())).thenReturn(mockFuture);
Mockito.when(mockDriver.isFinished()).thenReturn(false);
AbstractDriverThread executor =
- new DriverTaskThread(
- "0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
- DriverTask testTask =
- new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
+ new DriverTaskThread("0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
+ DriverTask testTask = new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
executor.execute(testTask);
Mockito.verify(mockDriver, Mockito.times(1)).processFor(Mockito.any());
Assert.assertNull(testTask.getAbortCause());
@@ -245,8 +233,7 @@ public class DriverTaskTimeoutSentinelThreadTest {
return true;
});
IndexedBlockingQueue<DriverTask> taskQueue =
- new L1PriorityQueue<>(
- 100, new DriverTask.TimeoutComparator(), new DriverTask());
+ new L1PriorityQueue<>(100, new DriverTask.TimeoutComparator(), new DriverTask());
// Mock the instance with a ready future
ListenableFuture<Void> mockFuture = Mockito.mock(ListenableFuture.class);
@@ -268,10 +255,8 @@ public class DriverTaskTimeoutSentinelThreadTest {
Mockito.when(mockDriver.processFor(Mockito.any())).thenReturn(mockFuture);
Mockito.when(mockDriver.isFinished()).thenReturn(false);
AbstractDriverThread executor =
- new DriverTaskThread(
- "0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
- DriverTask testTask =
- new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
+ new DriverTaskThread("0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
+ DriverTask testTask = new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
executor.execute(testTask);
Mockito.verify(mockDriver, Mockito.times(1)).processFor(Mockito.any());
Assert.assertNull(testTask.getAbortCause());
[iotdb] 01/02: add some comment and rename Driver related class
Posted by qi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch add_comment
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 45dc2b1a693123d5b1a52c34b88a2c80c3151280
Author: qiaojialin <64...@qq.com>
AuthorDate: Wed Apr 27 21:23:45 2022 +0800
add some comment and rename Driver related class
---
.../apache/iotdb/db/mpp/execution/DataDriver.java | 4 +-
.../org/apache/iotdb/db/mpp/execution/Driver.java | 31 +---
.../mpp/execution/FragmentInstanceExecution.java | 16 +-
.../db/mpp/execution/FragmentInstanceManager.java | 6 +-
.../org/apache/iotdb/db/mpp/execution/IDriver.java | 33 ++++
...ractExecutor.java => AbstractDriverThread.java} | 18 +--
...InstanceScheduler.java => DriverScheduler.java} | 112 ++++++-------
...anceTaskExecutor.java => DriverTaskThread.java} | 12 +-
...l.java => DriverTaskTimeoutSentinelThread.java} | 12 +-
.../iotdb/db/mpp/schedule/ExecutionContext.java | 4 +-
...nstanceScheduler.java => IDriverScheduler.java} | 6 +-
.../iotdb/db/mpp/schedule/ITaskScheduler.java | 40 ++---
.../{FragmentInstanceTask.java => DriverTask.java} | 40 ++---
...agmentInstanceTaskID.java => DriverTaskID.java} | 8 +-
...stanceTaskStatus.java => DriverTaskStatus.java} | 4 +-
.../java/org/apache/iotdb/db/service/DataNode.java | 4 +-
.../java/org/apache/iotdb/db/service/IoTDB.java | 4 +-
.../iotdb/db/mpp/execution/DataDriverTest.java | 2 +-
.../db/mpp/schedule/DefaultTaskSchedulerTest.java | 178 ++++++++++-----------
...SchedulerTest.java => DriverSchedulerTest.java} | 56 +++----
...va => DriverTaskTimeoutSentinelThreadTest.java} | 110 ++++++-------
21 files changed, 350 insertions(+), 350 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/DataDriver.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/DataDriver.java
index e6e8e8083a..899d9d2e0e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/DataDriver.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/DataDriver.java
@@ -99,7 +99,7 @@ public class DataDriver extends Driver {
List<DataSourceOperator> sourceOperators =
((DataDriverContext) driverContext).getSourceOperators();
if (sourceOperators != null && !sourceOperators.isEmpty()) {
- QueryDataSource dataSource = initQueryDataSourceCache();
+ QueryDataSource dataSource = initQueryDataSource();
sourceOperators.forEach(
sourceOperator -> {
// construct QueryDataSource for source operator
@@ -119,7 +119,7 @@ public class DataDriver extends Driver {
* The method is called in mergeLock() when executing query. This method will get all the
* QueryDataSource needed for this query
*/
- public QueryDataSource initQueryDataSourceCache() throws QueryProcessException {
+ private QueryDataSource initQueryDataSource() throws QueryProcessException {
DataDriverContext context = (DataDriverContext) driverContext;
DataRegion dataRegion = context.getDataRegion();
dataRegion.readLock();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/Driver.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/Driver.java
index 40da49ff15..27cb1d361f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/Driver.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/Driver.java
@@ -47,10 +47,7 @@ import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static java.lang.Boolean.TRUE;
import static org.apache.iotdb.db.mpp.operator.Operator.NOT_BLOCKED;
-/**
- * Driver encapsulates some methods which are necessary for execution scheduler to run a fragment
- * instance
- */
+
public abstract class Driver implements IDriver {
protected static final Logger LOGGER = LoggerFactory.getLogger(Driver.class);
@@ -83,11 +80,6 @@ public abstract class Driver implements IDriver {
driverBlockedFuture.set(future);
}
- /**
- * Used to judge whether this fragment instance should be scheduled for execution anymore
- *
- * @return true if the FragmentInstance is done or terminated due to failure, otherwise false.
- */
@Override
public boolean isFinished() {
checkLockNotHeld("Cannot check finished status while holding the driver lock");
@@ -107,16 +99,6 @@ public abstract class Driver implements IDriver {
/** release resource this driver used */
protected abstract void releaseResource();
- /**
- * run the fragment instance for {@param duration} time slice, the time of this run is likely not
- * to be equal to {@param duration}, the actual run time should be calculated by the caller
- *
- * @param duration how long should this fragment instance run
- * @return the returned ListenableFuture<Void> is used to represent status of this processing if
- * isDone() return true, meaning that this fragment instance is not blocked and is ready for
- * next processing otherwise, meaning that this fragment instance is blocked and not ready for
- * next processing.
- */
@Override
public ListenableFuture<Void> processFor(Duration duration) {
@@ -154,17 +136,11 @@ public abstract class Driver implements IDriver {
return result.orElse(NOT_BLOCKED);
}
- /**
- * the id information about this Fragment Instance.
- *
- * @return a {@link FragmentInstanceId} instance.
- */
@Override
public FragmentInstanceId getInfo() {
return driverContext.getId();
}
- /** clear resource used by this fragment instance */
@Override
public void close() {
// mark the service for destruction
@@ -178,11 +154,6 @@ public abstract class Driver implements IDriver {
tryWithLockUnInterruptibly(() -> TRUE);
}
- /**
- * fail current driver
- *
- * @param t reason cause this failure
- */
@Override
public void failed(Throwable t) {
driverContext.failed(t);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceExecution.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceExecution.java
index 15c85c4d99..be2fcc6d68 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceExecution.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceExecution.java
@@ -20,7 +20,7 @@ package org.apache.iotdb.db.mpp.execution;
import org.apache.iotdb.db.mpp.buffer.ISinkHandle;
import org.apache.iotdb.db.mpp.common.FragmentInstanceId;
-import org.apache.iotdb.db.mpp.schedule.IFragmentInstanceScheduler;
+import org.apache.iotdb.db.mpp.schedule.IDriverScheduler;
import com.google.common.collect.ImmutableList;
import io.airlift.stats.CounterStat;
@@ -30,8 +30,6 @@ import static org.apache.iotdb.db.mpp.execution.FragmentInstanceState.FAILED;
public class FragmentInstanceExecution {
- private final IFragmentInstanceScheduler scheduler;
-
private final FragmentInstanceId instanceId;
private final FragmentInstanceContext context;
@@ -44,31 +42,29 @@ public class FragmentInstanceExecution {
private long lastHeartbeat;
public static FragmentInstanceExecution createFragmentInstanceExecution(
- IFragmentInstanceScheduler scheduler,
+ IDriverScheduler scheduler,
FragmentInstanceId instanceId,
FragmentInstanceContext context,
IDriver driver,
FragmentInstanceStateMachine stateMachine,
CounterStat failedInstances) {
FragmentInstanceExecution execution =
- new FragmentInstanceExecution(scheduler, instanceId, context, driver, stateMachine);
- execution.initialize(failedInstances);
+ new FragmentInstanceExecution(instanceId, context, driver, stateMachine);
+ execution.initialize(failedInstances, scheduler);
+ scheduler.submitDrivers(instanceId.getQueryId(), ImmutableList.of(driver));
return execution;
}
private FragmentInstanceExecution(
- IFragmentInstanceScheduler scheduler,
FragmentInstanceId instanceId,
FragmentInstanceContext context,
IDriver driver,
FragmentInstanceStateMachine stateMachine) {
- this.scheduler = scheduler;
this.instanceId = instanceId;
this.context = context;
this.driver = driver;
this.sinkHandle = driver.getSinkHandle();
this.stateMachine = stateMachine;
- scheduler.submitFragmentInstances(instanceId.getQueryId(), ImmutableList.of(driver));
}
public void recordHeartbeat() {
@@ -101,7 +97,7 @@ public class FragmentInstanceExecution {
}
// this is a separate method to ensure that the `this` reference is not leaked during construction
- private void initialize(CounterStat failedInstances) {
+ private void initialize(CounterStat failedInstances, IDriverScheduler scheduler) {
requireNonNull(failedInstances, "failedInstances is null");
stateMachine.addStateChangeListener(
newState -> {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceManager.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceManager.java
index 438e1bea0e..a3abfe65e1 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceManager.java
@@ -22,8 +22,8 @@ import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.metadata.schemaregion.ISchemaRegion;
import org.apache.iotdb.db.mpp.common.FragmentInstanceId;
-import org.apache.iotdb.db.mpp.schedule.FragmentInstanceScheduler;
-import org.apache.iotdb.db.mpp.schedule.IFragmentInstanceScheduler;
+import org.apache.iotdb.db.mpp.schedule.DriverScheduler;
+import org.apache.iotdb.db.mpp.schedule.IDriverScheduler;
import org.apache.iotdb.db.mpp.sql.planner.LocalExecutionPlanner;
import org.apache.iotdb.db.mpp.sql.planner.plan.FragmentInstance;
@@ -49,7 +49,7 @@ public class FragmentInstanceManager {
private final Map<FragmentInstanceId, FragmentInstanceContext> instanceContext;
private final Map<FragmentInstanceId, FragmentInstanceExecution> instanceExecution;
private final LocalExecutionPlanner planner = LocalExecutionPlanner.getInstance();
- private final IFragmentInstanceScheduler scheduler = FragmentInstanceScheduler.getInstance();
+ private final IDriverScheduler scheduler = DriverScheduler.getInstance();
private final ScheduledExecutorService instanceManagementExecutor;
private final ExecutorService instanceNotificationExecutor;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/IDriver.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/IDriver.java
index 54c31901a2..4914760351 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/IDriver.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/IDriver.java
@@ -24,17 +24,50 @@ import org.apache.iotdb.db.mpp.common.FragmentInstanceId;
import com.google.common.util.concurrent.ListenableFuture;
import io.airlift.units.Duration;
+/**
+ * IDriver encapsulates some methods which are necessary for FragmentInstanceTaskExecutor to run a fragment
+ * instance
+ */
public interface IDriver {
+ /**
+ * Used to judge whether this IDriver should be scheduled for execution anymore
+ *
+ * @return true if the IDriver is done or terminated due to failure, otherwise false.
+ */
boolean isFinished();
+ /**
+ * run the IDriver for {@param duration} time slice, the time of this run is likely not
+ * to be equal to {@param duration}, the actual run time should be calculated by the caller
+ *
+ * @param duration how long should this IDriver run
+ * @return the returned ListenableFuture<Void> is used to represent status of this processing if
+ * isDone() return true, meaning that this IDriver is not blocked and is ready for
+ * next processing. Otherwise, meaning that this IDriver is blocked and not ready for
+ * next processing.
+ */
ListenableFuture<Void> processFor(Duration duration);
+ /**
+ * the id information about this IDriver.
+ *
+ * @return a {@link FragmentInstanceId} instance.
+ */
FragmentInstanceId getInfo();
+ /** clear resource used by this fragment instance */
void close();
+ /**
+ * fail current driver
+ *
+ * @param t reason cause this failure
+ */
void failed(Throwable t);
+ /**
+ * @return get SinkHandle of current IDriver
+ */
ISinkHandle getSinkHandle();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/AbstractExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/AbstractDriverThread.java
similarity index 79%
rename from server/src/main/java/org/apache/iotdb/db/mpp/schedule/AbstractExecutor.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/schedule/AbstractDriverThread.java
index d3df9bbb1e..13118452a8 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/AbstractExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/AbstractDriverThread.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.db.mpp.schedule;
import org.apache.iotdb.db.mpp.schedule.queue.IndexedBlockingQueue;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -28,18 +28,18 @@ import java.io.Closeable;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
-/** an abstract executor for {@link FragmentInstanceTask} */
-public abstract class AbstractExecutor extends Thread implements Closeable {
+/** an abstract executor for {@link DriverTask} */
+public abstract class AbstractDriverThread extends Thread implements Closeable {
- private static final Logger logger = LoggerFactory.getLogger(AbstractExecutor.class);
- private final IndexedBlockingQueue<FragmentInstanceTask> queue;
+ private static final Logger logger = LoggerFactory.getLogger(AbstractDriverThread.class);
+ private final IndexedBlockingQueue<DriverTask> queue;
protected final ITaskScheduler scheduler;
private volatile boolean closed;
- public AbstractExecutor(
+ public AbstractDriverThread(
String workerId,
ThreadGroup tg,
- IndexedBlockingQueue<FragmentInstanceTask> queue,
+ IndexedBlockingQueue<DriverTask> queue,
ITaskScheduler scheduler) {
super(tg, workerId);
this.queue = queue;
@@ -51,7 +51,7 @@ public abstract class AbstractExecutor extends Thread implements Closeable {
public void run() {
while (!closed && !Thread.currentThread().isInterrupted()) {
try {
- FragmentInstanceTask next = queue.poll();
+ DriverTask next = queue.poll();
execute(next);
} catch (InterruptedException e) {
break;
@@ -62,7 +62,7 @@ public abstract class AbstractExecutor extends Thread implements Closeable {
}
/** Processing a task. */
- protected abstract void execute(FragmentInstanceTask task)
+ protected abstract void execute(DriverTask task)
throws InterruptedException, ExecutionException;
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceScheduler.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/DriverScheduler.java
similarity index 68%
rename from server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceScheduler.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/schedule/DriverScheduler.java
index 880041924a..1089c3410c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/DriverScheduler.java
@@ -30,9 +30,9 @@ import org.apache.iotdb.db.mpp.execution.IDriver;
import org.apache.iotdb.db.mpp.schedule.queue.IndexedBlockingQueue;
import org.apache.iotdb.db.mpp.schedule.queue.L1PriorityQueue;
import org.apache.iotdb.db.mpp.schedule.queue.L2PriorityQueue;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTaskID;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTaskStatus;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTask;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTaskID;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTaskStatus;
import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstanceId;
import org.slf4j.Logger;
@@ -49,18 +49,18 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
/** the manager of fragment instances scheduling */
-public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IService {
+public class DriverScheduler implements IDriverScheduler, IService {
- private static final Logger logger = LoggerFactory.getLogger(FragmentInstanceScheduler.class);
+ private static final Logger logger = LoggerFactory.getLogger(DriverScheduler.class);
- public static FragmentInstanceScheduler getInstance() {
+ public static DriverScheduler getInstance() {
return InstanceHolder.instance;
}
- private final IndexedBlockingQueue<FragmentInstanceTask> readyQueue;
- private final IndexedBlockingQueue<FragmentInstanceTask> timeoutQueue;
- private final Set<FragmentInstanceTask> blockedTasks;
- private final Map<QueryId, Set<FragmentInstanceTask>> queryMap;
+ private final IndexedBlockingQueue<DriverTask> readyQueue;
+ private final IndexedBlockingQueue<DriverTask> timeoutQueue;
+ private final Set<DriverTask> blockedTasks;
+ private final Map<QueryId, Set<DriverTask>> queryMap;
private final ITaskScheduler scheduler;
private IDataBlockManager blockManager; // TODO: init with real IDataBlockManager
@@ -68,17 +68,17 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
private static final int WORKER_THREAD_NUM = 4; // TODO: load from config files
private static final int QUERY_TIMEOUT_MS = 10_000; // TODO: load from config files or requests
private final ThreadGroup workerGroups;
- private final List<AbstractExecutor> threads;
+ private final List<AbstractDriverThread> threads;
- private FragmentInstanceScheduler() {
+ private DriverScheduler() {
this.readyQueue =
new L2PriorityQueue<>(
MAX_CAPACITY,
- new FragmentInstanceTask.SchedulePriorityComparator(),
- new FragmentInstanceTask());
+ new DriverTask.SchedulePriorityComparator(),
+ new DriverTask());
this.timeoutQueue =
new L1PriorityQueue<>(
- MAX_CAPACITY, new FragmentInstanceTask.TimeoutComparator(), new FragmentInstanceTask());
+ MAX_CAPACITY, new DriverTask.TimeoutComparator(), new DriverTask());
this.queryMap = new ConcurrentHashMap<>();
this.blockedTasks = Collections.synchronizedSet(new HashSet<>());
this.scheduler = new Scheduler();
@@ -90,14 +90,14 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
@Override
public void start() throws StartupException {
for (int i = 0; i < WORKER_THREAD_NUM; i++) {
- AbstractExecutor t =
- new FragmentInstanceTaskExecutor(
+ AbstractDriverThread t =
+ new DriverTaskThread(
"Worker-Thread-" + i, workerGroups, readyQueue, scheduler);
threads.add(t);
t.start();
}
- AbstractExecutor t =
- new FragmentInstanceTimeoutSentinel(
+ AbstractDriverThread t =
+ new DriverTaskTimeoutSentinelThread(
"Sentinel-Thread", workerGroups, timeoutQueue, scheduler);
threads.add(t);
t.start();
@@ -121,20 +121,20 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
}
@Override
- public void submitFragmentInstances(QueryId queryId, List<IDriver> instances) {
- List<FragmentInstanceTask> tasks =
+ public void submitDrivers(QueryId queryId, List<IDriver> instances) {
+ List<DriverTask> tasks =
instances.stream()
.map(
v ->
- new FragmentInstanceTask(v, QUERY_TIMEOUT_MS, FragmentInstanceTaskStatus.READY))
+ new DriverTask(v, QUERY_TIMEOUT_MS, DriverTaskStatus.READY))
.collect(Collectors.toList());
queryMap
.computeIfAbsent(queryId, v -> Collections.synchronizedSet(new HashSet<>()))
.addAll(tasks);
- for (FragmentInstanceTask task : tasks) {
+ for (DriverTask task : tasks) {
task.lock();
try {
- if (task.getStatus() != FragmentInstanceTaskStatus.READY) {
+ if (task.getStatus() != DriverTaskStatus.READY) {
continue;
}
timeoutQueue.push(task);
@@ -147,9 +147,9 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
@Override
public void abortQuery(QueryId queryId) {
- Set<FragmentInstanceTask> queryRelatedTasks = queryMap.remove(queryId);
+ Set<DriverTask> queryRelatedTasks = queryMap.remove(queryId);
if (queryRelatedTasks != null) {
- for (FragmentInstanceTask task : queryRelatedTasks) {
+ for (DriverTask task : queryRelatedTasks) {
task.lock();
try {
task.setAbortCause(FragmentInstanceAbortedException.BY_QUERY_CASCADING_ABORTED);
@@ -163,7 +163,7 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
@Override
public void abortFragmentInstance(FragmentInstanceId instanceId) {
- FragmentInstanceTask task = timeoutQueue.get(new FragmentInstanceTaskID(instanceId));
+ DriverTask task = timeoutQueue.get(new DriverTaskID(instanceId));
if (task == null) {
return;
}
@@ -178,7 +178,7 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
@Override
public double getSchedulePriority(FragmentInstanceId instanceId) {
- FragmentInstanceTask task = timeoutQueue.get(new FragmentInstanceTaskID(instanceId));
+ DriverTask task = timeoutQueue.get(new DriverTaskID(instanceId));
if (task == null) {
throw new IllegalStateException(
"the fragmentInstance " + instanceId.getFullId() + " has been cleared");
@@ -186,9 +186,9 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
return task.getSchedulePriority();
}
- private void clearFragmentInstanceTask(FragmentInstanceTask task) {
- if (task.getStatus() != FragmentInstanceTaskStatus.FINISHED) {
- task.setStatus(FragmentInstanceTaskStatus.ABORTED);
+ private void clearFragmentInstanceTask(DriverTask task) {
+ if (task.getStatus() != DriverTaskStatus.FINISHED) {
+ task.setStatus(DriverTaskStatus.ABORTED);
}
if (task.getAbortCause() != null) {
task.getFragmentInstance()
@@ -196,7 +196,7 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
new FragmentInstanceAbortedException(
task.getFragmentInstance().getInfo(), task.getAbortCause()));
}
- if (task.getStatus() == FragmentInstanceTaskStatus.ABORTED) {
+ if (task.getStatus() == DriverTaskStatus.ABORTED) {
blockManager.forceDeregisterFragmentInstance(
new TFragmentInstanceId(
task.getId().getQueryId().getId(),
@@ -206,7 +206,7 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
readyQueue.remove(task.getId());
timeoutQueue.remove(task.getId());
blockedTasks.remove(task);
- Set<FragmentInstanceTask> tasks = queryMap.get(task.getId().getQueryId());
+ Set<DriverTask> tasks = queryMap.get(task.getId().getQueryId());
if (tasks != null) {
tasks.remove(task);
if (tasks.isEmpty()) {
@@ -220,22 +220,22 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
}
@TestOnly
- IndexedBlockingQueue<FragmentInstanceTask> getReadyQueue() {
+ IndexedBlockingQueue<DriverTask> getReadyQueue() {
return readyQueue;
}
@TestOnly
- IndexedBlockingQueue<FragmentInstanceTask> getTimeoutQueue() {
+ IndexedBlockingQueue<DriverTask> getTimeoutQueue() {
return timeoutQueue;
}
@TestOnly
- Set<FragmentInstanceTask> getBlockedTasks() {
+ Set<DriverTask> getBlockedTasks() {
return blockedTasks;
}
@TestOnly
- Map<QueryId, Set<FragmentInstanceTask>> getQueryMap() {
+ Map<QueryId, Set<DriverTask>> getQueryMap() {
return queryMap;
}
@@ -248,18 +248,18 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
private InstanceHolder() {}
- private static final FragmentInstanceScheduler instance = new FragmentInstanceScheduler();
+ private static final DriverScheduler instance = new DriverScheduler();
}
/** the default scheduler implementation */
private class Scheduler implements ITaskScheduler {
@Override
- public void blockedToReady(FragmentInstanceTask task) {
+ public void blockedToReady(DriverTask task) {
task.lock();
try {
- if (task.getStatus() != FragmentInstanceTaskStatus.BLOCKED) {
+ if (task.getStatus() != DriverTaskStatus.BLOCKED) {
return;
}
- task.setStatus(FragmentInstanceTaskStatus.READY);
+ task.setStatus(DriverTaskStatus.READY);
readyQueue.push(task);
blockedTasks.remove(task);
} finally {
@@ -268,13 +268,13 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
}
@Override
- public boolean readyToRunning(FragmentInstanceTask task) {
+ public boolean readyToRunning(DriverTask task) {
task.lock();
try {
- if (task.getStatus() != FragmentInstanceTaskStatus.READY) {
+ if (task.getStatus() != DriverTaskStatus.READY) {
return false;
}
- task.setStatus(FragmentInstanceTaskStatus.RUNNING);
+ task.setStatus(DriverTaskStatus.RUNNING);
} finally {
task.unlock();
}
@@ -282,14 +282,14 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
}
@Override
- public void runningToReady(FragmentInstanceTask task, ExecutionContext context) {
+ public void runningToReady(DriverTask task, ExecutionContext context) {
task.lock();
try {
- if (task.getStatus() != FragmentInstanceTaskStatus.RUNNING) {
+ if (task.getStatus() != DriverTaskStatus.RUNNING) {
return;
}
task.updateSchedulePriority(context);
- task.setStatus(FragmentInstanceTaskStatus.READY);
+ task.setStatus(DriverTaskStatus.READY);
readyQueue.push(task);
} finally {
task.unlock();
@@ -297,14 +297,14 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
}
@Override
- public void runningToBlocked(FragmentInstanceTask task, ExecutionContext context) {
+ public void runningToBlocked(DriverTask task, ExecutionContext context) {
task.lock();
try {
- if (task.getStatus() != FragmentInstanceTaskStatus.RUNNING) {
+ if (task.getStatus() != DriverTaskStatus.RUNNING) {
return;
}
task.updateSchedulePriority(context);
- task.setStatus(FragmentInstanceTaskStatus.BLOCKED);
+ task.setStatus(DriverTaskStatus.BLOCKED);
blockedTasks.add(task);
} finally {
task.unlock();
@@ -312,14 +312,14 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
}
@Override
- public void runningToFinished(FragmentInstanceTask task, ExecutionContext context) {
+ public void runningToFinished(DriverTask task, ExecutionContext context) {
task.lock();
try {
- if (task.getStatus() != FragmentInstanceTaskStatus.RUNNING) {
+ if (task.getStatus() != DriverTaskStatus.RUNNING) {
return;
}
task.updateSchedulePriority(context);
- task.setStatus(FragmentInstanceTaskStatus.FINISHED);
+ task.setStatus(DriverTaskStatus.FINISHED);
clearFragmentInstanceTask(task);
} finally {
task.unlock();
@@ -327,7 +327,7 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
}
@Override
- public void toAborted(FragmentInstanceTask task) {
+ public void toAborted(DriverTask task) {
task.lock();
try {
// If a task is already in an end state, it indicates that the task is finalized in other
@@ -343,9 +343,9 @@ public class FragmentInstanceScheduler implements IFragmentInstanceScheduler, IS
task.unlock();
}
QueryId queryId = task.getId().getQueryId();
- Set<FragmentInstanceTask> queryRelatedTasks = queryMap.remove(queryId);
+ Set<DriverTask> queryRelatedTasks = queryMap.remove(queryId);
if (queryRelatedTasks != null) {
- for (FragmentInstanceTask otherTask : queryRelatedTasks) {
+ for (DriverTask otherTask : queryRelatedTasks) {
if (task.equals(otherTask)) {
continue;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceTaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/DriverTaskThread.java
similarity index 88%
rename from server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceTaskExecutor.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/schedule/DriverTaskThread.java
index 25b4e29a0a..9c227cf100 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceTaskExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/DriverTaskThread.java
@@ -20,7 +20,7 @@ package org.apache.iotdb.db.mpp.schedule;
import org.apache.iotdb.db.mpp.execution.IDriver;
import org.apache.iotdb.db.mpp.schedule.queue.IndexedBlockingQueue;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTask;
import org.apache.iotdb.db.utils.stats.CpuTimer;
import com.google.common.util.concurrent.ListenableFuture;
@@ -30,24 +30,24 @@ import io.airlift.units.Duration;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
-/** the worker thread of {@link FragmentInstanceTask} */
-public class FragmentInstanceTaskExecutor extends AbstractExecutor {
+/** the worker thread of {@link DriverTask} */
+public class DriverTaskThread extends AbstractDriverThread {
public static final Duration EXECUTION_TIME_SLICE = new Duration(100, TimeUnit.MILLISECONDS);
// As the callback is lightweight enough, there's no need to use another one thread to execute.
private static final Executor listeningExecutor = MoreExecutors.directExecutor();
- public FragmentInstanceTaskExecutor(
+ public DriverTaskThread(
String workerId,
ThreadGroup tg,
- IndexedBlockingQueue<FragmentInstanceTask> queue,
+ IndexedBlockingQueue<DriverTask> queue,
ITaskScheduler scheduler) {
super(workerId, tg, queue, scheduler);
}
@Override
- public void execute(FragmentInstanceTask task) throws InterruptedException {
+ public void execute(DriverTask task) throws InterruptedException {
// try to switch it to RUNNING
if (!scheduler.readyToRunning(task)) {
return;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceTimeoutSentinel.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/DriverTaskTimeoutSentinelThread.java
similarity index 81%
rename from server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceTimeoutSentinel.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/schedule/DriverTaskTimeoutSentinelThread.java
index e7aaaf4e47..d0b691f09d 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceTimeoutSentinel.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/DriverTaskTimeoutSentinelThread.java
@@ -19,21 +19,21 @@
package org.apache.iotdb.db.mpp.schedule;
import org.apache.iotdb.db.mpp.schedule.queue.IndexedBlockingQueue;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTask;
-/** the thread for watching the timeout of {@link FragmentInstanceTask} */
-public class FragmentInstanceTimeoutSentinel extends AbstractExecutor {
+/** the thread for watching the timeout of {@link DriverTask} */
+public class DriverTaskTimeoutSentinelThread extends AbstractDriverThread {
- public FragmentInstanceTimeoutSentinel(
+ public DriverTaskTimeoutSentinelThread(
String workerId,
ThreadGroup tg,
- IndexedBlockingQueue<FragmentInstanceTask> queue,
+ IndexedBlockingQueue<DriverTask> queue,
ITaskScheduler scheduler) {
super(workerId, tg, queue, scheduler);
}
@Override
- public void execute(FragmentInstanceTask task) throws InterruptedException {
+ public void execute(DriverTask task) throws InterruptedException {
task.lock();
try {
// if this task is already in an end state, it means that the resource releasing will be
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/ExecutionContext.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/ExecutionContext.java
index e8cd091d2d..10d9f3d597 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/ExecutionContext.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/ExecutionContext.java
@@ -18,12 +18,12 @@
*/
package org.apache.iotdb.db.mpp.schedule;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTask;
import org.apache.iotdb.db.utils.stats.CpuTimer;
import io.airlift.units.Duration;
-/** The execution context of a {@link FragmentInstanceTask} */
+/** The execution context of a {@link DriverTask} */
public class ExecutionContext {
private CpuTimer.CpuDuration cpuDuration;
private Duration timeSlice;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IFragmentInstanceScheduler.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IDriverScheduler.java
similarity index 90%
rename from server/src/main/java/org/apache/iotdb/db/mpp/schedule/IFragmentInstanceScheduler.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/schedule/IDriverScheduler.java
index b8d5970ee9..b92f663eb2 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IFragmentInstanceScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IDriverScheduler.java
@@ -25,7 +25,7 @@ import org.apache.iotdb.db.mpp.execution.IDriver;
import java.util.List;
/** the interface of fragment instance scheduling */
-public interface IFragmentInstanceScheduler {
+public interface IDriverScheduler {
/**
* Submit one or more {@link org.apache.iotdb.db.mpp.execution.IDriver} in one query for later
@@ -34,7 +34,7 @@ public interface IFragmentInstanceScheduler {
* @param queryId the queryId these instances belong to.
* @param instances the submitted instances.
*/
- void submitFragmentInstances(QueryId queryId, List<IDriver> instances);
+ void submitDrivers(QueryId queryId, List<IDriver> instances);
/**
* Abort all the instances in this query.
@@ -44,7 +44,7 @@ public interface IFragmentInstanceScheduler {
void abortQuery(QueryId queryId);
/**
- * Abort the fragment instance. If the instance is not existed, nothing will happen.
+ * Abort all Drivers of the fragment instance. If the instance is not existed, nothing will happen.
*
* @param instanceId the id of the fragment instance to be aborted.
*/
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/ITaskScheduler.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/ITaskScheduler.java
index f4a008b4a2..7f1b973c01 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/ITaskScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/ITaskScheduler.java
@@ -18,60 +18,60 @@
*/
package org.apache.iotdb.db.mpp.schedule;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTaskStatus;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTask;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTaskStatus;
-/** the scheduler interface of {@link FragmentInstanceTask} */
+/** the scheduler interface of {@link DriverTask} */
public interface ITaskScheduler {
/**
- * Switch a task from {@link FragmentInstanceTaskStatus#BLOCKED} to {@link
- * FragmentInstanceTaskStatus#READY}.
+ * Switch a task from {@link DriverTaskStatus#BLOCKED} to {@link
+ * DriverTaskStatus#READY}.
*
* @param task the task to be switched.
*/
- void blockedToReady(FragmentInstanceTask task);
+ void blockedToReady(DriverTask task);
/**
- * Switch a task from {@link FragmentInstanceTaskStatus#READY} to {@link
- * FragmentInstanceTaskStatus#RUNNING}.
+ * Switch a task from {@link DriverTaskStatus#READY} to {@link
+ * DriverTaskStatus#RUNNING}.
*
* @param task the task to be switched.
* @return true if it's switched to the target status successfully, otherwise false.
*/
- boolean readyToRunning(FragmentInstanceTask task);
+ boolean readyToRunning(DriverTask task);
/**
- * Switch a task from {@link FragmentInstanceTaskStatus#RUNNING} to {@link
- * FragmentInstanceTaskStatus#READY}.
+ * Switch a task from {@link DriverTaskStatus#RUNNING} to {@link
+ * DriverTaskStatus#READY}.
*
* @param task the task to be switched.
* @param context the execution context of last running.
*/
- void runningToReady(FragmentInstanceTask task, ExecutionContext context);
+ void runningToReady(DriverTask task, ExecutionContext context);
/**
- * Switch a task from {@link FragmentInstanceTaskStatus#RUNNING} to {@link
- * FragmentInstanceTaskStatus#BLOCKED}.
+ * Switch a task from {@link DriverTaskStatus#RUNNING} to {@link
+ * DriverTaskStatus#BLOCKED}.
*
* @param task the task to be switched.
* @param context the execution context of last running.
*/
- void runningToBlocked(FragmentInstanceTask task, ExecutionContext context);
+ void runningToBlocked(DriverTask task, ExecutionContext context);
/**
- * Switch a task from {@link FragmentInstanceTaskStatus#RUNNING} to {@link
- * FragmentInstanceTaskStatus#FINISHED}.
+ * Switch a task from {@link DriverTaskStatus#RUNNING} to {@link
+ * DriverTaskStatus#FINISHED}.
*
* @param task the task to be switched.
* @param context the execution context of last running.
*/
- void runningToFinished(FragmentInstanceTask task, ExecutionContext context);
+ void runningToFinished(DriverTask task, ExecutionContext context);
/**
- * Switch a task to {@link FragmentInstanceTaskStatus#ABORTED}.
+ * Switch a task to {@link DriverTaskStatus#ABORTED}.
*
* @param task the task to be switched.
*/
- void toAborted(FragmentInstanceTask task);
+ void toAborted(DriverTask task);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/FragmentInstanceTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/DriverTask.java
similarity index 81%
rename from server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/FragmentInstanceTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/DriverTask.java
index fc322fee28..4844a0de89 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/FragmentInstanceTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/DriverTask.java
@@ -24,7 +24,7 @@ import org.apache.iotdb.db.mpp.common.PlanFragmentId;
import org.apache.iotdb.db.mpp.common.QueryId;
import org.apache.iotdb.db.mpp.execution.IDriver;
import org.apache.iotdb.db.mpp.schedule.ExecutionContext;
-import org.apache.iotdb.db.mpp.schedule.FragmentInstanceTaskExecutor;
+import org.apache.iotdb.db.mpp.schedule.DriverTaskThread;
import org.apache.iotdb.db.mpp.schedule.queue.ID;
import org.apache.iotdb.db.mpp.schedule.queue.IDIndexedAccessible;
@@ -37,13 +37,13 @@ import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
- * the scheduling element of {@link FragmentInstanceTaskExecutor}. It wraps a single
- * FragmentInstance.
+ * the scheduling element of {@link DriverTaskThread}. It wraps a single
+ * Driver.
*/
-public class FragmentInstanceTask implements IDIndexedAccessible {
+public class DriverTask implements IDIndexedAccessible {
- private FragmentInstanceTaskID id;
- private FragmentInstanceTaskStatus status;
+ private DriverTaskID id;
+ private DriverTaskStatus status;
private final IDriver fragmentInstance;
// the higher this field is, the higher probability it will be scheduled.
@@ -57,42 +57,42 @@ public class FragmentInstanceTask implements IDIndexedAccessible {
private String abortCause;
/** Initialize a dummy instance for queryHolder */
- public FragmentInstanceTask() {
+ public DriverTask() {
this(new StubFragmentInstance(), 0L, null);
}
- public FragmentInstanceTask(IDriver instance, long timeoutMs, FragmentInstanceTaskStatus status) {
+ public DriverTask(IDriver instance, long timeoutMs, DriverTaskStatus status) {
this.fragmentInstance = instance;
- this.id = new FragmentInstanceTaskID(instance.getInfo());
+ this.id = new DriverTaskID(instance.getInfo());
this.setStatus(status);
this.schedulePriority = 0.0D;
this.ddl = System.currentTimeMillis() + timeoutMs;
this.lock = new ReentrantLock();
}
- public FragmentInstanceTaskID getId() {
+ public DriverTaskID getId() {
return id;
}
@Override
public void setId(ID id) {
- this.id = (FragmentInstanceTaskID) id;
+ this.id = (DriverTaskID) id;
}
- public FragmentInstanceTaskStatus getStatus() {
+ public DriverTaskStatus getStatus() {
return status;
}
public boolean isEndState() {
- return status == FragmentInstanceTaskStatus.ABORTED
- || status == FragmentInstanceTaskStatus.FINISHED;
+ return status == DriverTaskStatus.ABORTED
+ || status == DriverTaskStatus.FINISHED;
}
public IDriver getFragmentInstance() {
return fragmentInstance;
}
- public void setStatus(FragmentInstanceTaskStatus status) {
+ public void setStatus(DriverTaskStatus status) {
this.status = status;
}
@@ -139,7 +139,7 @@ public class FragmentInstanceTask implements IDIndexedAccessible {
@Override
public boolean equals(Object o) {
- return o instanceof FragmentInstanceTask && ((FragmentInstanceTask) o).getId().equals(id);
+ return o instanceof DriverTask && ((DriverTask) o).getId().equals(id);
}
public String getAbortCause() {
@@ -151,10 +151,10 @@ public class FragmentInstanceTask implements IDIndexedAccessible {
}
/** a comparator of ddl, the less the ddl is, the low order it has. */
- public static class TimeoutComparator implements Comparator<FragmentInstanceTask> {
+ public static class TimeoutComparator implements Comparator<DriverTask> {
@Override
- public int compare(FragmentInstanceTask o1, FragmentInstanceTask o2) {
+ public int compare(DriverTask o1, DriverTask o2) {
if (o1.getId().equals(o2.getId())) {
return 0;
}
@@ -169,10 +169,10 @@ public class FragmentInstanceTask implements IDIndexedAccessible {
}
/** a comparator of ddl, the higher the schedulePriority is, the low order it has. */
- public static class SchedulePriorityComparator implements Comparator<FragmentInstanceTask> {
+ public static class SchedulePriorityComparator implements Comparator<DriverTask> {
@Override
- public int compare(FragmentInstanceTask o1, FragmentInstanceTask o2) {
+ public int compare(DriverTask o1, DriverTask o2) {
if (o1.getId().equals(o2.getId())) {
return 0;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/FragmentInstanceTaskID.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/DriverTaskID.java
similarity index 85%
rename from server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/FragmentInstanceTaskID.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/DriverTaskID.java
index a0081b177c..fe7421667b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/FragmentInstanceTaskID.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/DriverTaskID.java
@@ -26,17 +26,17 @@ import org.apache.iotdb.db.mpp.schedule.queue.ID;
import org.jetbrains.annotations.NotNull;
/** the class of id of the fragment instance task */
-public class FragmentInstanceTaskID implements ID, Comparable<FragmentInstanceTaskID> {
+public class DriverTaskID implements ID, Comparable<DriverTaskID> {
private final FragmentInstanceId id;
- public FragmentInstanceTaskID(FragmentInstanceId id) {
+ public DriverTaskID(FragmentInstanceId id) {
this.id = id;
}
@Override
public boolean equals(Object o) {
- return o instanceof FragmentInstanceTaskID && ((FragmentInstanceTaskID) o).id.equals(id);
+ return o instanceof DriverTaskID && ((DriverTaskID) o).id.equals(id);
}
@Override
@@ -62,7 +62,7 @@ public class FragmentInstanceTaskID implements ID, Comparable<FragmentInstanceTa
// This is the default comparator of FragmentInstanceID
@Override
- public int compareTo(@NotNull FragmentInstanceTaskID o) {
+ public int compareTo(@NotNull DriverTaskID o) {
return String.CASE_INSENSITIVE_ORDER.compare(this.toString(), o.toString());
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/FragmentInstanceTaskStatus.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/DriverTaskStatus.java
similarity index 92%
rename from server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/FragmentInstanceTaskStatus.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/DriverTaskStatus.java
index f50dc4dfdc..09dbeac9c0 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/FragmentInstanceTaskStatus.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/task/DriverTaskStatus.java
@@ -18,8 +18,8 @@
*/
package org.apache.iotdb.db.mpp.schedule.task;
-/** the status enum of {@link FragmentInstanceTask} */
-public enum FragmentInstanceTaskStatus {
+/** the status enum of {@link DriverTask} */
+public enum DriverTaskStatus {
/* Ready to be executed */
READY,
diff --git a/server/src/main/java/org/apache/iotdb/db/service/DataNode.java b/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
index 0c474d6334..84d5fb254c 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -44,7 +44,7 @@ import org.apache.iotdb.db.engine.flush.FlushManager;
import org.apache.iotdb.db.engine.trigger.service.TriggerRegistrationService;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.mpp.buffer.DataBlockService;
-import org.apache.iotdb.db.mpp.schedule.FragmentInstanceScheduler;
+import org.apache.iotdb.db.mpp.schedule.DriverScheduler;
import org.apache.iotdb.db.protocol.influxdb.meta.InfluxDBMetaManager;
import org.apache.iotdb.db.protocol.rest.RestService;
import org.apache.iotdb.db.query.udf.service.TemporaryQueryDataFileService;
@@ -249,7 +249,7 @@ public class DataNode implements DataNodeMBean {
registerManager.register(StorageEngineV2.getInstance());
registerManager.register(DataBlockService.getInstance());
registerManager.register(InternalService.getInstance());
- registerManager.register(FragmentInstanceScheduler.getInstance());
+ registerManager.register(DriverScheduler.getInstance());
IoTDBDescriptor.getInstance()
.getConfig()
.setRpcImplClassName(DataNodeTSIServiceImpl.class.getName());
diff --git a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
index ae8b8d0b1e..d60c167a00 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
@@ -40,7 +40,7 @@ import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.localconfignode.LocalConfigNode;
import org.apache.iotdb.db.metadata.LocalSchemaProcessor;
import org.apache.iotdb.db.mpp.buffer.DataBlockService;
-import org.apache.iotdb.db.mpp.schedule.FragmentInstanceScheduler;
+import org.apache.iotdb.db.mpp.schedule.DriverScheduler;
import org.apache.iotdb.db.protocol.influxdb.meta.InfluxDBMetaManager;
import org.apache.iotdb.db.protocol.rest.RestService;
import org.apache.iotdb.db.query.udf.service.TemporaryQueryDataFileService;
@@ -147,7 +147,7 @@ public class IoTDB implements IoTDBMBean {
registerManager.register(StorageEngineV2.getInstance());
registerManager.register(DataBlockService.getInstance());
registerManager.register(InternalService.getInstance());
- registerManager.register(FragmentInstanceScheduler.getInstance());
+ registerManager.register(DriverScheduler.getInstance());
IoTDBDescriptor.getInstance()
.getConfig()
.setRpcImplClassName(DataNodeTSIServiceImpl.class.getName());
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java
index 65cca8e51c..5f03cd366a 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java
@@ -61,7 +61,7 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import static org.apache.iotdb.db.mpp.execution.FragmentInstanceContext.createFragmentInstanceContext;
-import static org.apache.iotdb.db.mpp.schedule.FragmentInstanceTaskExecutor.EXECUTION_TIME_SLICE;
+import static org.apache.iotdb.db.mpp.schedule.DriverTaskThread.EXECUTION_TIME_SLICE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DefaultTaskSchedulerTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DefaultTaskSchedulerTest.java
index a93ab988ea..5689663133 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DefaultTaskSchedulerTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DefaultTaskSchedulerTest.java
@@ -23,8 +23,8 @@ import org.apache.iotdb.db.mpp.common.FragmentInstanceId;
import org.apache.iotdb.db.mpp.common.PlanFragmentId;
import org.apache.iotdb.db.mpp.common.QueryId;
import org.apache.iotdb.db.mpp.execution.IDriver;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTaskStatus;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTask;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTaskStatus;
import org.apache.iotdb.db.utils.stats.CpuTimer;
import org.apache.iotdb.mpp.rpc.thrift.InternalService;
@@ -40,7 +40,7 @@ import java.util.concurrent.TimeUnit;
public class DefaultTaskSchedulerTest {
- private final FragmentInstanceScheduler manager = FragmentInstanceScheduler.getInstance();
+ private final DriverScheduler manager = DriverScheduler.getInstance();
@After
public void tearDown() {
@@ -57,17 +57,17 @@ public class DefaultTaskSchedulerTest {
FragmentInstanceId instanceId =
new FragmentInstanceId(new PlanFragmentId(queryId, 0), "inst-0");
Mockito.when(mockDriver.getInfo()).thenReturn(instanceId);
- FragmentInstanceTaskStatus[] invalidStates =
- new FragmentInstanceTaskStatus[] {
- FragmentInstanceTaskStatus.FINISHED,
- FragmentInstanceTaskStatus.ABORTED,
- FragmentInstanceTaskStatus.READY,
- FragmentInstanceTaskStatus.RUNNING,
+ DriverTaskStatus[] invalidStates =
+ new DriverTaskStatus[] {
+ DriverTaskStatus.FINISHED,
+ DriverTaskStatus.ABORTED,
+ DriverTaskStatus.READY,
+ DriverTaskStatus.RUNNING,
};
- for (FragmentInstanceTaskStatus status : invalidStates) {
- FragmentInstanceTask testTask = new FragmentInstanceTask(mockDriver, 100L, status);
+ for (DriverTaskStatus status : invalidStates) {
+ DriverTask testTask = new DriverTask(mockDriver, 100L, status);
manager.getBlockedTasks().add(testTask);
- Set<FragmentInstanceTask> taskSet = new HashSet<>();
+ Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
manager.getQueryMap().put(queryId, taskSet);
manager.getTimeoutQueue().push(testTask);
@@ -80,15 +80,15 @@ public class DefaultTaskSchedulerTest {
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(testTask));
clear();
}
- FragmentInstanceTask testTask =
- new FragmentInstanceTask(mockDriver, 100L, FragmentInstanceTaskStatus.BLOCKED);
+ DriverTask testTask =
+ new DriverTask(mockDriver, 100L, DriverTaskStatus.BLOCKED);
manager.getBlockedTasks().add(testTask);
- Set<FragmentInstanceTask> taskSet = new HashSet<>();
+ Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
manager.getQueryMap().put(queryId, taskSet);
manager.getTimeoutQueue().push(testTask);
defaultScheduler.blockedToReady(testTask);
- Assert.assertEquals(FragmentInstanceTaskStatus.READY, testTask.getStatus());
+ Assert.assertEquals(DriverTaskStatus.READY, testTask.getStatus());
Assert.assertFalse(manager.getBlockedTasks().contains(testTask));
Assert.assertNotNull(manager.getReadyQueue().get(testTask.getId()));
Assert.assertNotNull(manager.getTimeoutQueue().get(testTask.getId()));
@@ -109,16 +109,16 @@ public class DefaultTaskSchedulerTest {
FragmentInstanceId instanceId =
new FragmentInstanceId(new PlanFragmentId(queryId, 0), "inst-0");
Mockito.when(mockDriver.getInfo()).thenReturn(instanceId);
- FragmentInstanceTaskStatus[] invalidStates =
- new FragmentInstanceTaskStatus[] {
- FragmentInstanceTaskStatus.FINISHED,
- FragmentInstanceTaskStatus.ABORTED,
- FragmentInstanceTaskStatus.BLOCKED,
- FragmentInstanceTaskStatus.RUNNING,
+ DriverTaskStatus[] invalidStates =
+ new DriverTaskStatus[] {
+ DriverTaskStatus.FINISHED,
+ DriverTaskStatus.ABORTED,
+ DriverTaskStatus.BLOCKED,
+ DriverTaskStatus.RUNNING,
};
- for (FragmentInstanceTaskStatus status : invalidStates) {
- FragmentInstanceTask testTask = new FragmentInstanceTask(mockDriver, 100L, status);
- Set<FragmentInstanceTask> taskSet = new HashSet<>();
+ for (DriverTaskStatus status : invalidStates) {
+ DriverTask testTask = new DriverTask(mockDriver, 100L, status);
+ Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
manager.getQueryMap().put(queryId, taskSet);
manager.getTimeoutQueue().push(testTask);
@@ -130,14 +130,14 @@ public class DefaultTaskSchedulerTest {
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(testTask));
clear();
}
- FragmentInstanceTask testTask =
- new FragmentInstanceTask(mockDriver, 100L, FragmentInstanceTaskStatus.READY);
- Set<FragmentInstanceTask> taskSet = new HashSet<>();
+ DriverTask testTask =
+ new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
+ Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
manager.getQueryMap().put(queryId, taskSet);
manager.getTimeoutQueue().push(testTask);
defaultScheduler.readyToRunning(testTask);
- Assert.assertEquals(FragmentInstanceTaskStatus.RUNNING, testTask.getStatus());
+ Assert.assertEquals(DriverTaskStatus.RUNNING, testTask.getStatus());
Assert.assertFalse(manager.getBlockedTasks().contains(testTask));
Assert.assertNotNull(manager.getTimeoutQueue().get(testTask.getId()));
Assert.assertTrue(manager.getQueryMap().containsKey(queryId));
@@ -156,16 +156,16 @@ public class DefaultTaskSchedulerTest {
FragmentInstanceId instanceId =
new FragmentInstanceId(new PlanFragmentId(queryId, 0), "inst-0");
Mockito.when(mockDriver.getInfo()).thenReturn(instanceId);
- FragmentInstanceTaskStatus[] invalidStates =
- new FragmentInstanceTaskStatus[] {
- FragmentInstanceTaskStatus.FINISHED,
- FragmentInstanceTaskStatus.ABORTED,
- FragmentInstanceTaskStatus.BLOCKED,
- FragmentInstanceTaskStatus.READY,
+ DriverTaskStatus[] invalidStates =
+ new DriverTaskStatus[] {
+ DriverTaskStatus.FINISHED,
+ DriverTaskStatus.ABORTED,
+ DriverTaskStatus.BLOCKED,
+ DriverTaskStatus.READY,
};
- for (FragmentInstanceTaskStatus status : invalidStates) {
- FragmentInstanceTask testTask = new FragmentInstanceTask(mockDriver, 100L, status);
- Set<FragmentInstanceTask> taskSet = new HashSet<>();
+ for (DriverTaskStatus status : invalidStates) {
+ DriverTask testTask = new DriverTask(mockDriver, 100L, status);
+ Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
manager.getQueryMap().put(queryId, taskSet);
manager.getTimeoutQueue().push(testTask);
@@ -178,9 +178,9 @@ public class DefaultTaskSchedulerTest {
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(testTask));
clear();
}
- FragmentInstanceTask testTask =
- new FragmentInstanceTask(mockDriver, 100L, FragmentInstanceTaskStatus.RUNNING);
- Set<FragmentInstanceTask> taskSet = new HashSet<>();
+ DriverTask testTask =
+ new DriverTask(mockDriver, 100L, DriverTaskStatus.RUNNING);
+ Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
manager.getQueryMap().put(queryId, taskSet);
manager.getTimeoutQueue().push(testTask);
@@ -189,7 +189,7 @@ public class DefaultTaskSchedulerTest {
context.setCpuDuration(new CpuTimer.CpuDuration());
defaultScheduler.runningToReady(testTask, context);
Assert.assertEquals(0.0D, testTask.getSchedulePriority(), 0.00001);
- Assert.assertEquals(FragmentInstanceTaskStatus.READY, testTask.getStatus());
+ Assert.assertEquals(DriverTaskStatus.READY, testTask.getStatus());
Assert.assertFalse(manager.getBlockedTasks().contains(testTask));
Assert.assertNotNull(manager.getReadyQueue().get(testTask.getId()));
Assert.assertNotNull(manager.getTimeoutQueue().get(testTask.getId()));
@@ -209,16 +209,16 @@ public class DefaultTaskSchedulerTest {
FragmentInstanceId instanceId =
new FragmentInstanceId(new PlanFragmentId(queryId, 0), "inst-0");
Mockito.when(mockDriver.getInfo()).thenReturn(instanceId);
- FragmentInstanceTaskStatus[] invalidStates =
- new FragmentInstanceTaskStatus[] {
- FragmentInstanceTaskStatus.FINISHED,
- FragmentInstanceTaskStatus.ABORTED,
- FragmentInstanceTaskStatus.BLOCKED,
- FragmentInstanceTaskStatus.READY,
+ DriverTaskStatus[] invalidStates =
+ new DriverTaskStatus[] {
+ DriverTaskStatus.FINISHED,
+ DriverTaskStatus.ABORTED,
+ DriverTaskStatus.BLOCKED,
+ DriverTaskStatus.READY,
};
- for (FragmentInstanceTaskStatus status : invalidStates) {
- FragmentInstanceTask testTask = new FragmentInstanceTask(mockDriver, 100L, status);
- Set<FragmentInstanceTask> taskSet = new HashSet<>();
+ for (DriverTaskStatus status : invalidStates) {
+ DriverTask testTask = new DriverTask(mockDriver, 100L, status);
+ Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
manager.getQueryMap().put(queryId, taskSet);
manager.getTimeoutQueue().push(testTask);
@@ -231,9 +231,9 @@ public class DefaultTaskSchedulerTest {
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(testTask));
clear();
}
- FragmentInstanceTask testTask =
- new FragmentInstanceTask(mockDriver, 100L, FragmentInstanceTaskStatus.RUNNING);
- Set<FragmentInstanceTask> taskSet = new HashSet<>();
+ DriverTask testTask =
+ new DriverTask(mockDriver, 100L, DriverTaskStatus.RUNNING);
+ Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
manager.getQueryMap().put(queryId, taskSet);
manager.getTimeoutQueue().push(testTask);
@@ -242,7 +242,7 @@ public class DefaultTaskSchedulerTest {
context.setCpuDuration(new CpuTimer.CpuDuration());
defaultScheduler.runningToBlocked(testTask, context);
Assert.assertEquals(0.0D, testTask.getSchedulePriority(), 0.00001);
- Assert.assertEquals(FragmentInstanceTaskStatus.BLOCKED, testTask.getStatus());
+ Assert.assertEquals(DriverTaskStatus.BLOCKED, testTask.getStatus());
Assert.assertTrue(manager.getBlockedTasks().contains(testTask));
Assert.assertNull(manager.getReadyQueue().get(testTask.getId()));
Assert.assertNotNull(manager.getTimeoutQueue().get(testTask.getId()));
@@ -262,16 +262,16 @@ public class DefaultTaskSchedulerTest {
FragmentInstanceId instanceId =
new FragmentInstanceId(new PlanFragmentId(queryId, 0), "inst-0");
Mockito.when(mockDriver.getInfo()).thenReturn(instanceId);
- FragmentInstanceTaskStatus[] invalidStates =
- new FragmentInstanceTaskStatus[] {
- FragmentInstanceTaskStatus.FINISHED,
- FragmentInstanceTaskStatus.ABORTED,
- FragmentInstanceTaskStatus.BLOCKED,
- FragmentInstanceTaskStatus.READY,
+ DriverTaskStatus[] invalidStates =
+ new DriverTaskStatus[] {
+ DriverTaskStatus.FINISHED,
+ DriverTaskStatus.ABORTED,
+ DriverTaskStatus.BLOCKED,
+ DriverTaskStatus.READY,
};
- for (FragmentInstanceTaskStatus status : invalidStates) {
- FragmentInstanceTask testTask = new FragmentInstanceTask(mockDriver, 100L, status);
- Set<FragmentInstanceTask> taskSet = new HashSet<>();
+ for (DriverTaskStatus status : invalidStates) {
+ DriverTask testTask = new DriverTask(mockDriver, 100L, status);
+ Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
manager.getQueryMap().put(queryId, taskSet);
manager.getTimeoutQueue().push(testTask);
@@ -284,9 +284,9 @@ public class DefaultTaskSchedulerTest {
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(testTask));
clear();
}
- FragmentInstanceTask testTask =
- new FragmentInstanceTask(mockDriver, 100L, FragmentInstanceTaskStatus.RUNNING);
- Set<FragmentInstanceTask> taskSet = new HashSet<>();
+ DriverTask testTask =
+ new DriverTask(mockDriver, 100L, DriverTaskStatus.RUNNING);
+ Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask);
manager.getQueryMap().put(queryId, taskSet);
manager.getTimeoutQueue().push(testTask);
@@ -295,7 +295,7 @@ public class DefaultTaskSchedulerTest {
context.setCpuDuration(new CpuTimer.CpuDuration());
defaultScheduler.runningToFinished(testTask, context);
Assert.assertEquals(0.0D, testTask.getSchedulePriority(), 0.00001);
- Assert.assertEquals(FragmentInstanceTaskStatus.FINISHED, testTask.getStatus());
+ Assert.assertEquals(DriverTaskStatus.FINISHED, testTask.getStatus());
Assert.assertFalse(manager.getBlockedTasks().contains(testTask));
Assert.assertNull(manager.getReadyQueue().get(testTask.getId()));
Assert.assertNull(manager.getTimeoutQueue().get(testTask.getId()));
@@ -319,15 +319,15 @@ public class DefaultTaskSchedulerTest {
FragmentInstanceId instanceId2 =
new FragmentInstanceId(new PlanFragmentId(queryId, 0), "inst-1");
Mockito.when(mockDriver2.getInfo()).thenReturn(instanceId2);
- FragmentInstanceTaskStatus[] invalidStates =
- new FragmentInstanceTaskStatus[] {
- FragmentInstanceTaskStatus.FINISHED, FragmentInstanceTaskStatus.ABORTED,
+ DriverTaskStatus[] invalidStates =
+ new DriverTaskStatus[] {
+ DriverTaskStatus.FINISHED, DriverTaskStatus.ABORTED,
};
- for (FragmentInstanceTaskStatus status : invalidStates) {
- FragmentInstanceTask testTask1 = new FragmentInstanceTask(mockDriver1, 100L, status);
- FragmentInstanceTask testTask2 =
- new FragmentInstanceTask(mockDriver2, 100L, FragmentInstanceTaskStatus.BLOCKED);
- Set<FragmentInstanceTask> taskSet = new HashSet<>();
+ for (DriverTaskStatus status : invalidStates) {
+ DriverTask testTask1 = new DriverTask(mockDriver1, 100L, status);
+ DriverTask testTask2 =
+ new DriverTask(mockDriver2, 100L, DriverTaskStatus.BLOCKED);
+ Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask1);
taskSet.add(testTask2);
manager.getQueryMap().put(queryId, taskSet);
@@ -337,7 +337,7 @@ public class DefaultTaskSchedulerTest {
defaultScheduler.toAborted(testTask1);
Assert.assertEquals(status, testTask1.getStatus());
- Assert.assertEquals(FragmentInstanceTaskStatus.BLOCKED, testTask2.getStatus());
+ Assert.assertEquals(DriverTaskStatus.BLOCKED, testTask2.getStatus());
Assert.assertFalse(manager.getBlockedTasks().contains(testTask1));
Assert.assertTrue(manager.getBlockedTasks().contains(testTask2));
Assert.assertNull(manager.getReadyQueue().get(testTask1.getId()));
@@ -352,23 +352,23 @@ public class DefaultTaskSchedulerTest {
Mockito.verify(mockDriver2, Mockito.never()).failed(Mockito.any());
clear();
}
- FragmentInstanceTaskStatus[] validStates =
- new FragmentInstanceTaskStatus[] {
- FragmentInstanceTaskStatus.RUNNING,
- FragmentInstanceTaskStatus.READY,
- FragmentInstanceTaskStatus.BLOCKED,
+ DriverTaskStatus[] validStates =
+ new DriverTaskStatus[] {
+ DriverTaskStatus.RUNNING,
+ DriverTaskStatus.READY,
+ DriverTaskStatus.BLOCKED,
};
- for (FragmentInstanceTaskStatus status : validStates) {
+ for (DriverTaskStatus status : validStates) {
Mockito.reset(mockDriver1);
Mockito.when(mockDriver1.getInfo()).thenReturn(instanceId1);
Mockito.reset(mockDriver2);
Mockito.when(mockDriver2.getInfo()).thenReturn(instanceId2);
- FragmentInstanceTask testTask1 = new FragmentInstanceTask(mockDriver1, 100L, status);
+ DriverTask testTask1 = new DriverTask(mockDriver1, 100L, status);
- FragmentInstanceTask testTask2 =
- new FragmentInstanceTask(mockDriver2, 100L, FragmentInstanceTaskStatus.BLOCKED);
- Set<FragmentInstanceTask> taskSet = new HashSet<>();
+ DriverTask testTask2 =
+ new DriverTask(mockDriver2, 100L, DriverTaskStatus.BLOCKED);
+ Set<DriverTask> taskSet = new HashSet<>();
taskSet.add(testTask1);
taskSet.add(testTask2);
manager.getQueryMap().put(queryId, taskSet);
@@ -381,8 +381,8 @@ public class DefaultTaskSchedulerTest {
Mockito.reset(mockDataBlockManager);
// An aborted fragment may cause others in the same query aborted.
- Assert.assertEquals(FragmentInstanceTaskStatus.ABORTED, testTask1.getStatus());
- Assert.assertEquals(FragmentInstanceTaskStatus.ABORTED, testTask2.getStatus());
+ Assert.assertEquals(DriverTaskStatus.ABORTED, testTask1.getStatus());
+ Assert.assertEquals(DriverTaskStatus.ABORTED, testTask2.getStatus());
Assert.assertFalse(manager.getBlockedTasks().contains(testTask1));
Assert.assertFalse(manager.getBlockedTasks().contains(testTask2));
Assert.assertNull(manager.getReadyQueue().get(testTask1.getId()));
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceSchedulerTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DriverSchedulerTest.java
similarity index 77%
rename from server/src/test/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceSchedulerTest.java
rename to server/src/test/java/org/apache/iotdb/db/mpp/schedule/DriverSchedulerTest.java
index f83d1213c6..07f3e67305 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceSchedulerTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DriverSchedulerTest.java
@@ -23,9 +23,9 @@ import org.apache.iotdb.db.mpp.common.FragmentInstanceId;
import org.apache.iotdb.db.mpp.common.PlanFragmentId;
import org.apache.iotdb.db.mpp.common.QueryId;
import org.apache.iotdb.db.mpp.execution.IDriver;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTaskID;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTaskStatus;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTask;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTaskID;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTaskStatus;
import org.junit.After;
import org.junit.Assert;
@@ -36,9 +36,9 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
-public class FragmentInstanceSchedulerTest {
+public class DriverSchedulerTest {
- private final FragmentInstanceScheduler manager = FragmentInstanceScheduler.getInstance();
+ private final DriverScheduler manager = DriverScheduler.getInstance();
@After
public void tearDown() {
@@ -62,40 +62,40 @@ public class FragmentInstanceSchedulerTest {
IDriver mockDriver2 = Mockito.mock(IDriver.class);
Mockito.when(mockDriver2.getInfo()).thenReturn(instanceId2);
List<IDriver> instances = Arrays.asList(mockDriver1, mockDriver2);
- manager.submitFragmentInstances(queryId, instances);
+ manager.submitDrivers(queryId, instances);
Assert.assertTrue(manager.getBlockedTasks().isEmpty());
Assert.assertEquals(1, manager.getQueryMap().size());
Assert.assertTrue(manager.getQueryMap().containsKey(queryId));
Assert.assertEquals(2, manager.getQueryMap().get(queryId).size());
Assert.assertEquals(2, manager.getTimeoutQueue().size());
Assert.assertEquals(2, manager.getReadyQueue().size());
- FragmentInstanceTask task1 =
- manager.getTimeoutQueue().get(new FragmentInstanceTaskID(instanceId1));
+ DriverTask task1 =
+ manager.getTimeoutQueue().get(new DriverTaskID(instanceId1));
Assert.assertNotNull(task1);
- FragmentInstanceTask task2 =
- manager.getTimeoutQueue().get(new FragmentInstanceTaskID(instanceId2));
+ DriverTask task2 =
+ manager.getTimeoutQueue().get(new DriverTaskID(instanceId2));
Assert.assertNotNull(task2);
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(task1));
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(task2));
- Assert.assertEquals(FragmentInstanceTaskStatus.READY, task1.getStatus());
- Assert.assertEquals(FragmentInstanceTaskStatus.READY, task2.getStatus());
+ Assert.assertEquals(DriverTaskStatus.READY, task1.getStatus());
+ Assert.assertEquals(DriverTaskStatus.READY, task2.getStatus());
// Submit another task of the same query
IDriver mockDriver3 = Mockito.mock(IDriver.class);
FragmentInstanceId instanceId3 = new FragmentInstanceId(fragmentId, "inst-2");
Mockito.when(mockDriver3.getInfo()).thenReturn(instanceId3);
- manager.submitFragmentInstances(queryId, Collections.singletonList(mockDriver3));
+ manager.submitDrivers(queryId, Collections.singletonList(mockDriver3));
Assert.assertTrue(manager.getBlockedTasks().isEmpty());
Assert.assertEquals(1, manager.getQueryMap().size());
Assert.assertTrue(manager.getQueryMap().containsKey(queryId));
Assert.assertEquals(3, manager.getQueryMap().get(queryId).size());
Assert.assertEquals(3, manager.getTimeoutQueue().size());
Assert.assertEquals(3, manager.getReadyQueue().size());
- FragmentInstanceTask task3 =
- manager.getTimeoutQueue().get(new FragmentInstanceTaskID(instanceId3));
+ DriverTask task3 =
+ manager.getTimeoutQueue().get(new DriverTaskID(instanceId3));
Assert.assertNotNull(task3);
Assert.assertTrue(manager.getQueryMap().get(queryId).contains(task3));
- Assert.assertEquals(FragmentInstanceTaskStatus.READY, task3.getStatus());
+ Assert.assertEquals(DriverTaskStatus.READY, task3.getStatus());
// Submit another task of the different query
QueryId queryId2 = new QueryId("test2");
@@ -103,18 +103,18 @@ public class FragmentInstanceSchedulerTest {
FragmentInstanceId instanceId4 = new FragmentInstanceId(fragmentId2, "inst-0");
IDriver mockDriver4 = Mockito.mock(IDriver.class);
Mockito.when(mockDriver4.getInfo()).thenReturn(instanceId4);
- manager.submitFragmentInstances(queryId2, Collections.singletonList(mockDriver4));
+ manager.submitDrivers(queryId2, Collections.singletonList(mockDriver4));
Assert.assertTrue(manager.getBlockedTasks().isEmpty());
Assert.assertEquals(2, manager.getQueryMap().size());
Assert.assertTrue(manager.getQueryMap().containsKey(queryId2));
Assert.assertEquals(1, manager.getQueryMap().get(queryId2).size());
Assert.assertEquals(4, manager.getTimeoutQueue().size());
Assert.assertEquals(4, manager.getReadyQueue().size());
- FragmentInstanceTask task4 =
- manager.getTimeoutQueue().get(new FragmentInstanceTaskID(instanceId4));
+ DriverTask task4 =
+ manager.getTimeoutQueue().get(new DriverTaskID(instanceId4));
Assert.assertNotNull(task4);
Assert.assertTrue(manager.getQueryMap().get(queryId2).contains(task4));
- Assert.assertEquals(FragmentInstanceTaskStatus.READY, task4.getStatus());
+ Assert.assertEquals(DriverTaskStatus.READY, task4.getStatus());
// Abort one FragmentInstance
Mockito.reset(mockDriver1);
@@ -127,10 +127,10 @@ public class FragmentInstanceSchedulerTest {
Assert.assertTrue(manager.getQueryMap().containsKey(queryId));
Assert.assertEquals(3, manager.getTimeoutQueue().size());
Assert.assertEquals(3, manager.getReadyQueue().size());
- Assert.assertEquals(FragmentInstanceTaskStatus.ABORTED, task1.getStatus());
- Assert.assertEquals(FragmentInstanceTaskStatus.READY, task2.getStatus());
- Assert.assertEquals(FragmentInstanceTaskStatus.READY, task3.getStatus());
- Assert.assertEquals(FragmentInstanceTaskStatus.READY, task4.getStatus());
+ Assert.assertEquals(DriverTaskStatus.ABORTED, task1.getStatus());
+ Assert.assertEquals(DriverTaskStatus.READY, task2.getStatus());
+ Assert.assertEquals(DriverTaskStatus.READY, task3.getStatus());
+ Assert.assertEquals(DriverTaskStatus.READY, task4.getStatus());
Mockito.verify(mockDriver1, Mockito.times(1)).failed(Mockito.any());
Assert.assertEquals(
FragmentInstanceAbortedException.BY_FRAGMENT_ABORT_CALLED, task1.getAbortCause());
@@ -151,10 +151,10 @@ public class FragmentInstanceSchedulerTest {
Assert.assertFalse(manager.getQueryMap().containsKey(queryId));
Assert.assertEquals(1, manager.getTimeoutQueue().size());
Assert.assertEquals(1, manager.getReadyQueue().size());
- Assert.assertEquals(FragmentInstanceTaskStatus.ABORTED, task1.getStatus());
- Assert.assertEquals(FragmentInstanceTaskStatus.ABORTED, task2.getStatus());
- Assert.assertEquals(FragmentInstanceTaskStatus.ABORTED, task3.getStatus());
- Assert.assertEquals(FragmentInstanceTaskStatus.READY, task4.getStatus());
+ Assert.assertEquals(DriverTaskStatus.ABORTED, task1.getStatus());
+ Assert.assertEquals(DriverTaskStatus.ABORTED, task2.getStatus());
+ Assert.assertEquals(DriverTaskStatus.ABORTED, task3.getStatus());
+ Assert.assertEquals(DriverTaskStatus.READY, task4.getStatus());
Mockito.verify(mockDriver1, Mockito.never()).failed(Mockito.any());
Mockito.verify(mockDriver2, Mockito.times(1)).failed(Mockito.any());
Mockito.verify(mockDriver3, Mockito.times(1)).failed(Mockito.any());
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceTimeoutSentinelTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DriverTaskTimeoutSentinelThreadTest.java
similarity index 75%
rename from server/src/test/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceTimeoutSentinelTest.java
rename to server/src/test/java/org/apache/iotdb/db/mpp/schedule/DriverTaskTimeoutSentinelThreadTest.java
index 28365962b8..4ce669c1d2 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceTimeoutSentinelTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/schedule/DriverTaskTimeoutSentinelThreadTest.java
@@ -24,8 +24,8 @@ import org.apache.iotdb.db.mpp.common.QueryId;
import org.apache.iotdb.db.mpp.execution.IDriver;
import org.apache.iotdb.db.mpp.schedule.queue.IndexedBlockingQueue;
import org.apache.iotdb.db.mpp.schedule.queue.L1PriorityQueue;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTaskStatus;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTask;
+import org.apache.iotdb.db.mpp.schedule.task.DriverTaskStatus;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
@@ -36,7 +36,7 @@ import org.mockito.Mockito;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
-public class FragmentInstanceTimeoutSentinelTest {
+public class DriverTaskTimeoutSentinelThreadTest {
@Test
public void testHandleInvalidStateTask() throws ExecutionException, InterruptedException {
@@ -44,52 +44,52 @@ public class FragmentInstanceTimeoutSentinelTest {
Mockito.when(mockScheduler.readyToRunning(Mockito.any()))
.thenAnswer(
ans -> {
- FragmentInstanceTask task = ans.getArgument(0);
- if (task.getStatus() != FragmentInstanceTaskStatus.READY) {
+ DriverTask task = ans.getArgument(0);
+ if (task.getStatus() != DriverTaskStatus.READY) {
return false;
}
- task.setStatus(FragmentInstanceTaskStatus.RUNNING);
+ task.setStatus(DriverTaskStatus.RUNNING);
return true;
});
QueryId queryId = new QueryId("test");
PlanFragmentId fragmentId = new PlanFragmentId(queryId, 0);
FragmentInstanceId instanceId = new FragmentInstanceId(fragmentId, "inst-0");
- IndexedBlockingQueue<FragmentInstanceTask> taskQueue =
+ IndexedBlockingQueue<DriverTask> taskQueue =
new L1PriorityQueue<>(
- 100, new FragmentInstanceTask.TimeoutComparator(), new FragmentInstanceTask());
+ 100, new DriverTask.TimeoutComparator(), new DriverTask());
IDriver mockDriver = Mockito.mock(IDriver.class);
Mockito.when(mockDriver.getInfo()).thenReturn(instanceId);
- AbstractExecutor executor =
- new FragmentInstanceTaskExecutor(
+ AbstractDriverThread executor =
+ new DriverTaskThread(
"0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
// FINISHED status test
- FragmentInstanceTask testTask =
- new FragmentInstanceTask(mockDriver, 100L, FragmentInstanceTaskStatus.FINISHED);
+ DriverTask testTask =
+ new DriverTask(mockDriver, 100L, DriverTaskStatus.FINISHED);
executor.execute(testTask);
- Assert.assertEquals(FragmentInstanceTaskStatus.FINISHED, testTask.getStatus());
+ Assert.assertEquals(DriverTaskStatus.FINISHED, testTask.getStatus());
Mockito.verify(mockDriver, Mockito.never()).processFor(Mockito.any());
Mockito.verify(mockDriver, Mockito.never()).failed(Mockito.any());
// ABORTED status test
- testTask = new FragmentInstanceTask(mockDriver, 100L, FragmentInstanceTaskStatus.ABORTED);
+ testTask = new DriverTask(mockDriver, 100L, DriverTaskStatus.ABORTED);
executor.execute(testTask);
- Assert.assertEquals(FragmentInstanceTaskStatus.ABORTED, testTask.getStatus());
+ Assert.assertEquals(DriverTaskStatus.ABORTED, testTask.getStatus());
Mockito.verify(mockDriver, Mockito.never()).processFor(Mockito.any());
Mockito.verify(mockDriver, Mockito.never()).failed(Mockito.any());
// RUNNING status test
- testTask = new FragmentInstanceTask(mockDriver, 100L, FragmentInstanceTaskStatus.RUNNING);
+ testTask = new DriverTask(mockDriver, 100L, DriverTaskStatus.RUNNING);
executor.execute(testTask);
- Assert.assertEquals(FragmentInstanceTaskStatus.RUNNING, testTask.getStatus());
+ Assert.assertEquals(DriverTaskStatus.RUNNING, testTask.getStatus());
Mockito.verify(mockDriver, Mockito.never()).processFor(Mockito.any());
Mockito.verify(mockDriver, Mockito.never()).failed(Mockito.any());
// BLOCKED status test
- testTask = new FragmentInstanceTask(mockDriver, 100L, FragmentInstanceTaskStatus.BLOCKED);
+ testTask = new DriverTask(mockDriver, 100L, DriverTaskStatus.BLOCKED);
executor.execute(testTask);
- Assert.assertEquals(FragmentInstanceTaskStatus.BLOCKED, testTask.getStatus());
+ Assert.assertEquals(DriverTaskStatus.BLOCKED, testTask.getStatus());
Mockito.verify(mockDriver, Mockito.never()).processFor(Mockito.any());
Assert.assertNull(testTask.getAbortCause());
Mockito.verify(mockScheduler, Mockito.never()).toAborted(Mockito.any());
@@ -104,16 +104,16 @@ public class FragmentInstanceTimeoutSentinelTest {
Mockito.when(mockScheduler.readyToRunning(Mockito.any()))
.thenAnswer(
ans -> {
- FragmentInstanceTask task = ans.getArgument(0);
- if (task.getStatus() != FragmentInstanceTaskStatus.READY) {
+ DriverTask task = ans.getArgument(0);
+ if (task.getStatus() != DriverTaskStatus.READY) {
return false;
}
- task.setStatus(FragmentInstanceTaskStatus.RUNNING);
+ task.setStatus(DriverTaskStatus.RUNNING);
return true;
});
- IndexedBlockingQueue<FragmentInstanceTask> taskQueue =
+ IndexedBlockingQueue<DriverTask> taskQueue =
new L1PriorityQueue<>(
- 100, new FragmentInstanceTask.TimeoutComparator(), new FragmentInstanceTask());
+ 100, new DriverTask.TimeoutComparator(), new DriverTask());
// Mock the instance with a cancelled future
IDriver mockDriver = Mockito.mock(IDriver.class);
@@ -124,11 +124,11 @@ public class FragmentInstanceTimeoutSentinelTest {
Mockito.when(mockDriver.processFor(Mockito.any()))
.thenReturn(Futures.immediateCancelledFuture());
- AbstractExecutor executor =
- new FragmentInstanceTaskExecutor(
+ AbstractDriverThread executor =
+ new DriverTaskThread(
"0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
- FragmentInstanceTask testTask =
- new FragmentInstanceTask(mockDriver, 100L, FragmentInstanceTaskStatus.READY);
+ DriverTask testTask =
+ new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
executor.execute(testTask);
Mockito.verify(mockDriver, Mockito.times(1)).processFor(Mockito.any());
Assert.assertEquals(
@@ -146,16 +146,16 @@ public class FragmentInstanceTimeoutSentinelTest {
Mockito.when(mockScheduler.readyToRunning(Mockito.any()))
.thenAnswer(
ans -> {
- FragmentInstanceTask task = ans.getArgument(0);
- if (task.getStatus() != FragmentInstanceTaskStatus.READY) {
+ DriverTask task = ans.getArgument(0);
+ if (task.getStatus() != DriverTaskStatus.READY) {
return false;
}
- task.setStatus(FragmentInstanceTaskStatus.RUNNING);
+ task.setStatus(DriverTaskStatus.RUNNING);
return true;
});
- IndexedBlockingQueue<FragmentInstanceTask> taskQueue =
+ IndexedBlockingQueue<DriverTask> taskQueue =
new L1PriorityQueue<>(
- 100, new FragmentInstanceTask.TimeoutComparator(), new FragmentInstanceTask());
+ 100, new DriverTask.TimeoutComparator(), new DriverTask());
// Mock the instance with a cancelled future
IDriver mockDriver = Mockito.mock(IDriver.class);
@@ -165,11 +165,11 @@ public class FragmentInstanceTimeoutSentinelTest {
Mockito.when(mockDriver.getInfo()).thenReturn(instanceId);
Mockito.when(mockDriver.processFor(Mockito.any())).thenReturn(Futures.immediateVoidFuture());
Mockito.when(mockDriver.isFinished()).thenReturn(true);
- AbstractExecutor executor =
- new FragmentInstanceTaskExecutor(
+ AbstractDriverThread executor =
+ new DriverTaskThread(
"0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
- FragmentInstanceTask testTask =
- new FragmentInstanceTask(mockDriver, 100L, FragmentInstanceTaskStatus.READY);
+ DriverTask testTask =
+ new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
executor.execute(testTask);
Mockito.verify(mockDriver, Mockito.times(1)).processFor(Mockito.any());
Assert.assertNull(testTask.getAbortCause());
@@ -186,16 +186,16 @@ public class FragmentInstanceTimeoutSentinelTest {
Mockito.when(mockScheduler.readyToRunning(Mockito.any()))
.thenAnswer(
ans -> {
- FragmentInstanceTask task = ans.getArgument(0);
- if (task.getStatus() != FragmentInstanceTaskStatus.READY) {
+ DriverTask task = ans.getArgument(0);
+ if (task.getStatus() != DriverTaskStatus.READY) {
return false;
}
- task.setStatus(FragmentInstanceTaskStatus.RUNNING);
+ task.setStatus(DriverTaskStatus.RUNNING);
return true;
});
- IndexedBlockingQueue<FragmentInstanceTask> taskQueue =
+ IndexedBlockingQueue<DriverTask> taskQueue =
new L1PriorityQueue<>(
- 100, new FragmentInstanceTask.TimeoutComparator(), new FragmentInstanceTask());
+ 100, new DriverTask.TimeoutComparator(), new DriverTask());
// Mock the instance with a blocked future
ListenableFuture<Void> mockFuture = Mockito.mock(ListenableFuture.class);
@@ -216,11 +216,11 @@ public class FragmentInstanceTimeoutSentinelTest {
Mockito.when(mockDriver.getInfo()).thenReturn(instanceId);
Mockito.when(mockDriver.processFor(Mockito.any())).thenReturn(mockFuture);
Mockito.when(mockDriver.isFinished()).thenReturn(false);
- AbstractExecutor executor =
- new FragmentInstanceTaskExecutor(
+ AbstractDriverThread executor =
+ new DriverTaskThread(
"0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
- FragmentInstanceTask testTask =
- new FragmentInstanceTask(mockDriver, 100L, FragmentInstanceTaskStatus.READY);
+ DriverTask testTask =
+ new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
executor.execute(testTask);
Mockito.verify(mockDriver, Mockito.times(1)).processFor(Mockito.any());
Assert.assertNull(testTask.getAbortCause());
@@ -237,16 +237,16 @@ public class FragmentInstanceTimeoutSentinelTest {
Mockito.when(mockScheduler.readyToRunning(Mockito.any()))
.thenAnswer(
ans -> {
- FragmentInstanceTask task = ans.getArgument(0);
- if (task.getStatus() != FragmentInstanceTaskStatus.READY) {
+ DriverTask task = ans.getArgument(0);
+ if (task.getStatus() != DriverTaskStatus.READY) {
return false;
}
- task.setStatus(FragmentInstanceTaskStatus.RUNNING);
+ task.setStatus(DriverTaskStatus.RUNNING);
return true;
});
- IndexedBlockingQueue<FragmentInstanceTask> taskQueue =
+ IndexedBlockingQueue<DriverTask> taskQueue =
new L1PriorityQueue<>(
- 100, new FragmentInstanceTask.TimeoutComparator(), new FragmentInstanceTask());
+ 100, new DriverTask.TimeoutComparator(), new DriverTask());
// Mock the instance with a ready future
ListenableFuture<Void> mockFuture = Mockito.mock(ListenableFuture.class);
@@ -267,11 +267,11 @@ public class FragmentInstanceTimeoutSentinelTest {
Mockito.when(mockDriver.getInfo()).thenReturn(instanceId);
Mockito.when(mockDriver.processFor(Mockito.any())).thenReturn(mockFuture);
Mockito.when(mockDriver.isFinished()).thenReturn(false);
- AbstractExecutor executor =
- new FragmentInstanceTaskExecutor(
+ AbstractDriverThread executor =
+ new DriverTaskThread(
"0", new ThreadGroup("timeout-test"), taskQueue, mockScheduler);
- FragmentInstanceTask testTask =
- new FragmentInstanceTask(mockDriver, 100L, FragmentInstanceTaskStatus.READY);
+ DriverTask testTask =
+ new DriverTask(mockDriver, 100L, DriverTaskStatus.READY);
executor.execute(testTask);
Mockito.verify(mockDriver, Mockito.times(1)).processFor(Mockito.any());
Assert.assertNull(testTask.getAbortCause());