This is an automated email from the ASF dual-hosted git repository.
cecemei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 0ea17adccd4 fix: isMounted() should never be blocked by a lock (#19746)
0ea17adccd4 is described below
commit 0ea17adccd48cc176a97d28bccefecc9448a3deb
Author: Cece Mei <[email protected]>
AuthorDate: Mon Jul 27 10:19:26 2026 -0700
fix: isMounted() should never be blocked by a lock (#19746)
* mount
* mount2
* mount3
---
.../loading/PartialSegmentBundleCacheEntry.java | 19 ++--
.../loading/PartialSegmentMetadataCacheEntry.java | 31 +++---
.../segment/loading/SegmentLocalCacheManager.java | 11 +-
.../SegmentLocalCacheManagerConcurrencyTest.java | 118 +++++++++++++++++++++
4 files changed, 143 insertions(+), 36 deletions(-)
diff --git
a/server/src/main/java/org/apache/druid/segment/loading/PartialSegmentBundleCacheEntry.java
b/server/src/main/java/org/apache/druid/segment/loading/PartialSegmentBundleCacheEntry.java
index c2dc29c7516..a5fe0499828 100644
---
a/server/src/main/java/org/apache/druid/segment/loading/PartialSegmentBundleCacheEntry.java
+++
b/server/src/main/java/org/apache/druid/segment/loading/PartialSegmentBundleCacheEntry.java
@@ -175,8 +175,10 @@ public class PartialSegmentBundleCacheEntry implements
CacheEntry
// metadata + parents safe from drop-time unmap even if the dependency is
statically reserved.
@GuardedBy("entryLock")
private final List<Closeable> dependencyReferences = new ArrayList<>();
- @GuardedBy("entryLock")
- private boolean mounted;
+ // volatile so isMounted() can read it without blocking behind a concurrent
doActualUnmount() holding entryLock
+ // across container eviction. Other reads of this field still go through
entryLock since they're paired with other
+ // guarded state (e.g. doMount's mounted + location check).
+ private volatile boolean mounted;
// Reference-counted gate over the actual cleanup work (evict containers,
release parent holds, unregister from
// metadata). Set on successful mount; unmount() closes the wrapper which
defers running cleanup until all outstanding
@@ -217,13 +219,7 @@ public class PartialSegmentBundleCacheEntry implements
CacheEntry
@Override
public boolean isMounted()
{
- entryLock.lock();
- try {
- return mounted;
- }
- finally {
- entryLock.unlock();
- }
+ return mounted;
}
public SegmentId getSegmentId()
@@ -423,8 +419,11 @@ public class PartialSegmentBundleCacheEntry implements
CacheEntry
location = mountLocation;
holds.addAll(acquired);
dependencyReferences.addAll(acquiredRefs);
- mounted = true;
+ // Must be set before `mounted` is published below: `mounted` is
volatile and isMounted() reads it without
+ // entryLock, so a lock-free caller must never be able to observe
isMounted() == true before the reference
+ // gate exists.
references.set(new
ReferenceCountingCloseableObject<Closeable>(this::doActualUnmount) {});
+ mounted = true;
}
finally {
entryLock.unlock();
diff --git
a/server/src/main/java/org/apache/druid/segment/loading/PartialSegmentMetadataCacheEntry.java
b/server/src/main/java/org/apache/druid/segment/loading/PartialSegmentMetadataCacheEntry.java
index be1d871dc10..ed9bfebc289 100644
---
a/server/src/main/java/org/apache/druid/segment/loading/PartialSegmentMetadataCacheEntry.java
+++
b/server/src/main/java/org/apache/druid/segment/loading/PartialSegmentMetadataCacheEntry.java
@@ -118,9 +118,11 @@ public class PartialSegmentMetadataCacheEntry implements
SegmentCacheEntry, Resi
@GuardedBy("entryLock")
@Nullable
private StorageLocation location;
- @GuardedBy("entryLock")
+ // volatile so isMounted() can read it without blocking behind a concurrent
doActualUnmount() holding entryLock
+ // across fileMapper.close() + deleteHeaderFiles() (real file I/O). Other
reads of this field still go through
+ // entryLock because they do more than a single null-check (e.g.
isFullyDownloaded() also calls a method on it).
@Nullable
- private PartialSegmentFileMapperV10 fileMapper;
+ private volatile PartialSegmentFileMapperV10 fileMapper;
// Cached PartialQueryableIndex for the mounted file mapper. Built lazily on
first {@code acquireReference} call
// so concurrent acquireReference calls share the same memoized
column-holder suppliers. first read of a column
// deserializes once and the ColumnHolder stays cached for subsequent
queries against the same entry. Cleared
@@ -666,13 +668,7 @@ public class PartialSegmentMetadataCacheEntry implements
SegmentCacheEntry, Resi
@Override
public boolean isMounted()
{
- entryLock.lock();
- try {
- return fileMapper != null;
- }
- finally {
- entryLock.unlock();
- }
+ return fileMapper != null;
}
@Override
@@ -817,11 +813,13 @@ public class PartialSegmentMetadataCacheEntry implements
SegmentCacheEntry, Resi
entryLock.lock();
try {
location = mountLocation;
- fileMapper = mapper;
// Install (or re-install, after a previous mount/unmount cycle
terminated the prior Phaser) the
// reference-counted gate over cleanup. Future
acquireMetadataReference() / unmount() calls operate on this
- // instance.
+ // instance. Must be set before fileMapper is published below:
fileMapper is volatile and isMounted() reads
+ // it without entryLock, so a lock-free caller must never be able to
observe isMounted() == true before the
+ // reference gate exists.
references.set(new
ReferenceCountingCloseableObject<Closeable>(this::doActualUnmount) {});
+ fileMapper = mapper;
}
finally {
entryLock.unlock();
@@ -1172,13 +1170,10 @@ public class PartialSegmentMetadataCacheEntry
implements SegmentCacheEntry, Resi
@Override
public boolean isFullyDownloaded()
{
- entryLock.lock();
- try {
- return fileMapper != null && fileMapper.isFullyDownloaded();
- }
- finally {
- entryLock.unlock();
- }
+ // Lock-free like isMounted(): fileMapper is volatile, and its
isFullyDownloaded() shares no mutable state with
+ // close(), so it's safe to call on a mapper that's concurrently
unmounting.
+ final PartialSegmentFileMapperV10 mapper = fileMapper;
+ return mapper != null && mapper.isFullyDownloaded();
}
/**
diff --git
a/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
b/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
index ec99cd4c1c3..b48e4202f9b 100644
---
a/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
+++
b/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
@@ -1878,7 +1878,8 @@ public class SegmentLocalCacheManager implements
SegmentCacheManager
private SegmentLazyLoadFailCallback lazyLoadCallback =
SegmentLazyLoadFailCallback.NOOP;
private StorageLocation location;
private File storageDir;
- private ReferenceCountedSegmentProvider referenceProvider;
+ // volatile so isMounted() can read it without blocking behind a
concurrent mount()/unmount() holding entryLock.
+ private volatile ReferenceCountedSegmentProvider referenceProvider;
private final AtomicReference<Runnable> onUnmount = new
AtomicReference<>();
// switched from synchronized to use a ReentrantLock to avoid pinning
virtual threads to platform threads until
// https://openjdk.org/jeps/491, we could consider switching back after
java 24+ is the minimum version
@@ -1906,13 +1907,7 @@ public class SegmentLocalCacheManager implements
SegmentCacheManager
@Override
public boolean isMounted()
{
- entryLock.lock();
- try {
- return referenceProvider != null;
- }
- finally {
- entryLock.unlock();
- }
+ return referenceProvider != null;
}
@Override
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerConcurrencyTest.java
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerConcurrencyTest.java
index afbcc0bc750..5c2b482779c 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerConcurrencyTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerConcurrencyTest.java
@@ -19,6 +19,10 @@
package org.apache.druid.segment.loading;
+import com.fasterxml.jackson.annotation.JacksonInject;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
import com.fasterxml.jackson.databind.InjectableValues;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.jsontype.NamedType;
@@ -59,6 +63,8 @@ import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
@@ -92,6 +98,7 @@ class SegmentLocalCacheManagerConcurrencyTest
{
jsonMapper = new DefaultObjectMapper();
jsonMapper.registerSubtypes(new NamedType(LocalLoadSpec.class, "local"));
+ jsonMapper.registerSubtypes(new NamedType(SlowLoadSpec.class, "slow"));
jsonMapper.setInjectableValues(
new InjectableValues.Std().addValue(
LocalDataSegmentPuller.class,
@@ -254,6 +261,45 @@ class SegmentLocalCacheManagerConcurrencyTest
Assertions.assertEquals(8 * 1209, rows);
}
+ @Test
+ public void testAcquireCachedSegmentDoesNotBlockOnConcurrentMount() throws
Exception
+ {
+ final File localStorageFolder = new File(tempDir, "local_storage_folder");
+ final Interval interval = Intervals.of("2019-01-01/P1D");
+ final DataSegment segment = makeSlowSegment(localStorageFolder, interval);
+
+ final String key = segment.getId().toString();
+ final CountDownLatch loadStarted = new CountDownLatch(1);
+ final CountDownLatch releaseLoad = new CountDownLatch(1);
+ SlowLoadSpec.STARTED.put(key, loadStarted);
+ SlowLoadSpec.RELEASE.put(key, releaseLoad);
+
+ try {
+ final Future<?> loadFuture = executorService.submit(() ->
manager.load(segment));
+ Assertions.assertTrue(loadStarted.await(5, TimeUnit.SECONDS));
+
+ // isMounted() (and hence acquireCachedSegment) must return promptly
even while mount() is in flight and
+ // holding entryLock for this same entry; it must not block until the
mount finishes.
+ final Future<Optional<Segment>> readFuture =
+ executorService.submit(() ->
manager.acquireCachedSegment(segment.getId(), AcquireMode.FULL));
+ final Optional<Segment> whileMounting = readFuture.get(500,
TimeUnit.MILLISECONDS);
+ Assertions.assertTrue(whileMounting.isEmpty());
+
+ releaseLoad.countDown();
+ loadFuture.get(5, TimeUnit.SECONDS);
+
+ final Optional<Segment> afterMounting =
+ manager.acquireCachedSegment(segment.getId(), AcquireMode.FULL);
+ Assertions.assertTrue(afterMounting.isPresent());
+ afterMounting.get().close();
+ }
+ finally {
+ SlowLoadSpec.STARTED.remove(key);
+ SlowLoadSpec.RELEASE.remove(key);
+ segmentsToLoad.add(segment);
+ }
+ }
+
@Test
public void testAcquireSegmentFailTooManySegments() throws IOException
{
@@ -854,6 +900,36 @@ class SegmentLocalCacheManagerConcurrencyTest
}
}
+ /**
+ * Builds a segment whose load spec blocks in {@link
SlowLoadSpec#loadSegment} until released via
+ * {@link SlowLoadSpec#RELEASE}, keyed by {@link
SlowLoadSpec#STARTED}/{@code RELEASE}.get(segment.getId().toString()).
+ */
+ private DataSegment makeSlowSegment(File localStorageFolder, Interval
interval) throws IOException
+ {
+ final String segmentPath = Paths.get(
+ localStorageFolder.getCanonicalPath(),
+ dataSource,
+ StringUtils.format("%s_%s", interval.getStart().toString(),
interval.getEnd().toString()),
+ segmentVersion,
+ "0"
+ ).toString();
+ final File localSegmentFile = new File(localStorageFolder, segmentPath +
"_build");
+ final File indexZip = new File(new File(localStorageFolder, segmentPath),
"index.zip");
+ SegmentLocalCacheManagerTest.makeSegmentZip(localSegmentFile, indexZip);
+
+ final DataSegment segment = newSegment(interval, 0, 1000);
+ return segment.withLoadSpec(
+ ImmutableMap.of(
+ "type",
+ "slow",
+ "key",
+ segment.getId().toString(),
+ "path",
+ indexZip.getAbsolutePath()
+ )
+ );
+ }
+
private DataSegment newSegment(Interval interval, int partitionId, long size)
{
return DataSegment.builder()
@@ -890,6 +966,48 @@ class SegmentLocalCacheManagerConcurrencyTest
}
}
+ /**
+ * A {@link LoadSpec} that blocks in {@link #loadSegment} until released, so
tests can hold {@code entryLock} open
+ * across {@code CompleteSegmentCacheEntry.mount()} for as long as they
need. Looked up by a {@code key} property
+ * (rather than passed directly) since {@link LoadSpec} is deserialized from
{@link DataSegment#getLoadSpec()}.
+ */
+ @JsonTypeName("slow")
+ public static class SlowLoadSpec implements LoadSpec
+ {
+ static final ConcurrentHashMap<String, CountDownLatch> STARTED = new
ConcurrentHashMap<>();
+ static final ConcurrentHashMap<String, CountDownLatch> RELEASE = new
ConcurrentHashMap<>();
+
+ private final LocalDataSegmentPuller puller;
+ private final String key;
+ private final String path;
+
+ @JsonCreator
+ public SlowLoadSpec(
+ @JacksonInject LocalDataSegmentPuller puller,
+ @JsonProperty(value = "key", required = true) String key,
+ @JsonProperty(value = "path", required = true) String path
+ )
+ {
+ this.puller = puller;
+ this.key = key;
+ this.path = path;
+ }
+
+ @Override
+ public LoadSpecResult loadSegment(File outDir) throws
SegmentLoadingException
+ {
+ STARTED.get(key).countDown();
+ try {
+ RELEASE.get(key).await();
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ return new LoadSpecResult(puller.getSegmentFiles(new File(path),
outDir).size());
+ }
+ }
+
private static class Load implements Callable<Void>
{
private final SegmentLocalCacheManager segmentManager;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
