You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "lhotari (via GitHub)" <gi...@apache.org> on 2026/10/16 17:18:47 UTC
Re: [PR] [improve][broker] Write async REST responses on the web service executor [pulsar]
lhotari commented on code in PR #26946:
URL: https://github.com/apache/pulsar/pull/26946#discussion_r4297324178
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/WebExecutorThreadPool.java:
##########
@@ -60,4 +164,100 @@ public void join() throws InterruptedException {
super.join();
}
}
+
+ /**
+ * A {@link ThreadPoolExecutor} that starts a new thread, up to its maximum pool size, instead of queueing a
+ * task when no thread is idle.
+ */
+ private static final class GrowingThreadPoolExecutor extends ThreadPoolExecutor {
Review Comment:
The growable thread pool has been dropped from this PR in 1fbcbdc1303. It will be proposed separately, and I'll take the shrinking behaviour under steady traffic and the `QueuedThreadPool` comparison into account there.
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/WebExecutorThreadPool.java:
##########
@@ -19,27 +19,131 @@
package org.apache.pulsar.broker.web;
import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.eclipse.jetty.util.BlockingArrayQueue;
+import org.eclipse.jetty.util.MathUtils;
+import org.eclipse.jetty.util.ProcessorUtils;
import org.eclipse.jetty.util.thread.ExecutorThreadPool;
+import org.eclipse.jetty.util.thread.ThreadPoolBudget;
+/**
+ * The thread pool of the Jetty web servers.
+ *
+ * <p>The pool keeps {@code numThreads} threads. Unless {@code maxThreads} is larger, it is a fixed size pool that
+ * queues tasks when all of its threads are busy. When {@code maxThreads} is larger, the pool starts additional
+ * threads, up to {@code maxThreads}, when a task is submitted and no thread is idle, and only queues tasks once
+ * it has {@code maxThreads} threads. Threads beyond {@code numThreads} stop after being idle for the idle timeout,
+ * see {@link #setIdleTimeout(int)}, which is {@link #DEFAULT_IDLE_TIMEOUT_MS} by default. A plain
+ * {@link ThreadPoolExecutor} would start threads beyond its core size only once its queue is full.
+ */
public class WebExecutorThreadPool extends ExecutorThreadPool {
+ public static final int DEFAULT_IDLE_TIMEOUT_MS = 60_000;
+
+ private static final ThreadLocal<WebExecutorThreadPool> CURRENT_POOL = new ThreadLocal<>();
+
+ private final ThreadPoolExecutor executor;
private final ThreadFactory threadFactory;
private volatile boolean startAttempted;
- public WebExecutorThreadPool(int maxThreads, String namePrefix) {
- this(maxThreads, namePrefix, 8192);
+ public WebExecutorThreadPool(int numThreads, String namePrefix) {
+ this(numThreads, namePrefix, 8192);
+ }
+
+ public WebExecutorThreadPool(int numThreads, String namePrefix, int queueCapacity) {
+ this(numThreads, 0, namePrefix, queueCapacity);
+ }
+
+ /**
+ * @param numThreads the number of threads that the pool keeps
+ * @param maxThreads the maximum number of threads; a value not larger than {@code numThreads}, such as 0,
+ * means that the pool doesn't grow beyond {@code numThreads}
+ * @param namePrefix the name prefix of the threads
+ * @param queueCapacity the capacity of the queue for tasks that wait for a thread
+ */
+ public WebExecutorThreadPool(int numThreads, int maxThreads, String namePrefix, int queueCapacity) {
+ this(createExecutor(numThreads, maxThreads, queueCapacity), numThreads, namePrefix);
}
- public WebExecutorThreadPool(int maxThreads, String namePrefix, int queueCapacity) {
- super(maxThreads, Math.min(8, maxThreads), new BlockingArrayQueue<>(queueCapacity));
+ private WebExecutorThreadPool(ThreadPoolExecutor executor, int numThreads, String namePrefix) {
+ super(executor, -1, null);
+ this.executor = executor;
+ // the number of threads that are started when the pool starts
+ setMinThreads(Math.min(8, numThreads));
+ if (executor instanceof GrowingThreadPoolExecutor) {
+ // Jetty sizes the reserved thread executor from the maximum number of threads, and its threads wait for
+ // work without an idle timeout, so they wouldn't let the pool shrink. Size it from numThreads, like it
+ // is sized for a pool that doesn't grow.
+ setReservedThreads(Math.max(1,
+ MathUtils.ceilToNextPowerOfTwo(Math.min(ProcessorUtils.availableProcessors(), numThreads / 8))));
+ }
Review Comment:
Dropped from this PR in 1fbcbdc1303 together with the rest of the growable pool, so the selector count is unchanged here. I'll cover the selector sizing in the separate proposal.
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/WebExecutorThreadPool.java:
##########
@@ -19,27 +19,131 @@
package org.apache.pulsar.broker.web;
import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.eclipse.jetty.util.BlockingArrayQueue;
+import org.eclipse.jetty.util.MathUtils;
+import org.eclipse.jetty.util.ProcessorUtils;
import org.eclipse.jetty.util.thread.ExecutorThreadPool;
+import org.eclipse.jetty.util.thread.ThreadPoolBudget;
+/**
+ * The thread pool of the Jetty web servers.
+ *
+ * <p>The pool keeps {@code numThreads} threads. Unless {@code maxThreads} is larger, it is a fixed size pool that
+ * queues tasks when all of its threads are busy. When {@code maxThreads} is larger, the pool starts additional
+ * threads, up to {@code maxThreads}, when a task is submitted and no thread is idle, and only queues tasks once
+ * it has {@code maxThreads} threads. Threads beyond {@code numThreads} stop after being idle for the idle timeout,
+ * see {@link #setIdleTimeout(int)}, which is {@link #DEFAULT_IDLE_TIMEOUT_MS} by default. A plain
+ * {@link ThreadPoolExecutor} would start threads beyond its core size only once its queue is full.
+ */
public class WebExecutorThreadPool extends ExecutorThreadPool {
+ public static final int DEFAULT_IDLE_TIMEOUT_MS = 60_000;
+
+ private static final ThreadLocal<WebExecutorThreadPool> CURRENT_POOL = new ThreadLocal<>();
+
+ private final ThreadPoolExecutor executor;
private final ThreadFactory threadFactory;
private volatile boolean startAttempted;
- public WebExecutorThreadPool(int maxThreads, String namePrefix) {
- this(maxThreads, namePrefix, 8192);
+ public WebExecutorThreadPool(int numThreads, String namePrefix) {
+ this(numThreads, namePrefix, 8192);
+ }
+
+ public WebExecutorThreadPool(int numThreads, String namePrefix, int queueCapacity) {
+ this(numThreads, 0, namePrefix, queueCapacity);
+ }
+
+ /**
+ * @param numThreads the number of threads that the pool keeps
+ * @param maxThreads the maximum number of threads; a value not larger than {@code numThreads}, such as 0,
+ * means that the pool doesn't grow beyond {@code numThreads}
+ * @param namePrefix the name prefix of the threads
+ * @param queueCapacity the capacity of the queue for tasks that wait for a thread
+ */
+ public WebExecutorThreadPool(int numThreads, int maxThreads, String namePrefix, int queueCapacity) {
+ this(createExecutor(numThreads, maxThreads, queueCapacity), numThreads, namePrefix);
}
- public WebExecutorThreadPool(int maxThreads, String namePrefix, int queueCapacity) {
- super(maxThreads, Math.min(8, maxThreads), new BlockingArrayQueue<>(queueCapacity));
+ private WebExecutorThreadPool(ThreadPoolExecutor executor, int numThreads, String namePrefix) {
+ super(executor, -1, null);
+ this.executor = executor;
+ // the number of threads that are started when the pool starts
+ setMinThreads(Math.min(8, numThreads));
+ if (executor instanceof GrowingThreadPoolExecutor) {
+ // Jetty sizes the reserved thread executor from the maximum number of threads, and its threads wait for
+ // work without an idle timeout, so they wouldn't let the pool shrink. Size it from numThreads, like it
+ // is sized for a pool that doesn't grow.
+ setReservedThreads(Math.max(1,
+ MathUtils.ceilToNextPowerOfTwo(Math.min(ProcessorUtils.availableProcessors(), numThreads / 8))));
+ }
this.threadFactory = new DefaultThreadFactory(namePrefix);
}
+ private static ThreadPoolExecutor createExecutor(int numThreads, int maxThreads, int queueCapacity) {
+ if (maxThreads <= numThreads) {
+ // the same executor as ExecutorThreadPool(maxThreads, minThreads, queue) creates
+ return new ThreadPoolExecutor(numThreads, numThreads, DEFAULT_IDLE_TIMEOUT_MS, TimeUnit.MILLISECONDS,
+ new BlockingArrayQueue<>(queueCapacity));
+ }
+ return new GrowingThreadPoolExecutor(numThreads, maxThreads, queueCapacity);
+ }
+
@Override
protected Thread newThread(Runnable job) {
- return threadFactory.newThread(job);
+ return threadFactory.newThread(() -> {
+ CURRENT_POOL.set(this);
+ try {
+ job.run();
+ } finally {
+ if (executor instanceof GrowingThreadPoolExecutor growingExecutor) {
+ // this thread has stopped, so a queued task might have no thread to run it
+ growingExecutor.startThreadForQueuedTasks();
+ }
+ }
+ });
+ }
+
+ /**
+ * Sets the maximum number of threads. Unlike {@link ExecutorThreadPool#setMaxThreads(int)}, it doesn't change
+ * the number of threads that a growing pool keeps, which would stop it from growing.
+ */
+ @Override
+ public void setMaxThreads(int threads) {
+ if (executor instanceof GrowingThreadPoolExecutor) {
+ ThreadPoolBudget budget = getThreadPoolBudget();
+ if (budget != null) {
+ budget.check(threads);
+ }
+ executor.setMaximumPoolSize(threads);
+ } else {
+ super.setMaxThreads(threads);
+ }
+ }
Review Comment:
Dropped in 1fbcbdc1303: the `setMaxThreads()` override and its test were part of the growable pool, which is no longer in this PR.
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java:
##########
@@ -359,6 +359,18 @@ public class ServiceConfiguration implements PulsarConfiguration {
// waiting for another HTTP call to complete in same thread.
private int numHttpServerThreads = Math.max(8, 2 * Runtime.getRuntime().availableProcessors());
+ @FieldContext(
+ category = CATEGORY_SERVER,
+ doc = "Maximum number of threads to use for HTTP requests processing. When all of the"
+ + " `numHttpServerThreads` threads are busy, for example writing responses to slow clients, the"
+ + " HTTP server starts more threads, up to this number, before it queues requests. The threads"
+ + " beyond `numHttpServerThreads` stop after being idle for `httpServerThreadPoolIdleTimeoutMs`."
+ + " A value not larger than"
+ + " `numHttpServerThreads`, such as the default 0, means that the HTTP server uses"
+ + " `numHttpServerThreads` threads."
+ )
+ private int numHttpServerThreadsMax = 0;
Review Comment:
Agreed. The growable pool and all its settings in the broker, functions worker, proxy and WebSocket service have been dropped from this PR in 1fbcbdc1303. The pool will be proposed separately, with a dev@ discussion or PIP for the settings, and the docs there will describe the shrinking behaviour accurately.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org