You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2021/06/04 09:04:50 UTC
[iotdb] 01/01: format code
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch ShowLockInfo
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3932c9852b8ecb8092ee8fde73d84936121b2ac1
Author: JackieTien97 <Ja...@foxmail.com>
AuthorDate: Fri Jun 4 17:04:00 2021 +0800
format code
---
.../antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4 | 5 ++
.../engine/storagegroup/StorageGroupProcessor.java | 72 ++++++++++++----------
.../virtualSg/VirtualStorageGroupManager.java | 14 ++---
3 files changed, 52 insertions(+), 39 deletions(-)
diff --git a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4 b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4
index 18539c7..9d05e7a 100644
--- a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4
+++ b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4
@@ -76,6 +76,7 @@ statement
| SHOW VERSION #showVersion
| SHOW LATEST? TIMESERIES prefixPath? showWhereClause? limitClause? #showTimeseries
| SHOW STORAGE GROUP prefixPath? #showStorageGroup
+ | SHOW LOCK INFO prefixPath? #showLockInfo
| SHOW CHILD PATHS prefixPath? #showChildPaths
| SHOW CHILD NODES prefixPath? #showChildNodes
| SHOW DEVICES prefixPath? (WITH STORAGE GROUP)? limitClause? #showDevices
@@ -1235,6 +1236,10 @@ ANY
: A N Y
;
+LOCK
+ : L O C K
+ ;
+
//============================
// End of the keywords list
//============================
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index 26261df..1e75eea 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -272,6 +272,8 @@ public class StorageGroupProcessor {
// DEFAULT_POOL_TRIM_INTERVAL_MILLIS
private long timeWhenPoolNotEmpty = Long.MAX_VALUE;
+ private String insertWriteLockHolder = "";
+
/** get the direct byte buffer from pool, each fetch contains two ByteBuffer */
public ByteBuffer[] getWalDirectByteBuffer() {
ByteBuffer[] res = new ByteBuffer[2];
@@ -798,7 +800,7 @@ public class StorageGroupProcessor {
if (!isAlive(insertRowPlan.getTime())) {
throw new OutOfTTLException(insertRowPlan.getTime(), (System.currentTimeMillis() - dataTTL));
}
- writeLock();
+ writeLock("InsertRow");
try {
// init map
long timePartitionId = StorageEngine.getTimePartition(insertRowPlan.getTime());
@@ -840,7 +842,7 @@ public class StorageGroupProcessor {
public void insertTablet(InsertTabletPlan insertTabletPlan)
throws BatchProcessException, TriggerExecutionException {
- writeLock();
+ writeLock("insertTablet");
try {
TSStatus[] results = new TSStatus[insertTabletPlan.getRowCount()];
Arrays.fill(results, RpcUtils.SUCCESS_STATUS);
@@ -1122,7 +1124,7 @@ public class StorageGroupProcessor {
}
public void submitAFlushTaskWhenShouldFlush(TsFileProcessor tsFileProcessor) {
- writeLock();
+ writeLock("submitAFlushTaskWhenShouldFlush");
try {
// check memtable size and may asyncTryToFlush the work memtable
if (tsFileProcessor.shouldFlush()) {
@@ -1342,9 +1344,9 @@ public class StorageGroupProcessor {
"{} will close all files for deleting data folder {}",
logicalStorageGroupName + "-" + virtualStorageGroupId,
systemDir);
- writeLock();
- syncCloseAllWorkingTsFileProcessors();
+ writeLock("deleteFolder");
try {
+ syncCloseAllWorkingTsFileProcessors();
File storageGroupFolder =
SystemFileFactory.INSTANCE.getFile(systemDir, virtualStorageGroupId);
if (storageGroupFolder.exists()) {
@@ -1385,21 +1387,16 @@ public class StorageGroupProcessor {
logger.info(
"{} will close all files for deleting data files",
logicalStorageGroupName + "-" + virtualStorageGroupId);
- writeLock();
- syncCloseAllWorkingTsFileProcessors();
- // normally, mergingModification is just need to be closed by after a merge task is finished.
- // we close it here just for IT test.
- if (this.tsFileManagement.mergingModification != null) {
- try {
+ writeLock("syncDeleteDataFiles");
+ try {
+
+ syncCloseAllWorkingTsFileProcessors();
+ // normally, mergingModification is just need to be closed by after a merge task is finished.
+ // we close it here just for IT test.
+ if (this.tsFileManagement.mergingModification != null) {
this.tsFileManagement.mergingModification.close();
- } catch (IOException e) {
- logger.error(
- "Cannot close the mergingMod file {}",
- this.tsFileManagement.mergingModification.getFilePath(),
- e);
}
- }
- try {
+
closeAllResources();
List<String> folder = DirectoryManager.getInstance().getAllSequenceFileFolders();
folder.addAll(DirectoryManager.getInstance().getAllUnSequenceFileFolders());
@@ -1411,6 +1408,11 @@ public class StorageGroupProcessor {
this.partitionLatestFlushedTimeForEachDevice.clear();
this.globalLatestFlushedTimeForEachDevice.clear();
this.latestTimeForEachDevice.clear();
+ } catch (IOException e) {
+ logger.error(
+ "Cannot close the mergingMod file {}",
+ this.tsFileManagement.mergingModification.getFilePath(),
+ e);
} finally {
writeUnlock();
}
@@ -1462,7 +1464,7 @@ public class StorageGroupProcessor {
return;
}
- writeLock();
+ writeLock("checkFileTTL");
try {
// prevent new merges and queries from choosing this file
resource.setDeleted(true);
@@ -1522,7 +1524,7 @@ public class StorageGroupProcessor {
}
public void asyncCloseAllWorkingTsFileProcessors() {
- writeLock();
+ writeLock("asyncCloseAllWorkingTsFileProcessors");
try {
logger.info(
"async force close all files in storage group: {}",
@@ -1543,7 +1545,7 @@ public class StorageGroupProcessor {
}
public void forceCloseAllWorkingTsFileProcessors() throws TsFileProcessorException {
- writeLock();
+ writeLock("forceCloseAllWorkingTsFileProcessors");
try {
logger.info(
"force close all processors in storage group: {}",
@@ -1612,11 +1614,13 @@ public class StorageGroupProcessor {
insertLock.readLock().unlock();
}
- public void writeLock() {
+ public void writeLock(String holder) {
insertLock.writeLock().lock();
+ insertWriteLockHolder = holder;
}
public void writeUnlock() {
+ insertWriteLockHolder = "";
insertLock.writeLock().unlock();
}
@@ -1705,7 +1709,7 @@ public class StorageGroupProcessor {
// TODO: how to avoid partial deletion?
// FIXME: notice that if we may remove a SGProcessor out of memory, we need to close all opened
// mod files in mergingModification, sequenceFileList, and unsequenceFileList
- writeLock();
+ writeLock("delete");
// record files which are updated so that we can roll back them in case of exception
List<ModificationFile> updatedModFiles = new ArrayList<>();
@@ -2049,7 +2053,7 @@ public class StorageGroupProcessor {
upgradeFileCount.getAndAdd(-1);
// load all upgraded resources in this sg to tsFileManagement
if (upgradeFileCount.get() == 0) {
- writeLock();
+ writeLock("upgradeTsFileResourceCallBack");
try {
loadUpgradedResources(upgradeSeqFileList, true);
loadUpgradedResources(upgradeUnseqFileList, false);
@@ -2111,7 +2115,7 @@ public class StorageGroupProcessor {
}
public void merge(boolean isFullMerge) {
- writeLock();
+ writeLock("merge");
try {
for (long timePartitionId : partitionLatestFlushedTimeForEachDevice.keySet()) {
executeCompaction(timePartitionId, isFullMerge);
@@ -2135,7 +2139,7 @@ public class StorageGroupProcessor {
public void loadNewTsFileForSync(TsFileResource newTsFileResource) throws LoadFileException {
File tsfileToBeInserted = newTsFileResource.getTsFile();
long newFilePartitionId = newTsFileResource.getTimePartitionWithCheck();
- writeLock();
+ writeLock("loadNewTsFileForSync");
try {
if (loadTsFileByType(
LoadTsFileType.LOAD_SEQUENCE,
@@ -2205,7 +2209,7 @@ public class StorageGroupProcessor {
public void loadNewTsFile(TsFileResource newTsFileResource) throws LoadFileException {
File tsfileToBeInserted = newTsFileResource.getTsFile();
long newFilePartitionId = newTsFileResource.getTimePartitionWithCheck();
- writeLock();
+ writeLock("loadNewTsFile");
try {
List<TsFileResource> sequenceList = tsFileManagement.getTsFileList(true);
@@ -2381,7 +2385,7 @@ public class StorageGroupProcessor {
*/
@SuppressWarnings("unused")
public void removeFullyOverlapFiles(TsFileResource resource) {
- writeLock();
+ writeLock("removeFullyOverlapFiles");
try {
Iterator<TsFileResource> iterator = tsFileManagement.getIterator(true);
removeFullyOverlapFiles(resource, iterator, true);
@@ -2654,7 +2658,7 @@ public class StorageGroupProcessor {
* module.
*/
public boolean deleteTsfile(File tsfieToBeDeleted) {
- writeLock();
+ writeLock("deleteTsfile");
TsFileResource tsFileResourceToBeDeleted = null;
try {
Iterator<TsFileResource> sequenceIterator = tsFileManagement.getIterator(true);
@@ -2708,7 +2712,7 @@ public class StorageGroupProcessor {
* @return whether the file to be moved exists. @UsedBy load external tsfile module.
*/
public boolean moveTsfile(File fileToBeMoved, File targetDir) {
- writeLock();
+ writeLock("moveTsfile");
TsFileResource tsFileResourceToBeMoved = null;
try {
Iterator<TsFileResource> sequenceIterator = tsFileManagement.getIterator(true);
@@ -2839,7 +2843,7 @@ public class StorageGroupProcessor {
/** remove all partitions that satisfy a filter. */
public void removePartitions(TimePartitionFilter filter) {
// this requires blocking all other activities
- writeLock();
+ writeLock("removePartitions");
try {
// abort ongoing comapctions and merges
CompactionMergeTaskPoolManager.getInstance().abortCompaction(logicalStorageGroupName);
@@ -2896,7 +2900,7 @@ public class StorageGroupProcessor {
public void insert(InsertRowsOfOneDevicePlan insertRowsOfOneDevicePlan)
throws WriteProcessException, TriggerExecutionException {
- writeLock();
+ writeLock("InsertRowsOfOneDevice");
try {
boolean isSequence = false;
InsertRowPlan[] rowPlans = insertRowsOfOneDevicePlan.getRowPlans();
@@ -2989,4 +2993,8 @@ public class StorageGroupProcessor {
boolean satisfy(String storageGroupName, long timePartitionId);
}
+
+ public String getInsertWriteLockHolder() {
+ return insertWriteLockHolder;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/VirtualStorageGroupManager.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/VirtualStorageGroupManager.java
index e8d1f77..b7c86b1 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/VirtualStorageGroupManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/VirtualStorageGroupManager.java
@@ -201,7 +201,7 @@ public class VirtualStorageGroupManager {
isSeq);
}
- processor.writeLock();
+ processor.writeLock("VirtualCloseStorageGroupProcessor-204");
try {
if (isSeq) {
// to avoid concurrent modification problem, we need a new array list
@@ -239,13 +239,13 @@ public class VirtualStorageGroupManager {
processor.getVirtualStorageGroupId() + "-" + processor.getLogicalStorageGroupName(),
isSeq,
partitionId);
- processor.writeLock();
- // to avoid concurrent modification problem, we need a new array list
- List<TsFileProcessor> processors =
- isSeq
- ? new ArrayList<>(processor.getWorkSequenceTsFileProcessors())
- : new ArrayList<>(processor.getWorkUnsequenceTsFileProcessors());
+ processor.writeLock("VirtualCloseStorageGroupProcessor-242");
try {
+ // to avoid concurrent modification problem, we need a new array list
+ List<TsFileProcessor> processors =
+ isSeq
+ ? new ArrayList<>(processor.getWorkSequenceTsFileProcessors())
+ : new ArrayList<>(processor.getWorkUnsequenceTsFileProcessors());
for (TsFileProcessor tsfileProcessor : processors) {
if (tsfileProcessor.getTimeRangeId() == partitionId) {
if (isSync) {