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 <[email protected]> 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) {
