FrankChen021 commented on code in PR #20273:
URL: https://github.com/apache/druid/pull/20273#discussion_r3957843147
##########
processing/src/main/java/org/apache/druid/java/util/http/client/pool/ResourcePool.java:
##########
@@ -355,6 +451,184 @@ public void close()
}
}
+ private static class AdaptiveResourceHolderPerKey<K, V> extends
PooledResources<V>
+ {
+ private final int maxSize;
+ private final K key;
+ private final ResourceFactory<K, V> factory;
+ private final long unusedResourceTimeoutMillis;
+ private final Semaphore permits;
+ private final Deque<ResourceHolder<V>> idleResources = new
ConcurrentLinkedDeque<>();
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ private AdaptiveResourceHolderPerKey(
+ int maxSize,
+ long unusedResourceTimeoutMillis,
+ K key,
+ ResourceFactory<K, V> factory
+ )
+ {
+ this.maxSize = maxSize;
+ this.key = key;
+ this.factory = factory;
+ this.unusedResourceTimeoutMillis = unusedResourceTimeoutMillis;
+ this.permits = new Semaphore(maxSize);
+ }
+
+ /**
+ * Fills the pool to its maximum size, closing what it created if that
fails.
+ */
+ void preload()
+ {
+ try {
+ for (int i = 0; i < maxSize; i++) {
+ idleResources.addLast(new
ResourceHolder<>(System.currentTimeMillis(), generate()));
+ }
+ }
+ catch (Throwable t) {
+ closeIdleResources();
+ throw t;
+ }
+ }
+
+ @Nullable
+ @Override
+ V get()
+ {
+ if (!acquirePermit()) {
+ return null;
+ }
+
+ boolean lent = false;
+ try {
+ V resource = takeIdleResource();
+ if (resource == null) {
+ resource = createResource();
+ }
+ lent = true;
+ return resource;
+ }
+ finally {
+ if (!lent) {
+ permits.release();
+ }
+ }
+ }
+
+ @Override
+ void giveBack(V object)
+ {
+ Preconditions.checkNotNull(object, "object");
+
+ if (closed.get()) {
+ log.info("giveBack called after being closed. key[%s]", key);
+ closeQuietly(object);
+ permits.release();
+ return;
+ }
+
+ idleResources.addLast(new ResourceHolder<>(System.currentTimeMillis(),
object));
+ permits.release();
+
+ if (closed.get()) {
+ // close() may have drained the idle resources before this one was
parked.
+ closeIdleResources();
+ }
+ }
+
+ @Override
+ public void close()
+ {
+ if (closed.compareAndSet(false, true)) {
+ permits.release(maxSize);
Review Comment:
[P2] Wake zero-sized adaptive pools on close
`numConnections` is validated with `@Min(0)`, so an adaptive pool can
legitimately be constructed with `maxSize == 0`. Its semaphore then starts with
zero permits, and `close()` calls `release(maxSize)`, i.e. `release(0)`, which
wakes no thread blocked in `acquirePermit()`. A request already blocked in
`ResourcePool.take()` therefore remains stuck through client shutdown, whereas
the retaining implementation's `notifyAll()` unblocks it. Release at least one
permit for the zero-size case (or use an explicit close signal) so close can
always wake waiters.
--
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]
