[iotdb] branch ISSUE_5792 updated: fix task initialization

[ji...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jiangtian pushed a commit to branch ISSUE_5792
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/ISSUE_5792 by this push:
     new 4e25ecd2be fix task initialization
4e25ecd2be is described below

commit 4e25ecd2bef76e89c5d57e75652a9acf486f187c
Author: Tian Jiang <jt...@163.com>
AuthorDate: Fri Apr 21 12:03:40 2023 +0800

    fix task initialization
---
 .../commons/concurrent/dynamic/DynamicThreadGroup.java   |  4 +++-
 .../iotdb/db/engine/flush/MemTableFlushTaskV2.java       | 16 ++++++++++------
 .../apache/iotdb/db/engine/flush/tasks/DeviceIOTask.java |  4 ++--
 .../apache/iotdb/db/engine/flush/tasks/FinalTask.java    |  4 +---
 .../iotdb/db/engine/flush/tasks/SortSeriesTask.java      |  4 +---
 5 files changed, 17 insertions(+), 15 deletions(-)

diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/dynamic/DynamicThreadGroup.java b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/dynamic/DynamicThreadGroup.java
index 240917d076..bbbd62a7e4 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/dynamic/DynamicThreadGroup.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/dynamic/DynamicThreadGroup.java
@@ -18,12 +18,12 @@
  */
 package org.apache.iotdb.commons.concurrent.dynamic;
 
-import java.util.concurrent.ExecutionException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
@@ -51,7 +51,9 @@ public class DynamicThreadGroup {
     this.threadFactory = threadFactory;
     this.minThreadCnt = Math.max(1, minThreadCnt);
     this.maxThreadCnt = Math.max(this.minThreadCnt, maxThreadCnt);
+  }
 
+  public void init() {
     int initialThreadNum = (this.minThreadCnt + this.maxThreadCnt) / 2;
     for (int i = 0; i < initialThreadNum; i++) {
       addThread();
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTaskV2.java b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTaskV2.java
index 6a4b95eaa6..2d56412e0f 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTaskV2.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTaskV2.java
@@ -115,12 +115,16 @@ public class MemTableFlushTaskV2 {
             this::newEncodingThread,
             config.getFlushMemTableMinSubThread(),
             config.getFlushMemTableMaxSubThread());
-    this.ioTask = new DynamicThreadGroup(
-        storageGroup + "-" + dataRegionId + "-" + memTable,
-        SUB_TASK_POOL_MANAGER::submit,
-        this::newIOThread,
-        1,
-        1);
+    this.ioTask =
+        new DynamicThreadGroup(
+            storageGroup + "-" + dataRegionId + "-" + memTable,
+            SUB_TASK_POOL_MANAGER::submit,
+            this::newIOThread,
+            1,
+            1);
+    this.sortTasks.init();
+    this.encodingTasks.init();
+    this.ioTask.init();
     LOGGER.debug(
         "flush task of database {} memtable is created, flushing to file {}.",
         storageGroup,
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/flush/tasks/DeviceIOTask.java b/server/src/main/java/org/apache/iotdb/db/engine/flush/tasks/DeviceIOTask.java
index cc3c9ba3c6..7267af2a3d 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/flush/tasks/DeviceIOTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/flush/tasks/DeviceIOTask.java
@@ -29,8 +29,8 @@ import java.io.IOException;
  * A DeviceIOTask will check if the current device in the FlushContext (indicated by
  * FlushContext.cursor) is fully encoded. If it is fully encoded, it will be flushed to the
  * underlying file and the cursor will be moved to the next device.
- * <p>
- * The above procedure repeats until the current device is still being encoded or all devices are
+ *
+ * <p>The above procedure repeats until the current device is still being encoded or all devices are
  * flushed. If all devices are flushed, a FinalTask will be generated. Otherwise, no new task will
  * be generated.
  */
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/flush/tasks/FinalTask.java b/server/src/main/java/org/apache/iotdb/db/engine/flush/tasks/FinalTask.java
index 632f9b4ef3..de5d74d3ca 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/flush/tasks/FinalTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/flush/tasks/FinalTask.java
@@ -20,9 +20,7 @@ package org.apache.iotdb.db.engine.flush.tasks;
 
 import org.apache.iotdb.commons.concurrent.pipeline.Task;
 
-/**
- * FinalTask ends the current TaskRunner by interruption.
- */
+/** FinalTask ends the current TaskRunner by interruption. */
 public class FinalTask implements Task {
 
   @Override
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/flush/tasks/SortSeriesTask.java b/server/src/main/java/org/apache/iotdb/db/engine/flush/tasks/SortSeriesTask.java
index a5c529f6ed..055e6f56d7 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/flush/tasks/SortSeriesTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/flush/tasks/SortSeriesTask.java
@@ -23,9 +23,7 @@ import org.apache.iotdb.db.engine.memtable.IWritableMemChunk;
 import org.apache.iotdb.db.metadata.idtable.entry.IDeviceID;
 import org.apache.iotdb.db.service.metrics.WritingMetrics;
 
-/**
- * SortSeriesTask sorts a timeseries and generates the associated encoding task.
- */
+/** SortSeriesTask sorts a timeseries and generates the associated encoding task. */
 public class SortSeriesTask implements Task {
   private IDeviceID deviceId;
   private String seriesId;