This is an automated email from the ASF dual-hosted git repository.
xiangweiwei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new cc83ab87f3 [IOTDB-5449] Allow query scheduler to receive more tasks
(#8957)
cc83ab87f3 is described below
commit cc83ab87f3b98343179a7c291c371b24a18ecf72
Author: Xiangwei Wei <[email protected]>
AuthorDate: Mon Feb 6 15:07:15 2023 +0800
[IOTDB-5449] Allow query scheduler to receive more tasks (#8957)
---
.../db/mpp/execution/schedule/DriverScheduler.java | 41 ++++++++---
.../schedule/queue/IndexedBlockingQueue.java | 11 ++-
.../queue/IndexedBlockingReserveQueue.java | 81 ++++++++++++++++++++++
.../multilevelqueue/MultilevelPriorityQueue.java | 4 +-
.../schedule/queue/L1PriorityQueueTest.java | 24 -------
.../schedule/queue/L2PriorityQueueTest.java | 24 -------
6 files changed, 118 insertions(+), 67 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverScheduler.java
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverScheduler.java
index fb0fb7bf0c..7188717d20 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverScheduler.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverScheduler.java
@@ -29,6 +29,7 @@ import org.apache.iotdb.db.mpp.execution.driver.IDriver;
import org.apache.iotdb.db.mpp.execution.exchange.IMPPDataExchangeManager;
import org.apache.iotdb.db.mpp.execution.exchange.MPPDataExchangeService;
import org.apache.iotdb.db.mpp.execution.schedule.queue.IndexedBlockingQueue;
+import
org.apache.iotdb.db.mpp.execution.schedule.queue.IndexedBlockingReserveQueue;
import org.apache.iotdb.db.mpp.execution.schedule.queue.L1PriorityQueue;
import
org.apache.iotdb.db.mpp.execution.schedule.queue.multilevelqueue.DriverTaskHandle;
import
org.apache.iotdb.db.mpp.execution.schedule.queue.multilevelqueue.MultilevelPriorityQueue;
@@ -68,7 +69,7 @@ public class DriverScheduler implements IDriverScheduler,
IService {
return InstanceHolder.instance;
}
- private final IndexedBlockingQueue<DriverTask> readyQueue;
+ private final IndexedBlockingReserveQueue<DriverTask> readyQueue;
private final IndexedBlockingQueue<DriverTask> timeoutQueue;
private final Set<DriverTask> blockedTasks;
private final Map<QueryId, Map<FragmentInstanceId, Set<DriverTask>>>
queryMap;
@@ -77,10 +78,11 @@ public class DriverScheduler implements IDriverScheduler,
IService {
private final AtomicInteger nextDriverTaskHandleId = new AtomicInteger(0);
private IMPPDataExchangeManager blockManager;
- private static final int MAX_CAPACITY =
+ private static final int QUERY_MAX_CAPACITY =
IoTDBDescriptor.getInstance().getConfig().getMaxAllowedConcurrentQueries();
private static final int WORKER_THREAD_NUM =
IoTDBDescriptor.getInstance().getConfig().getQueryThreadCount();
+ private static final int TASK_MAX_CAPACITY = QUERY_MAX_CAPACITY *
WORKER_THREAD_NUM * 2;
private static final long QUERY_TIMEOUT_MS =
IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold();
private final ThreadGroup workerGroups;
@@ -88,9 +90,10 @@ public class DriverScheduler implements IDriverScheduler,
IService {
private DriverScheduler() {
this.readyQueue =
- new MultilevelPriorityQueue(LEVEL_TIME_MULTIPLIER, MAX_CAPACITY, new
DriverTask());
+ new MultilevelPriorityQueue(LEVEL_TIME_MULTIPLIER, TASK_MAX_CAPACITY,
new DriverTask());
this.timeoutQueue =
- new L1PriorityQueue<>(MAX_CAPACITY, new
DriverTask.TimeoutComparator(), new DriverTask());
+ new L1PriorityQueue<>(
+ QUERY_MAX_CAPACITY, new DriverTask.TimeoutComparator(), new
DriverTask());
this.queryMap = new ConcurrentHashMap<>();
this.blockedTasks = Collections.synchronizedSet(new HashSet<>());
this.scheduler = new Scheduler();
@@ -254,12 +257,30 @@ public class DriverScheduler implements IDriverScheduler,
IService {
private void clearDriverTask(DriverTask task) {
try (SetThreadName driverTaskName =
new SetThreadName(task.getDriver().getDriverTaskId().getFullId())) {
- if (task.getStatus() != DriverTaskStatus.FINISHED) {
- task.setStatus(DriverTaskStatus.ABORTED);
+ DriverTaskStatus status = task.getStatus();
+ switch (status) {
+ // If it has been aborted, return directly
+ case ABORTED:
+ return;
+ case READY:
+ task.setStatus(DriverTaskStatus.ABORTED);
+ readyQueue.remove(task.getDriverTaskId());
+ break;
+ case BLOCKED:
+ task.setStatus(DriverTaskStatus.ABORTED);
+ blockedTasks.remove(task);
+ readyQueue.decreaseReservedSize();
+ break;
+ case RUNNING:
+ task.setStatus(DriverTaskStatus.ABORTED);
+ readyQueue.decreaseReservedSize();
+ break;
+ case FINISHED:
+ readyQueue.decreaseReservedSize();
+ break;
}
- readyQueue.remove(task.getDriverTaskId());
+
timeoutQueue.remove(task.getDriverTaskId());
- blockedTasks.remove(task);
Map<FragmentInstanceId, Set<DriverTask>> queryRelatedTasks =
queryMap.get(task.getDriverTaskId().getQueryId());
if (queryRelatedTasks != null) {
@@ -361,7 +382,7 @@ public class DriverScheduler implements IDriverScheduler,
IService {
BLOCK_QUEUED_TIME, System.nanoTime() -
task.getLastEnterBlockQueueTime());
task.setLastEnterReadyQueueTime(System.nanoTime());
task.resetLevelScheduledTime();
- readyQueue.push(task);
+ readyQueue.repush(task);
blockedTasks.remove(task);
} finally {
task.unlock();
@@ -395,7 +416,7 @@ public class DriverScheduler implements IDriverScheduler,
IService {
task.updateSchedulePriority(context);
task.setStatus(DriverTaskStatus.READY);
task.setLastEnterReadyQueueTime(System.nanoTime());
- readyQueue.push(task);
+ readyQueue.repush(task);
} finally {
task.unlock();
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/queue/IndexedBlockingQueue.java
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/queue/IndexedBlockingQueue.java
index 1498bbbc30..6fdcbb0392 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/queue/IndexedBlockingQueue.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/queue/IndexedBlockingQueue.java
@@ -37,9 +37,9 @@ import com.google.common.base.Preconditions;
*/
public abstract class IndexedBlockingQueue<E extends IDIndexedAccessible> {
- private final int MAX_CAPACITY;
- private final E queryHolder;
- private int size;
+ protected final int MAX_CAPACITY;
+ protected final E queryHolder;
+ protected int size;
/**
* Init the queue with a max capacity. The queryHolder is just a simple
reused object in query to
@@ -85,10 +85,7 @@ public abstract class IndexedBlockingQueue<E extends
IDIndexedAccessible> {
if (element == null) {
throw new NullPointerException("pushed element is null");
}
- Preconditions.checkState(
- !contains(element),
- "The queue has already contained the element: " +
element.getDriverTaskId());
- Preconditions.checkState(size < MAX_CAPACITY, "The queue is full");
+ Preconditions.checkState(size < MAX_CAPACITY, "The system can't allow more
queries.");
pushToQueue(element);
size++;
this.notifyAll();
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/queue/IndexedBlockingReserveQueue.java
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/queue/IndexedBlockingReserveQueue.java
new file mode 100644
index 0000000000..ba14fe9f82
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/queue/IndexedBlockingReserveQueue.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.mpp.execution.schedule.queue;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * This class is different from <class>IndexedBlockingQueue</class> in that it
will reserve space
+ * for polled element in case it will be pushed again.
+ */
+public abstract class IndexedBlockingReserveQueue<E extends
IDIndexedAccessible>
+ extends IndexedBlockingQueue<E> {
+
+ // To avoid some elements can't join the queue again that are polled out for
running or blocked
+ private int reservedSize;
+
+ public IndexedBlockingReserveQueue(int maxCapacity, E queryHolder) {
+ super(maxCapacity, queryHolder);
+ }
+
+ /**
+ * Get and remove the first element of the queue. Reserve space for this
polled element.
+ *
+ * @return the queue head element.
+ */
+ public synchronized E poll() throws InterruptedException {
+ while (isEmpty()) {
+ this.wait();
+ }
+ E output = pollFirst();
+ size--;
+ reservedSize++;
+ return output;
+ }
+
+ public synchronized void push(E element) {
+ if (element == null) {
+ throw new NullPointerException("pushed element is null");
+ }
+ Preconditions.checkState(
+ size + reservedSize < MAX_CAPACITY, "The system can't allow more query
tasks.");
+ pushToQueue(element);
+ size++;
+ this.notifyAll();
+ }
+
+ /** RePush an element which is polled out for running or blocked before to
the queue. */
+ public synchronized void repush(E element) {
+ if (element == null) {
+ throw new NullPointerException("pushed element is null");
+ }
+ pushToQueue(element);
+ reservedSize--;
+ size++;
+ this.notifyAll();
+ }
+
+ /**
+ * For task that is not in readyQueue when it's cleared, it won't be added
into the queue again.
+ */
+ public synchronized void decreaseReservedSize() {
+ this.reservedSize--;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/queue/multilevelqueue/MultilevelPriorityQueue.java
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/queue/multilevelqueue/MultilevelPriorityQueue.java
index 97ecf3927e..2c5bc41fa1 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/queue/multilevelqueue/MultilevelPriorityQueue.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/queue/multilevelqueue/MultilevelPriorityQueue.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.db.mpp.execution.schedule.queue.multilevelqueue;
-import org.apache.iotdb.db.mpp.execution.schedule.queue.IndexedBlockingQueue;
+import
org.apache.iotdb.db.mpp.execution.schedule.queue.IndexedBlockingReserveQueue;
import org.apache.iotdb.db.mpp.execution.schedule.task.DriverTask;
import java.util.PriorityQueue;
@@ -34,7 +34,7 @@ import static java.util.concurrent.TimeUnit.SECONDS;
* This class is inspired by Trino <a
*
href="https://github.com/trinodb/trino/blob/master/core/trino-main/src/main/java/io/trino/execution/executor/MultilevelSplitQueue.java";>...</a>
*/
-public class MultilevelPriorityQueue extends IndexedBlockingQueue<DriverTask> {
+public class MultilevelPriorityQueue extends
IndexedBlockingReserveQueue<DriverTask> {
/** Scheduled time threshold of TASK in each level */
static final int[] LEVEL_THRESHOLD_SECONDS = {0, 1, 10, 60, 300};
diff --git
a/server/src/test/java/org/apache/iotdb/db/mpp/execution/schedule/queue/L1PriorityQueueTest.java
b/server/src/test/java/org/apache/iotdb/db/mpp/execution/schedule/queue/L1PriorityQueueTest.java
index 5d36688454..418ca48fe5 100644
---
a/server/src/test/java/org/apache/iotdb/db/mpp/execution/schedule/queue/L1PriorityQueueTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/mpp/execution/schedule/queue/L1PriorityQueueTest.java
@@ -108,30 +108,6 @@ public class L1PriorityQueueTest {
Assert.assertEquals(0, queue.size());
}
- @Test
- public void testPushSameElement() {
- IndexedBlockingQueue<QueueElement> queue =
- new L1PriorityQueue<>(
- 10,
- (o1, o2) -> {
- if (o1.equals(o2)) {
- return 0;
- }
- return Integer.compare(o1.getValue(), o2.getValue());
- },
- new QueueElement(new QueueElement.QueueElementID(0), 0));
- QueueElement e1 = new QueueElement(new QueueElement.QueueElementID(1), 10);
- queue.push(e1);
- Assert.assertEquals(1, queue.size());
- QueueElement e1e = new QueueElement(new QueueElement.QueueElementID(1), 5);
- try {
- queue.push(e1e);
- Assert.fail();
- } catch (IllegalStateException e) {
- Assert.assertTrue(e.getMessage().contains("has already contained"));
- }
- }
-
@Test
public void testRemoveElement() {
IndexedBlockingQueue<QueueElement> queue =
diff --git
a/server/src/test/java/org/apache/iotdb/db/mpp/execution/schedule/queue/L2PriorityQueueTest.java
b/server/src/test/java/org/apache/iotdb/db/mpp/execution/schedule/queue/L2PriorityQueueTest.java
index 2c762e7736..9acf787151 100644
---
a/server/src/test/java/org/apache/iotdb/db/mpp/execution/schedule/queue/L2PriorityQueueTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/mpp/execution/schedule/queue/L2PriorityQueueTest.java
@@ -123,30 +123,6 @@ public class L2PriorityQueueTest {
Assert.assertEquals(0, queue.size());
}
- @Test
- public void testPushSameElement() {
- IndexedBlockingQueue<QueueElement> queue =
- new L2PriorityQueue<>(
- 10,
- (o1, o2) -> {
- if (o1.equals(o2)) {
- return 0;
- }
- return Integer.compare(o1.getValue(), o2.getValue());
- },
- new QueueElement(new QueueElement.QueueElementID(0), 0));
- QueueElement e1 = new QueueElement(new QueueElement.QueueElementID(1), 10);
- queue.push(e1);
- Assert.assertEquals(1, queue.size());
- QueueElement e1e = new QueueElement(new QueueElement.QueueElementID(1), 5);
- try {
- queue.push(e1e);
- fail();
- } catch (IllegalStateException e) {
- Assert.assertTrue(e.getMessage().contains("has already contained"));
- }
- }
-
@Test
public void testRemoveElement() {
IndexedBlockingQueue<QueueElement> queue =
