You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2019/06/25 04:05:26 UTC
[incubator-iotdb] branch feature_async_close_tsfile updated
(8c30b4a -> e18401b)
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a change to branch feature_async_close_tsfile
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git.
from 8c30b4a Merge branch 'feature_async_close_tsfile' of https://github.com/apache/incubator-iotdb into feature_async_close_tsfile
new bc334d1 add test and log in memtable flush task
new 009eb86 call unsealedtsfileprocessor.asyncclose in flushandcheckshouldclose in filenodeprocessor
new e18401b Merge remote-tracking branch 'origin/feature_async_close_tsfile' into feature_async_close_tsfile
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../db/engine/filenodeV2/FileNodeProcessorV2.java | 4 +-
.../db/engine/memtable/MemTableFlushTaskV2.java | 1 +
.../iotdb/db/engine/memtable/MemTablePoolTest.java | 68 ++++++++++++++++++++++
3 files changed, 70 insertions(+), 3 deletions(-)
create mode 100644 iotdb/src/test/java/org/apache/iotdb/db/engine/memtable/MemTablePoolTest.java
[incubator-iotdb] 01/03: add test and log in memtable flush task
Posted by qi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch feature_async_close_tsfile
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit bc334d13455f5eecf739ec67a7a1a1985c680ee3
Author: qiaojialin <64...@qq.com>
AuthorDate: Tue Jun 25 11:59:12 2019 +0800
add test and log in memtable flush task
---
.../db/engine/memtable/MemTableFlushTaskV2.java | 12 +++
.../filenodeV2/FileNodeProcessorV2Test2.java | 109 +++++++++++++++++++++
.../iotdb/db/engine/memtable/MemTablePoolTest.java | 68 +++++++++++++
3 files changed, 189 insertions(+)
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/memtable/MemTableFlushTaskV2.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/memtable/MemTableFlushTaskV2.java
index 13c45a5..b8825d1 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/memtable/MemTableFlushTaskV2.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/memtable/MemTableFlushTaskV2.java
@@ -60,6 +60,7 @@ public class MemTableFlushTaskV2 {
private Thread memoryFlushThread = new Thread(() -> {
long memSerializeTime = 0;
LOGGER.info("Storage group {},start serialize data into mem.", storageGroup);
+ long waitTime = 0;
while (!stop) {
Object task = memoryTaskQueue.poll();
if (task == null) {
@@ -68,7 +69,12 @@ public class MemTableFlushTaskV2 {
} catch (InterruptedException e) {
LOGGER.error("Storage group {}, io flush task is interrupted.", storageGroup, e);
}
+ waitTime += 10;
+ if (waitTime % 1000 == 0) {
+ LOGGER.info("encoding thread has waited {}ms", waitTime);
+ }
} else {
+ waitTime = 0;
if (task instanceof String) {
ioTaskQueue.add(task);
} else if (task instanceof ChunkGroupIoTask) {
@@ -101,6 +107,7 @@ public class MemTableFlushTaskV2 {
private Thread ioFlushThread = new Thread(() -> {
long ioTime = 0;
LOGGER.info("Storage group {}, start io cost.", storageGroup);
+ long waitTime = 0;
while (!stop) {
Object seriesWriterOrEndChunkGroupTask = ioTaskQueue.poll();
if (seriesWriterOrEndChunkGroupTask == null) {
@@ -109,7 +116,12 @@ public class MemTableFlushTaskV2 {
} catch (InterruptedException e) {
LOGGER.error("Storage group {}, io flush task is interrupted.", storageGroup, e);
}
+ waitTime += 10;
+ if (waitTime % 1000 == 0) {
+ LOGGER.info("io thread has waited {}ms", waitTime);
+ }
} else {
+ waitTime = 0;
long starTime = System.currentTimeMillis();
try {
if (seriesWriterOrEndChunkGroupTask instanceof IChunkWriter) {
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorV2Test2.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorV2Test2.java
new file mode 100644
index 0000000..e53c351
--- /dev/null
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorV2Test2.java
@@ -0,0 +1,109 @@
+/**
+ * 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.engine.filenodeV2;
+
+import org.apache.iotdb.db.engine.MetadataManagerHelper;
+import org.apache.iotdb.db.engine.querycontext.QueryDataSourceV2;
+import org.apache.iotdb.db.exception.FileNodeProcessorException;
+import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.TSRecord;
+import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class FileNodeProcessorV2Test {
+
+ private String storageGroup = "storage_group1";
+ private String systemDir = "data/info";
+ private String deviceId = "root.vehicle.d0";
+ private String measurementId = "s0";
+ private FileNodeProcessorV2 processor;
+
+ @Before
+ public void setUp() throws Exception {
+ MetadataManagerHelper.initMetadata();
+ EnvironmentUtils.envSetUp();
+ processor = new FileNodeProcessorV2(systemDir, storageGroup);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+ }
+
+
+ @Test
+ public void testSequenceSyncClose() {
+ for (int j = 1; j <= 100; j++) {
+ TSRecord record = new TSRecord(j, deviceId);
+ record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(j)));
+ processor.insert(new InsertPlan(record));
+ processor.asyncForceClose();
+ }
+
+ processor.syncCloseFileNode();
+ QueryDataSourceV2 queryDataSource = null;
+ try {
+ queryDataSource = processor.query(deviceId, measurementId);
+ } catch (FileNodeProcessorException e) {
+ e.printStackTrace();
+ }
+ Assert.assertEquals(queryDataSource.getSeqResources().size(), 100);
+ for (TsFileResourceV2 resource : queryDataSource.getSeqResources()) {
+ Assert.assertTrue(resource.isClosed());
+ }
+ }
+
+
+ @Test
+ public void testSeqAndUnSeqSyncClose() throws FileNodeProcessorException {
+
+ for (int j = 21; j <= 30; j++) {
+ TSRecord record = new TSRecord(j, deviceId);
+ record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(j)));
+ processor.insert(new InsertPlan(record));
+ processor.asyncForceClose();
+ }
+ processor.syncCloseFileNode();
+
+ for (int j = 10; j >= 1; j--) {
+ TSRecord record = new TSRecord(j, deviceId);
+ record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(j)));
+ processor.insert(new InsertPlan(record));
+ processor.asyncForceClose();
+ }
+
+ processor.syncCloseFileNode();
+
+ QueryDataSourceV2 queryDataSource = processor.query(deviceId, measurementId);
+ Assert.assertEquals(10, queryDataSource.getSeqResources().size());
+ Assert.assertEquals(10, queryDataSource.getUnseqResources().size());
+ for (TsFileResourceV2 resource : queryDataSource.getSeqResources()) {
+ Assert.assertTrue(resource.isClosed());
+ }
+ for (TsFileResourceV2 resource : queryDataSource.getUnseqResources()) {
+ Assert.assertTrue(resource.isClosed());
+ }
+ }
+
+}
\ No newline at end of file
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/memtable/MemTablePoolTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/memtable/MemTablePoolTest.java
new file mode 100644
index 0000000..b743207
--- /dev/null
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/memtable/MemTablePoolTest.java
@@ -0,0 +1,68 @@
+package org.apache.iotdb.db.engine.memtable;
+
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class MemTablePoolTest {
+
+ private ConcurrentLinkedQueue<IMemTable> memTables;
+
+ @Before
+ public void setUp() throws Exception {
+ memTables = new ConcurrentLinkedQueue();
+ new ReturnThread().start();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ }
+
+ @Test
+ public void testGetAndRelease() {
+ for (int i = 0; i < 100; i++) {
+ IMemTable memTable = MemTablePool.getInstance().getEmptyMemTable("test case");
+ memTables.add(memTable);
+ }
+ }
+
+ @Test
+ public void testSort() {
+ long start = System.currentTimeMillis();
+ TreeMap<Long, Long> treeMap = new TreeMap<>();
+ for (int i = 0; i < 1000000; i++) {
+ treeMap.put((long)i, (long)i);
+ }
+ start = System.currentTimeMillis() - start;
+ System.out.println("time cost: " + start);
+ }
+
+
+ class ReturnThread extends Thread{
+
+ @Override
+ public void run() {
+ while (true) {
+ IMemTable memTable = memTables.poll();
+ if (memTable == null) {
+ try {
+ Thread.sleep(10);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ continue;
+ }
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ memTables.remove(memTable);
+ MemTablePool.getInstance().putBack(memTable, "test case");
+ }
+ }
+ }
+
+}
\ No newline at end of file
[incubator-iotdb] 02/03: call unsealedtsfileprocessor.asyncclose in
flushandcheckshouldclose in filenodeprocessor
Posted by qi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch feature_async_close_tsfile
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 009eb869b66eaa03a8ff12847937b199b61d6197
Merge: bc334d1 c3a36d1
Author: qiaojialin <64...@qq.com>
AuthorDate: Tue Jun 25 12:04:42 2019 +0800
call unsealedtsfileprocessor.asyncclose in flushandcheckshouldclose in filenodeprocessor
.../org/apache/iotdb/db/concurrent/ThreadName.java | 1 +
.../db/engine/filenode/FileNodeProcessor.java | 8 +-
.../db/engine/filenodeV2/FileNodeProcessorV2.java | 168 +++++++++---------
.../iotdb/db/engine/filenodeV2/FlushManager.java | 7 +-
.../filenodeV2/UnsealedTsFileProcessorV2.java | 25 +--
.../db/engine/memtable/MemTableFlushTask.java | 36 ++--
.../db/engine/memtable/MemTableFlushTaskV2.java | 191 ++++++++++++---------
...leFlushTaskV2.java => MemTableFlushTaskV3.java} | 166 +++++++++---------
.../db/engine/memtable/MemTableFlushUtil.java | 3 +-
.../iotdb/db/engine/memtable/MemTablePool.java | 2 +-
.../iotdb/db/engine/pool/FlushPoolManager.java | 6 +-
...olManager.java => FlushSubTaskPoolManager.java} | 51 ++----
.../EngineExecutorWithoutTimeGenerator.java | 3 +-
.../db/query/factory/ISeriesReaderFactory.java | 27 ++-
.../db/query/factory/SeriesReaderFactory.java | 6 +-
.../db/query/factory/SeriesReaderFactoryImpl.java | 97 +++++++----
.../java/org/apache/iotdb/db/query/fill/IFill.java | 2 +-
...a => AllDataReaderWithOptGlobalTimeFilter.java} | 11 +-
.../query/reader/AllDataReaderWithValueFilter.java | 74 ++++++++
.../query/timegenerator/EngineNodeConstructor.java | 3 +-
.../iotdb/db/writelog/recover/LogReplayer.java | 13 +-
...rPerformer.java => TsFileRecoverPerformer.java} | 12 +-
.../recover/UnSeqTsFileRecoverPerformer.java | 117 -------------
.../engine/filenodeV2/FileNodeProcessorV2Test.java | 2 +-
.../filenodeV2/FileNodeProcessorV2Test2.java | 109 ------------
.../iotdb/db/integration/IoTDBCompleteIT.java | 2 +
... AllDataReaderWithOptGlobalTimeFilterTest.java} | 10 +-
.../db/writelog/recover/SeqTsFileRecoverTest.java | 5 +-
.../writelog/recover/UnseqTsFileRecoverTest.java | 4 +-
.../apache/iotdb/tsfile/write/TsFileWriter.java | 2 +-
.../iotdb/tsfile/write/writer/TsFileIOWriter.java | 6 +-
.../iotdb/tsfile/write/TsFileIOWriterTest.java | 2 +-
32 files changed, 545 insertions(+), 626 deletions(-)
diff --cc iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorV2.java
index 61b25fa,d7f6f6b..a2f2f77
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorV2.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorV2.java
@@@ -481,14 -474,14 +474,12 @@@ public class FileNodeProcessorV2
closingUnSequenceTsFileProcessor.add(unsealedTsFileProcessor);
workUnSequenceTsFileProcessor = null;
}
-- unsealedTsFileProcessor.setCloseMark();
LOGGER.info("The file size {} reaches the threshold, async close tsfile: {}.",
unsealedTsFileProcessor.getTsFileResource().getFileSize(),
unsealedTsFileProcessor.getTsFileResource().getFile().getAbsolutePath());
}
-- unsealedTsFileProcessor.asyncFlush();
--
++ unsealedTsFileProcessor.asyncClose();
}
public void asyncForceClose() {
diff --cc iotdb/src/main/java/org/apache/iotdb/db/engine/memtable/MemTableFlushTaskV2.java
index b8825d1,dcf3ccd..d12f601
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/memtable/MemTableFlushTaskV2.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/memtable/MemTableFlushTaskV2.java
@@@ -51,55 -57,62 +57,63 @@@ public class MemTableFlushTaskV2
this.tsFileIoWriter = writer;
this.storageGroup = storageGroup;
this.flushCallBack = callBack;
- ioFlushThread.start();
- memoryFlushThread.start();
- LOGGER.info("flush task created in Storage group {} ", storageGroup);
+ this.memoryFlushTaskFuture = subTaskPoolManager.submit(memoryFlushTask);
+ this.ioFlushTaskFuture = subTaskPoolManager.submit(ioFlushTask);
+ LOGGER.info("flush task of Storage group {} memtable {} is created ",
+ storageGroup, memTable.getVersion());
}
+
- private Thread memoryFlushThread = new Thread(() -> {
- long memSerializeTime = 0;
- LOGGER.info("Storage group {},start serialize data into mem.", storageGroup);
- long waitTime = 0;
- while (!stop) {
- Object task = memoryTaskQueue.poll();
- if (task == null) {
- try {
- Thread.sleep(10);
- } catch (InterruptedException e) {
- LOGGER.error("Storage group {}, io flush task is interrupted.", storageGroup, e);
- }
- waitTime += 10;
- if (waitTime % 1000 == 0) {
- LOGGER.info("encoding thread has waited {}ms", waitTime);
- }
- } else {
- waitTime = 0;
- if (task instanceof String) {
- ioTaskQueue.add(task);
- } else if (task instanceof ChunkGroupIoTask) {
- ioTaskQueue.add(task);
- } else {
- long starTime = System.currentTimeMillis();
- Pair<List<TimeValuePair>, MeasurementSchema> memorySerializeTask = (Pair<List<TimeValuePair>, MeasurementSchema>) task;
- ChunkBuffer chunkBuffer = new ChunkBuffer(memorySerializeTask.right);
- IChunkWriter seriesWriter = new ChunkWriterImpl(memorySerializeTask.right, chunkBuffer,
- PAGE_SIZE_THRESHOLD);
+ private Runnable memoryFlushTask = new Runnable() {
+ @Override
+ public void run() {
+ long memSerializeTime = 0;
+ LOGGER.info("Storage group {} memtable {}, starts to serialize data into mem.", storageGroup,
+ memTable.getVersion());
+ while (!stop) {
+ Object task = memoryTaskQueue.poll();
+ if (task == null) {
try {
- writeOneSeries(memorySerializeTask.left, seriesWriter,
- memorySerializeTask.right.getType());
- ioTaskQueue.add(seriesWriter);
- } catch (IOException e) {
- LOGGER.error("Storage group {}, io error.", storageGroup, e);
- throw new RuntimeException(e);
+ Thread.sleep(10);
+ } catch (InterruptedException e) {
+ LOGGER.error("Storage group {} memtable {}, io flush task is interrupted.",
+ storageGroup, memTable.getVersion(), e);
+ }
+ } else {
+ if (task instanceof String) {
+ LOGGER.info("Storage group {} memtable {}, issues a start flush chunk group task.",
+ storageGroup, memTable.getVersion());
+ ioTaskQueue.add(task);
+ } else if (task instanceof ChunkGroupIoTask) {
+ LOGGER.info("Storage group {} memtable {}, issues a end flush chunk group task.",
+ storageGroup, memTable.getVersion());
+ ioTaskQueue.add(task);
+ } else {
+ long starTime = System.currentTimeMillis();
+ Pair<List<TimeValuePair>, MeasurementSchema> memorySerializeTask = (Pair<List<TimeValuePair>, MeasurementSchema>) task;
+ ChunkBuffer chunkBuffer = new ChunkBuffer(memorySerializeTask.right);
+ IChunkWriter seriesWriter = new ChunkWriterImpl(memorySerializeTask.right, chunkBuffer,
+ PAGE_SIZE_THRESHOLD);
+ try {
+ writeOneSeries(memorySerializeTask.left, seriesWriter,
+ memorySerializeTask.right.getType());
+ ioTaskQueue.add(seriesWriter);
+ } catch (IOException e) {
+ LOGGER.error("Storage group {} memtable {}, io error.", storageGroup,
+ memTable.getVersion(), e);
+ throw new RuntimeException(e);
+ }
+ LOGGER.info("Storage group {} memtable {}, issues a write chunk task.",
+ storageGroup, memTable.getVersion());
+ memSerializeTime += System.currentTimeMillis() - starTime;
}
- memSerializeTime += System.currentTimeMillis() - starTime;
}
}
+ LOGGER.info("Storage group {}, flushing memtable {} into disk: serialize data into mem cost "
+ + "{} ms.",
+ storageGroup, memTable.getVersion(), memSerializeTime);
}
- LOGGER.info("Storage group {}, flushing a memtable into disk: serialize data into mem cost {} ms.",
- storageGroup, memSerializeTime);
- });
+ };
//TODO a better way is: for each TsFile, assign it a Executors.singleThreadPool,
[incubator-iotdb] 03/03: Merge remote-tracking branch
'origin/feature_async_close_tsfile' into feature_async_close_tsfile
Posted by qi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch feature_async_close_tsfile
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit e18401bf6f010b48246cef75a12b2d6364246557
Merge: 009eb86 8c30b4a
Author: qiaojialin <64...@qq.com>
AuthorDate: Tue Jun 25 12:05:03 2019 +0800
Merge remote-tracking branch 'origin/feature_async_close_tsfile' into feature_async_close_tsfile
.../engine/bufferwrite/BufferWriteProcessor.java | 4 +-
.../db/engine/filenode/FileNodeProcessor.java | 2 +-
.../db/qp/executor/IQueryProcessExecutor.java | 8 +-
.../groupby/GroupByWithValueFilterDataSet.java | 2 +-
.../db/query/executor/AggregateEngineExecutor.java | 6 +-
...rWithTimeGenerator.java => EngineExecutor.java} | 63 ++++-
.../EngineExecutorWithoutTimeGenerator.java | 103 -------
.../iotdb/db/query/executor/EngineQueryRouter.java | 17 +-
.../db/query/executor/FillEngineExecutor.java | 2 +-
.../db/query/factory/ISeriesReaderFactory.java | 12 +-
.../db/query/factory/SeriesReaderFactory.java | 312 ---------------------
.../db/query/factory/SeriesReaderFactoryImpl.java | 8 +-
.../java/org/apache/iotdb/db/query/fill/IFill.java | 2 +-
.../query/reader/sequence/SealedTsFilesReader.java | 183 ------------
.../sequence/SealedTsFilesReaderByTimestamp.java | 136 ---------
.../query/reader/sequence/SequenceDataReader.java | 148 ----------
.../sequence/SequenceDataReaderByTimestamp.java | 108 -------
.../reader/sequence/UnSealedTsFileReader.java | 101 -------
.../sequence/UnSealedTsFilesReaderByTimestamp.java | 64 -----
.../query/timegenerator/EngineNodeConstructor.java | 3 +-
.../org/apache/iotdb/db/service/StartupChecks.java | 2 +-
.../org/apache/iotdb/db/service/TSServiceImpl.java | 4 +-
.../iotdb/db/integration/IoTDBAuthorizationIT.java | 2 +-
.../iotdb/db/integration/IoTDBLargeDataIT.java | 4 +-
.../iotdb/db/integration/IoTDBMultiSeriesIT.java | 4 +-
.../iotdb/db/integration/IoTDBSeriesReaderIT.java | 4 +-
.../iotdb/db/integration/IoTDBTimeZoneIT.java | 2 +-
.../org/apache/iotdb/db/qp/plan/QPUpdateTest.java | 2 +-
.../SequenceDataReaderByTimestampTest.java | 91 ------
...reateByTimestampReadersOfSelectedPathsTest.java | 2 +-
.../apache/iotdb/db/utils/EnvironmentUtils.java | 6 +
31 files changed, 103 insertions(+), 1304 deletions(-)