This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 9b77b54defc Fix concurrent access during TierManager folder reset
(#18223)
9b77b54defc is described below
commit 9b77b54defcf8ddad0a5be27d99e8ee82b8f7c36
Author: Jiang Tian <[email protected]>
AuthorDate: Thu Jul 16 17:01:58 2026 +0800
Fix concurrent access during TierManager folder reset (#18223)
* fix TierManager may access an empty list during re-initialization
* Fix concurrent access to tier directory maps
---
.../db/storageengine/rescon/disk/TierManager.java | 39 ++++--
.../storageengine/rescon/disk/TierManagerTest.java | 156 +++++++++++++++++++++
2 files changed, 183 insertions(+), 12 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java
index 3e47ce442f7..87c5cf00273 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java
@@ -62,21 +62,21 @@ public class TierManager {
/**
* seq folder manager of each storage tier, managing both data directories
and multi-dir strategy
*/
- private final List<FolderManager> seqTiers = new ArrayList<>();
+ private volatile List<FolderManager> seqTiers = new ArrayList<>();
/**
* unSeq folder manager of each storage tier, managing both data directories
and multi-dir
* strategy
*/
- private final List<FolderManager> unSeqTiers = new ArrayList<>();
+ private volatile List<FolderManager> unSeqTiers = new ArrayList<>();
- private final List<FolderManager> objectTiers = new ArrayList<>();
+ private volatile List<FolderManager> objectTiers = new ArrayList<>();
/** seq file folder's rawFsPath path -> tier level */
- private final Map<String, Integer> seqDir2TierLevel = new HashMap<>();
+ private volatile Map<String, Integer> seqDir2TierLevel = new HashMap<>();
/** unSeq file folder's rawFsPath path -> tier level */
- private final Map<String, Integer> unSeqDir2TierLevel = new HashMap<>();
+ private volatile Map<String, Integer> unSeqDir2TierLevel = new HashMap<>();
private List<String> objectDirs;
@@ -95,6 +95,15 @@ public class TierManager {
}
public synchronized void initFolders() {
+ initFolders(seqTiers, unSeqTiers, objectTiers, seqDir2TierLevel,
unSeqDir2TierLevel);
+ }
+
+ private void initFolders(
+ List<FolderManager> seqTiers,
+ List<FolderManager> unSeqTiers,
+ List<FolderManager> objectTiers,
+ Map<String, Integer> seqDir2TierLevel,
+ Map<String, Integer> unSeqDir2TierLevel) {
directoryStrategyType =
DirectoryStrategyType.fromClassName(config.getMultiDirStrategyClassName());
@@ -204,13 +213,19 @@ public class TierManager {
public synchronized void resetFolders() {
long startTime = System.currentTimeMillis();
- seqTiers.clear();
- unSeqTiers.clear();
- objectTiers.clear();
- seqDir2TierLevel.clear();
- unSeqDir2TierLevel.clear();
-
- initFolders();
+ List<FolderManager> newSeqTiers = new ArrayList<>();
+ List<FolderManager> newUnSeqTiers = new ArrayList<>();
+ List<FolderManager> newObjectTiers = new ArrayList<>();
+ Map<String, Integer> newSeqDir2TierLevel = new HashMap<>();
+ Map<String, Integer> newUnSeqDir2TierLevel = new HashMap<>();
+
+ initFolders(
+ newSeqTiers, newUnSeqTiers, newObjectTiers, newSeqDir2TierLevel,
newUnSeqDir2TierLevel);
+ seqTiers = newSeqTiers;
+ unSeqTiers = newUnSeqTiers;
+ objectTiers = newObjectTiers;
+ seqDir2TierLevel = newSeqDir2TierLevel;
+ unSeqDir2TierLevel = newUnSeqDir2TierLevel;
long endTime = System.currentTimeMillis();
logger.info(StorageEngineMessages.FOLDERS_RESET_SUCCESSFULLY, (endTime -
startTime));
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManagerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManagerTest.java
new file mode 100644
index 00000000000..046e41dce48
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManagerTest.java
@@ -0,0 +1,156 @@
+/*
+ * 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.rescon.disk;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+
+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.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Callable;
+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.AtomicInteger;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+public class TierManagerTest {
+
+ private static final int DATA_DIR_NUM = 16;
+ private static final int RESET_TIMES = 20;
+
+ @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ private String[][] originalTierDataDirs;
+ private TierManager tierManager;
+ private File sequenceFile;
+
+ @Before
+ public void setUp() throws Exception {
+ originalTierDataDirs = config.getTierDataDirs();
+ String[] dataDirs = new String[DATA_DIR_NUM];
+ for (int i = 0; i < DATA_DIR_NUM; i++) {
+ dataDirs[i] = new File(temporaryFolder.getRoot(), "data" +
i).getAbsolutePath();
+ }
+ config.setTierDataDirs(new String[][] {dataDirs});
+ tierManager = TierManager.getInstance();
+ tierManager.resetFolders();
+ sequenceFile =
+ new File(dataDirs[0], IoTDBConstant.SEQUENCE_FOLDER_NAME +
File.separator + "test.tsfile");
+ assertTrue(sequenceFile.createNewFile());
+ }
+
+ @After
+ public void tearDown() {
+ config.setTierDataDirs(originalTierDataDirs);
+ tierManager.resetFolders();
+ }
+
+ @Test
+ public void testConcurrentResetFoldersAndGetNextFolder() throws Exception {
+ List<Callable<?>> folderAccessors =
+ Arrays.asList(
+ () -> tierManager.getNextFolderForTsFile(0, true),
+ () -> tierManager.getNextFolderForTsFile(0, false),
+ tierManager::getNextFolderForObjectFile,
+ tierManager::getNextFolderForCopyToTargetFile,
+ () -> tierManager.getFolderManager(0, true),
+ () -> tierManager.getFolderManager(0, false),
+ () -> {
+ int tiersNum = tierManager.getTiersNum();
+ assertTrue(tiersNum > 0);
+ return tiersNum;
+ },
+ () -> assertFoldersAvailable(tierManager.getAllFilesFolders()),
+ () ->
assertFoldersAvailable(tierManager.getAllLocalFilesFolders()),
+ () ->
assertFoldersAvailable(tierManager.getAllSequenceFileFolders()),
+ () ->
assertFoldersAvailable(tierManager.getAllLocalSequenceFileFolders()),
+ () ->
assertFoldersAvailable(tierManager.getAllUnSequenceFileFolders()),
+ () ->
assertFoldersAvailable(tierManager.getAllLocalUnSequenceFileFolders()),
+ () -> tierManager.getFileTierLevel(sequenceFile));
+ ExecutorService executorService =
Executors.newFixedThreadPool(folderAccessors.size() + 1);
+ CountDownLatch startLatch = new CountDownLatch(1);
+ AtomicBoolean resetting = new AtomicBoolean(true);
+
+ Future<?> resetFuture =
+ executorService.submit(
+ () -> {
+ startLatch.await();
+ try {
+ for (int i = 0; i < RESET_TIMES; i++) {
+ tierManager.resetFolders();
+ }
+ } finally {
+ resetting.set(false);
+ }
+ return null;
+ });
+ List<AtomicInteger> accessorCounts = new ArrayList<>();
+ List<Future<?>> accessorFutures = new ArrayList<>();
+ for (Callable<?> folderAccessor : folderAccessors) {
+ AtomicInteger accessorCount = new AtomicInteger();
+ accessorCounts.add(accessorCount);
+ accessorFutures.add(
+ executorService.submit(
+ () -> {
+ startLatch.await();
+ do {
+ assertNotNull(folderAccessor.call());
+ accessorCount.incrementAndGet();
+ } while (resetting.get());
+ return null;
+ }));
+ }
+
+ startLatch.countDown();
+ try {
+ resetFuture.get(30, TimeUnit.SECONDS);
+ for (Future<?> accessorFuture : accessorFutures) {
+ accessorFuture.get(30, TimeUnit.SECONDS);
+ }
+ for (AtomicInteger accessorCount : accessorCounts) {
+ assertTrue(accessorCount.get() > 0);
+ }
+ } finally {
+ executorService.shutdownNow();
+ assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS));
+ }
+ }
+
+ private List<String> assertFoldersAvailable(List<String> folders) {
+ assertTrue(folders.size() >= DATA_DIR_NUM);
+ return folders;
+ }
+}
