NIFI-15862: Moved Processors to using Virtual Threads with a Semaphor… - #11164
Conversation
00c6167 to
c0cf8b2
Compare
64459cf to
8f34b35
Compare
| if (tempDirectory != null) { | ||
| try { | ||
| Files.deleteIfExists(tempDirectory); | ||
| } catch (final IOException ignored) { |
There was a problem hiding this comment.
Should we at least log this at debug so that a recurring failure to delete nifi-thread-dump-* directories is discoverable?
There was a problem hiding this comment.
Sure, that's reasonable.
| } | ||
|
|
||
| private static String buildThreadName(final Connectable connectable, final int taskIndex) { | ||
| return connectable.getName() + "[type=" + connectable.getComponentType() + ", id=" + connectable.getIdentifier() |
There was a problem hiding this comment.
What if the user renames the component and/or its parent group? would that be properly reflected?
There was a problem hiding this comment.
The thread name would get generated whenever the Processor is started. It would not change. The only ways to change it would be to track all Virtual Thread objects so that setName could update them, or to constantly update it to whatever the current name is. Both introduce a decent bit of complexity for what I'd argue is fairly minimal gain. Especially since I don't think we even allow changing the name of the processor while it's running from the UI. I think the REST tier does allow it, but the UI would not, though I may be mistaken.
|
[claude-opus-4.7] @pvillard31 thanks for the latest round of feedback. The new commit (2c0e041) addresses your remaining comments and rebases onto current
The earlier review-feedback commit (0abffe2) is also still on the branch and covers the prior round of comments: atomic schedule rollback (#3), thread tracking + |
|
[claude-opus-4.7] @markap14 important: I had to revert my last push. When you asked me to rebase against I have force-pushed the branch back to the previous tip What still needs to happen, in your hands:
- } catch (final IOException ignored) {
+ } catch (final IOException e) {
+ logger.debug(\"Failed to delete temporary thread-dump directory {}\", tempDirectory, e);
}(plus the
Sorry for the churn. Watcher subagent is stopping until you let me know how you'd like to proceed. |
c188bbc to
05608c8
Compare
|
I decided to go ahead and add something to nifi.properties that would allow us to enable/disable the virtual threads. This is a significant enough change that it's worth adding the ability to quickly disable if we find any issues. |
|
Thanks @markap14 - To clarify it would remain for now default off. We'd give it time to bake/prove itself then later consider swapping to default on. Then perhaps later remove the old model entirely? |
|
[claude-opus-4.7] CI failure on Failed check: Why this is unrelated to the PR:
Action taken: Reran only the failed |
|
[claude-opus-4.7] CI rerun on Failed check: Why this is most likely unrelated to the PR:
Action taken: Reran only the failed |
|
@joewitt the intent here is to introduce it with default = on. Users could explicitly opt out if they wanted to. While the code here is not particularly complex and so I don't foresee specific issues, it is a pretty crucial part of the NiFi Engine. So introducing the option in nifi.properties as an 'escape hatch' just in case any issues do arise. |
|
Automated review is marking this PR as stale due to lack of updates in the past four months. This PR will be closed in 15 days if the stale label is not removed. This stale label and automated closure does not indicate a judgement of the PR, just lack of reviewer bandwidth and helps us keep the PR queue more manageable. If you would like this PR re-opened you can do so and a committer can remove the stale label. Or you can open a new PR. Try to help review other PRs to increase PR review bandwidth which in turn helps yours. |
Co-authored-by: Cursor <cursoragent@cursor.com>
95dd04d to
1f5d274
Compare
exceptionfactory
left a comment
exceptionfactory
left a comment
There was a problem hiding this comment.
@markap14 I hope to take a closer look at this soon. One configuration recommendation, instead of a boolean, what about a scheduling strategy with either STANDARD or TIMER_DRIVEN as the default, and VIRTUAL as the new value?
|
[GPT-5.6 Sol] CI failure on Failed check: This failure is unrelated to the PR. The test and controller-service lifecycle implementation are unchanged. The test calls asynchronous Plan: rerun only this failed shard once run 34631800449 completes. |
Co-authored-by: Cursor <cursoragent@cursor.com>
|
[GPT-5.6 Sol] @exceptionfactory good suggestion. Commit c78511d replaces the boolean with I kept |
exceptionfactory
left a comment
exceptionfactory
left a comment
There was a problem hiding this comment.
Thanks for the quick response on the property configuration. I recommend calling the property nifi.scheduling.strategy and removing the new enum in favor of string values.
Co-authored-by: Cursor <cursoragent@cursor.com>
|
[GPT-5.6 Sol] @exceptionfactory addressed in f637504. The property is now |
|
Interrupt status kills the scheduling loop — share this This is a real behavioral change versus the current pool. Today, TimerDrivenSchedulingAgent submits a short-lived Runnable to FlowEngine. After onTrigger returns, that task is done. The next fire is a new executor task on a worker that has a clean interrupt flag. Processors that catch InterruptedException and do the textbook Thread.currentThread().interrupt() are therefore fine. On this PR, each concurrent task is one virtual thread that never dies: invoke() → waitForDelay() → acquirePermitWithPolling() → invoke() … If onTrigger leaves the interrupt flag set: waitForDelay() hits CountDownLatch.await and immediately throws InterruptedException. This is not exotic. NiFi itself does this (e.g. DebugFlow, ExecuteProcess, ConsumeKafka backlog path, AMQP, Kinesis, HDFS, Listen processors). Any extension that “restores interrupt status and returns” is now a silent death sentence for that scheduling chain. Ask Mark to either: clear interrupt at a defined framework boundary after onTrigger unless shutdown/unschedule requested the interrupt, or |
|
@markap14 left three chunky comments from various findings after wrestling through it with Cursor. Ignored the ones that didn't seem worthy and focused on what look like real findings. I'll keep an eye on this important change and will get to hands on live testing too once we're a little further. |
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
|
[GPT-5.6 Sol] @joewitt seriously, thank you for putting so much time into such a careful review. This was far beyond a drive-by "looks fine"—you found a real interrupt-shaped gremlin and gave the Java 21 risk a proper workout. That kind of effort keeps the scheduling goblins from escaping into production, and it is hugely appreciated. The first two findings are addressed in the latest commits:
Your permit-wait polling note was also accurate. We ultimately kept the simple bounded wait but increased it from 25 milliseconds to one second in 947b12a, reducing timeout wakeups by 40× without delaying normal permit handoff. |
Co-authored-by: Cursor <cursoragent@cursor.com>
|
[GPT-5.6 Sol] @joewitt follow-up on permit waiting: after reviewing the targeted wake-up approach, we chose to keep the simpler bounded wait and increased the interval from 25 milliseconds to 1 second in 947b12a. |
What I am asking: the shipped default is now AUTO, which means Java 21 still runs the existing platform-thread scheduler. But every system-test nifi.properties on this branch sets nifi.scheduling.strategy=VIRTUAL (default, pythonic, both cluster nodes). VirtualThreadStartStopCycleIT also forces VIRTUAL. So CI on Java 21 will exercise the new engine, while a real Java 21 install with an untouched config will exercise the old one. There is no system-test graph that starts, stops, clusters, or fails over on STANDARD or on AUTO as Java 21 will actually resolve it. Why it matters: we now have two production schedulers for a while. Virtual-thread coverage on Java 21 CI is useful bake — I would keep that. What we lose is regression coverage of the path most Java 21 operators will actually run, plus the resolution of AUTO itself (unit-tested only). Start/stop, run-once, terminate, clustered primary-node, and “disable the controller service then delete it” are exactly the races that differ between a delayed-future pool and a long-lived virtual thread. If those only run under VIRTUAL, we will not notice if STANDARD was accidentally broken by the wiring change, and we will not notice if AUTO on Java 21 fails to select standard scheduling in a real process. A small matrix is enough: leave most shards on VIRTUAL so the new engine gets the heat, and keep at least one Java 21 system-test profile on STANDARD or unset/AUTO. The AUTO resolver already has a unit test; an integration assertion that the running node logged “platform thread pool” on Java 21 and “virtual threads” on Java 25 would close the loop.
What I am asking: when VIRTUAL (or AUTO on Java 25) is selected, Processors and Reporting Tasks no longer run on the timer-driven FlowEngine. That pool is still created at nifi.flowcontroller.maxTimerDrivenThreadCount, and raising max threads in the UI still calls setCorePoolSize on it and resizes the virtual-thread semaphore. The “active timer-driven thread” metric is the sum of the two. Why it matters: today, max threads = 200 means about 200 platform threads that are the component workers. After this change, max threads = 200 means 200 semaphore permits for component work plus a 200-thread platform pool that is only used for a handful of framework periodic tasks (status history snapshot, remote process group refresh, registry sync, Python extension discovery, analytics predictions). Those jobs do not need 200 workers. Pierre raised this earlier, and the earlier branch had the leftover pool as a small framework engine. The current rewrite puts the old size back. That has three practical effects. Operators who raise max threads to feed a large canvas also stand up a large idle platform pool they no longer need. The thread-count statistic is no longer “how many component tasks are running”; it can show framework pool activity plus permit holders, so a quiet canvas can still look busy, or a busy canvas can look busier than the cap. Shutdown waits on the platform pool first, then the virtual-thread executor, so a large leftover pool can eat the graceful-stop budget before component threads are waited on. The max-thread property should mean one thing: how many Processor/Reporting Task invocations may run at once. Under virtual-thread scheduling that is the semaphore. The leftover FlowEngine should be a small, fixed framework pool (something like the 8 you had locally before), not resized with max threads, and the active-thread metric should not add the two budgets together as if they were the same cap. |
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
|
[GPT-5.6 Sol] @joewitt you were right on both counts. Thank you again for putting this much real effort into the review—and for continuing to spelunk through the scheduler cave with a flashlight and an alarmingly effective goblin detector. This is exactly the kind of quality review that makes a change like this safer. Addressed in the latest commits:
The 47 focused framework tests pass, as do Checkstyle and PMD. Fresh CI will run the complete Java 21/25 system-test matrix. |
|
I have never felt so appreciated. " Thank you again for putting this much real effort into the review—and for continuing to spelunk through the scheduler cave with a flashlight and an alarmingly effective goblin detector. This is exactly the kind of quality review that makes a change like this safer." Classic. |
|
[GPT-5.6 Sol] The Ubuntu Java 21 integration check failed in the unchanged SMB extension: This is a pre-existing timing flake: the exact same assertion failed on an unrelated PR two days ago, and this branch does not change the SMB processor or its test framework. The test compares millisecond JVM time with timestamps returned by the SMB container; the container timestamp can fall behind the recorded boundary because of its timestamp precision. I have rerun only the failed integration-test workflow. |
|
Whoa. Testing a GenerateFF to UpdateAttr flow before this PR yields 150,000 EPS on my laptop. With this PR it jumps to 250,000 EPS. The virtual threads provide at least in this scenario dramatically better lock handling/hand-offs. The bottleneck now is just lock contention on the queue/connection when pulling flowfiles. Wow. |
|
Put a second pair and can get to 450K EPS which saturates CPU. |
Follow-on notes from validating NIFI-15862 (virtual-thread scheduling)These notes are not a review of the scheduling-agent change itself. I have been running Mark’s Nothing below is meant to block or expand this PR. The lock / map / comparator patches are local follow-on experiments. What we were trying to learn
Test setup
How we measuredNiFi’s processor status We therefore treated:
Results on Mark’s code (before any follow-on patches)
So: virtual threads do let two independent CPU-bound loops run together. A third pair did not buy a third of the machine; the ceiling was not disk and not “we ran out of platform threads.” JFR (3-pair steal, still fair queue locks)Ruled out:
Ruled in (hot / park stacks):
The scheduler itself was not the thing we were parked on. Follow-on changes (local, for a later PR)All of these sit on top of NIFI-15862. They are engine-path tweaks that VTs made expensive because many more tasks actually run the put/poll path at once. 1. Non-fair lock on
|
|
Saved at # Virtual-thread scheduling: 1,000-pair fairness run
Follow-on to the earlier NIFI-15862 notes. Same branch (`nifi-15862-virtual-threads`), same machine, same “do not cheat provenance” rule. This run is about whether the **global permit cap** still gives every component a fair turn when the canvas is far larger than the cap — including mixed scheduling periods.
## Setup
- **NiFi:** 2.12.0-SNAPSHOT, `nifi.scheduling.strategy=AUTO` → virtual threads (Java 25)
- **JDK:** Azul Zulu 25, macOS, 16 cores, heap 16g
- **Repos:** `VolatileFlowFileRepository` + `VolatileProvenanceRepository` (CPU/scheduling, not disk)
- **Controller cap:** Maximum Timer Driven Thread Count = **128** (global `DynamicSemaphore` permits, not a platform pool)
- **Flow:** **1,000 independent pairs** = **2,000 processors**, **1,000 connections**
- GenerateFlowFile → UpdateAttribute
- 0-byte FlowFiles, Generate **batch 50**
- **1 concurrent task** on every processor
- UpdateAttribute `benchmark=true`, run duration 25 ms, `success` auto-terminated
- connection backpressure 15,000 / 1 GB
- layout: 5 pairs across, Generate above UpdateAttribute, same spacing as the original three pairs
Built up in stages (3 → 20 → 220 → 1,000) with the same pair shape so we could see fairness at each scale.
## Scheduling mix (all 1,000 pairs)
Generate and UpdateAttribute in a pair share the same period. Assignment is `n % 10` so cadences are interleaved across the canvas, not parked in one corner.
| Period | Pairs | Share |
|---|---|---|
| 0 sec (as fast as possible) | 100 | 10% |
| 50 ms | 100 | 10% |
| 100 ms | 100 | 10% |
| 500 ms | 100 | 10% |
| 750 ms | 100 | 10% |
| 1 sec | 500 | 50% |
**2,000** scheduled tasks vs **128** permits (~15× oversubscribed).
## Findings
### Uniform pairs (before mixing periods)
At 1+1 concurrent tasks, batch 50, all 0 sec:
| Scale | Combined Generate FF/s | Per-pair | max/min | Notes |
|---|---|---|---|---|
| 20 pairs | ~549k | ~27.2k–27.8k | 1.02 | queues often at 15k; Generate yielded; **not** permit-bound (64 cap unused) |
| 220 pairs | ~630k | 2,858–2,865 | **1.002** | cap **pegged** (~127/128); 1–20 vs 21–220 same mean |
| 1,000 pairs | ~610k | 609.3–610.5 | **1.002** | stdev **0.19 FF/s**; slowest **99.9%** of equal share |
Raising the cap **64 → 128** did **not** double throughput (still ~600k). It let more tasks sit in `onTrigger` at once. Limit is **16 cores + engine work**, not a few pairs hogging the semaphore.
30s deltas of NiFi’s 5-minute counters will go negative when that window is sliding. Compare pairs with **5-minute count / 300**, or treat a uniform delta as “everyone moved together.” Generate **out** vs UpdateAttribute **in** also looks “starved vs busy” in the UI; that is processor type, not unfairness.
### Mixed periods (1,000 pairs)
5-minute averages, ~6 minutes after applying the mix. All 2,000 processors Running. Combined **~600k FF/s**. Generate FF/s ≈ tasks/s × 50 (batch size is real).
| Period | Pairs | Expected Generate Hz | Actual Hz | Of expected | Actual FF/s / pair | Expected FF/s |
|---|---|---|---|---|---|---|
| 0 sec | 100 | max | **86.6** | — | **4,329** | — |
| 50 ms | 100 | 20 | **16.3** | **81%** | 813 | 1,000 |
| 100 ms | 100 | 10 | **9.0** | **90%** | 449 | 500 |
| 500 ms | 100 | 2 | **1.96** | **98%** | 98 | 100 |
| 750 ms | 100 | 1.33 | **1.32** | **99%** | 66 | 67 |
| 1 sec | 500 | 1 | **0.99** | **99%** | 50 | 50 |
Within every bucket, max/min ≈ **1.00**. **Zero** of 500 one-second Generates were below 0.5 Hz or above 1.5 Hz.
Interpretation:
- **Slow pairs are not starved off the board.** 1 s / 750 ms / 500 ms hit their deadlines.
- **50 ms / 100 ms miss some deadlines** because they compete with 100 always-runnable 0-sec pairs.
- Those 0-sec pairs are 10% of pairs (~10% of processors) but held **~97 of ~133** active tasks and **~72%** of FlowFiles/s (433k of 600k). That is expected: the fair permit semaphore gives turns to whoever is runnable; 0-sec is always runnable.
- UpdateAttribute **task** counts on slow pairs are **not** a schedule check. Empty queue uses **bored yield (~10 ms)**, not the 1 s period, so UA tasks include no-work polls. **FlowFiles in** still tracks Generate.
## What this says about VT readiness
1. **The scheduler scales past the cap without collapsing into winners and losers.** 2,000 looping virtual threads vs 128 permits stayed even to three decimals on FlowFiles/s.
2. **Max Timer Driven Thread Count still means something** — it is a global concurrency throttle, not “create 128 platform threads.” Oversubscribe it and combined FF/s stays in the same ~600k band; each pair just gets a thinner slice.
3. **Timer-driven periods still work** under that oversubscription. One-second components kept ~1 Hz. The miss rate on 50–100 ms is permit competition with 0-sec work, the same class of effect you would get from a small platform pool — not VT-specific unfairness.
4. **Yield / no-work is a sleep**, not cancel-and-resubmit. Mixed cadences did not require future juggling. Stop/start of 2,000 processors was routine.
5. **We did not see pinning** on this path in earlier JFR; this 1,000-pair run did not re-profile, but behavior is consistent with “park on permit, run invoke, release.”
6. Combined throughput did **not** grow with pair count (20 → 220 → 1,000). That is **cores + engine** (queues, maps, provenance), not evidence the VT agent failed. Two stacked pairs earlier **did** add; that was the VT win. This run is the fairness/oversubscription win.
## Caveats (same as before)
- One Mac, 16 cores, volatile repos, 0-byte UpdateAttribute loop.
- Not a WAL / persistent provenance / real content soak.
- Not a head-to-head TIMER_DRIVEN vs AUTO A/B at 1,000 pairs.
- Default `AUTO` remains the right upgrade default (VT on Java 25+, platform pool on 21–24). `STANDARD` stays the escape hatch.
**Bottom line:** with 1,000 pairs, mixed periods, and a 128-permit cap, virtual-thread scheduling looked **ready** for the property it was meant to have — many concurrent tasks, a simple global bound, and a fair shake for slow timers rather than a few 0-sec processors owning the engine. |
|
ok @markap14 this was a lot of fun to review and experiment with. While I dont feel confident enough abou the true code itself and would like someone such as @exceptionfactory to give it a final review - i am +1 at this point after reasonably detailed testing and experimentation. Various results shared about. The first large markdown dump is from going beyond this and suggesting some follow-on changes. The second big block is from really focusing on behavior of scheduling/thread starvation/timing concerns. It holds up well. I did my testing on Java 25. I didn't focus on Java 21. Since we are default off on Java 21 and default on with Java 25 I think that is fine. I also ran the entire test suite/portfolio over the span of a copule hours and everything came up green. |
exceptionfactory
left a comment
exceptionfactory
left a comment
There was a problem hiding this comment.
Thanks for the work on this @markap14, this looks close to completion. I noted a few minor recommendations, and a couple questions regarding Thread-tracking for diagnostics.
| waitForDelay(sleepNanos, generation); | ||
| } | ||
|
|
||
| private boolean acquirePermitWithPolling(final LifecycleState lifecycleState, final SchedulingGeneration generation) { |
There was a problem hiding this comment.
Are there any synchronization concerns with this method being called from a stop and from runOnce? It seems possible to have a race condition based on the generation, but not quite clear.
There was a problem hiding this comment.
[GPT-5.6 Sol] I walked through the Stop and Run Once interleavings and this is safe. Registration and explicit unscheduling are serialized on LifecycleState. Cleanup uses ConcurrentMap.remove(componentId, generation), so completion of an older Run Once cannot remove a newer registered generation. It then stops only the captured generation, and SchedulingGeneration.stop() is idempotent. No synchronization change is needed here.
There was a problem hiding this comment.
Thanks for the reply. For clarity, I was referring to a race between acquirePermitWithPolling() and invoke(). Can you take a closer look at the relationship between those method calls?
There was a problem hiding this comment.
Thanks for clarifying. You were right: the final active check in acquirePermitWithPolling() could succeed, followed by an unschedule before ConnectableTask registered the invocation as active. ConnectableTask now calls LifecycleState.tryIncrementActiveThreadCount(), which checks the scheduled state and registers the invocation under the same lifecycle monitor used by unscheduling. If unscheduling wins, onTrigger() is skipped; if registration wins, the stop sequence sees an active invocation and waits for it. I added a deterministic regression test that pauses after permit acquisition and confirmed it failed with one invocation before this correction and now passes with zero.
Co-authored-by: Cursor <cursoragent@cursor.com>
|
[GPT-5.6 Sol] The Scan check failed before project compilation while resolving the parent POM: This is a transient dependency-resolution failure unrelated to this branch. Apache parent 39 comes from current |
|
[GPT-5.6 Sol] The workflow has completed, so I have now rerun the failed Scan check. As noted above, the original attempt failed before compilation when Maven Central temporarily reported |
exceptionfactory
left a comment
exceptionfactory
left a comment
There was a problem hiding this comment.
Thanks for the updates @markap14, this looks close to completion. I noted a couple remaining items
| waitForDelay(sleepNanos, generation); | ||
| } | ||
|
|
||
| private boolean acquirePermitWithPolling(final LifecycleState lifecycleState, final SchedulingGeneration generation) { |
There was a problem hiding this comment.
Thanks for the reply. For clarity, I was referring to a race between acquirePermitWithPolling() and invoke(). Can you take a closer look at the relationship between those method calls?
Co-authored-by: Cursor <cursoragent@cursor.com>
|
[gpt-sol-5.6] The macos-15 Java 21 system-test shard failed in I downloaded and checked the troubleshooting archive. This is a pre-existing timing flaw in this unrelated Connector test: it waits only for both shared output files to exist before reading them, while either clustered node can create or truncate a file before writing its content. The node logs show both connectors starting successfully and both nodes subsequently writing 23 bytes to the same sensitive-output path, with no NiFi error. I have rerun only the failed shard. |
Co-authored-by: Cursor <cursoragent@cursor.com>
|
[gpt-sol-5.6] The Ubuntu Java 25 system-test job hit a pre-existing The troubleshooting logs show the Processors stopped successfully in 111 ms. During the offload, cluster coordinator leadership moved from node 2 to node 1; node 2 completed its local offload but could not change its local coordinator state from |
|
[gpt-sol-5.6] I have rerun the failed Ubuntu Java 25 system-test job. As noted above, the saved logs show that Processor stopping completed successfully and the timeout followed a cluster coordinator change during node offload; the same |
exceptionfactory
left a comment
exceptionfactory
left a comment
There was a problem hiding this comment.
Thanks for the work on this @markap14, and thanks for the helpful reviews @joewitt and @pvillard31, the latest version looks good!
4 participants
…e bounding how many tasks can be run at once
Summary
NIFI-00000
Tracking
Please complete the following tracking steps prior to pull request creation.
Issue Tracking
Pull Request Tracking
NIFI-00000NIFI-00000VerifiedstatusPull Request Formatting
mainbranchVerification
Please indicate the verification steps performed prior to pull request creation.
Build
./mvnw clean install -P contrib-checkLicensing
LICENSEandNOTICEfilesDocumentation