Repository navigation
Add an opt-in virtual threads flag for blocking I/O thread pools on Java 25 - #9097
GGraziadei wants to merge 19 commits into
Conversation
|
Currently traveling and unable to properly review this. Happy to do it once I'm able to |
|
This comment was drafted with the help of an LLM (Claude) and reviewed by me before posting. Reviewed this in detail — two real behavioral side effects worth addressing before merge, plus some smaller items. Ordered by importance: 1. DRPC_INVOCATIONS concurrency silently jumps 5→64 when the flag is enabled
2. Worker shared-executor pool size becomes topology-overridable // before: bare `conf` in this instance method resolves to this.conf (daemon-level config)
int threadPoolSize = ObjectReader.getInt(conf.get(Config.TOPOLOGY_WORKER_SHARED_THREAD_POOL_SIZE));
// after: explicitly passes this.topologyConf (merged: daemon conf + topology's own overrides)
return ImmutableMap.of(WorkerTopologyContext.SHARED_EXECUTOR, makeSharedExecutor(topologyConf));Pre-PR, 3. Virtual threads are always daemon threads, with no override
4. Benchmark's proxy handler misroutes if (method.getDeclaringClass() == Object.class) {
return method.invoke(inFlight, margs); // should be `proxy`, not `inFlight`
}
5.
Smaller cleanup, non-blocking:
|
Introduce storm.virtual.threads.enabled (default false) and StormThreadFactory, which returns a virtual-thread ThreadFactory when the flag is on and a non-daemon, normal-priority platform-thread factory otherwise. Thread names are <prefix>-<n> in both modes. The factory is injected into the existing pools, so core sizes, queues, rejection policy and error handling are unchanged and the *.threads settings keep their meaning as a concurrency bound: - Thrift server handlers for the SASL, TLS and simple transports (Nimbus, Supervisor and DRPC). With the simple transport an executor is now always supplied when the flag is on; without a configured queue size it uses the same unbounded queue THsHaServer builds by default. - AsyncLocalizer download and task executors. - Nimbus AssignmentDistributionService and Supervisor heartbeat pools. - DRPCSpout background executor. - Worker shared executor exposed through TopologyContext, read from the merged topology conf so a topology can override the flag. Spout/bolt executor threads, worker transfer, JCQueue, Netty and timer threads are deliberately left on platform threads. Round-trip tests through NimbusClient assert that Thrift handlers run on virtual threads when the flag is on and on platform threads by default; unit tests cover the factory, the localizer executor and the worker shared executor. ThriftHandlerVirtualThreadsBench in storm-perf measures the flag against an in-process Nimbus Thrift server whose handler simulates blocking I/O. At equal *.threads the throughput is unchanged, while server-side OS threads drop from *.threads to roughly the number of cores.
f6547f1 to
94a4adc
Compare
|
Hi @reiabreu, thanks for the review. I will fix asap. |
rzo1
left a comment
rzo1
left a comment
There was a problem hiding this comment.
Thanks for this. With the flag off it is indeed only thread names. I agree with @reiabreu's points, all of them are still open on the current head, so I only add what is not covered there. Two corrections to that list though:
- Reading
topology.worker.shared.thread.pool.sizefrom the topology conf is not a trust boundary change. The pool lives in the submitter's own worker, which runs their code anyway. It is rather a fix for atopology.*key that was silently ignored before. Still worth a line in the description. ObjectReader.getBooleanthrows on aString, so switching to it would break the-Dstorm.options=...=truecase. See inline.
On the design: one flag covers the Thrift handlers, the localizer and pools a topology controls. Given the thread dump issue below I would prefer separate switches, or at least start with the Thrift pools only. Open for discussion.
The benchmark only covers the simple transport (THsHaServer). SASL/TLS use TThreadPoolServer with a thread per connection, which is the setup most secure clusters run, and that is not measured. Also, rows 2 and 3 are well below min(clients, threads) * 1000 / io-ms (86% / 71% with the flag off), so they are not bounded by concurrency as stated, and +37% p99 at 2000 is more than "slightly". Single runs without variance, so I would not draw conclusions either way. The server OS thread column is not printed by the tool; I assume it is peak minus client threads, please say so.
| // THsHaServer builds its own platform-thread pool (core 5, unbounded LinkedBlockingQueue) when no | ||
| // executor is supplied, so when virtual threads are requested we always supply an executor; | ||
| // without a configured queue size we use the same unbounded queue so no request is rejected | ||
| // that would not be rejected today. The shipped defaults always configure a queue size. |
There was a problem hiding this comment.
Not true for DRPC_INVOCATIONS, its queue key is hard coded null in ThriftConnectionType. So with the flag on this branch is always taken there (the 5 -> 64 jump from @reiabreu's review), and also for anyone who removed nimbus.queue.size or supervisor.queue.size. Same statement in the javadoc of the new key.
| * the configured {@code *.threads} value. | ||
| * | ||
| * <p>All virtual threads in a JVM share one carrier scheduler sized to the number of available processors; | ||
| * blocking file I/O (for example blob downloads in the supervisor localizer) occupies a carrier, so |
There was a problem hiding this comment.
On 25 the scheduler compensates for blocking file I/O (temporarily adds carriers up to maxPoolSize), and since JEP 491 synchronized no longer pins. What still pins is native frames (Hadoop native libs, JNI in user code on the shared executor). Raising parallelism is not the right advice here.
What is missing instead: virtual threads are not time sliced, so CPU heavy Nimbus calls (getTopologyPageInfo, submitTopology on a big cluster) can hold all carriers and delay cheap ones. Probably rare, but should be mentioned.
And with the flag on, jstack and Thread.getAllStackTraces() do not show virtual threads. That makes Utils.threadDump() blind for these pools (stuck slot dump in ReadClusterState, the forced halt dump in Utils), and the jstack action from the UI via flight.bash as well. A hung blob download simply disappears from the dump. Only jcmd <pid> Thread.dump_to_file shows them. I think this needs at least a note here, better flight.bash using jcmd when the flag is on.
| * operators enabling this on I/O-heavy supervisors should size {@code jdk.virtualThreadScheduler.parallelism} | ||
| * / {@code jdk.virtualThreadScheduler.maxPoolSize} accordingly. | ||
| * | ||
| * <p>A topology may override this key in its own config to control its worker shared executor. |
There was a problem hiding this comment.
DRPCSpout also builds its factory from the conf passed to open(), so its pool is topology controlled too.
| return (Boolean) value; | ||
| } | ||
| if (value instanceof String) { | ||
| return Boolean.parseBoolean((String) value); |
There was a problem hiding this comment.
"yes" or "1" end up as false here without any warning, only non String types hit the WARN below. Daemon and topology confs both go through @IsBoolean validation anyway, so the String case should only come from storm.options. Fine to keep, but please don't replace it with ObjectReader.getBoolean, that one throws on a String.
| } No newline at end of file | ||
|
|
||
| @Test | ||
| public void simpleTransportRunsHandlerOnVirtualThreadWhenEnabled() throws Exception { |
There was a problem hiding this comment.
This does not test the no queue size path. withServer starts from ConfigUtils.readStormConfig(), which already has nimbus.queue.size: 100000 from defaults.yaml, so queueSize != null and it runs the same branch as the WithQueueSize test. The LinkedBlockingQueue arm is not covered at all, and that is exactly the DRPC case. Setting NIMBUS_QUEUE_SIZE to null in extra or using DRPC_INVOCATIONS should do. The description claims both are covered.
TLS has no test either.
| Map<String, Object> conf = new HashMap<>(); | ||
| conf.put(Config.TOPOLOGY_WORKER_SHARED_THREAD_POOL_SIZE, 2); | ||
| conf.put(Config.STORM_VIRTUAL_THREADS_ENABLED, true); | ||
| ExecutorService pool = WorkerState.makeSharedExecutor(conf); |
There was a problem hiding this comment.
Calls the static helper directly, so the wiring in makeDefaultResources (which conf is passed) is not tested. Reverting that line to conf would still pass.
| Thread serveThread = new Thread(server::serve, "bench-thrift-serve"); | ||
| serveThread.setDaemon(true); | ||
| serveThread.start(); | ||
| while (!server.isServing()) { |
There was a problem hiding this comment.
No timeout, if serve() throws this spins forever.
| long[][] latenciesNs, long elapsedNs, int failures, int threadsBefore, int peakThreads, | ||
| int peakInFlight, long rssBeforeKb, long rssAfterKb) { | ||
| long[] all = Arrays.stream(latenciesNs).flatMapToLong(Arrays::stream).filter(v -> v > 0).sorted().toArray(); | ||
| int total = clients * calls; |
There was a problem hiding this comment.
Counts failed calls into the throughput, while the latencies filter them out.
|
|
||
| public static void main(String[] args) throws Exception { | ||
| Map<String, String> opts = parseArgs(args); | ||
| String mode = opts.getOrDefault("mode", "both").toLowerCase(Locale.ROOT); |
There was a problem hiding this comment.
Default both runs both modes in one JVM (shared heap and JIT), while the javadoc recommends separate JVMs. I would default to one mode.
|
This comment was generated with the help of an LLM (Claude) and reviewed by me before posting. Two points in my earlier review were wrong; @rzo1 is right on both:
The rest of my review stands. |
|
Any news on this @GGraziadei (maybe with a 🍺 in Glasgow? ;-) ) |
With the simple transport and virtual threads on, an executor of *.threads handlers is always supplied, which raises the effective concurrency from THsHaServer's 5. The DRPC invocations port has no queue size key, so this always applies there (5 -> drpc.invocations.threads), and it also applies to any server whose *.queue.size was explicitly unset. Say so in the code comment and in the javadoc of storm.virtual.threads.enabled instead of describing it as a rare case.
…irtual threads too The javadoc advised raising the carrier scheduler parallelism for blocking file I/O, but on JDK 25 the scheduler already compensates for that and synchronized no longer pins. Describe what actually matters instead: native frames still pin a carrier, virtual threads are not time sliced so a CPU-heavy Nimbus call can delay cheap ones, and jstack and Thread.getAllStackTraces() do not show virtual threads, which makes the thread dumps Storm logs itself blind for these pools. Have the jstack action of bin/flight.bash also write a jcmd Thread.dump_to_file dump, which does include virtual threads.
…olled DRPCSpout builds its thread factory from the conf passed to open(), so a topology-level storm.virtual.threads.enabled affects that pool as well, not only the worker shared executor.
…r false Boolean.parseBoolean silently mapped values such as "yes" or "1" to false, while only non-String types hit the warning. String values only reach the factory through storm.options, since daemon and topology confs are validated as boolean, so keep accepting strings but log the ones that are neither true nor false.
The loop waiting for isServing() spun forever if serve() threw. Capture a failure from the serve thread and give up after 30 seconds or when the thread exits without serving.
Throughput divided clients * calls by the elapsed time, so a client that failed part way still counted all of its calls, while the latency percentiles only used the calls that completed. Use the completed count for both and report attempted, completed and failed clients separately.
The javadoc recommends running each mode in its own JVM, but the default --mode was both, which runs them back to back sharing heap and JIT state. Default to off, reject unknown modes, and describe both as a quick look only.
…tual thread tests The simple transport test meant to cover the path where no *.queue.size is configured, but withServer starts from readStormConfig(), which carries nimbus.queue.size from defaults.yaml, so both simple transport tests ran the ArrayBlockingQueue branch. Unset the key explicitly so the LinkedBlockingQueue branch, the one the DRPC invocations port always takes, is exercised. Add a round trip through a TLS ThriftServer asserting the handler runs on a virtual thread, which covers TThreadPoolServer with the TLS plugin. The client is a plain Thrift client over a trust-only SSL socket because the test certificate has no subject alternative name.
The previous tests called the static helper directly, so reverting makeDefaultResources to read the daemon conf would still have passed. Build a WorkerState with the flag only in the topology conf and take the executor from getDefaultSharedResources(), which fails if the wiring regresses. The helper no longer needs to be visible for testing.
…self equals, hashCode and toString were forwarded to the captured in-flight counter instead of the proxy.
The localizer threads were renamed from "AsyncLocalizer Download Executor - 0" to "AsyncLocalizer-Download-Executor-0", which breaks any external log or metrics tooling matching the old literal. Have StormThreadFactory use the given prefix verbatim so callers can reproduce an existing format, and pass the original names for the localizer pools. The new pools keep their <prefix>-<n> names.
The three transport plugins built the same name from type.name() by hand. Expose it as getHandlerThreadNamePrefix() next to the other per-type values.
AssignmentDistributionService.prepare receives a raw Map and needed an unchecked cast to call the factory. Take Map<?, ?> instead, which only needs get().
The platform branch of StormThreadFactory sets daemon(false), while Thread.Builder.OfVirtual has no such option. The Nimbus and Supervisor shutdown hooks close the pools without joining their threads and rely on the delayed force halt, so Storm itself does not depend on non-daemon pool threads, but say so in the javadoc for anyone embedding the daemons.
…ized test The four tests differed only in transport, extra conf and expected result. One @ParameterizedTest with a case list keeps the no-queue-size, queue-size, default and digest cases and the ReqContext principal check for the authenticated transport.
…estimate The benchmark only covered the simple transport (THsHaServer), while the SASL and TLS transports secure clusters run use TThreadPoolServer with a thread per connection. Add --transport simple|digest; the digest mode writes a throwaway JAAS file with a fixed user. Since every client holds its connection, and with it a handler thread, for the whole run, digest needs --threads >= --clients and the benchmark rejects anything else instead of timing out half the handshakes. Print the server-side platform thread estimate (peak minus before minus client threads) that the PR description quoted but the tool did not output, and show the first client failure when there is one.
With hundreds of clients opening connections at the same instant the server socket's default listen backlog overflows and part of the connections are reset before the run starts, which showed up as failed clients in the digest runs at 500 and 2000 clients. Connection setup is not what is measured, so limit concurrent connects to 16; calls still start only once every client is connected.
3 participants
What is the purpose of the change
Now that master requires Java 25, this change lets operators opt in to Java virtual threads for Storm's blocking, I/O-bound thread pools through a new cluster setting,
storm.virtual.threads.enabled(defaultfalse). With the flag off nothing changes apart from thread names.A new helper,
org.apache.storm.utils.StormThreadFactory, returns a virtual-threadThreadFactorywhen the flag is on and a non-daemon, normal-priority platform-thread factory otherwise. The factory is injected into the existing pools, so core sizes, queues, rejection policy and error handling are untouched and the*.threadssettings keep their meaning as a concurrency bound:THsHaServerbuilds by default.AsyncLocalizerdownload and task executors.AssignmentDistributionServiceand Supervisor heartbeat pools.DRPCSpoutbackground executor.TopologyContext, read from the merged topology conf so a topology can override the flag.Spout/bolt executor threads, worker transfer, JCQueue, Netty event loops and timers are deliberately left on platform threads: they are busy-poll hot loops that would lose throughput on virtual threads.
The javadoc of the new key documents the operational caveats: all virtual threads share one carrier scheduler sized to the core count, blocking file I/O (blob downloads) occupies a carrier, and with the simple transport and no configured queue size the effective handler concurrency goes from
THsHaServer's fallback of 5 to*.threads.What the benchmark shows
ThriftHandlerVirtualThreadsBench(added toexamples/storm-perf) runs an in-process Nimbus Thrift server whose handler simulates blocking I/O and hammers it with N clients, each mode in its own JVM. On a 20-core machine:*.threads/ io-msAt equal
*.threadsthe throughput is unchanged (within a few percent) because the workload is bounded by the configured concurrency either way. The gain is in resources: server-side OS threads drop from*.threadsto roughly the number of cores, and RSS drops at high concurrency. The tail latency grows slightly at very high concurrency. In short, this is a resource saving on the control plane that lets operators raisenimbus.thrift.threads/supervisor.thrift.threadsto absorb bursts without paying for native threads; with the shipped defaults it is neutral, and it is off by default.How was the change tested
StormThreadFactoryTest(flag on/off/missing/unexpected type, thread naming, forcedNORM_PRIORITY),WorkerStateTest(shared executor on virtual vs platform threads),AsyncLocalizerTest(download executor on virtual vs platform threads).AuthTestthrough a realThriftServerandNimbusClient, asserting from inside the Nimbus handler that it runs on a virtual thread when the flag is on (simple transport with and without a queue size, digest SASL transport) and on a platform thread by default; the digest case also checksReqContextstill carries the principal on a virtual handler thread.storm-clientandstorm-serversuites pass with the flag off (default): 650 and 487 tests respectively, 0 failures.-Dstorm.options=storm.virtual.threads.enabled=true:LocalNimbusTestpasses and the "platform thread by default" assertion inAuthTestfails as expected, proving the flag reaches the handler pool.