Merge branch '2.1.x'

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/WeightCalculatorWebFilter.java

@@ -48,6 +48,7 @@ import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.W
 
 /**
  * @author Spencer Gibb
+ * @author Alexey Nakidkin
  */
 public class WeightCalculatorWebFilter
 		implements WebFilter, Ordered, SmartApplicationListener {
@@ -175,38 +176,40 @@ public class WeightCalculatorWebFilter
 			groupWeights.put(group, c);
 		}
 		GroupWeightConfig config = c;
-		config.weights.put(weightConfig.getRouteId(), weightConfig.getWeight());
+		synchronized (config) {
+			config.weights.put(weightConfig.getRouteId(), weightConfig.getWeight());
 
-		// recalculate
+			// recalculate
 
-		// normalize weights
-		int weightsSum = config.weights.values().stream().mapToInt(Integer::intValue)
-				.sum();
+			// normalize weights
+			int weightsSum = config.weights.values().stream().mapToInt(Integer::intValue)
+					.sum();
 
-		final AtomicInteger index = new AtomicInteger(0);
-		config.weights.forEach((routeId, weight) -> {
-			Double nomalizedWeight = weight / (double) weightsSum;
-			config.normalizedWeights.put(routeId, nomalizedWeight);
+			final AtomicInteger index = new AtomicInteger(0);
+			config.weights.forEach((routeId, weight) -> {
+				Double nomalizedWeight = weight / (double) weightsSum;
+				config.normalizedWeights.put(routeId, nomalizedWeight);
 
-			// recalculate rangeIndexes
-			config.rangeIndexes.put(index.getAndIncrement(), routeId);
-		});
+				// recalculate rangeIndexes
+				config.rangeIndexes.put(index.getAndIncrement(), routeId);
+			});
 
-		// TODO: calculate ranges
-		config.ranges.clear();
+			// TODO: calculate ranges
+			config.ranges.clear();
 
-		config.ranges.add(0.0);
+			config.ranges.add(0.0);
 
-		List<Double> values = new ArrayList<>(config.normalizedWeights.values());
-		for (int i = 0; i < values.size(); i++) {
-			Double currentWeight = values.get(i);
-			Double previousRange = config.ranges.get(i);
-			Double range = previousRange + currentWeight;
-			config.ranges.add(range);
-		}
+			List<Double> values = new ArrayList<>(config.normalizedWeights.values());
+			for (int i = 0; i < values.size(); i++) {
+				Double currentWeight = values.get(i);
+				Double previousRange = config.ranges.get(i);
+				Double range = previousRange + currentWeight;
+				config.ranges.add(range);
+			}
 
-		if (log.isTraceEnabled()) {
-			log.trace("Recalculated group weight config " + config);
+			if (log.isTraceEnabled()) {
+				log.trace("Recalculated group weight config " + config);
+			}
 		}
 	}
 
@@ -230,18 +233,20 @@ public class WeightCalculatorWebFilter
 
 			double r = this.random.nextDouble();
 
-			List<Double> ranges = config.ranges;
+			synchronized (config) {
+				List<Double> ranges = config.ranges;
 
-			if (log.isTraceEnabled()) {
-				log.trace("Weight for group: " + group + ", ranges: " + ranges + ", r: "
-						+ r);
-			}
+				if (log.isTraceEnabled()) {
+					log.trace("Weight for group: " + group + ", ranges: " + ranges
+							+ ", r: " + r);
+				}
 
-			for (int i = 0; i < ranges.size() - 1; i++) {
-				if (r >= ranges.get(i) && r < ranges.get(i + 1)) {
-					String routeId = config.rangeIndexes.get(i);
-					weights.put(group, routeId);
-					break;
+				for (int i = 0; i < ranges.size() - 1; i++) {
+					if (r >= ranges.get(i) && r < ranges.get(i + 1)) {
+						String routeId = config.rangeIndexes.get(i);
+						weights.put(group, routeId);
+						break;
+					}
 				}
 			}
 		}

spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/WeightCalculatorWebFilterConcurrentTests.java

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2013-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.gateway.filter;
+
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import io.netty.util.internal.ThreadLocalRandom;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import reactor.core.publisher.Mono;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.gateway.event.WeightDefinedEvent;
+import org.springframework.cloud.gateway.support.WeightConfig;
+import org.springframework.cloud.gateway.test.BaseWebClientTests;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.annotation.Import;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.web.server.ServerWebExchange;
+import org.springframework.web.server.WebFilterChain;
+
+import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;
+
+/**
+ * Bug #1459: WeightCalculatorWebFilter is not thread safe
+ *
+ * @author Alexey Nakidkin
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(webEnvironment = RANDOM_PORT)
+@ActiveProfiles("weights-concurrent")
+@DirtiesContext
+public class WeightCalculatorWebFilterConcurrentTests {
+
+	@Value("${test.concurrent.execution.timeInSeconds}")
+	private int maxTestTimeSeconds;
+
+	@Autowired
+	private ApplicationEventPublisher eventPublisher;
+
+	@Autowired
+	private WeightCalculatorWebFilter weightCalculatorWebFilter;
+
+	private ExecutorService executorService;
+
+	private long startTime;
+
+	@Before
+	public void setUp() {
+		executorService = Executors.newSingleThreadExecutor();
+		startTime = System.currentTimeMillis();
+	}
+
+	@Test
+	public void WeightCalculatorWebFilter_threadSafeTest() {
+		generateEvents();
+
+		ServerWebExchange serverWebExchangeMock = Mockito.mock(ServerWebExchange.class);
+		WebFilterChain emptyWebFilterChain = serverWebExchange -> Mono.empty();
+
+		while (isContinue()) {
+			weightCalculatorWebFilter.filter(serverWebExchangeMock, emptyWebFilterChain);
+		}
+	}
+
+	private boolean isContinue() {
+		return (System.currentTimeMillis() - startTime) <
+				TimeUnit.SECONDS.toMillis(maxTestTimeSeconds);
+	}
+
+	private void generateEvents() {
+		executorService.execute(() -> {
+			while (isContinue()) {
+				eventPublisher.publishEvent(createWeightDefinedEvent());
+			}
+		});
+	}
+
+	private WeightDefinedEvent createWeightDefinedEvent() {
+		int weight = ThreadLocalRandom.current().nextInt() & Integer.MAX_VALUE;
+		WeightConfig config = new WeightConfig("group_1", UUID.randomUUID().toString(),
+				weight);
+		return new WeightDefinedEvent(new Object(), config);
+	}
+
+	@EnableAutoConfiguration
+	@SpringBootConfiguration
+	@Import(BaseWebClientTests.DefaultTestConfig.class)
+	public static class CustomConfig {
+
+	}
+
+}

spring-cloud-gateway-core/src/test/resources/application-weights-concurrent.yml

@@ -0,0 +1,4 @@
+test:
+  concurrent:
+    execution:
+      timeInSeconds: 10