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 3e47ce442f740..87c5cf002739f 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 seqTiers = new ArrayList<>(); + private volatile List seqTiers = new ArrayList<>(); /** * unSeq folder manager of each storage tier, managing both data directories and multi-dir * strategy */ - private final List unSeqTiers = new ArrayList<>(); + private volatile List unSeqTiers = new ArrayList<>(); - private final List objectTiers = new ArrayList<>(); + private volatile List objectTiers = new ArrayList<>(); /** seq file folder's rawFsPath path -> tier level */ - private final Map seqDir2TierLevel = new HashMap<>(); + private volatile Map seqDir2TierLevel = new HashMap<>(); /** unSeq file folder's rawFsPath path -> tier level */ - private final Map unSeqDir2TierLevel = new HashMap<>(); + private volatile Map unSeqDir2TierLevel = new HashMap<>(); private List objectDirs; @@ -95,6 +95,15 @@ private TierManager() { } public synchronized void initFolders() { + initFolders(seqTiers, unSeqTiers, objectTiers, seqDir2TierLevel, unSeqDir2TierLevel); + } + + private void initFolders( + List seqTiers, + List unSeqTiers, + List objectTiers, + Map seqDir2TierLevel, + Map unSeqDir2TierLevel) { directoryStrategyType = DirectoryStrategyType.fromClassName(config.getMultiDirStrategyClassName()); @@ -204,13 +213,19 @@ public synchronized void initFolders() { public synchronized void resetFolders() { long startTime = System.currentTimeMillis(); - seqTiers.clear(); - unSeqTiers.clear(); - objectTiers.clear(); - seqDir2TierLevel.clear(); - unSeqDir2TierLevel.clear(); - - initFolders(); + List newSeqTiers = new ArrayList<>(); + List newUnSeqTiers = new ArrayList<>(); + List newObjectTiers = new ArrayList<>(); + Map newSeqDir2TierLevel = new HashMap<>(); + Map 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 0000000000000..046e41dce4875 --- /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> 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 accessorCounts = new ArrayList<>(); + List> 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 assertFoldersAvailable(List folders) { + assertTrue(folders.size() >= DATA_DIR_NUM); + return folders; + } +}