You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2022/12/20 01:24:31 UTC
[iotdb] 02/02: s
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch tsbs
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 58267a5eb25a0d750a7753ff09e8b9268e9010b2
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Mon Dec 19 09:45:03 2022 +0800
s
---
.../mpp/execution/exchange/LocalSourceHandle.java | 38 +++++++++----------
.../mpp/execution/schedule/DriverTaskThread.java | 5 +--
.../mpp/execution/schedule/ExecutionContext.java | 1 -
.../db/mpp/execution/schedule/task/DriverTask.java | 3 +-
.../schedule/DefaultDriverSchedulerTest.java | 44 +++++++++++-----------
5 files changed, 45 insertions(+), 46 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/LocalSourceHandle.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/LocalSourceHandle.java
index dd0c88f0c1..be81208f00 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/LocalSourceHandle.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/LocalSourceHandle.java
@@ -88,26 +88,26 @@ public class LocalSourceHandle implements ISourceHandle {
@Override
public TsBlock receive() {
-// try (SetThreadName sourceHandleName = new SetThreadName(threadName)) {
- checkState();
+ // try (SetThreadName sourceHandleName = new SetThreadName(threadName)) {
+ checkState();
- if (!queue.isBlocked().isDone()) {
- throw new IllegalStateException("Source handle is blocked.");
- }
- TsBlock tsBlock;
- synchronized (queue) {
- tsBlock = queue.remove();
- }
- if (tsBlock != null) {
- logger.debug(
- "[GetTsBlockFromQueue] TsBlock:{} size:{}",
- currSequenceId,
- tsBlock.getRetainedSizeInBytes());
- currSequenceId++;
- }
- checkAndInvokeOnFinished();
- return tsBlock;
-// }
+ if (!queue.isBlocked().isDone()) {
+ throw new IllegalStateException("Source handle is blocked.");
+ }
+ TsBlock tsBlock;
+ synchronized (queue) {
+ tsBlock = queue.remove();
+ }
+ if (tsBlock != null) {
+ logger.debug(
+ "[GetTsBlockFromQueue] TsBlock:{} size:{}",
+ currSequenceId,
+ tsBlock.getRetainedSizeInBytes());
+ currSequenceId++;
+ }
+ checkAndInvokeOnFinished();
+ return tsBlock;
+ // }
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverTaskThread.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverTaskThread.java
index b55dcf9ef5..e28797590b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverTaskThread.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverTaskThread.java
@@ -24,7 +24,6 @@ import org.apache.iotdb.db.mpp.execution.driver.IDriver;
import org.apache.iotdb.db.mpp.execution.schedule.queue.IndexedBlockingQueue;
import org.apache.iotdb.db.mpp.execution.schedule.task.DriverTask;
import org.apache.iotdb.db.utils.SetThreadName;
-import org.apache.iotdb.db.utils.stats.CpuTimer;
import com.google.common.util.concurrent.ListenableFuture;
import io.airlift.units.Duration;
@@ -61,9 +60,9 @@ public class DriverTaskThread extends AbstractDriverThread {
}
IDriver instance = task.getFragmentInstance();
long startTime = System.nanoTime();
-// CpuTimer timer = new CpuTimer();
+ // CpuTimer timer = new CpuTimer();
ListenableFuture<?> future = instance.processFor(EXECUTION_TIME_SLICE);
-// CpuTimer.CpuDuration duration = timer.elapsedTime();
+ // 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()) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/ExecutionContext.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/ExecutionContext.java
index 070aa548b0..bdbfa2d353 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/ExecutionContext.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/ExecutionContext.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.mpp.execution.schedule;
import org.apache.iotdb.db.mpp.execution.schedule.task.DriverTask;
-import org.apache.iotdb.db.utils.stats.CpuTimer;
import io.airlift.units.Duration;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/task/DriverTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/task/DriverTask.java
index 1aa87a119d..47f25c36cd 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/task/DriverTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/task/DriverTask.java
@@ -102,7 +102,8 @@ public class DriverTask implements IDIndexedAccessible {
// 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() / context.getTimeSlice().getValue(TimeUnit.NANOSECONDS);
+ double penaltyFactor =
+ context.getCpuDuration() / 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;
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/schedule/DefaultDriverSchedulerTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/schedule/DefaultDriverSchedulerTest.java
index 10fd9fac0e..28616c2fd1 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/schedule/DefaultDriverSchedulerTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/schedule/DefaultDriverSchedulerTest.java
@@ -1,4 +1,4 @@
-///*
+/// *
// * 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
@@ -16,30 +16,30 @@
// * specific language governing permissions and limitations
// * under the License.
// */
-//package org.apache.iotdb.db.mpp.execution.schedule;
+// package org.apache.iotdb.db.mpp.execution.schedule;
//
-//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.driver.IDriver;
-//import org.apache.iotdb.db.mpp.execution.exchange.IMPPDataExchangeManager;
-//import org.apache.iotdb.db.mpp.execution.schedule.task.DriverTask;
-//import org.apache.iotdb.db.mpp.execution.schedule.task.DriverTaskStatus;
-//import org.apache.iotdb.db.utils.stats.CpuTimer;
-//import org.apache.iotdb.mpp.rpc.thrift.IDataNodeRPCService;
+// 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.driver.IDriver;
+// import org.apache.iotdb.db.mpp.execution.exchange.IMPPDataExchangeManager;
+// import org.apache.iotdb.db.mpp.execution.schedule.task.DriverTask;
+// import org.apache.iotdb.db.mpp.execution.schedule.task.DriverTaskStatus;
+// import org.apache.iotdb.db.utils.stats.CpuTimer;
+// import org.apache.iotdb.mpp.rpc.thrift.IDataNodeRPCService;
//
-//import io.airlift.units.Duration;
-//import org.junit.After;
-//import org.junit.Assert;
-//import org.junit.Test;
-//import org.mockito.Mockito;
+// import io.airlift.units.Duration;
+// import org.junit.After;
+// import org.junit.Assert;
+// import org.junit.Test;
+// import org.mockito.Mockito;
//
-//import java.io.IOException;
-//import java.util.HashSet;
-//import java.util.Set;
-//import java.util.concurrent.TimeUnit;
+// import java.io.IOException;
+// import java.util.HashSet;
+// import java.util.Set;
+// import java.util.concurrent.TimeUnit;
//
-//public class DefaultDriverSchedulerTest {
+// public class DefaultDriverSchedulerTest {
//
// private final DriverScheduler manager = DriverScheduler.getInstance();
//
@@ -404,4 +404,4 @@
// manager.getReadyQueue().clear();
// manager.getTimeoutQueue().clear();
// }
-//}
+// }