capistrant commented on code in PR #19717:
URL: https://github.com/apache/druid/pull/19717#discussion_r3626939407
##########
server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java:
##########
Review Comment:
nit: this is now slightly misleading since the permit acquire has moved
##########
server/src/main/java/org/apache/druid/segment/loading/StorageLoadingThreadPool.java:
##########
@@ -30,89 +27,107 @@
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
import org.apache.druid.java.util.common.logger.Logger;
-import org.apache.druid.segment.PartialBundleAcquirer;
import org.apache.druid.segment.loading.external.VirtualStorageManager;
import javax.annotation.Nullable;
import java.io.Closeable;
-import java.util.Collection;
-import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
/**
* Holds the thread pool used for background loading by {@link
SegmentLocalCacheManager} and
* {@link VirtualStorageManager}.
+ * <p>
+ * <b>Submissions are not automatically concurrency-bounded.</b> The executor
returned by {@link #getExecutorService()}
+ * runs submitted tasks as fast as it can (in the virtual-thread mode, one
virtual thread per task). Instead, the
+ * number of concurrent <em>deep-storage reads</em> is bounded by a permit
that callers acquire via
+ * {@link #acquireLoadPermit()} around <em>only</em> the actual I/O, never
around lock acquisition, reservation, or
+ * deserialization. Any new load path that reads from deep storage must
acquire a permit around that read, or it will
+ * be unbounded.
*/
public class StorageLoadingThreadPool
{
- private static final Logger log = new Logger(StorageLoadingThreadPool.class);
-
- private final ListeningExecutorService exec;
-
- public StorageLoadingThreadPool(
- @Nullable final ListeningExecutorService exec
- )
- {
- this.exec = exec;
- }
-
public static StorageLoadingThreadPool createFromConfig(final
SegmentLoaderConfig config)
{
- final ListeningExecutorService exec;
+ if (!config.isVirtualStorage()) {
+ return new StorageLoadingThreadPool(null, null);
+ }
- if (config.isVirtualStorage()) {
- if (config.getVirtualStorageLoadThreads() <= 0) {
- throw DruidException.forPersona(DruidException.Persona.OPERATOR)
- .ofCategory(DruidException.Category.INVALID_INPUT)
- .build(
- "virtualStorageLoadThreads must be greater
than 0, got [%d]",
- config.getVirtualStorageLoadThreads()
- );
- }
- if (config.isVirtualStorageUseVirtualThreads()) {
- log.info(
- "Using virtual storage mode with virtual threads - max concurrent
on demand loads: [%d].",
- config.getVirtualStorageLoadThreads()
- );
- exec = new PermitBoundedListeningExecutorService(
- MoreExecutors.listeningDecorator(
- Executors.newThreadPerTaskExecutor(
- Thread.ofVirtual()
- .name("VirtualStorageOnDemandLoadingThread-", 0)
- .factory()
- )
- ),
- new Semaphore(config.getVirtualStorageLoadThreads())
- );
- } else {
- log.info(
- "Using virtual storage mode with fixed platform thread pool - on
demand load threads: [%d].",
- config.getVirtualStorageLoadThreads()
- );
- exec = MoreExecutors.listeningDecorator(
- Executors.newFixedThreadPool(
- config.getVirtualStorageLoadThreads(),
-
Execs.makeThreadFactory("VirtualStorageOnDemandLoadingThread-%s")
- )
- );
- }
- } else {
- exec = null;
+ if (config.getVirtualStorageLoadThreads() <= 0) {
+ throw DruidException.forPersona(DruidException.Persona.OPERATOR)
+ .ofCategory(DruidException.Category.INVALID_INPUT)
+ .build(
+ "virtualStorageLoadThreads must be greater than
0, got [%d]",
+ config.getVirtualStorageLoadThreads()
+ );
}
- return new StorageLoadingThreadPool(exec);
+ final ListeningExecutorService exec;
+ final Semaphore permits;
+ if (config.isVirtualStorageUseVirtualThreads()) {
+ log.info(
+ "Using virtual storage mode with virtual threads - max concurrent on
demand loads: [%d].",
+ config.getVirtualStorageLoadThreads()
+ );
+ // Unbounded thread-per-virtual-thread executor; concurrency is bounded
by the permit count, acquired by callers
+ // via acquireLoadPermit() around the actual deep-storage reads.
+ exec = MoreExecutors.listeningDecorator(
+ Executors.newThreadPerTaskExecutor(
+ Thread.ofVirtual()
+ .name("VirtualStorageOnDemandLoadingThread-", 0)
+ .factory()
+ )
+ );
+ permits = new Semaphore(config.getVirtualStorageLoadThreads());
+ } else {
+ log.info(
+ "Using virtual storage mode with fixed platform thread pool - on
demand load threads: [%d].",
+ config.getVirtualStorageLoadThreads()
+ );
+ // Fixed pool: the thread count is the concurrency bound, so no separate
permit is needed.
+ exec = MoreExecutors.listeningDecorator(
+ Executors.newFixedThreadPool(
+ config.getVirtualStorageLoadThreads(),
+ Execs.makeThreadFactory("VirtualStorageOnDemandLoadingThread-%s")
+ )
+ );
+ permits = null;
+ }
+ return new StorageLoadingThreadPool(exec, permits);
}
/**
* Returns an instance representing "no thread pool". Calling {@link
#getExecutorService()} will return an error.
*/
public static StorageLoadingThreadPool none()
{
- return new StorageLoadingThreadPool(null);
+ return new StorageLoadingThreadPool(null, null);
+ }
+
+ private static final Logger log = new Logger(StorageLoadingThreadPool.class);
+
+ /**
+ * A permit handle whose {@code close()} releases nothing. Returned by
{@link #acquireLoadPermit()} when there is no
+ * semaphore (the fixed-thread-pool mode, where the thread count is the
bound, and the "no pool" instance).
+ */
+ private static final LoadPermit NOOP_PERMIT = () -> {};
+
+ @Nullable
+ private final ListeningExecutorService exec;
+ /**
+ * Bounds concurrent on-demand deep-storage reads in the virtual-thread
mode, where the executor is otherwise
+ * unbounded (one virtual thread per task). Null in the fixed-thread-pool
mode (the pool size is the bound) and in
+ * the "no pool" instance.
+ */
+ @Nullable
+ private final Semaphore permits;
Review Comment:
nit- can this stuff be at the top of the class
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
