This is an automated email from the ASF dual-hosted git repository. xiangweiwei pushed a commit to branch to0.12closeReader in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 4698e7d135750b4391fc09943c3483c6b2a19e8b Author: Alima777 <[email protected]> AuthorDate: Fri Nov 26 22:49:35 2021 +0800 close TsFileSequenceReader immediately if ref is zero --- .../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(); } }
