This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 48233ea92dd [Fix] Isolate FileReaderManager reader I/O with per-file
locks (#18788)
48233ea92dd is described below
commit 48233ea92dde92bbc13b0466c2b4ed0173676438
Author: Jackie Tien <[email protected]>
AuthorDate: Fri Oct 9 10:23:39 2026 +0800
[Fix] Isolate FileReaderManager reader I/O with per-file locks (#18788)
---
.../iotdb/db/i18n/StorageEngineMessages.java | 6 +-
.../iotdb/db/i18n/StorageEngineMessages.java | 6 +-
.../fragment/FakedFragmentInstanceContext.java | 84 +-
.../fragment/FragmentInstanceContext.java | 181 +++--
.../dataregion/read/control/FileReaderManager.java | 698 +++++++++++------
.../dataregion/tsfile/TsFileLock.java | 27 +-
.../fragment/FileReaderReferenceLifecycleTest.java | 189 +++++
.../RegionQueryInitializationLockTest.java | 280 +++++++
.../fragment/RegionScanTimeSliceTest.java | 96 +++
.../buffer/TimeSeriesMetadataCacheTest.java | 4 +-
.../control/FileReaderManagerConcurrencyTest.java | 871 +++++++++++++++++++++
.../dataregion/tsfile/TsFileLockTest.java | 108 +++
12 files changed, 2221 insertions(+), 329 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java
index 032db3e9bb1..2a440d6d32d 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java
@@ -1495,8 +1495,10 @@ public final class StorageEngineMessages {
"Current chunk %s cannot be aligned with time chunk: %s, all time chunk
in first batch is %s";
public static final String WAL_NODE_CLOSED_FMT = "wal node-%s has been
closed";
public static final String BROKEN_WAL_FILE_FMT = "Broken wal file %s, size
%d";
- public static final String TSFILE_READER_CLOSED_BECAUSE_NO_REFERENCE =
- "{} TsFileReader is closed because of no reference.";
+ public static final String
LOG_READER_FOR_CLOSED_TSFILE_ARG_IS_CLOSED_BECAUSE_ITS_REFERENCE_COUNT_REACHED_ZERO_C3B71A85
=
+ "Reader for closed TsFile {} is closed because its reference count
reached zero.";
+ public static final String
LOG_READER_FOR_UNCLOSED_TSFILE_ARG_IS_CLOSED_BECAUSE_ITS_REFERENCE_COUNT_REACHED_ZERO_088BDEF8
=
+ "Reader for unclosed TsFile {} is closed because its reference count
reached zero.";
public static final String CLOSED_TSFILE_READER_CLOSED =
"{} closedTsFileReader is closed.";
public static final String UNCLOSED_TSFILE_READER_CLOSED =
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java
index 43ebfbd35ab..b55c248397e 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java
@@ -1424,8 +1424,10 @@ public final class StorageEngineMessages {
"当前 chunk %s 无法与 time chunk：%s 对齐，第一批中的所有 time chunk 为 %s";
public static final String WAL_NODE_CLOSED_FMT = "wal node-%s 已关闭";
public static final String BROKEN_WAL_FILE_FMT = "WAL 文件 %s 损坏，大小为 %d";
- public static final String TSFILE_READER_CLOSED_BECAUSE_NO_REFERENCE =
- "{} TsFileReader 因没有引用已关闭。";
+ public static final String
LOG_READER_FOR_CLOSED_TSFILE_ARG_IS_CLOSED_BECAUSE_ITS_REFERENCE_COUNT_REACHED_ZERO_C3B71A85
=
+ "已封口 TsFile {} 的 reader 因引用计数归零而关闭。";
+ public static final String
LOG_READER_FOR_UNCLOSED_TSFILE_ARG_IS_CLOSED_BECAUSE_ITS_REFERENCE_COUNT_REACHED_ZERO_088BDEF8
=
+ "未封口 TsFile {} 的 reader 因引用计数归零而关闭。";
public static final String CLOSED_TSFILE_READER_CLOSED =
"{} closedTsFileReader 已关闭。";
public static final String UNCLOSED_TSFILE_READER_CLOSED =
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FakedFragmentInstanceContext.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FakedFragmentInstanceContext.java
index 99e95dbf26b..c2f29ed1ab8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FakedFragmentInstanceContext.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FakedFragmentInstanceContext.java
@@ -24,16 +24,21 @@ import org.apache.iotdb.commons.path.IFullPath;
import
org.apache.iotdb.db.queryengine.plan.planner.memory.FakedMemoryReservationManager;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
-import
org.apache.iotdb.db.storageengine.dataregion.read.control.FileReaderManager;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.tsfile.read.filter.basic.Filter;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
public class FakedFragmentInstanceContext extends FragmentInstanceContext {
+ // Retain the registration-time state: an unclosed file may be sealed before
the query finishes.
+ private final Set<TsFileResource> closedResources = new HashSet<>();
+ private final Set<TsFileResource> unclosedResources = new HashSet<>();
+
public FakedFragmentInstanceContext(Filter timeFilter, DataRegion
dataRegion) {
super(0, new FakedMemoryReservationManager(), timeFilter, dataRegion,
false, false);
}
@@ -50,52 +55,55 @@ public class FakedFragmentInstanceContext extends
FragmentInstanceContext {
dataRegion.tryReadLock(Long.MAX_VALUE);
try {
- this.sharedQueryDataSource =
- dataRegion.query(
- Collections.singletonList(sourcePath),
- sourcePath.getDeviceId(),
- this,
- getGlobalTimeFilter(),
- null,
- Long.MAX_VALUE);
+ try {
+ this.sharedQueryDataSource =
+ dataRegion.query(
+ Collections.singletonList(sourcePath),
+ sourcePath.getDeviceId(),
+ this,
+ getGlobalTimeFilter(),
+ null,
+ Long.MAX_VALUE);
- // used files should be added before mergeLock is unlocked, or they may
be deleted by
- // running merge
- if (sharedQueryDataSource != null) {
- ((QueryDataSource) sharedQueryDataSource).setSingleDevice(true);
- List<TsFileResource> tsFileList =
- ((QueryDataSource) sharedQueryDataSource).getSeqResources();
- if (tsFileList != null) {
- for (TsFileResource tsFile : tsFileList) {
-
FileReaderManager.getInstance().increaseFileReaderReference(tsFile,
tsFile.isClosed());
+ // used files should be added before mergeLock is unlocked, or they
may be deleted by
+ // running merge
+ if (sharedQueryDataSource != null) {
+ ((QueryDataSource) sharedQueryDataSource).setSingleDevice(true);
+ List<TsFileResource> tsFileList =
+ ((QueryDataSource) sharedQueryDataSource).getSeqResources();
+ if (tsFileList != null) {
+ for (TsFileResource tsFile : tsFileList) {
+ addFilePathToMap(tsFile, tsFile.isClosed(), closedResources,
unclosedResources);
+ }
}
- }
- tsFileList = ((QueryDataSource)
sharedQueryDataSource).getUnseqResources();
- if (tsFileList != null) {
- for (TsFileResource tsFile : tsFileList) {
-
FileReaderManager.getInstance().increaseFileReaderReference(tsFile,
tsFile.isClosed());
+ tsFileList = ((QueryDataSource)
sharedQueryDataSource).getUnseqResources();
+ if (tsFileList != null) {
+ for (TsFileResource tsFile : tsFileList) {
+ addFilePathToMap(tsFile, tsFile.isClosed(), closedResources,
unclosedResources);
+ }
}
}
+ } finally {
+ dataRegion.readUnlock();
}
- } finally {
- dataRegion.readUnlock();
+ } catch (RuntimeException | Error e) {
+ // Rollback can close readers or wait for another close; never retain
the region lock here.
+ try {
+ releaseSharedQueryDataSource();
+ } catch (RuntimeException | Error rollbackFailure) {
+ if (rollbackFailure != e) {
+ e.addSuppressed(rollbackFailure);
+ }
+ }
+ throw e;
}
}
public void releaseSharedQueryDataSource() {
- if (sharedQueryDataSource != null) {
- List<TsFileResource> tsFileList = ((QueryDataSource)
sharedQueryDataSource).getSeqResources();
- if (tsFileList != null) {
- for (TsFileResource tsFile : tsFileList) {
- FileReaderManager.getInstance().decreaseFileReaderReference(tsFile,
tsFile.isClosed());
- }
- }
- tsFileList = ((QueryDataSource)
sharedQueryDataSource).getUnseqResources();
- if (tsFileList != null) {
- for (TsFileResource tsFile : tsFileList) {
- FileReaderManager.getInstance().decreaseFileReaderReference(tsFile,
tsFile.isClosed());
- }
- }
+ try (DeferredFileReaderReleases releases = new
DeferredFileReaderReleases()) {
+ releases.addAll(closedResources, true);
+ releases.addAll(unclosedResources, false);
+ } finally {
sharedQueryDataSource = null;
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
index c76b15b0f0f..e46603cdf79 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
@@ -76,6 +76,7 @@ import java.time.format.DateTimeParseException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -87,6 +88,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
@@ -774,16 +776,16 @@ public class FragmentInstanceContext extends QueryContext
{
public boolean initQueryDataSource(List<IFullPath> sourcePaths) throws
QueryProcessException {
return initializeQueryDataSource(
sourcePaths,
- () -> {
+ deferredReleases -> {
sharedQueryDataSource = EMPTY_QUERY_DATA_SOURCE;
return true;
},
() -> false,
- dataSource -> {
+ (dataSource, deferredReleases) -> {
sharedQueryDataSource = dataSource;
closedFilePaths = new HashSet<>();
unClosedFilePaths = new HashSet<>();
- addUsedFilesForQuery(dataSource, closedFilePaths, unClosedFilePaths);
+ addUsedFilesForQuery(dataSource, closedFilePaths, unClosedFilePaths,
deferredReleases);
return true;
});
}
@@ -792,7 +794,8 @@ public class FragmentInstanceContext extends QueryContext {
throws QueryProcessException {
return initializeQueryDataSource(
sourcePaths,
- () -> createBatchQueryDataSourceLease(EMPTY_QUERY_DATA_SOURCE),
+ deferredReleases ->
+ createBatchQueryDataSourceLease(EMPTY_QUERY_DATA_SOURCE,
deferredReleases),
() -> {
recordQueryDataSourceRetry();
return null;
@@ -802,14 +805,14 @@ public class FragmentInstanceContext extends QueryContext
{
private <T> T initializeQueryDataSource(
List<IFullPath> sourcePaths,
- Supplier<T> emptyResultSupplier,
+ Function<DeferredFileReaderReleases, T> emptyResultSupplier,
Supplier<T> unfinishedResultSupplier,
- Function<QueryDataSource, T> initializer)
+ BiFunction<QueryDataSource, DeferredFileReaderReleases, T> initializer)
throws QueryProcessException {
long startTime = System.nanoTime();
- try {
+ try (DeferredFileReaderReleases deferredReleases = new
DeferredFileReaderReleases()) {
if (sourcePaths == null || sourcePaths.isEmpty()) {
- return emptyResultSupplier.get();
+ return emptyResultSupplier.apply(deferredReleases);
}
IDeviceID singleDeviceId = findSingleDeviceId(sourcePaths);
@@ -843,7 +846,7 @@ public class FragmentInstanceContext extends QueryContext {
dataSource.setSingleDevice(singleDeviceId != null);
// The initializer retains referenced files while the DataRegion read
lock is held;
// otherwise a concurrent merge may delete them after the lock is
released.
- return initializer.apply(dataSource);
+ return initializer.apply(dataSource, deferredReleases);
} finally {
dataRegion.readUnlock();
}
@@ -869,14 +872,23 @@ public class FragmentInstanceContext extends QueryContext
{
return selectedDeviceIdSet.iterator().next();
}
- private QueryDataSourceLease createBatchQueryDataSourceLease(QueryDataSource
dataSource) {
+ private QueryDataSourceLease createBatchQueryDataSourceLease(
+ QueryDataSource dataSource, DeferredFileReaderReleases deferredReleases)
{
Set<TsFileResource> closedResources = new HashSet<>();
Set<TsFileResource> unclosedResources = new HashSet<>();
- addUsedFilesForQuery(dataSource, closedResources, unclosedResources);
- QueryDataSourceLease lease =
- new QueryDataSourceLease(dataSource, closedResources,
unclosedResources, this);
- batchQueryDataSourceLeases.add(lease);
- return lease;
+ try {
+ addUsedFilesForQuery(dataSource, closedResources, unclosedResources,
deferredReleases);
+ QueryDataSourceLease lease =
+ new QueryDataSourceLease(dataSource, closedResources,
unclosedResources, this);
+ batchQueryDataSourceLeases.add(lease);
+ return lease;
+ } catch (RuntimeException | Error e) {
+ // No lease owns a partially initialized batch. Release its retained
files after region
+ // unlock.
+ deferredReleases.addAll(closedResources, true);
+ deferredReleases.addAll(unclosedResources, false);
+ throw e;
+ }
}
public boolean initRegionScanQueryDataSource(Map<IDeviceID, DeviceContext>
devicePathsToContext) {
@@ -885,8 +897,11 @@ public class FragmentInstanceContext extends QueryContext {
return true;
}
- long waitForLockTime = COMMON_CONFIG.getDriverTaskExecutionTimeSliceInMs();
- if (dataRegion.tryReadLock(waitForLockTime)) {
+ try (DeferredFileReaderReleases deferredReleases = new
DeferredFileReaderReleases()) {
+ long waitForLockTime =
COMMON_CONFIG.getDriverTaskExecutionTimeSliceInMs();
+ if (!dataRegion.tryReadLock(waitForLockTime)) {
+ return false;
+ }
try {
// minus already consumed time
waitForLockTime -= (System.nanoTime() - startTime) / 1_000_000;
@@ -906,19 +921,18 @@ public class FragmentInstanceContext extends QueryContext
{
if (sharedQueryDataSource != null) {
closedFilePaths = new HashSet<>();
unClosedFilePaths = new HashSet<>();
- addUsedFilesForRegionQuery((QueryDataSourceForRegionScan)
sharedQueryDataSource);
+ addUsedFilesForRegionQuery(
+ (QueryDataSourceForRegionScan) sharedQueryDataSource,
deferredReleases);
return true;
} else {
// failed to acquire lock within the specific time
return false;
}
} finally {
- addInitQueryDataSourceCost(System.nanoTime() - startTime);
dataRegion.readUnlock();
}
- } else {
+ } finally {
addInitQueryDataSourceCost(System.nanoTime() - startTime);
- return false;
}
}
@@ -927,16 +941,20 @@ public class FragmentInstanceContext extends QueryContext
{
if (pathList == null) {
return true;
}
- long waitForLockTime = COMMON_CONFIG.getDriverTaskExecutionTimeSliceInMs();
- if (dataRegion.tryReadLock(waitForLockTime)) {
- // minus already consumed time
- waitForLockTime -= (System.nanoTime() - startTime) / 1_000_000;
- // no remaining time slice
- if (waitForLockTime <= 0) {
+ try (DeferredFileReaderReleases deferredReleases = new
DeferredFileReaderReleases()) {
+ long waitForLockTime =
COMMON_CONFIG.getDriverTaskExecutionTimeSliceInMs();
+ if (!dataRegion.tryReadLock(waitForLockTime)) {
return false;
}
try {
+ // minus already consumed time
+ waitForLockTime -= (System.nanoTime() - startTime) / 1_000_000;
+
+ // no remaining time slice
+ if (waitForLockTime <= 0) {
+ return false;
+ }
this.sharedQueryDataSource =
dataRegion.queryForSeriesRegionScan(
pathList,
@@ -948,19 +966,18 @@ public class FragmentInstanceContext extends QueryContext
{
if (sharedQueryDataSource != null) {
closedFilePaths = new HashSet<>();
unClosedFilePaths = new HashSet<>();
- addUsedFilesForRegionQuery((QueryDataSourceForRegionScan)
sharedQueryDataSource);
+ addUsedFilesForRegionQuery(
+ (QueryDataSourceForRegionScan) sharedQueryDataSource,
deferredReleases);
return true;
} else {
// failed to acquire lock within the specific time
return false;
}
} finally {
- addInitQueryDataSourceCost(System.nanoTime() - startTime);
dataRegion.readUnlock();
}
- } else {
+ } finally {
addInitQueryDataSourceCost(System.nanoTime() - startTime);
- return false;
}
}
@@ -1021,24 +1038,78 @@ public class FragmentInstanceContext extends
QueryContext {
}
}
+ /** Owns pending file references until they can be released outside region
locks. */
+ static final class DeferredFileReaderReleases implements AutoCloseable {
+ private final List<Pair<TsFileResource, Boolean>> references = new
ArrayList<>();
+
+ private void add(TsFileResource resource, boolean isClosed,
Set<TsFileResource> owners) {
+ if (!owners.contains(resource)) {
+ return;
+ }
+ // Allocate the pending record before removing its old owner, including
on allocation failure.
+ references.add(new Pair<>(resource, isClosed));
+ if (!owners.remove(resource)) {
+ references.remove(references.size() - 1);
+ }
+ }
+
+ void addAll(Set<TsFileResource> owners, boolean isClosed) {
+ Iterator<TsFileResource> iterator = owners.iterator();
+ while (iterator.hasNext()) {
+ references.add(new Pair<>(iterator.next(), isClosed));
+ iterator.remove();
+ }
+ }
+
+ @Override
+ public void close() {
+ Throwable failure = null;
+ try {
+ for (Pair<TsFileResource, Boolean> reference : references) {
+ try {
+ FileReaderManager.getInstance()
+ .decreaseFileReaderReference(reference.left, reference.right);
+ } catch (RuntimeException | Error e) {
+ if (failure == null) {
+ failure = e;
+ } else if (failure != e) {
+ failure.addSuppressed(e);
+ }
+ }
+ }
+ } finally {
+ references.clear();
+ }
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ if (failure != null) {
+ throw (RuntimeException) failure;
+ }
+ }
+ }
+
/** Lock and check if tsFileResource is deleted */
- private boolean processTsFileResource(TsFileResource tsFileResource, boolean
isClosed) {
- return processTsFileResource(tsFileResource, isClosed, closedFilePaths,
unClosedFilePaths);
+ private boolean processTsFileResource(
+ TsFileResource tsFileResource,
+ boolean isClosed,
+ DeferredFileReaderReleases deferredReleases) {
+ return processTsFileResource(
+ tsFileResource, isClosed, closedFilePaths, unClosedFilePaths,
deferredReleases);
}
private boolean processTsFileResource(
TsFileResource tsFileResource,
boolean isClosed,
Set<TsFileResource> closedResources,
- Set<TsFileResource> unclosedResources) {
+ Set<TsFileResource> unclosedResources,
+ DeferredFileReaderReleases deferredReleases) {
addFilePathToMap(tsFileResource, isClosed, closedResources,
unclosedResources);
// this file may be deleted just before we lock it
if (tsFileResource.isDeleted()) {
Set<TsFileResource> pathSet = isClosed ? closedResources :
unclosedResources;
- // This resource may be removed by other threads of this query.
- if (pathSet.remove(tsFileResource)) {
-
FileReaderManager.getInstance().decreaseFileReaderReference(tsFileResource,
isClosed);
- }
+ // Keep the file reference until region unlock; the last release may
wait for reader I/O.
+ deferredReleases.add(tsFileResource, isClosed, pathSet);
return true;
} else {
return false;
@@ -1049,7 +1120,8 @@ public class FragmentInstanceContext extends QueryContext
{
private void addUsedFilesForQuery(
QueryDataSource dataSource,
Set<TsFileResource> closedResources,
- Set<TsFileResource> unclosedResources) {
+ Set<TsFileResource> unclosedResources,
+ DeferredFileReaderReleases deferredReleases) {
// sequence data
dataSource
@@ -1057,7 +1129,11 @@ public class FragmentInstanceContext extends
QueryContext {
.removeIf(
tsFileResource ->
processTsFileResource(
- tsFileResource, tsFileResource.isClosed(),
closedResources, unclosedResources));
+ tsFileResource,
+ tsFileResource.isClosed(),
+ closedResources,
+ unclosedResources,
+ deferredReleases));
// Record statistics of seqFiles
int unclosedSeqFileCount = unclosedResources.size();
@@ -1071,19 +1147,25 @@ public class FragmentInstanceContext extends
QueryContext {
.removeIf(
tsFileResource ->
processTsFileResource(
- tsFileResource, tsFileResource.isClosed(),
closedResources, unclosedResources));
+ tsFileResource,
+ tsFileResource.isClosed(),
+ closedResources,
+ unclosedResources,
+ deferredReleases));
// Record statistics of files of unseqFiles
unclosedUnseqFileNum.addAndGet(unclosedResources.size() -
unclosedSeqFileCount);
closedUnseqFileNum.addAndGet(closedResources.size() - closedSeqFileCount);
}
- private void addUsedFilesForRegionQuery(QueryDataSourceForRegionScan
dataSource) {
+ private void addUsedFilesForRegionQuery(
+ QueryDataSourceForRegionScan dataSource, DeferredFileReaderReleases
deferredReleases) {
dataSource
.getSeqFileScanHandles()
.removeIf(
fileScanHandle ->
- processTsFileResource(fileScanHandle.getTsResource(),
fileScanHandle.isClosed()));
+ processTsFileResource(
+ fileScanHandle.getTsResource(), fileScanHandle.isClosed(),
deferredReleases));
int unclosedSeqFileCount = unClosedFilePaths.size();
int closedSeqFileCount = closedFilePaths.size();
@@ -1094,7 +1176,8 @@ public class FragmentInstanceContext extends QueryContext
{
.getUnseqFileScanHandles()
.removeIf(
fileScanHandle ->
- processTsFileResource(fileScanHandle.getTsResource(),
fileScanHandle.isClosed()));
+ processTsFileResource(
+ fileScanHandle.getTsResource(), fileScanHandle.isClosed(),
deferredReleases));
unclosedUnseqFileNum.addAndGet(unClosedFilePaths.size() -
unclosedSeqFileCount);
closedUnseqFileNum.addAndGet(closedFilePaths.size() - closedSeqFileCount);
@@ -1110,14 +1193,20 @@ public class FragmentInstanceContext extends
QueryContext {
addFilePathToMap(tsFile, isClosed, closedFilePaths, unClosedFilePaths);
}
- private void addFilePathToMap(
+ static void addFilePathToMap(
TsFileResource tsFile,
boolean isClosed,
Set<TsFileResource> closedResources,
Set<TsFileResource> unclosedResources) {
Set<TsFileResource> pathSet = isClosed ? closedResources :
unclosedResources;
if (pathSet.add(tsFile)) {
- FileReaderManager.getInstance().increaseFileReaderReference(tsFile,
isClosed);
+ try {
+ FileReaderManager.getInstance().increaseFileReaderReference(tsFile,
isClosed);
+ } catch (RuntimeException | Error e) {
+ // Registration already rolled back its read lock. Do not release it a
second time later.
+ pathSet.remove(tsFile);
+ throw e;
+ }
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/control/FileReaderManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/control/FileReaderManager.java
index 429859f594f..a0b9d0f169f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/control/FileReaderManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/control/FileReaderManager.java
@@ -31,347 +31,589 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.util.Iterator;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.LongConsumer;
/**
- * {@link 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.
+ * Caches readers for managed TsFiles and external file paths, with reader I/O
serialized per file.
+ *
+ * <p>Each {@link Entry} is a stable lifecycle record and the monitor for its
reader slots. An
+ * operation pin keeps the entry registered while a manager call uses it or
waits for its monitor.
+ * Pins do not keep a returned reader open: callers retain a {@link
Slot#references file-reader
+ * reference} for the duration of reader use, separately from each manager
call's pin.
+ *
+ * <p>{@link #registryLock} protects registry membership, pins, and reference
counts. Entry monitors
+ * protect reader construction, close, and the cached reader fields, not reads
performed by callers
+ * after {@code get} returns. Never hold {@code registryLock} while acquiring
an entry monitor or
+ * doing I/O. An entry may briefly acquire {@code registryLock} to recheck
references, but must not
+ * acquire a resource lock, region lock, or another entry monitor. Reference
registration uses only
+ * the short registry critical section, so callers can register while holding
a region read lock.
*/
public class FileReaderManager {
-
private static final Logger logger =
LoggerFactory.getLogger(FileReaderManager.class);
private static final Logger resourceLogger =
LoggerFactory.getLogger("FileMonitor");
- private static final Logger DEBUG_LOGGER =
LoggerFactory.getLogger("QUERY_DEBUG");
-
- /** max number of file streams being cached, must be lower than 65535. */
private static final int MAX_CACHED_FILE_SIZE = 30000;
+ private static final int PRINT_INTERVAL = 10000;
/**
- * When number of file streams reached MAX_CACHED_FILE_SIZE, then we will
print a warning log each
- * PRINT_INTERVAL.
+ * Manager-wide monitor for both registries, {@link #clearing}, both levels
of pin counters, and
+ * {@link Slot#references}. It is never held across reader I/O or an
entry-monitor acquisition.
*/
- private static final int PRINT_INTERVAL = 10000;
+ private final Object registryLock = new Object();
+
+ /** Serializes entire global test-cleanup passes; ordinary reader operations
do not acquire it. */
+ private final Object clearLock = new Object();
/**
- * the key of closedFileReaderMap is the file path and the value of
closedFileReaderMap is the
- * corresponding reader.
+ * Managed files keyed by TsFileID, each with independent closed/unclosed
slots under one entry
+ * monitor. Guarded by {@link #registryLock}.
*/
- private Map<TsFileID, TsFileSequenceReader> closedFileReaderMap;
+ private final Map<TsFileID, Entry> internal = new HashMap<>();
/**
- * the key of unclosedFileReaderMap is the file path and the value of
unclosedFileReaderMap is the
- * corresponding reader.
+ * External files keyed by the supplied path string, in a namespace separate
from {@link
+ * #internal}. External entries use only their closed slot. Guarded by
{@link #registryLock}.
*/
- private Map<TsFileID, TsFileSequenceReader> unclosedFileReaderMap;
+ private final Map<String, Entry> external = new HashMap<>();
+
+ /** Cached closed-reader count for managed files; updated concurrently under
different entries. */
+ private final AtomicInteger closedCount = new AtomicInteger();
+
+ /** Cached unclosed-reader count for managed files; independent of query
references and pins. */
+ private final AtomicInteger unclosedCount = new AtomicInteger();
+
+ /** Cached-reader count for the external-path namespace. */
+ private final AtomicInteger externalCount = new AtomicInteger();
/**
- * the key of closedFileReaderMap is the file path and the value of
closedFileReaderMap is the
- * file's reference count.
+ * Test-cleanup admission gate: new acquisitions wait while releases remain
allowed. Guarded by
+ * {@link #registryLock}; it remains set through the final pin drain and
entry reclamation.
*/
- private Map<TsFileID, AtomicInteger> closedReferenceMap;
+ private boolean clearing;
/**
- * the key of unclosedFileReaderMap is the file path and the value of
unclosedFileReaderMap is the
- * file's reference count.
+ * Sum of {@link Entry#pins} across both registries, including operations
waiting for entry
+ * monitors. Global test cleanup drains this counter; zero does not imply
zero query references.
+ * Guarded by {@link #registryLock}.
*/
- private Map<TsFileID, AtomicInteger> unclosedReferenceMap;
-
- /** External TsFile readers. The key is the file path. */
- private Map<String, TsFileSequenceReader> externalFileReaderMap;
+ private int pins;
- /** Reference count of external TsFile readers. */
- private Map<String, AtomicInteger> externalReferenceMap;
-
- private FileReaderManager() {
- closedFileReaderMap = new ConcurrentHashMap<>();
- unclosedFileReaderMap = new ConcurrentHashMap<>();
- closedReferenceMap = new ConcurrentHashMap<>();
- unclosedReferenceMap = new ConcurrentHashMap<>();
- externalFileReaderMap = new ConcurrentHashMap<>();
- externalReferenceMap = new ConcurrentHashMap<>();
- }
+ private FileReaderManager() {}
public static FileReaderManager getInstance() {
return FileReaderManagerHelper.INSTANCE;
}
- public synchronized void closeFileAndRemoveReader(TsFileID tsFileID) throws
IOException {
- closedReferenceMap.remove(tsFileID);
- TsFileSequenceReader reader = closedFileReaderMap.remove(tsFileID);
- if (reader != null) {
- reader.close();
+ /** One reader state (closed or unclosed) and its independently maintained
query references. */
+ private static class Slot {
+ /**
+ * Nullable cached handle, created and discarded under the owning entry's
monitor. Reclamation
+ * may inspect it under registryLock after that entry's operations have
finished.
+ */
+ private TsFileSequenceReader reader;
+
+ /**
+ * Outstanding file-reader retains for this slot, guarded by registryLock.
May be positive
+ * before any reader is opened; unlike operation pins, these retains span
the caller's reader
+ * use.
+ */
+ private int references;
+
+ /**
+ * Shared cached-reader counter for this slot's namespace/state.
Publishing a reader increments
+ * it; discarding a handle after a close attempt decrements it, even when
close fails. This
+ * counts cached handles, not query references or all open OS descriptors.
+ */
+ private final AtomicInteger count;
+
+ private Slot(AtomicInteger count) {
+ this.count = count;
+ }
+
+ private void close() throws IOException {
+ if (reader != null) {
+ try {
+ reader.close();
+ } catch (IOException e) {
+ logger.error(
+ StorageEngineMessages.CANNOT_CLOSE_TSFILE_SEQUENCE_READER,
reader.getFileName(), e);
+ throw e;
+ } finally {
+ // A failed close must neither poison future reads nor retain an
unused entry forever.
+ // As with the original manager, discard the handle after a
best-effort close. Pins keep
+ // this entry stable until the attempt finishes, including when
close throws unchecked.
+ reader = null;
+ count.decrementAndGet();
+ }
+ }
+ }
+
+ private boolean empty() {
+ return reader == null && references == 0;
+ }
+ }
+
+ /**
+ * Stable per-key record whose monitor serializes reader lifecycle changes.
It cannot be reclaimed
+ * while an operation is pinned, even if both slots are empty, because a
waiter may still use this
+ * exact monitor. Replacing it early would allow two entries to open readers
for the same key.
+ */
+ private static class Entry {
+ /** Closed-file reader slot; also the sole slot used by external entries.
*/
+ private final Slot closed;
+
+ /** Unclosed-file reader slot for managed files; unused for external
entries. */
+ private final Slot unclosed;
+
+ /**
+ * Manager operations retaining this entry, including monitor waiters;
guarded by registryLock.
+ */
+ private int pins;
+
+ /**
+ * @param closedCount managed closed-reader counter or external-reader
counter, depending on the
+ * registry that owns this entry
+ * @param unclosedCount managed unclosed-reader counter; external entries
never use this slot
+ */
+ private Entry(AtomicInteger closedCount, AtomicInteger unclosedCount) {
+ closed = new Slot(closedCount);
+ unclosed = new Slot(unclosedCount);
+ }
+
+ private Slot slot(boolean isClosed) {
+ return isClosed ? closed : unclosed;
}
- unclosedReferenceMap.remove(tsFileID);
- reader = unclosedFileReaderMap.remove(tsFileID);
- if (reader != null) {
- reader.close();
+
+ private boolean empty() {
+ return closed.empty() && unclosed.empty();
}
}
/**
- * 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 .
- * Otherwise a new reader will be created and cached.
+ * Looks up an entry and retains its identity for one manager operation,
optionally creating it.
+ *
+ * <p>The operation is counted before the caller can wait for the entry
monitor, so reclamation
+ * cannot replace an entry that a waiter still holds. This method neither
acquires that monitor,
+ * opens a reader, nor registers a query reference. Every non-null result
must be paired with
+ * exactly one {@code unpin} in a {@code finally} block, after leaving the
entry monitor.
*
- * @param filePath the path of the tsfile
- * @param tsFileID the id of the tsfile, of which the reader is desired.
- * @param isClosed whether the corresponding file still receives insertions
or not.
- * @return the reader of the file specified by filePath.
- * @throws IOException when reader cannot be created.
+ * <p>Admission waits are uninterruptible: an interrupt is remembered and
restored before this
+ * method returns. Releases bypass the gate so they can complete during
global test cleanup.
+ *
+ * @param <K> TsFileID for the internal registry, or String for the external
registry
+ * @param registry exactly {@link #internal} or {@link #external}; also
selects the cached-reader
+ * counter for a newly created entry
+ * @param key non-null file identity within the selected registry
+ * @param admission true to wait until global test cleanup reopens
admission; false for reference
+ * releases that must remain allowed during cleanup
+ * @param create true to install an empty entry if the key is absent; false
to return null instead
+ * @return the entry with both its own and the global operation-pin count
incremented, or null if
+ * the key is absent and creation is disabled
*/
- @SuppressWarnings("squid:S2095")
- public synchronized TsFileSequenceReader get(String filePath, TsFileID
tsFileID, boolean isClosed)
- throws IOException {
- return get(filePath, tsFileID, isClosed, null);
+ private <K> Entry pin(Map<K, Entry> registry, K key, boolean admission,
boolean create) {
+ Objects.requireNonNull(key);
+ boolean interrupted = false;
+ synchronized (registryLock) {
+ while (admission && clearing) {
+ try {
+ registryLock.wait();
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ Entry entry = registry.get(key);
+ if (entry == null && create) {
+ entry = new Entry(registry == internal ? closedCount : externalCount,
unclosedCount);
+ registry.put(key, entry);
+ }
+ if (entry != null) {
+ entry.pins++;
+ pins++;
+ }
+ return entry;
+ }
}
/**
- * 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 .
- * Otherwise a new reader will be created and cached.
+ * Releases the matching operation pin without changing query references or
closing a reader.
+ * Outside global cleanup, an empty entry is reclaimed only when no
operation still retains it.
+ * Call after leaving the entry monitor, including when the operation fails.
*
- * @param filePath the path of the tsfile
- * @param tsFileID the id of the tsfile, of which the reader is desired.
- * @param isClosed whether the corresponding file still receives insertions
or not.
- * @param ioSizeRecorder can be null
- * @return the reader of the file specified by filePath.
- * @throws IOException when reader cannot be created.
+ * @param <K> key type of the selected registry
+ * @param registry the same registry supplied to the matching {@code pin}
+ * @param key the same file identity supplied to the matching {@code pin}
+ * @param entry the exact non-null entry returned by that call, not a fresh
registry lookup
*/
- @SuppressWarnings("squid:S2095")
- public synchronized TsFileSequenceReader get(
+ private <K> void unpin(Map<K, Entry> registry, K key, Entry entry) {
+ synchronized (registryLock) {
+ entry.pins--;
+ pins--;
+ // At zero pins all slot mutations have completed and published through
registryLock.
+ // Keep entries stable during clear, including those used by releases
entering mid-clear.
+ if (!clearing && entry.pins == 0 && entry.empty()) {
+ registry.remove(key, entry);
+ }
+ if (clearing && pins == 0) {
+ registryLock.notifyAll();
+ }
+ }
+ }
+
+ /** Caller must exclude new references and finish existing reader use
(resource write lock). */
+ public void closeFileAndRemoveReader(TsFileID tsFileID) throws IOException {
+ Entry entry = pin(internal, tsFileID, true, false);
+ if (entry == null) {
+ return;
+ }
+ try {
+ synchronized (entry) {
+ closeEntry(entry);
+ }
+ } finally {
+ unpin(internal, tsFileID, entry);
+ }
+ }
+
+ private void closeEntry(Entry entry) throws IOException {
+ synchronized (registryLock) {
+ entry.closed.references = 0;
+ entry.unclosed.references = 0;
+ }
+ IOException failure = null;
+ try {
+ try {
+ closeSlot(entry.closed,
StorageEngineMessages.CLOSED_TSFILE_READER_CLOSED);
+ } catch (IOException e) {
+ failure = mergeFailure(failure, e);
+ }
+ } finally {
+ // Also attempt the second slot if the first close throws an unchecked
exception.
+ try {
+ closeSlot(entry.unclosed,
StorageEngineMessages.UNCLOSED_TSFILE_READER_CLOSED);
+ } catch (IOException e) {
+ failure = mergeFailure(failure, e);
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ private static void closeSlot(Slot slot, String message) throws IOException {
+ TsFileSequenceReader reader = slot.reader;
+ slot.close();
+ if (reader != null && resourceLogger.isDebugEnabled()) {
+ resourceLogger.debug(message, reader.getFileName());
+ }
+ }
+
+ private static IOException mergeFailure(IOException failure, IOException
next) {
+ if (failure == null) {
+ return new IOException(next);
+ }
+ // Never attach other files' errors to an exception supplied by a reader.
+ failure.addSuppressed(next);
+ return failure;
+ }
+
+ public TsFileSequenceReader get(String filePath, TsFileID tsFileID, boolean
isClosed)
+ throws IOException {
+ return get(filePath, tsFileID, isClosed, null);
+ }
+
+ public TsFileSequenceReader get(
String filePath, TsFileID tsFileID, boolean isClosed, LongConsumer
ioSizeRecorder)
throws IOException {
return get(filePath, tsFileID, isClosed, ioSizeRecorder, false);
}
- @SuppressWarnings("squid:S2095")
- public synchronized TsFileSequenceReader get(
+ /** Get does not acquire a query reference. A failed constructor leaves
registered refs intact. */
+ public TsFileSequenceReader get(
String filePath,
TsFileID tsFileID,
boolean isClosed,
LongConsumer ioSizeRecorder,
boolean isExternalTsFile)
throws IOException {
- if (isExternalTsFile) {
- return getExternalTsFileReader(filePath, ioSizeRecorder);
- }
-
- Map<TsFileID, TsFileSequenceReader> readerMap =
- !isClosed ? unclosedFileReaderMap : closedFileReaderMap;
- if (!readerMap.containsKey(tsFileID)) {
- int currentOpenedReaderCount = readerMap.size();
- if (currentOpenedReaderCount >= MAX_CACHED_FILE_SIZE
- && (currentOpenedReaderCount % PRINT_INTERVAL == 0)) {
- logger.warn(StorageEngineMessages.QUERY_OPENED_FILES,
readerMap.size());
- }
-
- TsFileSequenceReader tsFileReader = null;
- // check if the file is old version
- if (!isClosed) {
- tsFileReader =
- new UnClosedTsFileReader(
- filePath,
- EncryptDBUtils.getFirstEncryptParamFromTSFilePath(filePath),
- ioSizeRecorder);
- } else {
- // already do the version check in TsFileSequenceReader's constructor
- tsFileReader =
- new TsFileSequenceReader(
- filePath,
- ioSizeRecorder,
- EncryptDBUtils.getFirstEncryptParamFromTSFilePath(filePath));
- }
- readerMap.put(tsFileID, tsFileReader);
- return tsFileReader;
- }
-
- return readerMap.get(tsFileID);
+ return isExternalTsFile
+ ? get(external, filePath, filePath, true, ioSizeRecorder)
+ : get(internal, tsFileID, filePath, isClosed, ioSizeRecorder);
}
- private TsFileSequenceReader getExternalTsFileReader(String filePath,
LongConsumer ioSizeRecorder)
+ @SuppressWarnings("squid:S2095")
+ private <K> TsFileSequenceReader get(
+ Map<K, Entry> registry, K key, String path, boolean isClosed,
LongConsumer recorder)
throws IOException {
- TsFileSequenceReader reader = externalFileReaderMap.get(filePath);
- if (reader == null) {
- int currentOpenedReaderCount = externalFileReaderMap.size();
- if (currentOpenedReaderCount >= MAX_CACHED_FILE_SIZE
- && (currentOpenedReaderCount % PRINT_INTERVAL == 0)) {
- logger.warn(StorageEngineMessages.QUERY_OPENED_FILES,
externalFileReaderMap.size());
+ Entry entry = pin(registry, key, true, true);
+ try {
+ synchronized (entry) {
+ Slot slot = entry.slot(isClosed);
+ if (slot.reader == null) {
+ slot.reader =
+ isClosed
+ ? new TsFileSequenceReader(
+ path, recorder,
EncryptDBUtils.getFirstEncryptParamFromTSFilePath(path))
+ : new UnClosedTsFileReader(
+ path,
EncryptDBUtils.getFirstEncryptParamFromTSFilePath(path), recorder);
+ int count = slot.count.incrementAndGet();
+ if (count >= MAX_CACHED_FILE_SIZE && count % PRINT_INTERVAL == 0) {
+ logger.warn(StorageEngineMessages.QUERY_OPENED_FILES, count);
+ }
+ }
+ return slot.reader;
}
- reader =
- new TsFileSequenceReader(
- filePath,
- ioSizeRecorder,
- EncryptDBUtils.getFirstEncryptParamFromTSFilePath(filePath));
- externalFileReaderMap.put(filePath, reader);
+ } finally {
+ unpin(registry, key, entry);
}
- return reader;
}
/**
- * Increase the reference count of the reader specified by filePath. Only
when the reference count
- * of a reader equals zero, the reader can be closed and removed.
+ * Retains one reader reference and the resource read lock until the
matching release.
+ *
+ * @param tsFile resource whose reader will be used
+ * @param isClosed slot to retain; pass this same value when releasing, even
if the file is sealed
+ * in the meantime
*/
public void increaseFileReaderReference(TsFileResource tsFile, boolean
isClosed) {
tsFile.readLock();
- synchronized (this) {
- if (!isClosed) {
- unclosedReferenceMap
- .computeIfAbsent(tsFile.getTsFileID(), k -> new AtomicInteger())
- .getAndIncrement();
- } else {
- closedReferenceMap
- .computeIfAbsent(tsFile.getTsFileID(), k -> new AtomicInteger())
- .getAndIncrement();
+ boolean registered = false;
+ try {
+ increase(internal, tsFile.getTsFileID(), isClosed);
+ registered = true;
+ } finally {
+ if (!registered) {
+ tsFile.readUnlock();
}
}
}
- public synchronized void increaseExternalFileReaderReference(String
filePath) {
- externalReferenceMap.computeIfAbsent(filePath, k -> new
AtomicInteger()).getAndIncrement();
+ public void increaseExternalFileReaderReference(String filePath) {
+ increase(external, filePath, true);
+ }
+
+ private <K> void increase(Map<K, Entry> registry, K key, boolean isClosed) {
+ // One short registry critical section, with no wait for an opening or
closing reader.
+ synchronized (registryLock) {
+ Entry entry = pin(registry, key, true, true);
+ try {
+ entry.slot(isClosed).references++;
+ } finally {
+ unpin(registry, key, entry);
+ }
+ }
}
/**
- * Decrease the reference count of the reader specified by filePath. This
method is latch-free.
- * Only when the reference count of a reader equals zero, the reader can be
closed and removed.
+ * Releases one retained reference and its resource read lock. The resource
lock is released even
+ * if global test cleanup has already removed the reader reference.
+ *
+ * @param tsFile resource from the matching successful registration; release
it exactly once
+ * @param isClosed slot selected at registration, not the file's current
sealed state
*/
public void decreaseFileReaderReference(TsFileResource tsFile, boolean
isClosed) {
- synchronized (this) {
- if (!isClosed && unclosedReferenceMap.containsKey(tsFile.getTsFileID()))
{
- if (unclosedReferenceMap.get(tsFile.getTsFileID()).decrementAndGet()
== 0) {
- closeUnUsedReaderAndRemoveRef(tsFile.getTsFilePath(),
tsFile.getTsFileID(), false);
- }
- } else if (closedReferenceMap.containsKey(tsFile.getTsFileID())
- && (closedReferenceMap.get(tsFile.getTsFileID()).decrementAndGet()
== 0)) {
- closeUnUsedReaderAndRemoveRef(tsFile.getTsFilePath(),
tsFile.getTsFileID(), true);
- }
+ try {
+ decrease(internal, tsFile.getTsFileID(), isClosed);
+ } finally {
+ tsFile.readUnlock();
}
- tsFile.readUnlock();
}
- public synchronized void decreaseExternalFileReaderReference(String
filePath) {
- AtomicInteger reference = externalReferenceMap.get(filePath);
- if (reference != null && reference.decrementAndGet() == 0) {
- closeUnUsedExternalReaderAndRemoveRef(filePath);
- }
+ public void decreaseExternalFileReaderReference(String filePath) {
+ decrease(external, filePath, true);
}
- private void closeUnUsedExternalReaderAndRemoveRef(String readerKey) {
- synchronized (this) {
- AtomicInteger reference = externalReferenceMap.get(readerKey);
- if (reference != null && reference.get() != 0) {
- return;
+ private <K> void decrease(Map<K, Entry> registry, K key, boolean isClosed) {
+ Entry entry = pin(registry, key, false, false);
+ if (entry == null) {
+ return;
+ }
+ try {
+ Slot slot;
+ synchronized (registryLock) {
+ // Sealing a file does not migrate existing references between slots.
+ slot = entry.slot(isClosed);
+ if (slot.references == 0 || --slot.references != 0) {
+ return;
+ }
}
-
- TsFileSequenceReader reader = externalFileReaderMap.get(readerKey);
- if (reader != null) {
+ synchronized (entry) {
+ synchronized (registryLock) {
+ // A new query may have registered while this release waited for
reader I/O.
+ if (slot.references != 0) {
+ return;
+ }
+ }
+ // New references may now register, but their get() waits for this
entry's close to finish.
try {
- reader.close();
+ closeSlot(
+ slot,
+ slot == entry.closed
+ ? StorageEngineMessages
+
.LOG_READER_FOR_CLOSED_TSFILE_ARG_IS_CLOSED_BECAUSE_ITS_REFERENCE_COUNT_REACHED_ZERO_C3B71A85
+ : StorageEngineMessages
+
.LOG_READER_FOR_UNCLOSED_TSFILE_ARG_IS_CLOSED_BECAUSE_ITS_REFERENCE_COUNT_REACHED_ZERO_088BDEF8);
} catch (IOException e) {
- logger.error(
- StorageEngineMessages.CANNOT_CLOSE_TSFILE_SEQUENCE_READER,
reader.getFileName(), e);
+ // Slot.close already logged the error and discarded the unusable
handle.
}
}
- externalFileReaderMap.remove(readerKey);
- externalReferenceMap.remove(readerKey);
- if (resourceLogger.isDebugEnabled()) {
- resourceLogger.debug(
- "{} externalTsFileReader is closed because of no reference.",
readerKey);
- }
+ } finally {
+ unpin(registry, key, entry);
}
}
- private void closeUnUsedReaderAndRemoveRef(
- String tsFilePath, TsFileID tsFileID, boolean isClosed) {
- Map<TsFileID, TsFileSequenceReader> readerMap =
- isClosed ? closedFileReaderMap : unclosedFileReaderMap;
- Map<TsFileID, AtomicInteger> refMap = isClosed ? closedReferenceMap :
unclosedReferenceMap;
- synchronized (this) {
- // check ref num again
- if (refMap.get(tsFileID).get() != 0) {
- return;
- }
-
- TsFileSequenceReader reader = readerMap.get(tsFileID);
- if (reader != null) {
+ /**
+ * Test cleanup only: callers must first stop reader use. Zero operation
pins does not mean zero
+ * query references. Acquisitions wait; releases remain admitted throughout
cleanup. Interrupts
+ * are restored only after cleanup and the final release-pin drain, never
reopening the gate
+ * early.
+ */
+ @TestOnly
+ public void closeAndRemoveAllOpenedReaders() throws IOException {
+ synchronized (clearLock) {
+ boolean interrupted = Thread.interrupted();
+ IOException failure = null;
+ try {
+ List<Entry> entries;
+ synchronized (registryLock) {
+ clearing = true;
+ interrupted |= drainPins();
+ entries = new ArrayList<>(internal.values());
+ entries.addAll(external.values());
+ }
+ for (Entry entry : entries) {
+ synchronized (entry) {
+ try {
+ closeEntry(entry);
+ } catch (IOException e) {
+ failure = mergeFailure(failure, e);
+ }
+ }
+ interrupted |= Thread.interrupted();
+ }
+ } finally {
try {
- reader.close();
- } catch (IOException e) {
- logger.error(
- StorageEngineMessages.CANNOT_CLOSE_TSFILE_SEQUENCE_READER,
reader.getFileName(), e);
+ synchronized (registryLock) {
+ try {
+ // Drain and reclaim before reopening admission: releases may
still retain an entry.
+ interrupted |= Thread.interrupted();
+ interrupted |= drainPins();
+ internal.values().removeIf(Entry::empty);
+ external.values().removeIf(Entry::empty);
+ } finally {
+ // Even a failed snapshot or reclamation must not permanently
close the test gate.
+ clearing = false;
+ registryLock.notifyAll();
+ }
+ }
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
}
}
- readerMap.remove(tsFileID);
- refMap.remove(tsFileID);
- if (resourceLogger.isDebugEnabled()) {
- resourceLogger.debug(
- StorageEngineMessages.TSFILE_READER_CLOSED_BECAUSE_NO_REFERENCE,
tsFilePath);
+ if (failure != null) {
+ throw failure;
}
}
}
/**
- * Only for <code>EnvironmentUtils.cleanEnv</code> method. To make sure that
unit tests and
- * integration tests will not conflict with each other.
+ * Waits for all currently admitted manager operations to finish during
global test cleanup.
+ * Requires registryLock; waiting releases that monitor so operations can
finish and unpin. This
+ * drains operation pins, not query references, and does not acquire any
entry monitor.
*
- * @throws IOException if failed to close file handlers, IOException will be
thrown
+ * @return whether an interrupt was consumed while waiting; the cleanup
caller restores it after
+ * completing cleanup and reopening admission
*/
- public synchronized void closeAndRemoveAllOpenedReaders() throws IOException
{
- Iterator<Map.Entry<TsFileID, TsFileSequenceReader>> iterator =
- closedFileReaderMap.entrySet().iterator();
- while (iterator.hasNext()) {
- Map.Entry<TsFileID, TsFileSequenceReader> entry = iterator.next();
- entry.getValue().close();
- if (resourceLogger.isDebugEnabled()) {
-
resourceLogger.debug(StorageEngineMessages.CLOSED_TSFILE_READER_CLOSED,
entry.getKey());
+ private boolean drainPins() {
+ boolean interrupted = false;
+ while (pins != 0) {
+ try {
+ registryLock.wait();
+ } catch (InterruptedException e) {
+ interrupted = true;
}
- closedReferenceMap.remove(entry.getKey());
- iterator.remove();
}
- iterator = unclosedFileReaderMap.entrySet().iterator();
- while (iterator.hasNext()) {
- Map.Entry<TsFileID, TsFileSequenceReader> entry = iterator.next();
- entry.getValue().close();
- if (resourceLogger.isDebugEnabled()) {
-
resourceLogger.debug(StorageEngineMessages.UNCLOSED_TSFILE_READER_CLOSED,
entry.getKey());
- }
- unclosedReferenceMap.remove(entry.getKey());
- iterator.remove();
+ return interrupted;
+ }
+
+ @TestOnly
+ public boolean contains(TsFileResource tsFile, boolean isClosed) {
+ TsFileID key = tsFile.getTsFileID();
+ Entry entry = pin(internal, key, true, false);
+ if (entry == null) {
+ return false;
}
- Iterator<Map.Entry<String, TsFileSequenceReader>> externalIterator =
- externalFileReaderMap.entrySet().iterator();
- while (externalIterator.hasNext()) {
- Map.Entry<String, TsFileSequenceReader> entry = externalIterator.next();
- entry.getValue().close();
- if (resourceLogger.isDebugEnabled()) {
- resourceLogger.debug("{} externalTsFileReader is closed.",
entry.getKey());
+ try {
+ synchronized (entry) {
+ return entry.slot(isClosed).reader != null;
}
- externalReferenceMap.remove(entry.getKey());
- externalIterator.remove();
+ } finally {
+ unpin(internal, key, entry);
}
}
- /** This method is only for unit tests. */
- public synchronized boolean contains(TsFileResource tsFile, boolean
isClosed) {
- return (isClosed && closedFileReaderMap.containsKey(tsFile.getTsFileID()))
- || (!isClosed &&
unclosedFileReaderMap.containsKey(tsFile.getTsFileID()));
- }
-
+ /** Snapshot for tests; modifying it does not mutate the manager. */
@TestOnly
public Map<TsFileID, TsFileSequenceReader> getClosedFileReaderMap() {
- return closedFileReaderMap;
+ return snapshot(true);
}
@TestOnly
public Map<TsFileID, TsFileSequenceReader> getUnclosedFileReaderMap() {
- return unclosedFileReaderMap;
+ return snapshot(false);
}
- private static class FileReaderManagerHelper {
+ private Map<TsFileID, TsFileSequenceReader> snapshot(boolean isClosed) {
+ List<TsFileID> keys;
+ synchronized (registryLock) {
+ keys = new ArrayList<>(internal.keySet());
+ }
+ Map<TsFileID, TsFileSequenceReader> result = new HashMap<>();
+ for (TsFileID key : keys) {
+ Entry entry = pin(internal, key, true, false);
+ if (entry != null) {
+ try {
+ synchronized (entry) {
+ if (entry.slot(isClosed).reader != null) {
+ result.put(key, entry.slot(isClosed).reader);
+ }
+ }
+ } finally {
+ unpin(internal, key, entry);
+ }
+ }
+ }
+ return result;
+ }
+ @TestOnly
+ public void setReaderForTest(TsFileID key, boolean isClosed,
TsFileSequenceReader reader)
+ throws IOException {
+ Entry entry = pin(internal, key, true, true);
+ try {
+ synchronized (entry) {
+ Slot slot = entry.slot(isClosed);
+ slot.close();
+ slot.reader = reader;
+ if (reader != null) {
+ slot.count.incrementAndGet();
+ }
+ }
+ } finally {
+ unpin(internal, key, entry);
+ }
+ }
+
+ private static class FileReaderManagerHelper {
private static final FileReaderManager INSTANCE = new FileReaderManager();
private FileReaderManagerHelper() {}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileLock.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileLock.java
index ff5fe58919f..3fddd51f4b7 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileLock.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileLock.java
@@ -39,18 +39,20 @@ public class TsFileLock {
private volatile int writeCnt;
public void readLock() {
+ boolean interrupted = false;
synchronized (this) {
while (writeCnt > 0) {
try {
- this.wait(1);
+ this.wait();
} catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- // ignore
+ interrupted = true;
}
}
-
readCnt++;
}
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
}
public void readUnlock() {
@@ -63,18 +65,21 @@ public class TsFileLock {
}
public void writeLock() {
+ boolean interrupted = false;
synchronized (this) {
- try {
- while (writeCnt > 0 || readCnt > 0) {
- this.wait(1);
+ while (writeCnt > 0 || readCnt > 0) {
+ try {
+ this.wait();
+ } catch (InterruptedException e) {
+ // This API is uninterruptible: preserve exclusion, then restore the
interrupt status.
+ interrupted = true;
}
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- // ignore
}
-
writeCnt++;
}
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
}
public void writeUnlock() {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FileReaderReferenceLifecycleTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FileReaderReferenceLifecycleTest.java
new file mode 100644
index 00000000000..97ba208663a
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FileReaderReferenceLifecycleTest.java
@@ -0,0 +1,189 @@
+/*
+ * 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.queryengine.execution.fragment;
+
+import org.apache.iotdb.commons.path.IFullPath;
+import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
+import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
+import
org.apache.iotdb.db.storageengine.dataregion.read.control.FileReaderManager;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import
org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
+
+import org.apache.tsfile.read.TsFileSequenceReader;
+import org.junit.After;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class FileReaderReferenceLifecycleTest {
+ @Rule public TemporaryFolder folder = new TemporaryFolder();
+ private final FileReaderManager manager = FileReaderManager.getInstance();
+
+ @After
+ public void tearDown() throws Exception {
+ manager.closeAndRemoveAllOpenedReaders();
+ }
+
+ private TsFileResource resource(int id) {
+ return new TsFileResource(new File(folder.getRoot(), id +
"-1-0-0.tsfile"));
+ }
+
+ @Test
+ public void failedRegistrationIsRemovedFromFragmentReleaseSet() throws
Exception {
+ TsFileResource resource = spy(resource(1));
+ Set<TsFileResource> closed = new HashSet<>();
+ Set<TsFileResource> unclosed = new HashSet<>();
+ // An unrelated query already owns a read lock. A failed registration must
not repay that lock.
+ resource.readLock();
+ try {
+ doThrow(new OutOfMemoryError("injected registration failure"))
+ .doCallRealMethod()
+ .when(resource)
+ .getTsFileID();
+ assertThrows(
+ OutOfMemoryError.class,
+ () -> FragmentInstanceContext.addFilePathToMap(resource, true,
closed, unclosed));
+ assertTrue(closed.isEmpty());
+ closed.forEach(file -> manager.decreaseFileReaderReference(file, true));
+ assertFalse(resource.tryWriteLock());
+ } finally {
+ resource.readUnlock();
+ }
+ assertTrue(resource.tryWriteLock());
+ resource.writeUnlock();
+ }
+
+ private FakedFragmentInstanceContext context(TsFileResource... resources)
throws Exception {
+ DataRegion region = mock(DataRegion.class);
+ when(region.tryReadLock(anyLong())).thenReturn(true);
+ when(region.query(anyList(), any(), any(), any(), any(), anyLong()))
+ .thenReturn(new QueryDataSource(Arrays.asList(resources),
Collections.emptyList()));
+ return new FakedFragmentInstanceContext(null, region);
+ }
+
+ @Test
+ public void fakedContextReleasesTheRegistrationTimeSlotAfterSealing() throws
Exception {
+ TsFileResource resource = resource(1);
+ resource.setStatus(TsFileResourceStatus.UNCLOSED);
+ FakedFragmentInstanceContext context = context(resource);
+ context.initQueryDataSource(mock(IFullPath.class));
+ TsFileSequenceReader unclosed = mock(TsFileSequenceReader.class);
+ manager.setReaderForTest(resource.getTsFileID(), false, unclosed);
+ resource.setStatus(TsFileResourceStatus.NORMAL);
+ manager.increaseFileReaderReference(resource, true);
+ TsFileSequenceReader closed = mock(TsFileSequenceReader.class);
+ manager.setReaderForTest(resource.getTsFileID(), true, closed);
+ try {
+ context.releaseSharedQueryDataSource();
+ context.releaseSharedQueryDataSource();
+ verify(unclosed).close();
+ verify(closed, never()).close();
+ assertFalse(manager.contains(resource, false));
+ assertTrue(manager.contains(resource, true));
+ assertFalse(resource.tryWriteLock());
+ } finally {
+ manager.decreaseFileReaderReference(resource, true);
+ }
+ verify(closed).close();
+ assertTrue(resource.tryWriteLock());
+ resource.writeUnlock();
+ }
+
+ @Test
+ public void fakedContextRollsBackOnlySuccessfulRegistrations() throws
Exception {
+ TsFileResource first = resource(1);
+ TsFileResource failing = spy(resource(2));
+ FakedFragmentInstanceContext context = context(first, failing);
+ failing.readLock();
+ try {
+ doThrow(new OutOfMemoryError("injected registration failure"))
+ .doCallRealMethod()
+ .when(failing)
+ .getTsFileID();
+ assertThrows(
+ OutOfMemoryError.class, () ->
context.initQueryDataSource(mock(IFullPath.class)));
+ context.releaseSharedQueryDataSource();
+ assertTrue(first.tryWriteLock());
+ first.writeUnlock();
+ assertFalse(failing.tryWriteLock());
+ } finally {
+ failing.readUnlock();
+ }
+ assertTrue(failing.tryWriteLock());
+ failing.writeUnlock();
+ }
+
+ @Test
+ public void fakedRollbackPreservesOriginalFailureAndReleasesEveryReference()
throws Exception {
+ TsFileResource first = resource(1);
+ TsFileResource second = resource(2);
+ TsFileResource failing = spy(resource(3));
+ first.setStatus(TsFileResourceStatus.NORMAL);
+ second.setStatus(TsFileResourceStatus.NORMAL);
+ FakedFragmentInstanceContext context = context(first, second, failing);
+ TsFileSequenceReader firstReader = mock(TsFileSequenceReader.class);
+ TsFileSequenceReader secondReader = mock(TsFileSequenceReader.class);
+ manager.setReaderForTest(first.getTsFileID(), true, firstReader);
+ manager.setReaderForTest(second.getTsFileID(), true, secondReader);
+ RuntimeException registrationFailure =
+ new IllegalStateException("injected registration failure");
+ RuntimeException closeFailure = new IllegalStateException("injected
unchecked close failure");
+ doThrow(registrationFailure).when(failing).getTsFileID();
+ doThrow(closeFailure).when(firstReader).close();
+ first.readLock();
+ try {
+ RuntimeException actual =
+ assertThrows(
+ RuntimeException.class, () ->
context.initQueryDataSource(mock(IFullPath.class)));
+ assertSame(registrationFailure, actual);
+ assertEquals(1, actual.getSuppressed().length);
+ assertSame(closeFailure, actual.getSuppressed()[0]);
+ verify(firstReader).close();
+ verify(secondReader).close();
+ assertTrue(second.tryWriteLock());
+ second.writeUnlock();
+ context.releaseSharedQueryDataSource();
+ assertFalse(first.tryWriteLock());
+ assertTrue(manager.getClosedFileReaderMap().isEmpty());
+ } finally {
+ first.readUnlock();
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/RegionQueryInitializationLockTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/RegionQueryInitializationLockTest.java
new file mode 100644
index 00000000000..9fbac4790cc
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/RegionQueryInitializationLockTest.java
@@ -0,0 +1,280 @@
+/*
+ * 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.queryengine.execution.fragment;
+
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.path.IFullPath;
+import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
+import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
+import
org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSourceForRegionScan;
+import
org.apache.iotdb.db.storageengine.dataregion.read.control.FileReaderManager;
+import
org.apache.iotdb.db.storageengine.dataregion.read.filescan.IFileScanHandle;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import
org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
+
+import org.apache.tsfile.read.TsFileSequenceReader;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class RegionQueryInitializationLockTest {
+ @Rule public TemporaryFolder folder = new TemporaryFolder();
+ private final FileReaderManager manager = FileReaderManager.getInstance();
+ private ExecutorService workers;
+ private int originalTimeSlice;
+
+ @Before
+ public void setUp() throws Exception {
+ manager.closeAndRemoveAllOpenedReaders();
+ originalTimeSlice =
+
CommonDescriptor.getInstance().getConfig().getDriverTaskExecutionTimeSliceInMs();
+
CommonDescriptor.getInstance().getConfig().setDriverTaskExecutionTimeSliceInMs(10_000);
+ workers = Executors.newFixedThreadPool(3);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ workers.shutdownNow();
+ assertTrue(workers.awaitTermination(10, TimeUnit.SECONDS));
+ manager.closeAndRemoveAllOpenedReaders();
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .setDriverTaskExecutionTimeSliceInMs(originalTimeSlice);
+ }
+
+ private enum QueryMode {
+ SHARED,
+ BATCH,
+ DEVICE_SCAN,
+ SERIES_SCAN,
+ FAKED_FAILURE,
+ BATCH_FAILURE
+ }
+
+ private TsFileResource resource(int id) {
+ TsFileResource resource = new TsFileResource(new File(folder.getRoot(), id
+ "-1-0-0.tsfile"));
+ resource.setStatusForTest(TsFileResourceStatus.NORMAL);
+ return resource;
+ }
+
+ @Test
+ public void sharedQueryReleasesDeletedFilesOutsideRegionLock() throws
Exception {
+ releasesOutsideRegionLock(QueryMode.SHARED);
+ }
+
+ @Test
+ public void batchQueryReleasesDeletedFilesOutsideRegionLock() throws
Exception {
+ releasesOutsideRegionLock(QueryMode.BATCH);
+ }
+
+ @Test
+ public void deviceScanReleasesDeletedFilesOutsideRegionLock() throws
Exception {
+ releasesOutsideRegionLock(QueryMode.DEVICE_SCAN);
+ }
+
+ @Test
+ public void seriesScanReleasesDeletedFilesOutsideRegionLock() throws
Exception {
+ releasesOutsideRegionLock(QueryMode.SERIES_SCAN);
+ }
+
+ @Test
+ public void fakedQueryRollsBackOutsideRegionLock() throws Exception {
+ releasesOutsideRegionLock(QueryMode.FAKED_FAILURE);
+ }
+
+ @Test
+ public void failedBatchReleasesAllReferencesOutsideRegionLock() throws
Exception {
+ releasesOutsideRegionLock(QueryMode.BATCH_FAILURE);
+ }
+
+ private void releasesOutsideRegionLock(QueryMode mode) throws Exception {
+ DataRegion region = spy(new DataRegion("root.region_release", "1"));
+ FakedFragmentInstanceContext context = new
FakedFragmentInstanceContext(null, region);
+ IFullPath path = mock(IFullPath.class);
+ TsFileResource resource = resource(1);
+ if (mode != QueryMode.FAKED_FAILURE) {
+ resource.setStatusForTest(TsFileResourceStatus.DELETED);
+ }
+ TsFileResource failing = spy(resource(2));
+ TsFileResource retained = resource(3);
+ RuntimeException failure = new IllegalStateException("injected reference
registration failure");
+ doThrow(failure).when(failing).getTsFileID();
+ ArrayList<TsFileResource> seq = new
ArrayList<>(Collections.singletonList(resource));
+ if (mode == QueryMode.FAKED_FAILURE || mode == QueryMode.BATCH_FAILURE) {
+ seq.add(failing);
+ }
+ if (mode == QueryMode.BATCH_FAILURE) {
+ seq.add(0, retained);
+ }
+ QueryDataSource dataSource = new QueryDataSource(seq, new ArrayList<>());
+ doReturn(dataSource).when(region).query(anyList(), any(), any(), any(),
any(), anyLong());
+ IFileScanHandle handle = mock(IFileScanHandle.class);
+ when(handle.getTsResource()).thenReturn(resource);
+ when(handle.isClosed()).thenReturn(true);
+ QueryDataSourceForRegionScan scanSource =
+ new QueryDataSourceForRegionScan(
+ new ArrayList<>(Collections.singletonList(handle)), new
ArrayList<>());
+ doReturn(scanSource)
+ .when(region)
+ .queryForDeviceRegionScan(any(), any(), any(), any(), anyLong());
+ doReturn(scanSource)
+ .when(region)
+ .queryForSeriesRegionScan(anyList(), any(), any(), any(), anyLong());
+
+ TsFileSequenceReader reader = mock(TsFileSequenceReader.class);
+ manager.setReaderForTest(resource.getTsFileID(), true, reader);
+ manager.increaseFileReaderReference(resource, true);
+ CountDownLatch closing = new CountDownLatch(1);
+ CountDownLatch finishClose = new CountDownLatch(1);
+ AtomicReference<Thread> closeThread = new AtomicReference<>();
+ AtomicReference<Thread> queryThread = new AtomicReference<>();
+ CountDownLatch queryStarted = new CountDownLatch(1);
+ doAnswer(
+ invocation -> {
+ closeThread.set(Thread.currentThread());
+ closing.countDown();
+ assertTrue(finishClose.await(20, TimeUnit.SECONDS));
+ return null;
+ })
+ .when(reader)
+ .close();
+ Future<?> release = workers.submit(() ->
manager.decreaseFileReaderReference(resource, true));
+ Future<?> query = null;
+ try {
+ assertTrue(closing.await(10, TimeUnit.SECONDS));
+ query =
+ workers.submit(
+ () -> {
+ queryThread.set(Thread.currentThread());
+ queryStarted.countDown();
+ switch (mode) {
+ case SHARED:
+
assertTrue(context.initQueryDataSource(Collections.singletonList(path)));
+ break;
+ case BATCH:
+ try (QueryDataSourceLease lease =
+
context.initBatchQueryDataSource(Collections.singletonList(path))) {
+
assertTrue(lease.getDataSource().getSeqResources().isEmpty());
+ }
+ break;
+ case DEVICE_SCAN:
+
assertTrue(context.initRegionScanQueryDataSource(Collections.emptyMap()));
+ break;
+ case SERIES_SCAN:
+ assertTrue(
+
context.initRegionScanQueryDataSource(Collections.singletonList(path)));
+ break;
+ case FAKED_FAILURE:
+ assertSame(
+ failure,
+ assertThrows(
+ RuntimeException.class, () ->
context.initQueryDataSource(path)));
+ break;
+ case BATCH_FAILURE:
+ assertSame(
+ failure,
+ assertThrows(
+ RuntimeException.class,
+ () ->
+
context.initBatchQueryDataSource(Collections.singletonList(path))));
+ break;
+ default:
+ throw new AssertionError(mode);
+ }
+ return null;
+ });
+ assertTrue(queryStarted.await(10, TimeUnit.SECONDS));
+ awaitEntryWait(queryThread.get(), closeThread.get());
+ Future<?> writer =
+ workers.submit(
+ () -> {
+ region.writeLock("deferred reference release test");
+ region.writeUnlock();
+ });
+ writer.get(10, TimeUnit.SECONDS);
+ assertFalse(query.isDone());
+ } finally {
+ finishClose.countDown();
+ release.get(10, TimeUnit.SECONDS);
+ if (query != null) {
+ query.get(10, TimeUnit.SECONDS);
+ }
+ }
+ verify(reader).close();
+ if (mode == QueryMode.SHARED || mode == QueryMode.BATCH) {
+ assertTrue(dataSource.getSeqResources().isEmpty());
+ } else if (mode == QueryMode.DEVICE_SCAN || mode == QueryMode.SERIES_SCAN)
{
+ assertTrue(scanSource.getSeqFileScanHandles().isEmpty());
+ }
+ for (TsFileResource file : Arrays.asList(resource, failing, retained)) {
+ assertTrue(file.tryWriteLock());
+ file.writeUnlock();
+ }
+ assertTrue(manager.getClosedFileReaderMap().isEmpty());
+ assertTrue(manager.getUnclosedFileReaderMap().isEmpty());
+ if (mode == QueryMode.FAKED_FAILURE) {
+ context.releaseSharedQueryDataSource();
+ }
+ }
+
+ private static void awaitEntryWait(Thread waiter, Thread owner) throws
Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (System.nanoTime() < deadline) {
+ ThreadInfo info =
ManagementFactory.getThreadMXBean().getThreadInfo(waiter.getId());
+ if (info != null
+ && info.getThreadState() == Thread.State.BLOCKED
+ && info.getLockOwnerId() == owner.getId()) {
+ return;
+ }
+ Thread.sleep(1);
+ }
+ fail("query did not reach the entry held by the closing reader");
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/RegionScanTimeSliceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/RegionScanTimeSliceTest.java
new file mode 100644
index 00000000000..23db46d121e
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/RegionScanTimeSliceTest.java
@@ -0,0 +1,96 @@
+/*
+ * 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.queryengine.execution.fragment;
+
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.path.IFullPath;
+import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
+
+import com.google.common.util.concurrent.Uninterruptibles;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+public class RegionScanTimeSliceTest {
+ private int originalTimeSlice;
+
+ @Before
+ public void setUp() {
+ originalTimeSlice =
+
CommonDescriptor.getInstance().getConfig().getDriverTaskExecutionTimeSliceInMs();
+ }
+
+ @After
+ public void tearDown() {
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .setDriverTaskExecutionTimeSliceInMs(originalTimeSlice);
+ }
+
+ @Test
+ public void exhaustedSeriesScanTimeSliceStillUnlocksRegion() throws
Exception {
+ DataRegion realRegion = new DataRegion("root.region_release", "1");
+ DataRegion region = spy(realRegion);
+ FakedFragmentInstanceContext context = new
FakedFragmentInstanceContext(null, region);
+
CommonDescriptor.getInstance().getConfig().setDriverTaskExecutionTimeSliceInMs(1);
+ AtomicBoolean locked = new AtomicBoolean();
+ doAnswer(
+ invocation -> {
+ assertTrue(realRegion.tryReadLock(10_000));
+ locked.set(true);
+ Uninterruptibles.sleepUninterruptibly(10, TimeUnit.MILLISECONDS);
+ return true;
+ })
+ .when(region)
+ .tryReadLock(anyLong());
+ doAnswer(
+ invocation -> {
+ invocation.callRealMethod();
+ locked.set(false);
+ return null;
+ })
+ .when(region)
+ .readUnlock();
+ try {
+ assertFalse(
+
context.initRegionScanQueryDataSource(Collections.singletonList(mock(IFullPath.class))));
+ assertFalse(
+ "region read lock must be released on the exhausted-time-slice
return", locked.get());
+ verify(region).readUnlock();
+ assertEquals(0, context.getInitQueryDataSourceRetryCount());
+ } finally {
+ if (locked.get()) {
+ region.readUnlock();
+ }
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/buffer/TimeSeriesMetadataCacheTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/buffer/TimeSeriesMetadataCacheTest.java
index 751eae090ae..87526eb594b 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/buffer/TimeSeriesMetadataCacheTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/buffer/TimeSeriesMetadataCacheTest.java
@@ -238,7 +238,7 @@ public class TimeSeriesMetadataCacheTest {
new TimeSeriesMetadataCacheKey(tsFileID, deviceID, measurement);
TsFileSequenceReader reader = Mockito.mock(TsFileSequenceReader.class);
- fileReaderManager.getClosedFileReaderMap().put(tsFileID, reader);
+ fileReaderManager.setReaderForTest(tsFileID, true, reader);
Mockito.when(reader.readBloomFilter(Mockito.any(LongConsumer.class))).thenReturn(null);
Mockito.when(
reader.readTimeseriesMetadata(
@@ -274,7 +274,7 @@ public class TimeSeriesMetadataCacheTest {
} finally {
cache.clear();
bloomFilterCache.clear();
- fileReaderManager.getClosedFileReaderMap().remove(tsFileID);
+ fileReaderManager.closeFileAndRemoveReader(tsFileID);
IoTDBDescriptor.getInstance()
.getMemoryConfig()
.setMayCacheNonExistSeries(mayCacheNonExistSeries);
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/control/FileReaderManagerConcurrencyTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/control/FileReaderManagerConcurrencyTest.java
new file mode 100644
index 00000000000..67a480f43d2
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/control/FileReaderManagerConcurrencyTest.java
@@ -0,0 +1,871 @@
+/*
+ * 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.storageengine.dataregion.read.control;
+
+import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileID;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+
+import org.apache.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.tsfile.fileSystem.fileInputFactory.FileInputFactory;
+import org.apache.tsfile.read.TsFileSequenceReader;
+import org.apache.tsfile.read.reader.TsFileInput;
+import org.apache.tsfile.write.writer.TsFileIOWriter;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doCallRealMethod;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+public class FileReaderManagerConcurrencyTest {
+ @Rule public TemporaryFolder folder = new TemporaryFolder();
+ private final FileReaderManager manager = FileReaderManager.getInstance();
+ private ExecutorService workers;
+
+ @Before
+ public void setUp() throws Exception {
+ manager.closeAndRemoveAllOpenedReaders();
+ workers = Executors.newCachedThreadPool();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ workers.shutdownNow();
+ assertTrue(workers.awaitTermination(15, TimeUnit.SECONDS));
+ manager.closeAndRemoveAllOpenedReaders();
+ assertEmpty();
+ }
+
+ private TsFileResource file(int id) throws Exception {
+ File f = new File(folder.getRoot(), id + "-1-0-0.tsfile");
+ try (TsFileIOWriter writer = new TsFileIOWriter(f)) {
+ writer.endFile();
+ }
+ return new TsFileResource(f);
+ }
+
+ private TsFileSequenceReader get(TsFileResource file, boolean closed) throws
IOException {
+ return manager.get(file.getTsFilePath(), file.getTsFileID(), closed);
+ }
+
+ private static void done(Future<?> task) throws Exception {
+ task.get(10, TimeUnit.SECONDS);
+ }
+
+ private static class Gate implements AutoCloseable {
+ final CountDownLatch entered = new CountDownLatch(1);
+ final CountDownLatch open = new CountDownLatch(1);
+ final AtomicReference<Thread> owner = new AtomicReference<>();
+
+ void block() {
+ owner.set(Thread.currentThread());
+ entered.countDown();
+ boolean interrupted = false;
+ try {
+ while (true) {
+ try {
+ if (!open.await(20, TimeUnit.SECONDS)) {
+ throw new AssertionError("gate not released");
+ }
+ return;
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ void await() throws Exception {
+ assertTrue(entered.await(10, TimeUnit.SECONDS));
+ }
+
+ @Override
+ public void close() {
+ open.countDown();
+ }
+ }
+
+ // Prove lock contention and its owner, rather than inferring it from a
Future timeout.
+ private static void blocked(Thread waiter, Thread owner) throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (System.nanoTime() < deadline) {
+ ThreadInfo info =
ManagementFactory.getThreadMXBean().getThreadInfo(waiter.getId());
+ if (info != null
+ && info.getThreadState() == Thread.State.BLOCKED
+ && info.getLockOwnerId() == owner.getId()) {
+ assertNotNull(info.getLockInfo());
+ return;
+ }
+ Thread.sleep(1);
+ }
+ fail("worker did not block on the expected monitor owner");
+ }
+
+ @Test
+ public void slowClosedOpenDoesNotBlockOtherFiles() throws Exception {
+ slowOpen(true, false);
+ }
+
+ @Test
+ public void slowUnclosedOpenDoesNotBlockOtherFiles() throws Exception {
+ slowOpen(false, false);
+ }
+
+ @Test
+ public void slowExternalOpenDoesNotBlockOtherFiles() throws Exception {
+ slowOpen(true, true);
+ }
+
+ private void slowOpen(boolean closed, boolean external) throws Exception {
+ TsFileResource a = file(1);
+ TsFileResource b = file(2);
+ TsFileResource c = file(3);
+ assertNotEquals(a.getTsFileID(), b.getTsFileID());
+ TsFileSequenceReader warm = get(b, true);
+ try (Gate gate = new Gate()) {
+ Future<TsFileSequenceReader> opening =
+ workers.submit(
+ () ->
+ manager.get(
+ a.getTsFilePath(), a.getTsFileID(), closed, n ->
gate.block(), external));
+ gate.await();
+ AtomicReference<Thread> waiter = new AtomicReference<>();
+ CountDownLatch started = new CountDownLatch(1);
+ Future<TsFileSequenceReader> same =
+ workers.submit(
+ () -> {
+ waiter.set(Thread.currentThread());
+ started.countDown();
+ return manager.get(a.getTsFilePath(), a.getTsFileID(), closed,
null, external);
+ });
+ assertTrue(started.await(10, TimeUnit.SECONDS));
+ blocked(waiter.get(), gate.owner.get());
+ unrelated(b, c, warm);
+ gate.close();
+ assertSame(opening.get(10, TimeUnit.SECONDS), same.get(10,
TimeUnit.SECONDS));
+ }
+ }
+
+ private void unrelated(TsFileResource b, TsFileResource c,
TsFileSequenceReader warm)
+ throws Exception {
+ done(
+ workers.submit(
+ () -> {
+ assertSame(warm, get(b, true));
+ assertNotNull(get(c, true));
+ assertNotNull(get(c, false));
+ assertNotNull(manager.get(c.getTsFilePath(), c.getTsFileID(),
true, null, true));
+ manager.increaseFileReaderReference(b, true);
+ manager.decreaseFileReaderReference(b, true);
+ manager.increaseExternalFileReaderReference(c.getTsFilePath());
+ manager.decreaseExternalFileReaderReference(c.getTsFilePath());
+ return null;
+ }));
+ }
+
+ @Test
+ public void slowForceCloseDoesNotBlockOtherFiles() throws Exception {
+ slowClose(false);
+ }
+
+ @Test
+ public void slowLastReleaseDoesNotBlockOtherFiles() throws Exception {
+ slowClose(true);
+ }
+
+ private void slowClose(boolean release) throws Exception {
+ TsFileResource a = file(1);
+ TsFileResource b = file(2);
+ TsFileResource c = file(3);
+ TsFileSequenceReader warm = get(b, true);
+ TsFileSequenceReader mock = mock(TsFileSequenceReader.class);
+ manager.setReaderForTest(a.getTsFileID(), true, mock);
+ if (release) {
+ manager.increaseFileReaderReference(a, true);
+ }
+ try (Gate gate = new Gate()) {
+ doAnswer(
+ invocation -> {
+ gate.block();
+ return null;
+ })
+ .when(mock)
+ .close();
+ Future<?> closing =
+ workers.submit(
+ () -> {
+ if (release) {
+ manager.decreaseFileReaderReference(a, true);
+ } else {
+ manager.closeFileAndRemoveReader(a.getTsFileID());
+ }
+ return null;
+ });
+ gate.await();
+ AtomicReference<Thread> waiter = new AtomicReference<>();
+ CountDownLatch started = new CountDownLatch(1);
+ Future<?> acquire =
+ workers.submit(
+ () -> {
+ waiter.set(Thread.currentThread());
+ started.countDown();
+ manager.increaseFileReaderReference(a, true);
+ try {
+ assertNotSame(mock, get(a, true));
+ } finally {
+ manager.decreaseFileReaderReference(a, true);
+ }
+ return null;
+ });
+ assertTrue(started.await(10, TimeUnit.SECONDS));
+ blocked(waiter.get(), gate.owner.get());
+ unrelated(b, c, warm);
+ gate.close();
+ done(closing);
+ done(acquire);
+ verify(mock, times(1)).close();
+ assertFalse(manager.contains(a, true));
+ }
+ }
+
+ @Test
+ public void failedConstructionPreservesReferencesAndRetries() throws
Exception {
+ TsFileResource a = file(1);
+ manager.increaseFileReaderReference(a, true);
+ manager.increaseFileReaderReference(a, true);
+ try {
+ manager.get(a.getTsFilePath() + ".missing", a.getTsFileID(), true);
+ fail("missing file opened");
+ } catch (IOException expected) {
+ assertFalse(manager.contains(a, true));
+ }
+ TsFileSequenceReader reader = get(a, true);
+ manager.decreaseFileReaderReference(a, true);
+ assertSame(reader, get(a, true));
+ manager.decreaseFileReaderReference(a, true);
+ assertEmpty();
+ }
+
+ @Test
+ public void closeFailureDoesNotPoisonOverlappingQueries() throws Exception {
+ TsFileResource a = file(1);
+ TsFileSequenceReader reader = mock(TsFileSequenceReader.class);
+ doThrow(new IOException("injected close failure")).when(reader).close();
+ manager.setReaderForTest(a.getTsFileID(), true, reader);
+ manager.increaseFileReaderReference(a, true);
+ manager.decreaseFileReaderReference(a, true);
+ assertTrue(a.tryWriteLock());
+ a.writeUnlock();
+ assertEmpty();
+ // A long-running query registers this file before reaching it. Other
queries must recover
+ // without waiting for its reference (and all the overlapping queries'
references) to drain.
+ manager.increaseFileReaderReference(a, true);
+ try {
+ for (int i = 0; i < 5; i++) {
+ manager.increaseFileReaderReference(a, true);
+ try {
+ TsFileSequenceReader reopened = get(a, true);
+ assertNotSame(reader, reopened);
+ assertTrue(reopened.fileSize() > 0);
+ } finally {
+ manager.decreaseFileReaderReference(a, true);
+ }
+ }
+ } finally {
+ manager.decreaseFileReaderReference(a, true);
+ }
+ verify(reader, times(1)).close();
+ assertEmpty();
+ }
+
+ @Test
+ public void failedUnclosedCloseDoesNotLeakAfterSealing() throws Exception {
+ TsFileResource a = file(1);
+ TsFileSequenceReader unclosed = mock(TsFileSequenceReader.class);
+ doThrow(new IOException("injected unclosed close
failure")).when(unclosed).close();
+ manager.setReaderForTest(a.getTsFileID(), false, unclosed);
+ manager.increaseFileReaderReference(a, false);
+ manager.decreaseFileReaderReference(a, false);
+ assertEmpty();
+ // All subsequent queries address the closed slot, never the failed
unclosed slot again.
+ manager.increaseFileReaderReference(a, true);
+ assertTrue(get(a, true).fileSize() > 0);
+ manager.decreaseFileReaderReference(a, true);
+ assertEmpty();
+ }
+
+ @Test
+ public void failedExternalCloseDoesNotPoisonNextQuery() throws Exception {
+ TsFileResource a = file(1);
+ FileInputFactory original = FSFactoryProducer.getFileInputFactory();
+ TsFileInput input = spy(original.getTsFileInput(a.getTsFilePath()));
+ AtomicBoolean first = new AtomicBoolean(true);
+ doThrow(new IOException("injected external close
failure")).when(input).close();
+ FSFactoryProducer.setFileInputFactory(
+ path -> first.getAndSet(false) ? input :
original.getTsFileInput(path));
+ try {
+ manager.increaseExternalFileReaderReference(a.getTsFilePath());
+ TsFileSequenceReader failed = manager.get(a.getTsFilePath(), null, true,
null, true);
+ manager.decreaseExternalFileReaderReference(a.getTsFilePath());
+ assertEmpty();
+ manager.increaseExternalFileReaderReference(a.getTsFilePath());
+ TsFileSequenceReader reopened = manager.get(a.getTsFilePath(), null,
true, null, true);
+ assertNotSame(failed, reopened);
+ assertTrue(reopened.fileSize() > 0);
+ manager.decreaseExternalFileReaderReference(a.getTsFilePath());
+ assertEmpty();
+ } finally {
+ FSFactoryProducer.setFileInputFactory(original);
+ // The injected exception deliberately prevented closing the real input.
+ doCallRealMethod().when(input).close();
+ input.close();
+ }
+ }
+
+ @Test
+ public void slowOpenDoesNotHoldRegionLockDuringReferenceRegistration()
throws Exception {
+ referenceRegistrationDuringIo(false);
+ }
+
+ @Test
+ public void slowCloseDoesNotHoldRegionLockDuringReferenceRegistration()
throws Exception {
+ referenceRegistrationDuringIo(true);
+ }
+
+ private void referenceRegistrationDuringIo(boolean close) throws Exception {
+ TsFileResource a = file(1);
+ DataRegion region = new DataRegion("root.reference_registration", "1");
+ try (Gate gate = new Gate()) {
+ Future<?> io;
+ if (close) {
+ TsFileSequenceReader reader = mock(TsFileSequenceReader.class);
+ manager.setReaderForTest(a.getTsFileID(), true, reader);
+ manager.increaseFileReaderReference(a, true);
+ doAnswer(
+ invocation -> {
+ gate.block();
+ return null;
+ })
+ .when(reader)
+ .close();
+ io = workers.submit(() -> manager.decreaseFileReaderReference(a,
true));
+ } else {
+ io =
+ workers.submit(
+ () -> manager.get(a.getTsFilePath(), a.getTsFileID(), true, n
-> gate.block()));
+ }
+ gate.await();
+ done(
+ workers.submit(
+ () -> {
+ assertTrue(region.tryReadLock(10_000));
+ try {
+ manager.increaseFileReaderReference(a, true);
+ } finally {
+ region.readUnlock();
+ }
+ }));
+ // A writer must progress while reader I/O is still blocked on the same
file.
+ done(
+ workers.submit(
+ () -> {
+ region.writeLock("reference registration test");
+ region.writeUnlock();
+ }));
+ assertFalse(io.isDone());
+ gate.close();
+ done(io);
+ assertTrue(get(a, true).fileSize() > 0);
+ assertFalse(a.tryWriteLock());
+ manager.decreaseFileReaderReference(a, true);
+ assertTrue(a.tryWriteLock());
+ a.writeUnlock();
+ assertEmpty();
+ }
+ }
+
+ @Test
+ public void lastReleaseRechecksReferencesAfterWaitingForOpen() throws
Exception {
+ TsFileResource a = file(1);
+ manager.increaseFileReaderReference(a, true);
+ try (Gate gate = new Gate()) {
+ Future<TsFileSequenceReader> open =
+ workers.submit(
+ () -> manager.get(a.getTsFilePath(), a.getTsFileID(), true, n ->
gate.block()));
+ gate.await();
+ AtomicReference<Thread> releaseThread = new AtomicReference<>();
+ CountDownLatch started = new CountDownLatch(1);
+ Future<?> release =
+ workers.submit(
+ () -> {
+ releaseThread.set(Thread.currentThread());
+ started.countDown();
+ manager.decreaseFileReaderReference(a, true);
+ });
+ assertTrue(started.await(10, TimeUnit.SECONDS));
+ blocked(releaseThread.get(), gate.owner.get());
+ done(workers.submit(() -> manager.increaseFileReaderReference(a, true)));
+ gate.close();
+ TsFileSequenceReader reader = open.get(10, TimeUnit.SECONDS);
+ done(release);
+ assertSame(reader, get(a, true));
+ assertTrue(reader.fileSize() > 0);
+ assertFalse(a.tryWriteLock());
+ manager.decreaseFileReaderReference(a, true);
+ assertEmpty();
+ }
+ }
+
+ @Test
+ public void namespacesAndSlotReferencesAreIndependent() throws Exception {
+ TsFileResource a = file(1);
+ manager.increaseFileReaderReference(a, true);
+ manager.increaseFileReaderReference(a, false);
+ manager.increaseExternalFileReaderReference(a.getTsFilePath());
+ TsFileSequenceReader closed = get(a, true);
+ TsFileSequenceReader unclosed = get(a, false);
+ TsFileSequenceReader external =
+ manager.get(a.getTsFilePath(), a.getTsFileID(), true, null, true);
+ assertNotSame(closed, unclosed);
+ assertNotSame(closed, external);
+ assertSame(external, manager.get(new String(a.getTsFilePath()), null,
false, null, true));
+ manager.decreaseFileReaderReference(a, true);
+ assertFalse(manager.contains(a, true));
+ assertSame(unclosed, get(a, false));
+ assertTrue(unclosed.fileSize() > 0);
+ manager.decreaseFileReaderReference(a, false);
+ assertFalse(manager.contains(a, false));
+ // Forced internal-reader cleanup must not affect the separately retained
external reader.
+ assertNotNull(get(a, true));
+ assertNotNull(get(a, false));
+ manager.closeFileAndRemoveReader(a.getTsFileID());
+ assertFalse(manager.contains(a, true));
+ assertFalse(manager.contains(a, false));
+ assertSame(external, manager.get(a.getTsFilePath(), null, true, null,
true));
+ assertTrue(external.fileSize() > 0);
+ manager.decreaseExternalFileReaderReference(a.getTsFilePath());
+ assertEmpty();
+ }
+
+ @Test
+ public void lateUnclosedReleaseDoesNotCloseNewClosedReader() throws
Exception {
+ TsFileResource a = file(1);
+ manager.increaseFileReaderReference(a, false);
+ assertNotNull(get(a, false));
+ // Reader use has stopped, but this query still owes its resource
read-lock release.
+ manager.closeAndRemoveAllOpenedReaders();
+ manager.increaseFileReaderReference(a, true);
+ try {
+ TsFileSequenceReader closed = get(a, true);
+ manager.decreaseFileReaderReference(a, false);
+ // The old unclosed reference was already cleared. Its release must not
consume the new
+ // query's closed reference or close that query's reader.
+ assertTrue(manager.contains(a, true));
+ assertSame(closed, get(a, true));
+ assertTrue(closed.fileSize() > 0);
+ assertFalse(a.tryWriteLock());
+ } finally {
+ manager.decreaseFileReaderReference(a, true);
+ }
+ assertTrue(a.tryWriteLock());
+ a.writeUnlock();
+ assertEmpty();
+ }
+
+ @Test
+ public void clearDrainsReleasePinsEvenWhenInterrupted() throws Exception {
+ TsFileResource a = file(1);
+ TsFileSequenceReader reader = mock(TsFileSequenceReader.class);
+ manager.setReaderForTest(a.getTsFileID(), true, reader);
+ Object registry = field("internal");
+ Method pin =
+ FileReaderManager.class.getDeclaredMethod(
+ "pin", Map.class, Object.class, boolean.class, boolean.class);
+ pin.setAccessible(true);
+ Object entry = null;
+ try (Gate gate = new Gate()) {
+ doAnswer(
+ invocation -> {
+ gate.block();
+ return null;
+ })
+ .when(reader)
+ .close();
+ Future<Boolean> clear =
+ workers.submit(
+ () -> {
+ manager.closeAndRemoveAllOpenedReaders();
+ return Thread.interrupted();
+ });
+ gate.await();
+ // Hold a release pin beyond the close phase: a deterministic model of a
descheduled release.
+ entry = pin.invoke(manager, registry, a.getTsFileID(), false, false);
+ assertNotNull(entry);
+ gate.owner.get().interrupt();
+ Future<?> acquire = acquireDuringClear(a, true);
+ gate.close();
+ awaitClearDrain(gate.owner.get());
+ assertFalse(clear.isDone());
+ assertFalse(acquire.isDone());
+ Method unpin =
+ FileReaderManager.class.getDeclaredMethod(
+ "unpin", Map.class, Object.class, entry.getClass());
+ unpin.setAccessible(true);
+ unpin.invoke(manager, registry, a.getTsFileID(), entry);
+ entry = null;
+ assertTrue(clear.get(10, TimeUnit.SECONDS));
+ done(acquire);
+ } finally {
+ if (entry != null) {
+ Method unpin =
+ FileReaderManager.class.getDeclaredMethod(
+ "unpin", Map.class, Object.class, entry.getClass());
+ unpin.setAccessible(true);
+ unpin.invoke(manager, registry, a.getTsFileID(), entry);
+ }
+ }
+ }
+
+ private Future<TsFileSequenceReader> acquireDuringClear(TsFileResource
resource, boolean closed)
+ throws Exception {
+ AtomicReference<Thread> thread = new AtomicReference<>();
+ CountDownLatch started = new CountDownLatch(1);
+ Future<TsFileSequenceReader> acquire =
+ workers.submit(
+ () -> {
+ thread.set(Thread.currentThread());
+ started.countDown();
+ return get(resource, closed);
+ });
+ assertTrue(started.await(10, TimeUnit.SECONDS));
+ // An unfinished Future alone does not prove that the worker has reached
the admission gate.
+ awaitRegistryWait(thread.get(), "pin");
+ return acquire;
+ }
+
+ private void awaitClearDrain(Thread thread) throws Exception {
+ awaitRegistryWait(thread, "drainPins");
+ }
+
+ private void awaitRegistryWait(Thread thread, String method) throws
Exception {
+ Object registryLock = field("registryLock");
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (System.nanoTime() < deadline) {
+ ThreadInfo info =
ManagementFactory.getThreadMXBean().getThreadInfo(thread.getId(), 30);
+ if (info != null
+ && info.getThreadState() == Thread.State.WAITING
+ && info.getLockInfo() != null
+ && info.getLockInfo().getIdentityHashCode() ==
System.identityHashCode(registryLock)) {
+ for (StackTraceElement frame : info.getStackTrace()) {
+ if (frame.getClassName().equals(FileReaderManager.class.getName())
+ && frame.getMethodName().equals(method)) {
+ return;
+ }
+ }
+ }
+ Thread.sleep(1);
+ }
+ fail("worker did not wait on registryLock in " + method);
+ }
+
+ @Test
+ public void concurrentClearAndReleaseComplete() throws Exception {
+ TsFileResource a = file(1);
+ TsFileSequenceReader reader = mock(TsFileSequenceReader.class);
+ manager.setReaderForTest(a.getTsFileID(), true, reader);
+ manager.increaseFileReaderReference(a, true);
+ try (Gate gate = new Gate()) {
+ doAnswer(
+ invocation -> {
+ gate.block();
+ return null;
+ })
+ .when(reader)
+ .close();
+ Future<?> first =
+ workers.submit(
+ () -> {
+ manager.closeAndRemoveAllOpenedReaders();
+ return null;
+ });
+ gate.await();
+ AtomicReference<Thread> releaseThread = new AtomicReference<>();
+ CountDownLatch started = new CountDownLatch(1);
+ Future<?> release =
+ workers.submit(
+ () -> {
+ releaseThread.set(Thread.currentThread());
+ started.countDown();
+ manager.decreaseFileReaderReference(a, true);
+ });
+ assertTrue(started.await(10, TimeUnit.SECONDS));
+ // Clear already reset the reference count. This release no longer waits
for reader I/O.
+ done(release);
+ AtomicReference<Thread> secondThread = new AtomicReference<>();
+ CountDownLatch secondStarted = new CountDownLatch(1);
+ Future<?> second =
+ workers.submit(
+ () -> {
+ secondThread.set(Thread.currentThread());
+ secondStarted.countDown();
+ manager.closeAndRemoveAllOpenedReaders();
+ return null;
+ });
+ assertTrue(secondStarted.await(10, TimeUnit.SECONDS));
+ blocked(secondThread.get(), gate.owner.get());
+ gate.close();
+ done(first);
+ done(second);
+ done(release);
+ verify(reader, times(1)).close();
+ assertEmpty();
+ }
+ }
+
+ @Test
+ public void concurrentChurnReclaimsEntriesAndReaders() throws Exception {
+ TsFileResource[] files = {file(1), file(2), file(3)};
+ List<Future<?>> tasks = new ArrayList<>();
+ for (int t = 0; t < 8; t++) {
+ tasks.add(
+ workers.submit(
+ () -> {
+ for (int i = 0; i < 500; i++) {
+ TsFileResource f = files[i % files.length];
+ boolean closed = (i & 1) == 0;
+ manager.increaseFileReaderReference(f, closed);
+ try {
+ assertSame(get(f, closed), get(f, closed));
+ } finally {
+ manager.decreaseFileReaderReference(f, closed);
+ }
+
manager.increaseExternalFileReaderReference(f.getTsFilePath());
+ try {
+ assertNotNull(manager.get(f.getTsFilePath(), null, true,
null, true));
+ } finally {
+
manager.decreaseExternalFileReaderReference(f.getTsFilePath());
+ }
+ }
+ return null;
+ }));
+ }
+ for (Future<?> task : tasks) {
+ task.get(60, TimeUnit.SECONDS);
+ }
+ assertEmpty();
+ }
+
+ @Test
+ public void forceCloseAttemptsBothSlotsAndDiscardsFailure() throws Exception
{
+ TsFileResource a = file(1);
+ TsFileSequenceReader closed = mock(TsFileSequenceReader.class);
+ TsFileSequenceReader unclosed = mock(TsFileSequenceReader.class);
+ manager.setReaderForTest(a.getTsFileID(), true, closed);
+ manager.setReaderForTest(a.getTsFileID(), false, unclosed);
+ doThrow(new IOException("close failed")).doNothing().when(closed).close();
+ try {
+ manager.closeFileAndRemoveReader(a.getTsFileID());
+ fail("close failure lost");
+ } catch (IOException expected) {
+ verify(unclosed).close();
+ assertFalse(manager.contains(a, false));
+ }
+ manager.closeAndRemoveAllOpenedReaders();
+ verify(closed, times(1)).close();
+ assertEmpty();
+ }
+
+ @Test
+ public void closeAggregationDoesNotMutateReaderExceptions() throws Exception
{
+ TsFileResource a = file(1);
+ TsFileSequenceReader closed = mock(TsFileSequenceReader.class);
+ TsFileSequenceReader unclosed = mock(TsFileSequenceReader.class);
+ IOException first = new IOException("first close failure");
+ IOException second = new IOException("second close failure");
+ doThrow(first).when(closed).close();
+ doThrow(second).when(unclosed).close();
+ manager.setReaderForTest(a.getTsFileID(), true, closed);
+ manager.setReaderForTest(a.getTsFileID(), false, unclosed);
+ IOException aggregate =
+ assertThrows(IOException.class, () ->
manager.closeFileAndRemoveReader(a.getTsFileID()));
+ assertSame(first, aggregate.getCause());
+ assertSame(second, aggregate.getSuppressed()[0]);
+ assertEquals(0, first.getSuppressed().length);
+ assertEquals(0, second.getSuppressed().length);
+ assertEmpty();
+ }
+
+ @Test
+ public void clearReopensAdmissionAfterSnapshotOrReclamationError() throws
Exception {
+ Field internal = FileReaderManager.class.getDeclaredField("internal");
+ internal.setAccessible(true);
+ Object original = internal.get(manager);
+ for (int failureCall : new int[] {1, 2}) {
+ Map<TsFileID, Object> failing =
+ new HashMap<TsFileID, Object>() {
+ private int calls;
+
+ @Override
+ public Collection<Object> values() {
+ if (++calls == failureCall) {
+ throw new AssertionError("injected registry traversal
failure");
+ }
+ return super.values();
+ }
+ };
+ internal.set(manager, failing);
+ try {
+ assertThrows(AssertionError.class,
manager::closeAndRemoveAllOpenedReaders);
+ assertEquals(false, field("clearing"));
+ } finally {
+ internal.set(manager, original);
+ }
+ }
+ TsFileResource a = file(1);
+ assertNotNull(get(a, true));
+ }
+
+ @Test
+ public void pinnedEmptyEntryCannotBeReplaced() throws Exception {
+ TsFileResource a = file(1);
+ Object registry = field("internal");
+ Method pin =
+ FileReaderManager.class.getDeclaredMethod(
+ "pin", Map.class, Object.class, boolean.class, boolean.class);
+ pin.setAccessible(true);
+ Object entry = pin.invoke(manager, registry, a.getTsFileID(), true, true);
+ try {
+ manager.closeFileAndRemoveReader(a.getTsFileID());
+ done(workers.submit(() -> get(a, true)));
+ assertSame(entry, ((Map<?, ?>) registry).get(a.getTsFileID()));
+ manager.closeFileAndRemoveReader(a.getTsFileID());
+ assertSame(entry, ((Map<?, ?>) registry).get(a.getTsFileID()));
+ } finally {
+ Method unpin =
+ FileReaderManager.class.getDeclaredMethod(
+ "unpin", Map.class, Object.class, entry.getClass());
+ unpin.setAccessible(true);
+ unpin.invoke(manager, registry, a.getTsFileID(), entry);
+ }
+ assertEmpty();
+ }
+
+ @Test
+ public void failedRegistrationReleasesResourceLock() throws Exception {
+ TsFileResource resource = mock(TsFileResource.class);
+ org.mockito.Mockito.when(resource.getTsFileID())
+ .thenThrow(new IllegalStateException("injected"));
+ try {
+ manager.increaseFileReaderReference(resource, true);
+ fail("registration failure lost");
+ } catch (IllegalStateException expected) {
+ verify(resource).readLock();
+ verify(resource).readUnlock();
+ }
+ assertEmpty();
+ }
+
+ @Test
+ public void interruptedClearWaitsForExistingOperation() throws Exception {
+ TsFileResource a = file(1);
+ try (Gate gate = new Gate()) {
+ Future<?> open =
+ workers.submit(
+ () -> manager.get(a.getTsFilePath(), a.getTsFileID(), true, n ->
gate.block()));
+ gate.await();
+ AtomicReference<Thread> clearThread = new AtomicReference<>();
+ CountDownLatch started = new CountDownLatch(1);
+ Future<Boolean> clear =
+ workers.submit(
+ () -> {
+ clearThread.set(Thread.currentThread());
+ started.countDown();
+ manager.closeAndRemoveAllOpenedReaders();
+ return Thread.interrupted();
+ });
+ assertTrue(started.await(10, TimeUnit.SECONDS));
+ awaitClearDrain(clearThread.get());
+ clearThread.get().interrupt();
+ Future<?> acquire = acquireDuringClear(a, false);
+ assertFalse(clear.isDone());
+ assertFalse(acquire.isDone());
+ gate.close();
+ done(open);
+ assertTrue(clear.get(10, TimeUnit.SECONDS));
+ done(acquire);
+ assertFalse(manager.contains(a, true));
+ assertTrue(manager.contains(a, false));
+ }
+ }
+
+ private Object field(String name) throws Exception {
+ Field field = FileReaderManager.class.getDeclaredField(name);
+ field.setAccessible(true);
+ return field.get(manager);
+ }
+
+ private void assertEmpty() throws Exception {
+ assertEquals(0, field("pins"));
+ assertTrue(((Map<?, ?>) field("internal")).isEmpty());
+ assertTrue(((Map<?, ?>) field("external")).isEmpty());
+ for (String name : new String[] {"closedCount", "unclosedCount",
"externalCount"}) {
+ assertEquals(0, ((java.util.concurrent.atomic.AtomicInteger)
field(name)).get());
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileLockTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileLockTest.java
new file mode 100644
index 00000000000..d574070003a
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileLockTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.storageengine.dataregion.tsfile;
+
+import org.junit.Test;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class TsFileLockTest {
+ @Test
+ public void interruptedWriterStillWaitsForReaders() throws Exception {
+ interruptedWaiter(true);
+ }
+
+ @Test
+ public void interruptedReaderStillWaitsForWriter() throws Exception {
+ interruptedWaiter(false);
+ }
+
+ private void interruptedWaiter(boolean writer) throws Exception {
+ TsFileLock lock = new TsFileLock();
+ ExecutorService worker = Executors.newSingleThreadExecutor();
+ AtomicReference<Thread> waiter = new AtomicReference<>();
+ CountDownLatch started = new CountDownLatch(1);
+ if (writer) {
+ lock.readLock();
+ } else {
+ lock.writeLock();
+ }
+ boolean held = true;
+ try {
+ Future<Boolean> acquired =
+ worker.submit(
+ () -> {
+ waiter.set(Thread.currentThread());
+ started.countDown();
+ if (writer) {
+ lock.writeLock();
+ } else {
+ lock.readLock();
+ }
+ try {
+ return Thread.currentThread().isInterrupted();
+ } finally {
+ if (writer) {
+ lock.writeUnlock();
+ } else {
+ lock.readUnlock();
+ }
+ }
+ });
+ assertTrue(started.await(10, TimeUnit.SECONDS));
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (waiter.get().getState() != Thread.State.WAITING
+ && waiter.get().getState() != Thread.State.TIMED_WAITING) {
+ if (System.nanoTime() > deadline) {
+ fail("waiter never reached the resource lock");
+ }
+ Thread.sleep(1);
+ }
+ waiter.get().interrupt();
+ assertThrows(TimeoutException.class, () -> acquired.get(100,
TimeUnit.MILLISECONDS));
+ if (writer) {
+ lock.readUnlock();
+ } else {
+ lock.writeUnlock();
+ }
+ held = false;
+ assertTrue(acquired.get(10, TimeUnit.SECONDS));
+ } finally {
+ if (held) {
+ if (writer) {
+ lock.readUnlock();
+ } else {
+ lock.writeUnlock();
+ }
+ }
+ worker.shutdownNow();
+ assertTrue(worker.awaitTermination(10, TimeUnit.SECONDS));
+ }
+ }
+}
