markap14 commented on code in PR #11730:
URL: https://github.com/apache/nifi/pull/11730#discussion_r4340751151
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/VirtualThreadSchedulingAgent.java:
##########
@@ -585,10 +1042,453 @@ private static OffsetDateTime getNextCronSchedule(final
OffsetDateTime currentSc
return cronExpression.next(now.isAfter(currentSchedule) ? now :
currentSchedule);
}
- private static class SchedulingGeneration {
+ private static class ProcessorTaskWorker {
+ private final long identifier;
+ private final AtomicBoolean retired = new AtomicBoolean();
+
+ private ProcessorTaskWorker(final long identifier) {
+ this.identifier = identifier;
+ }
+ }
+
+ private record WaitingProcessorTask(Thread thread, ProcessorTaskWorker
worker) {
+ }
+
+ private class ProcessorAutoSchedulingState implements
ConcurrentTaskMoveCoordinator.Participant {
+ private final Connectable connectable;
+ private final ConnectableTask connectableTask;
+ private final LifecycleState lifecycleState;
+ private final SchedulingGeneration generation;
+ private final int maxConcurrentTasks;
+ private final boolean batchingSupported;
+ private final boolean sourceProcessor;
+ private final StandardAutoSchedulingController controller;
+ private final ProcessorSchedulingMeasurements measurements = new
ProcessorSchedulingMeasurements();
+ private final AtomicReference<SchedulingSettings> schedulingSettings;
+ private final AtomicInteger activeProcessorInvocations = new
AtomicInteger();
+ private final ConcurrentMap<Long, ProcessorTaskWorker> workers = new
ConcurrentHashMap<>();
+ private final AtomicLong workerSequence = new AtomicLong();
+ private final AtomicBoolean sourceWorkCheckInProgress = new
AtomicBoolean();
+ private final AtomicLong resetSequence = new AtomicLong();
+ private final long evaluationIntervalNanos;
+ private final long baseSourceDelayNanos;
+ private final long maximumSourceDelayNanos;
+
+ private volatile long nextEvaluationNanos;
+ private volatile long lastEvaluationNanos = System.nanoTime();
+ private volatile long lastReadinessMeasurementNanos =
System.nanoTime();
+ private volatile long previousInputQueueSize;
+ private volatile long sourceDelayNanos;
+ private volatile long nextSourceWorkCheckNanos;
+ private volatile long lastSourceActivityNanos;
+ private volatile ProcessorSchedulingDecision lastDecision;
+ private volatile ProcessorSchedulingDecision lastChangeDecision;
+ private volatile ProcessorSchedulingSnapshot
lastProcessorSchedulingSnapshot;
+ private volatile boolean previousPrimaryNode;
+ private volatile boolean measurementsSupported;
+ private volatile double demandScore;
+
+ private ProcessorAutoSchedulingState(final Connectable connectable,
final ConnectableTask connectableTask,
+ final LifecycleState
lifecycleState, final SchedulingGeneration generation,
+ final int maxConcurrentTasks,
final boolean batchingSupported,
+ final boolean
processorTriggeredSerially) {
+ this.connectable = connectable;
+ this.connectableTask = connectableTask;
+ this.lifecycleState = lifecycleState;
+ this.generation = generation;
+ this.maxConcurrentTasks = maxConcurrentTasks;
+ this.batchingSupported = batchingSupported;
+ schedulingSettings = new AtomicReference<>(new
SchedulingSettings(1, batchingSupported ? TimeUnit.MILLISECONDS.toNanos(25L) :
0L));
+ sourceProcessor = connectable.isTriggerWhenEmpty() ||
!connectable.hasIncomingConnection() ||
!Connectables.hasNonLoopConnection(connectable);
+ controller = new
StandardAutoSchedulingController(maxConcurrentTasks, batchingSupported,
processorTriggeredSerially,
+ autoMaxConcurrentTasksBasedOnAvailableProcessors,
availableProcessorCount);
+ final double stagger =
((Math.floorMod(connectable.getIdentifier().hashCode(), 201) - 100) / 1000D);
+ this.evaluationIntervalNanos = (long)
(TimeUnit.SECONDS.toNanos(1L) * (1D + stagger));
+ this.nextEvaluationNanos = lastEvaluationNanos +
evaluationIntervalNanos;
+ this.baseSourceDelayNanos = Math.max(noWorkYieldNanos,
TimeUnit.MILLISECONDS.toNanos(1L));
+ this.maximumSourceDelayNanos = Math.max(baseSourceDelayNanos,
TimeUnit.MILLISECONDS.toNanos(100L));
+ this.previousPrimaryNode = flowController.isPrimary();
+ }
+
+ synchronized List<ProcessorTaskWorker> resizeWorkers(final int
targetWorkerCount) {
+ final List<ProcessorTaskWorker> activeWorkers = new ArrayList<>();
+ for (final ProcessorTaskWorker worker : workers.values()) {
+ if (!worker.retired.get()) {
+ activeWorkers.add(worker);
+ }
+ }
+
+ if (activeWorkers.size() > targetWorkerCount) {
+ activeWorkers.sort(Comparator.comparingLong(worker ->
worker.identifier));
+ for (int index = targetWorkerCount; index <
activeWorkers.size(); index++) {
+ activeWorkers.get(index).retired.set(true);
+ }
+
+ generation.signalChange();
+ generation.signalConcurrentTaskSlotChange();
+ return List.of();
+ }
+
+ final List<ProcessorTaskWorker> newWorkers = new ArrayList<>();
+ for (int index = activeWorkers.size(); index < targetWorkerCount;
index++) {
+ final ProcessorTaskWorker worker = new
ProcessorTaskWorker(workerSequence.getAndIncrement());
+ workers.put(worker.identifier, worker);
+ newWorkers.add(worker);
+ }
+
+ return newWorkers;
+ }
+
+ void workerStopped(final ProcessorTaskWorker worker) {
+ workers.remove(worker.identifier, worker);
+ generation.signalChange();
+ generation.signalConcurrentTaskSlotChange();
+ }
+
+ boolean tryAcquireConcurrentTaskSlot(final SchedulingSettings
expectedSettings) {
+ while (true) {
+ if (schedulingSettings.get() != expectedSettings) {
+ return false;
+ }
+
+ final int currentActiveProcessorInvocations =
activeProcessorInvocations.get();
+ if (currentActiveProcessorInvocations >=
expectedSettings.concurrentTasks()) {
+ return false;
+ }
+
+ if
(activeProcessorInvocations.compareAndSet(currentActiveProcessorInvocations,
currentActiveProcessorInvocations + 1)) {
+ if (schedulingSettings.get() == expectedSettings) {
+ return true;
+ }
+
+ activeProcessorInvocations.decrementAndGet();
+ generation.signalConcurrentTaskSlotChange();
+ return false;
+ }
+ }
+ }
+
+ void releaseConcurrentTaskSlot() {
+ activeProcessorInvocations.decrementAndGet();
+ generation.signalConcurrentTaskSlotChange();
+ }
+
+ boolean isSourceWorkCheckAllowed() {
+ return !sourceProcessor ||
connectableTask.hasLocallyConsumableInput() || sourceDelayNanos == 0L
+ || (System.nanoTime() >= nextSourceWorkCheckNanos &&
!sourceWorkCheckInProgress.get());
+ }
+
+ boolean reserveSourceWorkCheck() {
+ return requiresSourceWorkCheck() &&
sourceWorkCheckInProgress.compareAndSet(false, true);
+ }
+
+ boolean requiresSourceWorkCheck() {
+ return sourceProcessor &&
!connectableTask.hasLocallyConsumableInput() && sourceDelayNanos > 0L;
+ }
+
+ void releaseSourceWorkCheck() {
+ sourceWorkCheckInProgress.set(false);
+ generation.signalChange();
+ }
+
+ long getSourceWorkCheckDelayNanos() {
+ return Math.max(0L, nextSourceWorkCheckNanos - System.nanoTime());
+ }
+
+ void recordInvocationResult(final InvocationResult result) {
+ if (!sourceProcessor) {
+ return;
+ }
+
+ if (result.getOutcome() ==
InvocationOutcome.INVOKED_WITH_ACTIVITY) {
+ sourceDelayNanos = 0L;
+ nextSourceWorkCheckNanos = 0L;
+ lastSourceActivityNanos = System.nanoTime();
+ generation.signalChange();
+ } else if (result.getOutcome() ==
InvocationOutcome.INVOKED_WITHOUT_ACTIVITY
+ && activeProcessorInvocations.get() <= 1
+ && !connectableTask.hasLocallyConsumableInput()) {
+ sourceDelayNanos = sourceDelayNanos == 0L ?
baseSourceDelayNanos : Math.min(maximumSourceDelayNanos, sourceDelayNanos * 2L);
+ nextSourceWorkCheckNanos = System.nanoTime() +
sourceDelayNanos;
+ }
+ }
+
+ Instant getNextQueueDeadline() {
+ Instant earliestDeadline = Instant.EPOCH;
+ for (final FlowFileQueue queue : generation.queues) {
+ final Instant deadline =
queue.getNextFlowFileAvailabilityTime();
+ if (deadline.isAfter(Instant.EPOCH) &&
(earliestDeadline.equals(Instant.EPOCH) ||
deadline.isBefore(earliestDeadline))) {
+ earliestDeadline = deadline;
+ }
+ }
+
+ return earliestDeadline;
+ }
+
+ boolean isEvaluationDue(final long nowNanos) {
+ if (nowNanos < nextEvaluationNanos) {
+ return false;
+ }
+
+ nextEvaluationNanos = nowNanos + evaluationIntervalNanos;
+ return true;
+ }
+
+ void recordReadiness(final long nowNanos) {
+ final long measurementNanos = Math.max(1L, nowNanos -
lastReadinessMeasurementNanos);
+ lastReadinessMeasurementNanos = nowNanos;
+
measurements.recordReadiness(connectableTask.getReadinessOutcome(),
measurementNanos);
+ }
+
+ ProcessorSchedulingSnapshot captureProcessorSchedulingSnapshot(final
long nowNanos, final boolean globalCapacityTestAllowed) {
+ final SchedulingSettings currentSettings =
schedulingSettings.get();
+ long inputQueueSize = 0L;
+ for (final Connection connection :
connectable.getIncomingConnections()) {
Review Comment:
[Claude Opus 5.5] Good catch, fixed in
https://github.com/apache/nifi/pull/11730/commits/b8400cdc5a123ddfb39b767f7715db7a1eca758a.
`getLocalQueueSize()` now reads the queue's running size counter, which
includes swapped out FlowFiles, so it no longer builds diagnostics, scans the
active queue, or reads swap files. For load-balanced queues it returns the same
count for the local partition. I strengthened the existing queue tests to check
that swapped out FlowFiles are counted.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/VirtualThreadSchedulingAgent.java:
##########
@@ -54,23 +92,56 @@ public class VirtualThreadSchedulingAgent implements
SchedulingAgent {
private static final Logger logger =
LoggerFactory.getLogger(VirtualThreadSchedulingAgent.class);
private static final long PERMIT_POLL_INTERVAL_NANOS =
TimeUnit.SECONDS.toNanos(1L);
+ private static final long NON_BATCHED_PROCESSOR_BURST_NANOS =
TimeUnit.MILLISECONDS.toNanos(10L);
private final FlowController flowController;
private final RepositoryContextFactory contextFactory;
private final DynamicSemaphore globalSemaphore;
+ private final int autoMaxConcurrentTasks;
+ private final int availableProcessorCount;
+ private final boolean autoMaxConcurrentTasksBasedOnAvailableProcessors;
private final long noWorkYieldNanos;
private final ExecutorService executorService;
+ private final ScheduledExecutorService schedulingEvaluationExecutor;
+ private final SystemSchedulingMetrics systemSchedulingMetrics;
private final ConcurrentMap<String, SchedulingGeneration>
schedulingGenerations = new ConcurrentHashMap<>();
+ private final ConcurrentTaskMoveCoordinator concurrentTaskMoveCoordinator
= new ConcurrentTaskMoveCoordinator();
+ private final LongAdder globalPermitHoldNanos = new LongAdder();
+ private final LongAdder globalPermitWaitNanos = new LongAdder();
+ private final LongAdder globalTaskSlotsFullyUsedSamples = new LongAdder();
+ private final LongAdder globalTaskUsageSamples = new LongAdder();
private final AtomicBoolean shutdown = new AtomicBoolean();
private final AtomicInteger runningThreadCount = new AtomicInteger();
+ private volatile SystemSchedulingSnapshot systemSchedulingSnapshot =
SystemSchedulingSnapshot.createBuilder()
+ .setCpuLoad(-1D)
+ .setAverageCpuLoad(-1D)
+ .setMaxGlobalConcurrentTasks(1)
+ .build();
+ private volatile long lastSystemSchedulingSnapshotNanos =
System.nanoTime();
private volatile String adminYieldDuration = "1 sec";
private volatile long adminYieldNanos = TimeUnit.SECONDS.toNanos(1L);
public VirtualThreadSchedulingAgent(final FlowController flowController,
final RepositoryContextFactory contextFactory,
final NiFiProperties nifiProperties,
final int maxThreadCount) {
+ this(flowController, contextFactory, nifiProperties, maxThreadCount,
nifiProperties.getProcessorAutoMaxConcurrentTasks());
+ }
+
+ public VirtualThreadSchedulingAgent(final FlowController flowController,
final RepositoryContextFactory contextFactory,
+ final NiFiProperties nifiProperties,
final int maxThreadCount, final int autoMaxConcurrentTasks) {
+ this(flowController, contextFactory, nifiProperties, maxThreadCount,
autoMaxConcurrentTasks, new SystemSchedulingMetrics());
+ }
+
+ VirtualThreadSchedulingAgent(final FlowController flowController, final
RepositoryContextFactory contextFactory, final NiFiProperties nifiProperties,
+ final int maxThreadCount, final int
autoMaxConcurrentTasks, final SystemSchedulingMetrics systemSchedulingMetrics) {
+ this.systemSchedulingMetrics = systemSchedulingMetrics;
this.flowController = flowController;
this.contextFactory = contextFactory;
this.globalSemaphore = new DynamicSemaphore(maxThreadCount);
+ this.autoMaxConcurrentTasks = autoMaxConcurrentTasks;
+ availableProcessorCount = Runtime.getRuntime().availableProcessors();
+ final String configuredMaximum =
nifiProperties.getProperty(NiFiProperties.PROCESSOR_AUTO_MAX_CONCURRENT_TASKS);
Review Comment:
[Claude Opus 5.5] Yes, that could fail. Fixed in
https://github.com/apache/nifi/pull/11730/commits/b8400cdc5a123ddfb39b767f7715db7a1eca758a.
The value is now read through
`NiFiProperties.getProcessorAutoMaxConcurrentTasks()`, which uses
`getProperty(key, default)`, so a blank value falls back to the default of 12.
A value that is not a whole number, or is less than 1, fails with a clear
message naming the property. The limit based on CPU count is now applied in the
scheduling agent rather than in `NiFiProperties`.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java:
##########
@@ -1312,39 +1373,36 @@ public void onLocalNodeIdentifierSet(final
NodeIdentifier localNodeId) {
} finally {
partitionWriteLock.unlock();
}
+
+ notifySchedulingListeners();
}
@Override
public void onNodeStateChange(final NodeIdentifier nodeId, final
NodeConnectionState newState) {
- partitionWriteLock.lock();
- try {
- if (!offloaded) {
- switch (newState) {
- case OFFLOADING:
- onNodeRemoved(nodeId);
- break;
- case CONNECTED:
- onNodeAdded(nodeId);
- break;
- }
- } else {
- switch (newState) {
- case CONNECTED:
- if (nodeId != null &&
nodeId.equals(clusterCoordinator.getLocalNodeIdentifier())) {
- // the node with this queue was connected to
the cluster, make sure the queue is not offloaded
- resetOffloadedQueue();
- }
- break;
- case OFFLOADED:
- case OFFLOADING:
- case DISCONNECTED:
- case DISCONNECTING:
- onNodeRemoved(nodeId);
- break;
- }
+ if (!offloaded) {
Review Comment:
[Claude Opus 5.5] No, that wasn't intentional. Thanks for spotting it. Fixed
in
https://github.com/apache/nifi/pull/11730/commits/b8400cdc5a123ddfb39b767f7715db7a1eca758a:
the state transition in `onNodeStateChange` holds `partitionWriteLock` again,
so `resetOffloadedQueue()` can't run at the same time as `offloadQueue()`.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/VirtualThreadSchedulingAgent.java:
##########
@@ -158,6 +263,155 @@ public void schedule(final Connectable connectable, final
LifecycleState lifecyc
}
}
+ private void registerQueueListeners(final Connectable connectable, final
SchedulingGeneration generation) {
+ final Set<FlowFileQueue> queues = new HashSet<>();
+ for (final Connection connection :
connectable.getIncomingConnections()) {
+ queues.add(connection.getFlowFileQueue());
+ }
+
+ for (final Connection connection : connectable.getConnections()) {
+ queues.add(connection.getFlowFileQueue());
+ }
+
+ for (final FlowFileQueue queue : queues) {
+ generation.addQueue(queue);
+
generation.addQueueRegistration(queue.addSchedulingListener(generation::signalChange));
+ }
+ }
+
+ private void startAutoWorkers(final ProcessorAutoSchedulingState state,
final int desiredWorkerCount) {
+ final List<ProcessorTaskWorker> workers =
state.resizeWorkers(desiredWorkerCount);
+ for (final ProcessorTaskWorker worker : workers) {
+ final String threadName = buildThreadName(state.connectable,
Math.toIntExact(worker.identifier));
+ submitTask(threadName, state.generation, () ->
runAutoSchedulingLoop(state, worker));
+ }
+ }
+
+ private void runAutoSchedulingLoop(final ProcessorAutoSchedulingState
state, final ProcessorTaskWorker worker) {
Review Comment:
[Claude Opus 5.5] Yes, it should. Fixed in
https://github.com/apache/nifi/pull/11730/commits/b8400cdc5a123ddfb39b767f7715db7a1eca758a.
If an automatic scheduling worker hits an unexpected error, it now logs the
error, yields the Processor for the administrative yield duration, waits that
long, and then keeps going, the same way the existing scheduling loop does. It
stops only once the worker is no longer active.
--
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]
