This is an automated email from the ASF dual-hosted git repository.
HappenLee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new ce6c03a95c6 [Fix](profile) Prevent concurrent historical profile
loaders (#67012)
ce6c03a95c6 is described below
commit ce6c03a95c6812c1683c9762a19f57bffdaf1f29
Author: linrrarity <[email protected]>
AuthorDate: Mon Aug 24 11:21:06 2026 +0800
[Fix](profile) Prevent concurrent historical profile loaders (#67012)
### What problem does this PR solve?
Problem Summary:
`ProfileManager` periodically calls
`loadProfilesFromStorageIfFirstTime(false)` after FE becomes ready.
The method previously tracked only whether historical profiles had
finished loading. Because the storage scan runs asynchronously, repeated
calls made before the first scan completed could each create a new
`profile-loader` thread. This resulted in concurrent and duplicated
profile storage scans.
This PR adds an atomic loading flag to ensure that only one historical
profile loader can run at a time:
- Atomically acquire the loading state before starting the loader.
- Ignore repeated calls while a loader is running.
- Clear the loading state after success or failure so failed loads can
be retried.
- Recheck the loaded state after acquiring the flag to handle the race
where the previous loader finishes between the initial check and the CAS
operation.
The existing loading logic and synchronous waiting behavior remain
unchanged.
---
.../doris/common/profile/ProfileManager.java | 58 +++++++---------
.../doris/common/profile/ProfileManagerTest.java | 81 +++++++++++++++++++---
2 files changed, 97 insertions(+), 42 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 927d5863d5a..232096be42b 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;
@@ -573,23 +578,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();
- if (files != null) {
- for (File file : files) {
- if (file.isFile()) {
- res.add(file.getAbsolutePath());
- }
+ 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;
@@ -599,7 +602,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;
}
@@ -653,15 +656,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);
}
});
@@ -1119,12 +1118,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 7c681814f60..8ed48a2d63e 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.planner.PlanFragmentId;
import org.apache.doris.thrift.TCounter;
@@ -54,7 +55,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 {
@@ -77,7 +81,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;
}
@@ -518,7 +522,7 @@ class ProfileManagerTest {
@Test
void testLoadProfile() throws IOException {
- profileManager.isProfileLoaded = false;
+ profileManager.profileLoadStatus.set(ProfileLoadStatus.UNLOADED);
try {
// Create some test profile files
@@ -529,7 +533,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) {
@@ -682,7 +686,7 @@ class ProfileManagerTest {
}
// Execute deletion
- profileManager.isProfileLoaded = true;
+ profileManager.profileLoadStatus.set(ProfileLoadStatus.LOADED);
profileManager.deleteOutdatedProfilesFromStorage();
// Verify correct profiles were deleted
@@ -764,7 +768,7 @@ class ProfileManagerTest {
}
// Delete broken profiles
- profileManager.isProfileLoaded = true;
+ profileManager.profileLoadStatus.set(ProfileLoadStatus.LOADED);
profileManager.deleteBrokenProfiles();
// Verify normal files still exist
@@ -799,6 +803,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
@@ -819,7 +880,7 @@ class ProfileManagerTest {
}
// Trigger cleanup
- profileManager.isProfileLoaded = true;
+ profileManager.profileLoadStatus.set(ProfileLoadStatus.LOADED);
profileManager.deleteOutdatedProfilesFromStorage();
// Verify number of profiles is within limits
@@ -844,7 +905,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
@@ -1018,7 +1079,7 @@ class ProfileManagerTest {
}
// Trigger cleanup - should move old profiles to pending and
possibly archive
- profileManager.isProfileLoaded = true;
+ profileManager.profileLoadStatus.set(ProfileLoadStatus.LOADED);
profileManager.deleteOutdatedProfilesFromStorage();
// Verify storage directory only has max allowed profiles
@@ -1086,7 +1147,7 @@ class ProfileManagerTest {
}
// Trigger cleanup - should directly delete old profiles
- profileManager.isProfileLoaded = true;
+ profileManager.profileLoadStatus.set(ProfileLoadStatus.LOADED);
profileManager.deleteOutdatedProfilesFromStorage();
// Verify delete invocations
@@ -1143,7 +1204,7 @@ class ProfileManagerTest {
// Simulate periodic cleanup via runAfterCatalogReady
// Note: The first call will trigger cleanup since
lastArchiveCleanupTime is 0
- profileManager.isProfileLoaded = true;
+ profileManager.profileLoadStatus.set(ProfileLoadStatus.LOADED);
profileManager.runAfterCatalogReady();
// Verify that the old archive was deleted by runAfterCatalogReady
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
