morningman commented on a change in pull request #665: Change routine load task
sheduler interval to 0
URL: https://github.com/apache/incubator-doris/pull/665#discussion_r260567637
##########
File path:
fe/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
##########
@@ -28,52 +28,74 @@
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import java.sql.Date;
import java.util.Queue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
/**
* Routine load task scheduler is a function which allocate task to be.
* Step1: get total idle task num of backends.
+ * Step1.1: if total idle task num == 0, exit this round and switch to the
next round immediately
* Step2: equally divide to be
+ * Step2.1: if there is no task in queue, waiting task until an element
becomes available.
+ * Step2.2: divide task to be
* Step3: submit tasks to be
*/
-// TODO(ml): change interval ms in constructor
public class RoutineLoadTaskScheduler extends Daemon {
private static final Logger LOG =
LogManager.getLogger(RoutineLoadTaskScheduler.class);
- private RoutineLoadManager routineLoadManager =
Catalog.getInstance().getRoutineLoadManager();
+ private RoutineLoadManager routineLoadManager;
+ private LinkedBlockingQueue<RoutineLoadTaskInfo> needSchedulerTasksQueue;
+
+ public RoutineLoadTaskScheduler() {
+ super("routine load task", 0);
+ routineLoadManager = Catalog.getInstance().getRoutineLoadManager();
+ needSchedulerTasksQueue = (LinkedBlockingQueue)
routineLoadManager.getNeedSchedulerTasksQueue();
+ }
@Override
protected void runOneCycle() {
try {
process();
} catch (Throwable e) {
LOG.warn("Failed to process one round of RoutineLoadTaskScheduler
with error message {}",
- e.getMessage(), e);
+ e.getMessage(), e);
}
}
private void process() throws LoadException {
// update current beIdMaps for tasks
routineLoadManager.updateBeIdTaskMaps();
- // get idle be task num
+ LOG.info("There are {} need scheduler task in queue when {}",
+ needSchedulerTasksQueue.size(), System.currentTimeMillis());
+ AgentBatchTask batchTask = new AgentBatchTask();
+ int sizeOfTasksQueue = needSchedulerTasksQueue.size();
int clusterIdleSlotNum = routineLoadManager.getClusterIdleSlotNum();
+ int needScheduledTaskNum = sizeOfTasksQueue < clusterIdleSlotNum ?
sizeOfTasksQueue : clusterIdleSlotNum;
int scheduledTaskNum = 0;
- Queue<RoutineLoadTaskInfo> needSchedulerTasksQueue =
routineLoadManager.getNeedSchedulerTasksQueue();
- AgentBatchTask batchTask = new AgentBatchTask();
-
+ // get idle be task num
// allocate task to be
- while (clusterIdleSlotNum > 0) {
- if (needSchedulerTasksQueue.peek() != null) {
- RoutineLoadTaskInfo routineLoadTaskInfo =
needSchedulerTasksQueue.poll();
+ while (needScheduledTaskNum > 0) {
+ RoutineLoadTaskInfo routineLoadTaskInfo = null;
+ try {
+ routineLoadTaskInfo = needSchedulerTasksQueue.take();
+ } catch (InterruptedException e) {
+ LOG.warn("Taking routine load task from queue has been
interrupted with error msg {}",
+ e.getMessage());
+ return;
+ }
+
+ if (clusterIdleSlotNum > 0) {
Review comment:
You don't need to check if clusterIdleSlotNum > 0 here. Because if
clusterIdleSlotNum <= 0, you will not enter this while loop at all. And you
already make sure that clusterIdleSlotNum is always larger than
needScheduledTaskNum.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
