You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2021/11/13 04:32:56 UTC
[iotdb] branch master updated: [IoTDB-1865]【Cluster】Compaction is blocking when removing old files (#4365)
This is an automated email from the ASF dual-hosted git repository.
qiaojialin 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 4f615ea [IoTDB-1865]【Cluster】Compaction is blocking when removing old files (#4365)
4f615ea is described below
commit 4f615ea1a67a628266379365e23ce0d8ee0310e2
Author: Potato <TX...@gmail.com>
AuthorDate: Sat Nov 13 12:32:21 2021 +0800
[IoTDB-1865]【Cluster】Compaction is blocking when removing old files (#4365)
---
.../iotdb/cluster/query/reader/DataSourceInfo.java | 5 +--
.../query/reader/mult/MultDataSourceInfo.java | 5 +--
.../iotdb/db/query/control/QueryFileManager.java | 38 ++++++++++++----------
3 files changed, 27 insertions(+), 21 deletions(-)
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/DataSourceInfo.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/DataSourceInfo.java
index 02be077..57eb9d8 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/DataSourceInfo.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/DataSourceInfo.java
@@ -95,8 +95,6 @@ public class DataSourceInfo {
if (newReaderId != null) {
logger.debug("get a readerId {} for {} from {}", newReaderId, request.path, node);
if (newReaderId != -1) {
- // register the node so the remote resources can be released
- context.registerRemoteNode(node, partitionGroup.getHeader());
this.readerId = newReaderId;
this.curSource = node;
this.curPos = nextNodePos;
@@ -114,6 +112,9 @@ public class DataSourceInfo {
logger.error("Cannot query {} from {}", this.request.path, node, e);
} catch (Exception e) {
logger.error("Cannot query {} from {}", this.request.path, node, e);
+ } finally {
+ // register the node so the remote resources can be released
+ context.registerRemoteNode(node, partitionGroup.getHeader());
}
nextNodePos = (nextNodePos + 1) % this.nodes.size();
if (nextNodePos == this.curPos) {
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultDataSourceInfo.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultDataSourceInfo.java
index e849f04..1521ba3 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultDataSourceInfo.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultDataSourceInfo.java
@@ -100,8 +100,6 @@ public class MultDataSourceInfo {
if (newReaderId != null) {
logger.debug("get a readerId {} for {} from {}", newReaderId, request.path, node);
if (newReaderId != -1) {
- // register the node so the remote resources can be released
- context.registerRemoteNode(node, partitionGroup.getHeader());
this.readerId = newReaderId;
this.curSource = node;
this.curPos = nextNodePos;
@@ -119,6 +117,9 @@ public class MultDataSourceInfo {
logger.error("Cannot query {} from {}", this.request.path, node, e);
} catch (Exception e) {
logger.error("Cannot query {} from {}", this.request.path, node, e);
+ } finally {
+ // register the node so the remote resources can be released
+ context.registerRemoteNode(node, partitionGroup.getHeader());
}
nextNodePos = (nextNodePos + 1) % this.nodes.size();
if (nextNodePos == this.curPos) {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/control/QueryFileManager.java b/server/src/main/java/org/apache/iotdb/db/query/control/QueryFileManager.java
index d75bbe3..5fbe3ce 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/control/QueryFileManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/control/QueryFileManager.java
@@ -21,11 +21,9 @@ package org.apache.iotdb.db.query.control;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
/**
@@ -35,10 +33,10 @@ import java.util.concurrent.ConcurrentHashMap;
*/
public class QueryFileManager {
- /** Map<queryId, Set<filePaths>> */
- private Map<Long, Set<TsFileResource>> sealedFilePathsMap;
+ /** Map<queryId, Map<filePath,filePath>> */
+ private Map<Long, Map<TsFileResource, TsFileResource>> sealedFilePathsMap;
- private Map<Long, Set<TsFileResource>> unsealedFilePathsMap;
+ private Map<Long, Map<TsFileResource, TsFileResource>> unsealedFilePathsMap;
QueryFileManager() {
sealedFilePathsMap = new ConcurrentHashMap<>();
@@ -50,8 +48,8 @@ public class QueryFileManager {
* must be invoked.
*/
void addQueryId(long queryId) {
- sealedFilePathsMap.computeIfAbsent(queryId, x -> new HashSet<>());
- unsealedFilePathsMap.computeIfAbsent(queryId, x -> new HashSet<>());
+ sealedFilePathsMap.computeIfAbsent(queryId, x -> new ConcurrentHashMap<>());
+ unsealedFilePathsMap.computeIfAbsent(queryId, x -> new ConcurrentHashMap<>());
}
/** Add the unique file paths to sealedFilePathsMap and unsealedFilePathsMap. */
@@ -73,10 +71,10 @@ public class QueryFileManager {
// this file may be deleted just before we lock it
if (tsFileResource.isDeleted()) {
- Map<Long, Set<TsFileResource>> pathMap =
+ Map<Long, Map<TsFileResource, TsFileResource>> pathMap =
!isClosed ? unsealedFilePathsMap : sealedFilePathsMap;
// This resource may be removed by other threads of this query.
- if (pathMap.get(queryId).remove(tsFileResource)) {
+ if (pathMap.get(queryId).remove(tsFileResource) != null) {
FileReaderManager.getInstance().decreaseFileReaderReference(tsFileResource, isClosed);
}
iterator.remove();
@@ -93,7 +91,7 @@ public class QueryFileManager {
sealedFilePathsMap.computeIfPresent(
queryId,
(k, v) -> {
- for (TsFileResource tsFile : v) {
+ for (TsFileResource tsFile : v.keySet()) {
FileReaderManager.getInstance().decreaseFileReaderReference(tsFile, true);
}
return null;
@@ -101,7 +99,7 @@ public class QueryFileManager {
unsealedFilePathsMap.computeIfPresent(
queryId,
(k, v) -> {
- for (TsFileResource tsFile : v) {
+ for (TsFileResource tsFile : v.keySet()) {
FileReaderManager.getInstance().decreaseFileReaderReference(tsFile, false);
}
return null;
@@ -115,11 +113,17 @@ public class QueryFileManager {
* not return null.
*/
void addFilePathToMap(long queryId, TsFileResource tsFile, boolean isClosed) {
- Map<Long, Set<TsFileResource>> pathMap = isClosed ? sealedFilePathsMap : unsealedFilePathsMap;
- // TODO this is not an atomic operation, is there concurrent problem?
- if (!pathMap.get(queryId).contains(tsFile)) {
- pathMap.get(queryId).add(tsFile);
- FileReaderManager.getInstance().increaseFileReaderReference(tsFile, isClosed);
- }
+ Map<Long, Map<TsFileResource, TsFileResource>> pathMap =
+ isClosed ? sealedFilePathsMap : unsealedFilePathsMap;
+ // Although there are no concurrency issues here at the moment, I've implemented thread-safe
+ // code here to avoid leaving holes for future newcomers.
+ pathMap
+ .get(queryId)
+ .computeIfAbsent(
+ tsFile,
+ k -> {
+ FileReaderManager.getInstance().increaseFileReaderReference(tsFile, isClosed);
+ return k;
+ });
}
}