Fix concurrent access during TierManager folder reset

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java

@@ -62,21 +62,21 @@
   /**
    * 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 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 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));
   }

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;
+  }
+}