Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-362: Ensure that we synchronize on ScheduleState before modifying it

This commit is contained in:
Mark Payne 2015-02-22 10:59:14 -05:00
parent 4cc106a54d
commit f1e74cc041
1 changed files with 14 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java
View File

@ -132,8 +132,13 @@ public class TimerDrivenSchedulingAgent implements SchedulingAgent {
final long yieldNanos = TimeUnit.MILLISECONDS.toNanos(yieldMillis); final long yieldNanos = TimeUnit.MILLISECONDS.toNanos(yieldMillis);
final ScheduledFuture<?> newFuture = flowEngine.scheduleWithFixedDelay(this, yieldNanos, final ScheduledFuture<?> newFuture = flowEngine.scheduleWithFixedDelay(this, yieldNanos,
connectable.getSchedulingPeriod(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS); connectable.getSchedulingPeriod(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS);
scheduleState.replaceFuture(scheduledFuture, newFuture);
futureRef.set(newFuture); synchronized (scheduleState) {
if ( scheduleState.isScheduled() ) {
scheduleState.replaceFuture(scheduledFuture, newFuture);
futureRef.set(newFuture);
}
}
} }
} else if ( shouldYield ) { } else if ( shouldYield ) {
// Component itself didn't yield but there was no work to do, so the framework will choose // Component itself didn't yield but there was no work to do, so the framework will choose
@ -149,8 +154,13 @@ public class TimerDrivenSchedulingAgent implements SchedulingAgent {
if (scheduledFuture.cancel(false)) { if (scheduledFuture.cancel(false)) {
final ScheduledFuture<?> newFuture = flowEngine.scheduleWithFixedDelay(this, NO_WORK_YIELD_NANOS, final ScheduledFuture<?> newFuture = flowEngine.scheduleWithFixedDelay(this, NO_WORK_YIELD_NANOS,
connectable.getSchedulingPeriod(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS); connectable.getSchedulingPeriod(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS);
scheduleState.replaceFuture(scheduledFuture, newFuture);
futureRef.set(newFuture); synchronized (scheduleState) {
if ( scheduleState.isScheduled() ) {
scheduleState.replaceFuture(scheduledFuture, newFuture);
futureRef.set(newFuture);
}
}
} }
} }
} }