This is an automated email from the ASF dual-hosted git repository. ericpai pushed a commit to branch new_mpp in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0b5356fd6fdd67454e6a2bf57bdef684ec14d9b7 Author: ericpai <[email protected]> AuthorDate: Mon Mar 21 10:09:37 2022 +0800 Implement phase 1 --- .../apache/iotdb/commons/service/ServiceType.java | 2 +- .../iotdb/db/mpp/buffer/IDataBlockManager.java | 6 +- ...anceTaskExecutor.java => AbstractExecutor.java} | 40 ++- .../iotdb/db/mpp/schedule/ExecutionContext.java | 24 +- .../db/mpp/schedule/FragmentInstanceManager.java | 270 ++++++++++++++------- .../mpp/schedule/FragmentInstanceTaskCallback.java | 27 --- .../mpp/schedule/FragmentInstanceTaskExecutor.java | 66 +++-- .../schedule/FragmentInstanceTimeoutSentinel.java | 66 ++--- .../db/mpp/schedule/IFragmentInstanceManager.java | 24 +- .../iotdb/db/mpp/schedule/ITaskScheduler.java | 77 ++++++ .../db/mpp/schedule/task/FragmentInstanceTask.java | 83 +++++-- .../org/apache/iotdb/db/utils/stats/CpuTimer.java | 156 ++++++++++++ 12 files changed, 622 insertions(+), 219 deletions(-) diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java b/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java index a5ad95d..e583447 100644 --- a/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java +++ b/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java @@ -65,7 +65,7 @@ public enum ServiceType { CLUSTER_META_ENGINE("Cluster Meta Engine", "ClusterMetaEngine"), CLUSTER_DATA_ENGINE("Cluster Data Engine", "ClusterDataEngine"), REST_SERVICE("REST Service", "REST Service"), - CONFIG_NODE_SERVICE("Config Node service", "ConfigNodeRPCServer"); + CONFIG_NODE_SERVICE("Config Node service", "ConfigNodeRPCServer"), FRAGMENT_INSTANCE_MANAGER_SERVICE("Fragment instance manager", "FragmentInstanceManager"); private final String name; diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/IDataBlockManager.java b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/IDataBlockManager.java index 391db95..5f45ee8 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/IDataBlockManager.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/IDataBlockManager.java @@ -19,9 +19,9 @@ package org.apache.iotdb.db.mpp.buffer; +import org.apache.iotdb.db.mpp.common.TsBlock; import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceID; import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask; -import org.apache.iotdb.mpp.common.ITSBlock; public interface IDataBlockManager { @@ -59,7 +59,7 @@ public interface IDataBlockManager { * @param instanceID ID of fragment instance that generates the block. * @return If there are enough memory for the next block. */ - boolean putDataBlock(FragmentInstanceID instanceID, ITSBlock block); + boolean putDataBlock(FragmentInstanceID instanceID, TsBlock block); /** * Check if there are data blocks from the specified upstream fragment instance. @@ -76,5 +76,5 @@ public interface IDataBlockManager { * @param instanceID ID of the upstream fragment instance. * @return A data block. */ - ITSBlock getDataBlock(FragmentInstanceID instanceID); + TsBlock getDataBlock(FragmentInstanceID instanceID); } 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/AbstractExecutor.java similarity index 54% copy from server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceTaskExecutor.java copy to server/src/main/java/org/apache/iotdb/db/mpp/schedule/AbstractExecutor.java index 2d3e606..f15b192 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/AbstractExecutor.java @@ -24,29 +24,53 @@ import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -/** the worker thread of {@link FragmentInstanceTask} */ -public class FragmentInstanceTaskExecutor extends Thread { +import java.io.Closeable; +import java.io.IOException; +import java.util.concurrent.ExecutionException; - private static final Logger logger = LoggerFactory.getLogger(FragmentInstanceTaskExecutor.class); +/** an abstract executor for {@link FragmentInstanceTask} */ +public abstract class AbstractExecutor extends Thread implements Closeable { + private static final Logger logger = LoggerFactory.getLogger(AbstractExecutor.class); private final IndexedBlockingQueue<FragmentInstanceTask> queue; + private final ITaskScheduler scheduler; + private volatile boolean closed; - public FragmentInstanceTaskExecutor( - String workerId, ThreadGroup tg, IndexedBlockingQueue<FragmentInstanceTask> queue) { + public AbstractExecutor( + String workerId, + ThreadGroup tg, + IndexedBlockingQueue<FragmentInstanceTask> queue, + ITaskScheduler scheduler) { super(tg, workerId); this.queue = queue; + this.scheduler = scheduler; + this.closed = false; } @Override public void run() { - while (true) { + while (!closed && !Thread.currentThread().isInterrupted()) { try { FragmentInstanceTask next = queue.poll(); - // do logic here + execute(next); } catch (InterruptedException e) { - logger.info("{} is interrupted.", this.getName()); break; + } catch (Exception e) { + logger.error("Executor " + this.getName() + " processes failed", e); } } } + + protected ITaskScheduler getScheduler() { + return scheduler; + } + + /** Processing a task. */ + protected abstract void execute(FragmentInstanceTask task) + throws InterruptedException, ExecutionException; + + @Override + public void close() throws IOException { + closed = true; + } } 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 adf7874..e8cd091 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 @@ -19,6 +19,28 @@ package org.apache.iotdb.db.mpp.schedule; import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask; +import org.apache.iotdb.db.utils.stats.CpuTimer; + +import io.airlift.units.Duration; /** The execution context of a {@link FragmentInstanceTask} */ -public class ExecutionContext {} +public class ExecutionContext { + private CpuTimer.CpuDuration cpuDuration; + private Duration timeSlice; + + public CpuTimer.CpuDuration getCpuDuration() { + return cpuDuration; + } + + public void setCpuDuration(CpuTimer.CpuDuration cpuDuration) { + this.cpuDuration = cpuDuration; + } + + public Duration getTimeSlice() { + return timeSlice; + } + + public void setTimeSlice(Duration timeSlice) { + this.timeSlice = timeSlice; + } +} diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceManager.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceManager.java index bf38396..1e22368 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceManager.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceManager.java @@ -18,34 +18,49 @@ */ package org.apache.iotdb.db.mpp.schedule; -import org.apache.iotdb.db.exception.StartupException; +import org.apache.iotdb.commons.exception.StartupException; +import org.apache.iotdb.commons.service.IService; +import org.apache.iotdb.commons.service.ServiceType; +import org.apache.iotdb.db.mpp.execution.ExecFragmentInstance; 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.FragmentInstanceID; import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask; import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTaskStatus; -import org.apache.iotdb.db.service.IService; -import org.apache.iotdb.db.service.ServiceType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; /** the manager of fragment instances scheduling */ public class FragmentInstanceManager implements IFragmentInstanceManager, IService { + private static final Logger logger = LoggerFactory.getLogger(FragmentInstanceManager.class); + public static IFragmentInstanceManager getInstance() { return InstanceHolder.instance; } private final IndexedBlockingQueue<FragmentInstanceTask> readyQueue; private final IndexedBlockingQueue<FragmentInstanceTask> timeoutQueue; - private final Map<String, List<FragmentInstanceTask>> queryMap; + private final Set<FragmentInstanceTask> blockedTasks; + private final Map<String, Set<FragmentInstanceTask>> queryMap; + private final ITaskScheduler scheduler; private static final int MAX_CAPACITY = 1000; // TODO: load from config files private static final int WORKER_THREAD_NUM = 4; // TODO: load from config files - private final ThreadGroup workerGroups = new ThreadGroup("ScheduleThreads"); + private static final int QUERY_TIMEOUT_MS = 10000; // TODO: load from config files or requests + private final ThreadGroup workerGroups; + private final List<AbstractExecutor> threads; public FragmentInstanceManager() { this.readyQueue = @@ -59,21 +74,38 @@ public class FragmentInstanceManager implements IFragmentInstanceManager, IServi new FragmentInstanceTask.SchedulePriorityComparator(), new FragmentInstanceTask()); this.queryMap = new ConcurrentHashMap<>(); + this.blockedTasks = Collections.synchronizedSet(new HashSet<>()); + this.scheduler = new Scheduler(); + this.workerGroups = new ThreadGroup("ScheduleThreads"); + this.threads = new ArrayList<>(); } @Override public void start() throws StartupException { for (int i = 0; i < WORKER_THREAD_NUM; i++) { - new FragmentInstanceTaskExecutor("Worker-Thread-" + i, workerGroups, readyQueue).start(); + AbstractExecutor t = + new FragmentInstanceTaskExecutor( + "Worker-Thread-" + i, workerGroups, readyQueue, scheduler); + threads.add(t); + t.start(); } - new FragmentInstanceTimeoutSentinel( - "Sentinel-Thread", workerGroups, timeoutQueue, this::abortFragmentInstanceTask) - .start(); + AbstractExecutor t = + new FragmentInstanceTimeoutSentinel( + "Sentinel-Thread", workerGroups, timeoutQueue, scheduler); + threads.add(t); + t.start(); } @Override public void stop() { - workerGroups.interrupt(); + this.threads.forEach( + t -> { + try { + t.close(); + } catch (IOException e) { + // Only a field is set, there's no chance to throw an IOException + } + }); } @Override @@ -82,97 +114,173 @@ public class FragmentInstanceManager implements IFragmentInstanceManager, IServi } @Override - public void submitFragmentInstance() { - // TODO: pass a real task - FragmentInstanceTask task = new FragmentInstanceTask(); - - task.lock(); - try { - timeoutQueue.push(task); - // TODO: if no upstream deps, set to ready - task.setStatus(FragmentInstanceTaskStatus.READY); - readyQueue.push(task); - } finally { - task.unlock(); + public void submitFragmentInstances(String queryId, List<ExecFragmentInstance> instances) { + Set<FragmentInstanceTask> tasks = + instances.stream() + .map( + v -> + new FragmentInstanceTask(v, QUERY_TIMEOUT_MS, FragmentInstanceTaskStatus.READY)) + .collect(Collectors.toSet()); + queryMap.put(queryId, Collections.synchronizedSet(tasks)); + for (FragmentInstanceTask task : tasks) { + task.lock(); + try { + timeoutQueue.push(task); + readyQueue.push(task); + } finally { + task.unlock(); + } } } @Override - public void inputBlockAvailable( - FragmentInstanceID instanceID, FragmentInstanceID upstreamInstanceId) { - FragmentInstanceTask task = timeoutQueue.get(instanceID); - if (task == null) { - return; - } - task.lock(); - try { - if (task.getStatus() != FragmentInstanceTaskStatus.BLOCKED) { - return; - } - task.inputReady(instanceID); - if (task.getStatus() == FragmentInstanceTaskStatus.READY) { - readyQueue.push(task); + public void abortQuery(String queryId) { + Set<FragmentInstanceTask> queryRelatedTasks = queryMap.remove(queryId); + if (queryRelatedTasks != null) { + for (FragmentInstanceTask task : queryRelatedTasks) { + task.lock(); + try { + clearFragmentInstanceTask(task); + } finally { + task.unlock(); + } } - } finally { - task.unlock(); } } @Override - public void outputBlockAvailable(FragmentInstanceID instanceID) { - FragmentInstanceTask task = timeoutQueue.get(instanceID); - if (task == null) { - return; + public void fetchFragmentInstance(ExecFragmentInstance instance) {} + + private void clearFragmentInstanceTask(FragmentInstanceTask task) { + if (task.getStatus() != FragmentInstanceTaskStatus.FINISHED) { + task.setStatus(FragmentInstanceTaskStatus.ABORTED); } - task.lock(); - try { - if (task.getStatus() != FragmentInstanceTaskStatus.BLOCKED) { - return; - } - task.outputReady(); - if (task.getStatus() == FragmentInstanceTaskStatus.READY) { - readyQueue.push(task); - } - } finally { - task.unlock(); + if (task.getStatus() == FragmentInstanceTaskStatus.ABORTED) { + // TODO: remember to call the implementation + // IDataBlockManager.forceDeregisterFragmentInstance(task); + } + readyQueue.remove(task.getId()); + timeoutQueue.remove(task.getId()); + blockedTasks.remove(task); + Set<FragmentInstanceTask> tasks = queryMap.get(task.getId().getQueryId()); + tasks.remove(task); + if (tasks.isEmpty()) { + queryMap.remove(task.getId().getQueryId()); } } - /** abort a {@link FragmentInstanceTask} */ - void abortFragmentInstanceTask(FragmentInstanceTask task) { - List<FragmentInstanceTask> queryRelatedTasks = queryMap.remove(task.getId().getQueryId()); - clearFragmentInstanceTask(task); - if (queryRelatedTasks != null) { - // if queryRelatedTask is not null, it means that the clean request comes from this node, not - // coordinator. - // TODO: tell coordinator - for (FragmentInstanceTask otherTask : queryRelatedTasks) { - clearFragmentInstanceTask(otherTask); + private static class InstanceHolder { + + private InstanceHolder() {} + + private static final IFragmentInstanceManager instance = new FragmentInstanceManager(); + } + /** the default scheduler implementation */ + private class Scheduler implements ITaskScheduler { + @Override + public void blockedToReady(FragmentInstanceTask task) { + task.lock(); + try { + if (task.getStatus() != FragmentInstanceTaskStatus.BLOCKED) { + return; + } + task.setStatus(FragmentInstanceTaskStatus.READY); + readyQueue.push(task); + blockedTasks.remove(task); + } finally { + task.unlock(); } } - // TODO: call LocalMemoryManager to release resources - } - private void clearFragmentInstanceTask(FragmentInstanceTask task) { - task.lock(); - try { - if (task.getStatus() != FragmentInstanceTaskStatus.FINISHED) { - task.setStatus(FragmentInstanceTaskStatus.ABORTED); + @Override + public boolean readyToRunning(FragmentInstanceTask task) { + task.lock(); + try { + if (task.getStatus() != FragmentInstanceTaskStatus.READY) { + return false; + } + task.setStatus(FragmentInstanceTaskStatus.RUNNING); + } finally { + task.unlock(); } - readyQueue.remove(task.getId()); - timeoutQueue.remove(task.getId()); - } finally { - task.unlock(); + return true; } - } - @Override - public void abortQuery(String queryId) {} + @Override + public void runningToReady(FragmentInstanceTask task, ExecutionContext context) { + task.lock(); + try { + if (task.getStatus() != FragmentInstanceTaskStatus.RUNNING) { + return; + } + task.updateSchedulePriority(context); + task.setStatus(FragmentInstanceTaskStatus.READY); + readyQueue.push(task); + } finally { + task.unlock(); + } + } - private static class InstanceHolder { + @Override + public void runningToBlocked(FragmentInstanceTask task, ExecutionContext context) { + task.lock(); + try { + if (task.getStatus() != FragmentInstanceTaskStatus.RUNNING) { + return; + } + task.updateSchedulePriority(context); + task.setStatus(FragmentInstanceTaskStatus.BLOCKED); + blockedTasks.add(task); + } finally { + task.unlock(); + } + } - private InstanceHolder() {} + @Override + public void runningToFinished(FragmentInstanceTask task, ExecutionContext context) { + task.lock(); + try { + if (task.getStatus() != FragmentInstanceTaskStatus.RUNNING) { + return; + } + task.updateSchedulePriority(context); + task.setStatus(FragmentInstanceTaskStatus.FINISHED); + clearFragmentInstanceTask(task); + } finally { + task.unlock(); + } + } - private static final IFragmentInstanceManager instance = new FragmentInstanceManager(); + @Override + public void toAborted(FragmentInstanceTask task) { + task.lock(); + try { + // If a task is already in an end state, it indicates that the task is finalized in other + // threads. + if (task.isEndState()) { + return; + } + logger.warn( + "The task {} is aborted. All other tasks in the same query will be cancelled", + task.getId().toString()); + clearFragmentInstanceTask(task); + } finally { + task.unlock(); + } + Set<FragmentInstanceTask> queryRelatedTasks = queryMap.get(task.getId().getQueryId()); + if (queryRelatedTasks != null) { + for (FragmentInstanceTask otherTask : queryRelatedTasks) { + if (task.equals(otherTask)) { + continue; + } + otherTask.lock(); + try { + clearFragmentInstanceTask(otherTask); + } finally { + otherTask.unlock(); + } + } + } + } } } diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceTaskCallback.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceTaskCallback.java deleted file mode 100644 index c3ba2e7..0000000 --- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/FragmentInstanceTaskCallback.java +++ /dev/null @@ -1,27 +0,0 @@ -/* - * 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.schedule; - -import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask; - -/** A common interface for {@link FragmentInstanceTask} business logic callback */ -@FunctionalInterface -public interface FragmentInstanceTaskCallback { - void call(FragmentInstanceTask task) throws Exception; -} 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/FragmentInstanceTaskExecutor.java index 2d3e606..d1d1b8c 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/FragmentInstanceTaskExecutor.java @@ -18,35 +18,67 @@ */ package org.apache.iotdb.db.mpp.schedule; +import org.apache.iotdb.db.mpp.execution.ExecFragmentInstance; import org.apache.iotdb.db.mpp.schedule.queue.IndexedBlockingQueue; import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask; +import org.apache.iotdb.db.utils.stats.CpuTimer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.MoreExecutors; +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 Thread { +public class FragmentInstanceTaskExecutor extends AbstractExecutor { - private static final Logger logger = LoggerFactory.getLogger(FragmentInstanceTaskExecutor.class); + private static final Duration EXECUTION_TIME_SLICE = new Duration(100, TimeUnit.MILLISECONDS); - private final IndexedBlockingQueue<FragmentInstanceTask> queue; + // 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( - String workerId, ThreadGroup tg, IndexedBlockingQueue<FragmentInstanceTask> queue) { - super(tg, workerId); - this.queue = queue; + String workerId, + ThreadGroup tg, + IndexedBlockingQueue<FragmentInstanceTask> queue, + ITaskScheduler scheduler) { + super(workerId, tg, queue, scheduler); } @Override - public void run() { - while (true) { - try { - FragmentInstanceTask next = queue.poll(); - // do logic here - } catch (InterruptedException e) { - logger.info("{} is interrupted.", this.getName()); - break; - } + public void execute(FragmentInstanceTask task) throws InterruptedException { + // try to switch it to RUNNING + if (!getScheduler().readyToRunning(task)) { + return; + } + ExecFragmentInstance instance = task.getFragmentInstance(); + CpuTimer timer = new CpuTimer(); + ListenableFuture<Void> future = instance.processFor(EXECUTION_TIME_SLICE); + CpuTimer.CpuDuration duration = timer.elapsedTime(); + // long cost = System.nanoTime() - startTime; + // If the future is cancelled, the task is in an error and should be thrown. + if (future.isCancelled()) { + getScheduler().toAborted(task); + return; + } + ExecutionContext context = new ExecutionContext(); + context.setCpuDuration(duration); + context.setTimeSlice(EXECUTION_TIME_SLICE); + if (instance.isFinished()) { + getScheduler().runningToFinished(task, context); + return; + } + + if (future.isDone()) { + getScheduler().runningToReady(task, context); + } else { + getScheduler().runningToBlocked(task, context); + future.addListener( + () -> { + getScheduler().blockedToReady(task); + }, + listeningExecutor); } } } 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/FragmentInstanceTimeoutSentinel.java index 75f9eba..5093ecc 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/FragmentInstanceTimeoutSentinel.java @@ -20,64 +20,36 @@ 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.FragmentInstanceTaskStatus; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** the thread for watching the timeout of {@link FragmentInstanceTask} */ -public class FragmentInstanceTimeoutSentinel extends Thread { - - private static final Logger logger = - LoggerFactory.getLogger(FragmentInstanceTimeoutSentinel.class); - - private final IndexedBlockingQueue<FragmentInstanceTask> queue; - private final FragmentInstanceTaskCallback timeoutCallback; - // the check interval in milliseconds if the queue head remains the same. - private static final int CHECK_INTERVAL = 100; +public class FragmentInstanceTimeoutSentinel extends AbstractExecutor { public FragmentInstanceTimeoutSentinel( String workerId, ThreadGroup tg, IndexedBlockingQueue<FragmentInstanceTask> queue, - FragmentInstanceTaskCallback timeoutCallback) { - super(tg, workerId); - this.queue = queue; - this.timeoutCallback = timeoutCallback; + ITaskScheduler scheduler) { + super(workerId, tg, queue, scheduler); } @Override - public void run() { - while (true) { - try { - FragmentInstanceTask next = queue.poll(); - next.lock(); - try { - // if this task is already in an end state, it means that the resource releasing will be - // handled by other threads, we don't care anymore. - if (next.isEndState()) { - continue; - } - // if this task is not in end state and not timeout, we should push it back to the queue. - if (next.getDDL() > System.currentTimeMillis()) { - queue.push(next); - Thread.sleep(CHECK_INTERVAL); - continue; - } - next.setStatus(FragmentInstanceTaskStatus.ABORTED); - } finally { - next.unlock(); - } - try { - // Or we should do something to abort - timeoutCallback.call(next); - } catch (Exception e) { - logger.error("Abort instance " + next.getId() + " failed", e); - } - } catch (InterruptedException e) { - logger.info("{} is interrupted.", this.getName()); - break; + public void execute(FragmentInstanceTask task) throws InterruptedException { + task.lock(); + try { + // if this task is already in an end state, it means that the resource releasing will be + // handled by other threads, we don't care anymore. + if (task.isEndState()) { + return; } + } finally { + task.unlock(); + } + // if this task is not timeout, we can wait it to timeout. + long waitTime = task.getDDL() - System.currentTimeMillis(); + if (waitTime > 0L) { + // After this time, the task must be timeout. + Thread.sleep(waitTime); } + getScheduler().toAborted(task); } } diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IFragmentInstanceManager.java b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IFragmentInstanceManager.java index 98dc0c4..6d1e34f 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IFragmentInstanceManager.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IFragmentInstanceManager.java @@ -18,28 +18,19 @@ */ package org.apache.iotdb.db.mpp.schedule; -import org.apache.iotdb.db.mpp.buffer.IDataBlockManager; -import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceID; +import org.apache.iotdb.db.mpp.execution.ExecFragmentInstance; + +import java.util.List; /** the interface of fragment instance scheduling */ public interface IFragmentInstanceManager { - void submitFragmentInstance(); - /** - * the notifying interface for {@link IDataBlockManager} when upstream data comes. + * Submit one or more {@link ExecFragmentInstance} in one query for later scheduling. * - * @param instanceID the fragment instance to be notified. - * @param upstreamInstanceId the upstream instance id. + * @param instances the submitted instances. */ - void inputBlockAvailable(FragmentInstanceID instanceID, FragmentInstanceID upstreamInstanceId); - - /** - * the notifying interface for {@link IDataBlockManager} when downstream data has been consumed. - * - * @param instanceID the fragment instance to be notified. - */ - void outputBlockAvailable(FragmentInstanceID instanceID); + void submitFragmentInstances(String queryId, List<ExecFragmentInstance> instances); /** * abort all the instances in this query @@ -47,4 +38,7 @@ public interface IFragmentInstanceManager { * @param queryId the id of the query to be aborted. */ void abortQuery(String queryId); + + /** Fetch an {@link ExecFragmentInstance}. */ + void fetchFragmentInstance(ExecFragmentInstance instance); } 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 new file mode 100644 index 0000000..328acb6 --- /dev/null +++ b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/ITaskScheduler.java @@ -0,0 +1,77 @@ +/* + * 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.schedule; + +import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask; +import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTaskStatus; + +/** the scheduler interface of {@link FragmentInstanceTask} */ +interface ITaskScheduler { + + /** + * Switch a task from {@link FragmentInstanceTaskStatus#BLOCKED} to {@link + * FragmentInstanceTaskStatus#READY}. + * + * @param task the task to be switched. + */ + void blockedToReady(FragmentInstanceTask task); + + /** + * Switch a task from {@link FragmentInstanceTaskStatus#READY} to {@link + * FragmentInstanceTaskStatus#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); + + /** + * Switch a task from {@link FragmentInstanceTaskStatus#RUNNING} to {@link + * FragmentInstanceTaskStatus#READY}. + * + * @param task the task to be switched. + * @param context the execution context of last running. + */ + void runningToReady(FragmentInstanceTask task, ExecutionContext context); + + /** + * Switch a task from {@link FragmentInstanceTaskStatus#RUNNING} to {@link + * FragmentInstanceTaskStatus#BLOCKED}. + * + * @param task the task to be switched. + * @param context the execution context of last running. + */ + void runningToBlocked(FragmentInstanceTask task, ExecutionContext context); + + /** + * Switch a task from {@link FragmentInstanceTaskStatus#RUNNING} to {@link + * FragmentInstanceTaskStatus#FINISHED}. + * + * @param task the task to be switched. + * @param context the execution context of last running. + */ + void runningToFinished(FragmentInstanceTask task, ExecutionContext context); + + /** + * Switch a task to {@link FragmentInstanceTaskStatus#ABORTED}. + * + * @param task the task to be switched. + */ + void toAborted(FragmentInstanceTask 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/FragmentInstanceTask.java index 3c26f33..65e2e4a 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/FragmentInstanceTask.java @@ -18,12 +18,17 @@ */ package org.apache.iotdb.db.mpp.schedule.task; +import org.apache.iotdb.db.mpp.execution.ExecFragmentInstance; import org.apache.iotdb.db.mpp.schedule.ExecutionContext; import org.apache.iotdb.db.mpp.schedule.FragmentInstanceTaskExecutor; import org.apache.iotdb.db.mpp.schedule.queue.ID; import org.apache.iotdb.db.mpp.schedule.queue.IDIndexedAccessible; +import com.google.common.util.concurrent.ListenableFuture; +import io.airlift.units.Duration; + import java.util.Comparator; +import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -35,23 +40,26 @@ public class FragmentInstanceTask implements IDIndexedAccessible { private FragmentInstanceID id; private FragmentInstanceTaskStatus status; - private final ExecutionContext executionContext; + private final ExecFragmentInstance fragmentInstance; // the higher this field is, the higher probability it will be scheduled. - private long schedulePriority; + private double schedulePriority; private final long ddl; private final Lock lock; + // Running stats + private long cpuWallNano; + /** Initialize a dummy instance for queryHolder */ public FragmentInstanceTask() { - this(null, 0L, null); + this(new StubFragmentInstance(), 0L, null); } public FragmentInstanceTask( - FragmentInstanceID id, long timeoutMs, FragmentInstanceTaskStatus status) { - this.id = id; + ExecFragmentInstance instance, long timeoutMs, FragmentInstanceTaskStatus status) { + this.fragmentInstance = instance; + this.id = new FragmentInstanceID(instance.getInfo(), instance.getInfo(), instance.getInfo()); this.setStatus(status); - this.executionContext = new ExecutionContext(); this.schedulePriority = 0L; this.ddl = System.currentTimeMillis() + timeoutMs; this.lock = new ReentrantLock(); @@ -75,26 +83,32 @@ public class FragmentInstanceTask implements IDIndexedAccessible { || status == FragmentInstanceTaskStatus.FINISHED; } - public void inputReady(FragmentInstanceID inputId) { - throw new UnsupportedOperationException("unsupported"); - } - - public void outputReady() { - throw new UnsupportedOperationException("unsupported"); + public ExecFragmentInstance getFragmentInstance() { + return fragmentInstance; } public void setStatus(FragmentInstanceTaskStatus status) { this.status = status; } - public ExecutionContext getExecutionContext() { - return executionContext; - } + /** + * Update the schedule priority according to the execution context. + * + * @param context the last execution context. + */ + public void updateSchedulePriority(ExecutionContext context) { + // TODO: need to implement more complex here + + // 1. The penalty factor means that if a task executes less time in one schedule, it will have a + // high schedule priority + double penaltyFactor = + context.getCpuDuration().getWall().getValue(TimeUnit.NANOSECONDS) + / context.getTimeSlice().getValue(TimeUnit.NANOSECONDS); + // 2. If a task is nearly timeout, it should be scheduled as soon as possible. + long base = System.currentTimeMillis() - ddl; - /** Update the schedule priority according to the execution context. */ - public void updateSchedulePriority() { - // TODO: need to implement here - this.schedulePriority = System.currentTimeMillis() - ddl; + // 3. Now the final schedulePriority is out, this may not be so reasonable. + this.schedulePriority = base * penaltyFactor; } public void lock() { @@ -113,6 +127,16 @@ public class FragmentInstanceTask implements IDIndexedAccessible { return ddl; } + @Override + public int hashCode() { + return id.hashCode(); + } + + @Override + public boolean equals(Object o) { + return o instanceof FragmentInstanceTask && ((FragmentInstanceTask) o).getId().equals(id); + } + /** a comparator of ddl, the less the ddl is, the low order it has. */ public static class TimeoutComparator implements Comparator<FragmentInstanceTask> { @@ -148,4 +172,25 @@ public class FragmentInstanceTask implements IDIndexedAccessible { return o1.getId().compareTo(o2); } } + + private static class StubFragmentInstance implements ExecFragmentInstance { + + @Override + public boolean isFinished() { + return false; + } + + @Override + public ListenableFuture<Void> processFor(Duration duration) { + return null; + } + + @Override + public String getInfo() { + return "stub"; + } + + @Override + public void close() {} + } } diff --git a/server/src/main/java/org/apache/iotdb/db/utils/stats/CpuTimer.java b/server/src/main/java/org/apache/iotdb/db/utils/stats/CpuTimer.java new file mode 100644 index 0000000..2e4947e --- /dev/null +++ b/server/src/main/java/org/apache/iotdb/db/utils/stats/CpuTimer.java @@ -0,0 +1,156 @@ +/* + * 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.utils.stats; + +import io.airlift.units.Duration; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadMXBean; + +import static com.google.common.base.MoreObjects.toStringHelper; +import static java.util.concurrent.TimeUnit.NANOSECONDS; + +/** + * this class is copied from + * https://github.com/airlift/airlift/blob/214/stats/src/main/java/io/airlift/stats/CpuTimer.java as + * it doesn't support Java8. + */ +public class CpuTimer { + private static final ThreadMXBean THREAD_MX_BEAN = ManagementFactory.getThreadMXBean(); + + private final long wallStartTime; + private final long cpuStartTime; + private final long userStartTime; + + private long intervalWallStart; + private long intervalCpuStart; + private long intervalUserStart; + + public CpuTimer() { + wallStartTime = System.nanoTime(); + cpuStartTime = THREAD_MX_BEAN.getCurrentThreadCpuTime(); + userStartTime = THREAD_MX_BEAN.getCurrentThreadUserTime(); + + intervalWallStart = wallStartTime; + intervalCpuStart = cpuStartTime; + intervalUserStart = userStartTime; + } + + public CpuDuration startNewInterval() { + long currentWallTime = System.nanoTime(); + long currentCpuTime = THREAD_MX_BEAN.getCurrentThreadCpuTime(); + long currentUserTime = THREAD_MX_BEAN.getCurrentThreadUserTime(); + + CpuDuration cpuDuration = + new CpuDuration( + nanosBetween(intervalWallStart, currentWallTime), + nanosBetween(intervalCpuStart, currentCpuTime), + nanosBetween(intervalUserStart, currentUserTime)); + + intervalWallStart = currentWallTime; + intervalCpuStart = currentCpuTime; + intervalUserStart = currentUserTime; + + return cpuDuration; + } + + public CpuDuration elapsedIntervalTime() { + long currentWallTime = System.nanoTime(); + long currentCpuTime = THREAD_MX_BEAN.getCurrentThreadCpuTime(); + long currentUserTime = THREAD_MX_BEAN.getCurrentThreadUserTime(); + + return new CpuDuration( + nanosBetween(intervalWallStart, currentWallTime), + nanosBetween(intervalCpuStart, currentCpuTime), + nanosBetween(intervalUserStart, currentUserTime)); + } + + public CpuDuration elapsedTime() { + long currentWallTime = System.nanoTime(); + long currentCpuTime = THREAD_MX_BEAN.getCurrentThreadCpuTime(); + long currentUserTime = THREAD_MX_BEAN.getCurrentThreadUserTime(); + + return new CpuDuration( + nanosBetween(wallStartTime, currentWallTime), + nanosBetween(cpuStartTime, currentCpuTime), + nanosBetween(userStartTime, currentUserTime)); + } + + private static Duration nanosBetween(long start, long end) { + return new Duration(Math.abs(end - start), NANOSECONDS); + } + + public static class CpuDuration { + private final Duration wall; + private final Duration cpu; + private final Duration user; + + public CpuDuration() { + this.wall = new Duration(0, NANOSECONDS); + this.cpu = new Duration(0, NANOSECONDS); + this.user = new Duration(0, NANOSECONDS); + } + + public CpuDuration(Duration wall, Duration cpu, Duration user) { + this.wall = wall; + this.cpu = cpu; + this.user = user; + } + + public Duration getWall() { + return wall; + } + + public Duration getCpu() { + return cpu; + } + + public Duration getUser() { + return user; + } + + public CpuDuration add(CpuDuration cpuDuration) { + return new CpuDuration( + addDurations(wall, cpuDuration.wall), + addDurations(cpu, cpuDuration.cpu), + addDurations(user, cpuDuration.user)); + } + + public CpuDuration subtract(CpuDuration cpuDuration) { + return new CpuDuration( + subtractDurations(wall, cpuDuration.wall), + subtractDurations(cpu, cpuDuration.cpu), + subtractDurations(user, cpuDuration.user)); + } + + private static Duration addDurations(Duration a, Duration b) { + return new Duration(a.getValue(NANOSECONDS) + b.getValue(NANOSECONDS), NANOSECONDS); + } + + private static Duration subtractDurations(Duration a, Duration b) { + return new Duration( + Math.max(0, a.getValue(NANOSECONDS) - b.getValue(NANOSECONDS)), NANOSECONDS); + } + + @Override + public String toString() { + return toStringHelper(this).add("wall", wall).add("cpu", cpu).add("user", user).toString(); + } + } +}
