This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new bf4fdf900ca branch-4.1: [Fix](profile) Prevent concurrent historical
profile loaders (#67012) (#67069)
bf4fdf900ca is described below
commit bf4fdf900ca28c6f5f61c7dcd1d81a451c7b7fc4
Author: linrrarity <[email protected]>
AuthorDate: Mon Sep 7 18:59:35 2026 +0800
branch-4.1: [Fix](profile) Prevent concurrent historical profile loaders
(#67012) (#67069)
pick: https://github.com/apache/doris/pull/67012
---
.../doris/common/profile/ProfileManager.java | 50 +++++++--------
.../doris/common/profile/ProfileManagerTest.java | 75 ++++++++++++++++++++--
2 files changed, 91 insertions(+), 34 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/profile/ProfileManager.java
b/fe/fe-core/src/main/java/org/apache/doris/common/profile/ProfileManager.java
index b9b30fff647..731e7997e0a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/common/profile/ProfileManager.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/common/profile/ProfileManager.java
@@ -56,6 +56,7 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
@@ -122,10 +123,14 @@ public class ProfileManager extends MasterDaemon {
}
}
- // this variable is assigned to true the first time the profile is loaded
from storage
- // no further write operation, so no data race
- private final ReentrantReadWriteLock isProfileLoadedLock = new
ReentrantReadWriteLock();
- volatile boolean isProfileLoaded = false;
+ enum ProfileLoadStatus {
+ UNLOADED,
+ LOADING,
+ LOADED
+ }
+
+ final AtomicReference<ProfileLoadStatus> profileLoadStatus =
+ new AtomicReference<>(ProfileLoadStatus.UNLOADED);
// only protect queryIdDeque; queryIdToProfileMap is concurrent, no need
to protect
private ReentrantReadWriteLock lock;
@@ -551,21 +556,21 @@ public class ProfileManager extends MasterDaemon {
// string will contain profile id and its storage timestamp
protected List<String> getOnStorageProfileInfos() {
List<String> res = Lists.newArrayList();
- try {
- File profileDir = new File(PROFILE_STORAGE_PATH);
- if (!profileDir.exists()) {
- LOG.warn("Profile storage directory {} does not exist",
PROFILE_STORAGE_PATH);
- return res;
- }
+ File profileDir = new File(PROFILE_STORAGE_PATH);
+ if (!profileDir.exists()) {
+ LOG.warn("Profile storage directory {} does not exist",
PROFILE_STORAGE_PATH);
+ return res;
+ }
- File[] files = profileDir.listFiles();
+ File[] files = profileDir.listFiles();
+ if (files != null) {
for (File file : files) {
if (file.isFile()) {
res.add(file.getAbsolutePath());
}
}
- } catch (Exception e) {
- LOG.error("Failed to get profile meta from storage", e);
+ } else {
+ throw new IllegalStateException("Failed to list profile storage
directory: " + PROFILE_STORAGE_PATH);
}
return res;
@@ -575,7 +580,7 @@ public class ProfileManager extends MasterDaemon {
// deserialize to an object Profile
// push them to memory structure of ProfileManager for index
protected void loadProfilesFromStorageIfFirstTime(boolean sync) {
- if (checkIfProfileLoaded()) {
+ if (!profileLoadStatus.compareAndSet(ProfileLoadStatus.UNLOADED,
ProfileLoadStatus.LOADING)) {
return;
}
@@ -629,15 +634,11 @@ public class ProfileManager extends MasterDaemon {
LOG.info("Load profiles into memory finished, costs {}ms",
System.currentTimeMillis() - startTime);
- // Set isProfileLoaded to true with write lock
- isProfileLoadedLock.writeLock().lock();
- try {
- this.isProfileLoaded = true;
- } finally {
- isProfileLoadedLock.writeLock().unlock();
- }
+ profileLoadStatus.set(ProfileLoadStatus.LOADED);
} catch (Exception e) {
LOG.error("Failed to load query profile from storage", e);
+ } finally {
+ profileLoadStatus.compareAndSet(ProfileLoadStatus.LOADING,
ProfileLoadStatus.UNLOADED);
}
});
@@ -1099,12 +1100,7 @@ public class ProfileManager extends MasterDaemon {
}
private boolean checkIfProfileLoaded() {
- isProfileLoadedLock.readLock().lock();
- try {
- return isProfileLoaded;
- } finally {
- isProfileLoadedLock.readLock().unlock();
- }
+ return profileLoadStatus.get() == ProfileLoadStatus.LOADED;
}
public void removeProfile(String profileId) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/common/profile/ProfileManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/common/profile/ProfileManagerTest.java
index 21096556a8f..65778b8c512 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/common/profile/ProfileManagerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/common/profile/ProfileManagerTest.java
@@ -19,6 +19,7 @@ package org.apache.doris.common.profile;
import org.apache.doris.common.Config;
import org.apache.doris.common.profile.ProfileManager.ProfileElement;
+import org.apache.doris.common.profile.ProfileManager.ProfileLoadStatus;
import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.thrift.TUniqueId;
@@ -44,7 +45,10 @@ import java.util.PriorityQueue;
import java.util.Random;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
@ResourceLock("global")
class ProfileManagerTest {
@@ -66,7 +70,7 @@ class ProfileManagerTest {
originalPath = ProfileManager.PROFILE_STORAGE_PATH;
ProfileManager.PROFILE_STORAGE_PATH = tempDir.getAbsolutePath();
profileManager.cleanProfile();
- profileManager.isProfileLoaded = false;
+ profileManager.profileLoadStatus.set(ProfileLoadStatus.UNLOADED);
originMaxProfiles = Config.max_query_profile_num;
}
@@ -370,7 +374,7 @@ class ProfileManagerTest {
@Test
void testLoadProfile() throws IOException {
- profileManager.isProfileLoaded = false;
+ profileManager.profileLoadStatus.set(ProfileLoadStatus.UNLOADED);
try {
// Create some test profile files
@@ -381,7 +385,7 @@ class ProfileManagerTest {
}
profileManager.loadProfilesFromStorageIfFirstTime(true);
- Assertions.assertTrue(profileManager.isProfileLoaded);
+ Assertions.assertEquals(ProfileLoadStatus.LOADED,
profileManager.profileLoadStatus.get());
Assertions.assertEquals(30,
profileManager.queryIdToProfileMap.size());
Assertions.assertEquals(0,
profileManager.queryIdToExecutionProfiles.size());
} catch (InterruptedException e) {
@@ -548,7 +552,7 @@ class ProfileManagerTest {
}
// Execute deletion
- profileManager.isProfileLoaded = true;
+ profileManager.profileLoadStatus.set(ProfileLoadStatus.LOADED);
profileManager.deleteOutdatedProfilesFromStorage();
// Verify correct profiles were deleted
@@ -626,7 +630,7 @@ class ProfileManagerTest {
}
// Delete broken profiles
- profileManager.isProfileLoaded = true;
+ profileManager.profileLoadStatus.set(ProfileLoadStatus.LOADED);
profileManager.deleteBrokenProfiles();
// Verify normal files still exist
@@ -661,6 +665,63 @@ class ProfileManagerTest {
Assertions.assertEquals(numProfiles,
profileManager.queryIdToProfileMap.size());
}
+ @Test
+ public void testOnlyOneProfileLoaderCanRun() throws Exception {
+ CountDownLatch loadStarted = new CountDownLatch(1);
+ CountDownLatch duplicateLoadStarted = new CountDownLatch(1);
+ CountDownLatch allowLoadToFinish = new CountDownLatch(1);
+ AtomicInteger scanCount = new AtomicInteger();
+ ProfileManager manager = new ProfileManager() {
+ @Override
+ protected List<String> getOnStorageProfileInfos() {
+ if (scanCount.incrementAndGet() > 1) {
+ duplicateLoadStarted.countDown();
+ }
+ loadStarted.countDown();
+ try {
+ allowLoadToFinish.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ return Lists.newArrayList();
+ }
+ };
+ Thread initialLoad = new Thread(() ->
manager.loadProfilesFromStorageIfFirstTime(true));
+
+ try {
+ initialLoad.start();
+ Assertions.assertTrue(loadStarted.await(5, TimeUnit.SECONDS));
+
+ for (int i = 0; i < 10; i++) {
+ manager.loadProfilesFromStorageIfFirstTime(false);
+ }
+
+ Assertions.assertFalse(duplicateLoadStarted.await(500,
TimeUnit.MILLISECONDS));
+ Assertions.assertEquals(1, scanCount.get());
+ } finally {
+ allowLoadToFinish.countDown();
+ initialLoad.join(5000);
+ Assertions.assertFalse(initialLoad.isAlive());
+ }
+ }
+
+ @Test
+ public void testProfileLoaderCanRetryAfterFailure() throws IOException {
+ File invalidProfileStorage = new File(tempDir, "not_a_directory");
+ Assertions.assertTrue(invalidProfileStorage.createNewFile());
+ ProfileManager.PROFILE_STORAGE_PATH =
invalidProfileStorage.getAbsolutePath();
+ ProfileManager manager = new ProfileManager();
+
+ manager.loadProfilesFromStorageIfFirstTime(true);
+ Assertions.assertEquals(ProfileLoadStatus.UNLOADED,
manager.profileLoadStatus.get());
+
+ Assertions.assertTrue(invalidProfileStorage.delete());
+ ProfileManager.PROFILE_STORAGE_PATH = tempDir.getAbsolutePath();
+ manager.loadProfilesFromStorageIfFirstTime(true);
+ Assertions.assertEquals(ProfileLoadStatus.LOADED,
manager.profileLoadStatus.get());
+ }
+
@Test
public void testProfileStorageLimit() throws Exception {
// Set small storage limit
@@ -681,7 +742,7 @@ class ProfileManagerTest {
}
// Trigger cleanup
- profileManager.isProfileLoaded = true;
+ profileManager.profileLoadStatus.set(ProfileLoadStatus.LOADED);
profileManager.deleteOutdatedProfilesFromStorage();
// Verify number of profiles is within limits
@@ -706,7 +767,7 @@ class ProfileManagerTest {
brokenFile.createNewFile();
// Trigger cleanup
- profileManager.isProfileLoaded = true;
+ profileManager.profileLoadStatus.set(ProfileLoadStatus.LOADED);
profileManager.deleteBrokenProfiles();
// Verify broken profile is removed but valid one remains
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
