You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2020/04/06 06:39:24 UTC
[camel] branch master updated: Camel-Base: Use
AtomicInteger/AtomicLong instead of volatile primitives,
because increments on primitive fields are not atomic operations and can
lead to lost updates.
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/master by this push:
new 074cd4b Camel-Base: Use AtomicInteger/AtomicLong instead of volatile primitives, because increments on primitive fields are not atomic operations and can lead to lost updates.
new 31f01fb Merge pull request #3712 from PascalSchumacher/atomic_integer
074cd4b is described below
commit 074cd4b0b6f0d12e9717bc1640637fd33f6f0d9b
Author: Pascal Schumacher <pa...@gmx.net>
AuthorDate: Sun Apr 5 20:54:13 2020 +0200
Camel-Base: Use AtomicInteger/AtomicLong instead of volatile primitives, because increments on primitive fields are not atomic operations and can lead to lost updates.
---
.../camel/impl/engine/DurationRoutePolicy.java | 9 +++---
.../health/RoutePerformanceCounterEvaluators.java | 37 ++++++++++------------
.../org/apache/camel/processor/SendProcessor.java | 10 +++---
3 files changed, 28 insertions(+), 28 deletions(-)
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/engine/DurationRoutePolicy.java b/core/camel-base/src/main/java/org/apache/camel/impl/engine/DurationRoutePolicy.java
index ee208cf..73e470d 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/engine/DurationRoutePolicy.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/engine/DurationRoutePolicy.java
@@ -20,6 +20,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
@@ -46,8 +47,8 @@ public class DurationRoutePolicy extends org.apache.camel.support.RoutePolicySup
private String routeId;
private ScheduledExecutorService executorService;
private volatile ScheduledFuture task;
- private volatile int doneMessages;
- private AtomicBoolean actionDone = new AtomicBoolean();
+ private final AtomicInteger doneMessages = new AtomicInteger();
+ private final AtomicBoolean actionDone = new AtomicBoolean();
private Action action = Action.STOP_ROUTE;
private int maxMessages;
@@ -129,9 +130,9 @@ public class DurationRoutePolicy extends org.apache.camel.support.RoutePolicySup
@Override
public void onExchangeDone(Route route, Exchange exchange) {
- doneMessages++;
+ int newDoneMessages = doneMessages.incrementAndGet();
- if (maxMessages > 0 && doneMessages >= maxMessages) {
+ if (maxMessages > 0 && newDoneMessages >= maxMessages) {
if (actionDone.compareAndSet(false, true)) {
performMaxMessagesAction();
if (task != null && !task.isDone()) {
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/health/RoutePerformanceCounterEvaluators.java b/core/camel-base/src/main/java/org/apache/camel/impl/health/RoutePerformanceCounterEvaluators.java
index 0913c54..5f7f903 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/health/RoutePerformanceCounterEvaluators.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/health/RoutePerformanceCounterEvaluators.java
@@ -17,6 +17,7 @@
package org.apache.camel.impl.health;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.api.management.mbean.ManagedRouteMBean;
import org.apache.camel.health.HealthCheckResultBuilder;
@@ -157,8 +158,7 @@ public final class RoutePerformanceCounterEvaluators {
public static final class LastProcessingTime implements PerformanceCounterEvaluator<ManagedRouteMBean> {
private final long timeThreshold;
private final int failuresThreshold;
-
- private volatile int failureCount;
+ private final AtomicInteger failureCount = new AtomicInteger();
public LastProcessingTime(long timeThreshold, int failuresThreshold) {
this.timeThreshold = timeThreshold;
@@ -170,13 +170,13 @@ public final class RoutePerformanceCounterEvaluators {
try {
long value = counter.getLastProcessingTime();
if (value > timeThreshold) {
- failureCount++;
+ int newFailureCount = failureCount.incrementAndGet();
- if (failureCount > failuresThreshold) {
+ if (newFailureCount > failuresThreshold) {
builder.down();
}
} else {
- failureCount = 0;
+ failureCount.set(0);
}
builder.detail("exchanges.last-processing-time", value);
@@ -190,8 +190,7 @@ public final class RoutePerformanceCounterEvaluators {
public static final class MinProcessingTime implements PerformanceCounterEvaluator<ManagedRouteMBean> {
private final long timeThreshold;
private final int failuresThreshold;
-
- private volatile int failureCount;
+ private final AtomicInteger failureCount = new AtomicInteger();
public MinProcessingTime(long timeThreshold, int failuresThreshold) {
this.timeThreshold = timeThreshold;
@@ -203,13 +202,13 @@ public final class RoutePerformanceCounterEvaluators {
try {
long value = counter.getMinProcessingTime();
if (value > timeThreshold) {
- failureCount++;
+ int newFailureCount = failureCount.incrementAndGet();
- if (failureCount > failuresThreshold) {
+ if (newFailureCount > failuresThreshold) {
builder.down();
}
} else {
- failureCount = 0;
+ failureCount.set(0);
}
builder.detail("exchanges.min-processing-time", value);
@@ -223,8 +222,7 @@ public final class RoutePerformanceCounterEvaluators {
public static final class MeanProcessingTime implements PerformanceCounterEvaluator<ManagedRouteMBean> {
private final long timeThreshold;
private final int failuresThreshold;
-
- private volatile int failureCount;
+ private final AtomicInteger failureCount = new AtomicInteger();
public MeanProcessingTime(long timeThreshold, int failuresThreshold) {
this.timeThreshold = timeThreshold;
@@ -236,13 +234,13 @@ public final class RoutePerformanceCounterEvaluators {
try {
long value = counter.getMeanProcessingTime();
if (value > timeThreshold) {
- failureCount++;
+ int newFailureCount = failureCount.incrementAndGet();
- if (failureCount > failuresThreshold) {
+ if (newFailureCount > failuresThreshold) {
builder.down();
}
} else {
- failureCount = 0;
+ failureCount.set(0);
}
builder.detail("exchanges.mean-processing-time", value);
@@ -256,8 +254,7 @@ public final class RoutePerformanceCounterEvaluators {
public static final class MaxProcessingTime implements PerformanceCounterEvaluator<ManagedRouteMBean> {
private final long timeThreshold;
private final int failuresThreshold;
-
- private volatile int failureCount;
+ private final AtomicInteger failureCount = new AtomicInteger();
public MaxProcessingTime(long timeThreshold, int failuresThreshold) {
this.timeThreshold = timeThreshold;
@@ -269,13 +266,13 @@ public final class RoutePerformanceCounterEvaluators {
try {
long value = counter.getMaxProcessingTime();
if (value > timeThreshold) {
- failureCount++;
+ int newFailureCount = failureCount.incrementAndGet();
- if (failureCount > failuresThreshold) {
+ if (newFailureCount > failuresThreshold) {
builder.down();
}
} else {
- failureCount = 0;
+ failureCount.set(0);
}
builder.detail("exchanges.max-processing-time", value);
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/SendProcessor.java b/core/camel-base/src/main/java/org/apache/camel/processor/SendProcessor.java
index 48455e5..c9164e4 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/SendProcessor.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/SendProcessor.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.processor;
+import java.util.concurrent.atomic.AtomicLong;
+
import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProducer;
import org.apache.camel.Endpoint;
@@ -56,7 +58,7 @@ public class SendProcessor extends AsyncProcessorSupport implements Traceable, E
protected ExchangePattern destinationExchangePattern;
protected String id;
protected String routeId;
- protected volatile long counter;
+ protected final AtomicLong counter = new AtomicLong();
public SendProcessor(Endpoint destination) {
this(destination, null);
@@ -121,7 +123,7 @@ public class SendProcessor extends AsyncProcessorSupport implements Traceable, E
// if you want to permanently to change the MEP then use .setExchangePattern in the DSL
final ExchangePattern existingPattern = exchange.getPattern();
- counter++;
+ counter.incrementAndGet();
// if we have a producer then use that as its optimized
if (producer != null) {
@@ -199,11 +201,11 @@ public class SendProcessor extends AsyncProcessorSupport implements Traceable, E
}
public long getCounter() {
- return counter;
+ return counter.get();
}
public void reset() {
- counter = 0;
+ counter.set(0);
}
@Override