Copilot commented on code in PR #7663:
URL: https://github.com/apache/kyuubi/pull/7663#discussion_r3888088201
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/ui/JettyServer.scala:
##########
@@ -88,11 +89,17 @@ object JettyServer {
host: String,
port: Int,
poolSize: Int,
+ useVirtualThreads: Boolean,
stopTimeout: Long,
sendServerVersion: Boolean): JettyServer = {
- val pool = new QueuedThreadPool(poolSize)
- pool.setName(name)
- pool.setDaemon(true)
+ val pool: ThreadPool = if (useVirtualThreads) {
+ new VirtualThreadPool(poolSize, name)
Review Comment:
This does not preserve the REST/Trino queue and rejection semantics required
by #7662. Their previous `QueuedThreadPool(poolSize)` has Jetty 9.4.57's
expandable job queue, whereas `VirtualThreadPool` uses the zero-queue bounded
executor and rejects the `(poolSize + 1)`th concurrent submission immediately.
Bursts that were previously queued can therefore fail HTTP work. Please make
queue behavior configurable and use queued semantics for REST/Trino while
retaining zero-queue behavior for the Thrift HTTP frontend.
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/http/VirtualThreadPool.scala:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.server.http
+
+import java.util.concurrent.{ExecutorService, RejectedExecutionException,
TimeUnit}
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.eclipse.jetty.util.component.AbstractLifeCycle
+import org.eclipse.jetty.util.thread.ThreadPool
+
+import org.apache.kyuubi.util.ThreadUtils
+
+/**
+ * Adapts a virtual-thread-per-task executor to Jetty's [[ThreadPool]]
contract.
+ *
+ * Despite the name, virtual threads are never pooled or reused. Each task
runs on a new virtual
+ * thread, while `maxThreads` only limits concurrent tasks to preserve
frontend backpressure.
+ *
+ * Jetty 12 provides a native
`org.eclipse.jetty.util.thread.VirtualThreadPool`, which can replace
+ * this adapter after Kyuubi upgrades to that version.
+ */
+private[server] class VirtualThreadPool(maxThreads: Int, name: String)
+ extends AbstractLifeCycle with ThreadPool.SizedThreadPool {
+
+ private val activeThreads = new AtomicInteger()
+ @volatile private var executor: ExecutorService = _
+
+ override protected def doStart(): Unit = {
+ executor = ThreadUtils.newBoundedVirtualThreadPerTaskExecutor(maxThreads,
name)
+ }
+
+ override def execute(command: Runnable): Unit = {
+ val currentExecutor = executor
+ if (!isRunning || currentExecutor == null) {
+ throw new RejectedExecutionException(s"$name is not running")
+ }
+ currentExecutor.execute(() => {
+ activeThreads.incrementAndGet()
+ try {
+ command.run()
+ } finally {
+ activeThreads.decrementAndGet()
+ }
+ })
+ }
+
+ override def join(): Unit = {
+ val currentExecutor = executor
+ if (currentExecutor != null) {
+ currentExecutor.awaitTermination(Long.MaxValue, TimeUnit.NANOSECONDS)
+ }
+ }
+
+ override def getThreads: Int = activeThreads.get()
+
+ override def getIdleThreads: Int = Math.max(maxThreads -
activeThreads.get(), 0)
Review Comment:
Jetty defines `getIdleThreads` as the number of actual idle threads, but
this adapter never retains idle workers and currently reports unused
concurrency permits instead. An idle pool consequently exposes `maxThreads`
phantom idle threads through Jetty diagnostics/JMX. Return zero and continue
reporting saturation through `isLowOnThreads`.
--
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]
