This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch rel/0.12
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/0.12 by this push:
new 1761a79 close TsFileSequenceReader immediately if ref is zero (#4481)
1761a79 is described below
commit 1761a79083412cb2b41aad2e2a5423a83a97b18e
Author: Xiangwei Wei <[email protected]>
AuthorDate: Sat Nov 27 10:02:17 2021 +0800
close TsFileSequenceReader immediately if ref is zero (#4481)
---
.../iotdb/db/query/control/FileReaderManager.java | 112 ++++++---------------
.../iotdb/db/engine/cache/ChunkCacheTest.java | 5 +-
.../compaction/LevelCompactionRecoverTest.java | 4 +-
.../compaction/LevelCompactionRestoreTest.java | 1 -
.../db/engine/compaction/LevelCompactionTest.java | 1 -
.../apache/iotdb/db/engine/merge/MergeTest.java | 1 -
.../db/query/control/FileReaderManagerTest.java | 19 +---
.../query/reader/series/SeriesReaderTestUtil.java | 1 -
8 files changed, 42 insertions(+), 102 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/control/FileReaderManager.java
b/server/src/main/java/org/apache/iotdb/db/query/control/FileReaderManager.java
index cf329af..0046bf0 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/control/FileReaderManager.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/control/FileReaderManager.java
@@ -18,11 +18,7 @@
*/
package org.apache.iotdb.db.query.control;
-import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.service.IService;
-import org.apache.iotdb.db.service.ServiceType;
import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import org.apache.iotdb.tsfile.read.UnClosedTsFileReader;
@@ -35,15 +31,13 @@ import java.io.IOException;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
* FileReaderManager is a singleton, which is used to manage all file
readers(opened file streams)
* to ensure that each file is opened at most once.
*/
-public class FileReaderManager implements IService {
+public class FileReaderManager {
private static final Logger logger =
LoggerFactory.getLogger(FileReaderManager.class);
private static final Logger resourceLogger =
LoggerFactory.getLogger("FileMonitor");
@@ -73,16 +67,11 @@ public class FileReaderManager implements IService {
*/
private Map<String, AtomicInteger> unclosedReferenceMap;
- private ScheduledExecutorService executorService;
-
private FileReaderManager() {
closedFileReaderMap = new ConcurrentHashMap<>();
unclosedFileReaderMap = new ConcurrentHashMap<>();
closedReferenceMap = new ConcurrentHashMap<>();
unclosedReferenceMap = new ConcurrentHashMap<>();
- executorService = IoTDBThreadPoolFactory.newScheduledThreadPool(1,
"open-files-manager");
-
- clearUnUsedFilesInFixTime();
}
public static FileReaderManager getInstance() {
@@ -102,46 +91,6 @@ public class FileReaderManager implements IService {
}
}
- private void clearUnUsedFilesInFixTime() {
-
- long examinePeriod =
IoTDBDescriptor.getInstance().getConfig().getCacheFileReaderClearPeriod();
-
- executorService.scheduleAtFixedRate(
- () -> {
- synchronized (this) {
- clearMap(closedFileReaderMap, closedReferenceMap);
- clearMap(unclosedFileReaderMap, unclosedReferenceMap);
- }
- },
- 0,
- examinePeriod,
- TimeUnit.MILLISECONDS);
- }
-
- private void clearMap(
- Map<String, TsFileSequenceReader> readerMap, Map<String, AtomicInteger>
refMap) {
- Iterator<Map.Entry<String, TsFileSequenceReader>> iterator =
readerMap.entrySet().iterator();
- while (iterator.hasNext()) {
- Map.Entry<String, TsFileSequenceReader> entry = iterator.next();
- TsFileSequenceReader reader = entry.getValue();
- AtomicInteger refAtom = refMap.get(entry.getKey());
-
- if (refAtom != null && refAtom.get() == 0) {
- try {
- reader.close();
- } catch (IOException e) {
- logger.error("Can not close TsFileSequenceReader {} !",
reader.getFileName(), e);
- }
- iterator.remove();
- refMap.remove(entry.getKey());
- if (resourceLogger.isDebugEnabled()) {
- resourceLogger.debug(
- "{} TsFileReader is closed because of no reference.",
entry.getKey());
- }
- }
- }
- }
-
/**
* Get the reader of the file(tsfile or unseq tsfile) indicated by filePath.
If the reader already
* exists, just get it from closedFileReaderMap or unclosedFileReaderMap
depending on isClosing .
@@ -213,14 +162,44 @@ public class FileReaderManager implements IService {
void decreaseFileReaderReference(TsFileResource tsFile, boolean isClosed) {
synchronized (this) {
if (!isClosed &&
unclosedReferenceMap.containsKey(tsFile.getTsFilePath())) {
- unclosedReferenceMap.get(tsFile.getTsFilePath()).decrementAndGet();
+ if (unclosedReferenceMap.get(tsFile.getTsFilePath()).decrementAndGet()
== 0) {
+ closeUnUsedReaderAndRemoveRef(tsFile.getTsFilePath(), false);
+ }
} else if (closedReferenceMap.containsKey(tsFile.getTsFilePath())) {
- closedReferenceMap.get(tsFile.getTsFilePath()).decrementAndGet();
+ if (closedReferenceMap.get(tsFile.getTsFilePath()).decrementAndGet()
== 0) {
+ closeUnUsedReaderAndRemoveRef(tsFile.getTsFilePath(), true);
+ }
}
}
tsFile.readUnlock();
}
+ private void closeUnUsedReaderAndRemoveRef(String tsFilePath, boolean
isClosed) {
+ Map<String, TsFileSequenceReader> readerMap =
+ isClosed ? closedFileReaderMap : unclosedFileReaderMap;
+ Map<String, AtomicInteger> refMap = isClosed ? closedReferenceMap :
unclosedReferenceMap;
+ synchronized (this) {
+ // check ref num again
+ if (refMap.get(tsFilePath).get() != 0) {
+ return;
+ }
+
+ TsFileSequenceReader reader = readerMap.get(tsFilePath);
+ if (reader != null) {
+ try {
+ reader.close();
+ } catch (IOException e) {
+ logger.error("Can not close TsFileSequenceReader {} !",
reader.getFileName(), e);
+ }
+ }
+ readerMap.remove(tsFilePath);
+ refMap.remove(tsFilePath);
+ if (resourceLogger.isDebugEnabled()) {
+ resourceLogger.debug("{} TsFileReader is closed because of no
reference.", tsFilePath);
+ }
+ }
+ }
+
/**
* Only for <code>EnvironmentUtils.cleanEnv</code> method. To make sure that
unit tests and
* integration tests will not conflict with each other.
@@ -255,31 +234,6 @@ public class FileReaderManager implements IService {
|| (!isClosed &&
unclosedFileReaderMap.containsKey(tsFile.getTsFilePath()));
}
- @Override
- public void start() {
- // Do nothing
- }
-
- @Override
- public void stop() {
- if (executorService == null || executorService.isShutdown()) {
- return;
- }
-
- executorService.shutdown();
- try {
- executorService.awaitTermination(10, TimeUnit.SECONDS);
- } catch (InterruptedException e) {
- logger.error("StatMonitor timing service could not be shutdown.", e);
- Thread.currentThread().interrupt();
- }
- }
-
- @Override
- public ServiceType getID() {
- return ServiceType.FILE_READER_MANAGER_SERVICE;
- }
-
private static class FileReaderManagerHelper {
private static final FileReaderManager INSTANCE = new FileReaderManager();
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/cache/ChunkCacheTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/cache/ChunkCacheTest.java
index 1a0d25d..e374f14 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/cache/ChunkCacheTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/cache/ChunkCacheTest.java
@@ -50,7 +50,9 @@ import org.junit.Test;
import java.io.File;
import java.io.IOException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
import static org.apache.iotdb.db.conf.IoTDBConstant.PATH_SEPARATOR;
import static org.junit.Assert.assertTrue;
@@ -251,6 +253,5 @@ public class ChunkCacheTest {
resourceFile.delete();
}
FileReaderManager.getInstance().closeAndRemoveAllOpenedReaders();
- FileReaderManager.getInstance().stop();
}
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionRecoverTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionRecoverTest.java
index d26f1c2..0d2d64f 100644
---
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionRecoverTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionRecoverTest.java
@@ -74,7 +74,8 @@ import static
org.apache.iotdb.db.conf.IoTDBConstant.PATH_SEPARATOR;
import static
org.apache.iotdb.db.engine.compaction.utils.CompactionLogger.COMPACTION_LOG_NAME;
import static
org.apache.iotdb.db.engine.compaction.utils.CompactionLogger.SOURCE_NAME;
import static
org.apache.iotdb.db.engine.compaction.utils.CompactionLogger.TARGET_NAME;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
public class LevelCompactionRecoverTest {
@@ -188,7 +189,6 @@ public class LevelCompactionRecoverTest {
resourceFile.delete();
}
FileReaderManager.getInstance().closeAndRemoveAllOpenedReaders();
- FileReaderManager.getInstance().stop();
}
void prepareFile(TsFileResource tsFileResource, long timeOffset, long ptNum,
long valueOffset)
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionRestoreTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionRestoreTest.java
index f884448..8c42dd5 100644
---
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionRestoreTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionRestoreTest.java
@@ -183,7 +183,6 @@ public class LevelCompactionRestoreTest {
resourceFile.delete();
}
FileReaderManager.getInstance().closeAndRemoveAllOpenedReaders();
- FileReaderManager.getInstance().stop();
}
void prepareFiles(int seqFileNum, int unseqFileNum) throws IOException,
WriteProcessException {
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionTest.java
index d614144..3a8f468 100644
---
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionTest.java
@@ -206,7 +206,6 @@ abstract class LevelCompactionTest {
resourceFile.delete();
}
FileReaderManager.getInstance().closeAndRemoveAllOpenedReaders();
- FileReaderManager.getInstance().stop();
}
void prepareFile(TsFileResource tsFileResource, long timeOffset, long ptNum,
long valueOffset)
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTest.java
index 699e3a8..18970d7 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTest.java
@@ -199,7 +199,6 @@ abstract class MergeTest {
}
FileReaderManager.getInstance().closeAndRemoveAllOpenedReaders();
- FileReaderManager.getInstance().stop();
}
void prepareFile(TsFileResource tsFileResource, long timeOffset, long ptNum,
long valueOffset)
diff --git
a/server/src/test/java/org/apache/iotdb/db/query/control/FileReaderManagerTest.java
b/server/src/test/java/org/apache/iotdb/db/query/control/FileReaderManagerTest.java
index ea5b75b..663b6d8 100644
---
a/server/src/test/java/org/apache/iotdb/db/query/control/FileReaderManagerTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/query/control/FileReaderManagerTest.java
@@ -123,25 +123,14 @@ public class FileReaderManagerTest {
t1.join();
t2.join();
+ Thread.sleep(1000);
+ // Since we have closed the reader after reading the file, it should be
false that the file is
+ // still contained by manager
for (int i = 1; i <= MAX_FILE_SIZE; i++) {
TsFileResource tsFile = new
TsFileResource(SystemFileFactory.INSTANCE.getFile(filePath + i));
- Assert.assertTrue(manager.contains(tsFile, false));
+ Assert.assertFalse(manager.contains(tsFile, false));
}
- // the code below is not valid because the cacheFileReaderClearPeriod
config in this class is
- // not valid
-
- // TimeUnit.SECONDS.sleep(5);
- //
- // for (int i = 1; i <= MAX_FILE_SIZE; i++) {
- //
- // if (i == 4 || i == 5 || i == 6) {
- // Assert.assertTrue(manager.contains(filePath + i));
- // } else {
- // Assert.assertFalse(manager.contains(filePath + i));
- // }
- // }
-
FileReaderManager.getInstance().closeAndRemoveAllOpenedReaders();
for (int i = 1; i < MAX_FILE_SIZE; i++) {
File file = SystemFileFactory.INSTANCE.getFile(filePath + i);
diff --git
a/server/src/test/java/org/apache/iotdb/db/query/reader/series/SeriesReaderTestUtil.java
b/server/src/test/java/org/apache/iotdb/db/query/reader/series/SeriesReaderTestUtil.java
index f865ee3..142e309 100644
---
a/server/src/test/java/org/apache/iotdb/db/query/reader/series/SeriesReaderTestUtil.java
+++
b/server/src/test/java/org/apache/iotdb/db/query/reader/series/SeriesReaderTestUtil.java
@@ -235,6 +235,5 @@ public class SeriesReaderTestUtil {
}
FileReaderManager.getInstance().closeAndRemoveAllOpenedReaders();
- FileReaderManager.getInstance().stop();
}
}
