From c773e05787f38b77c3dd64ef3ae62c250d3e9cf4 Mon Sep 17 00:00:00 2001 From: Alexey Nakidkin Date: Tue, 10 Dec 2019 09:11:01 +0300 Subject: [PATCH 1/2] Fixes gh-1459 Thread-safe WeightCalculatorWebFilter --- .../filter/WeightCalculatorWebFilter.java | 73 ++++++----- ...ghtCalculatorWebFilterConcurrentTests.java | 118 ++++++++++++++++++ .../application-weights-concurrent.yml | 4 + 3 files changed, 161 insertions(+), 34 deletions(-) create mode 100644 spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/WeightCalculatorWebFilterConcurrentTests.java create mode 100644 spring-cloud-gateway-core/src/test/resources/application-weights-concurrent.yml diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/WeightCalculatorWebFilter.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/WeightCalculatorWebFilter.java index 58d83f32..2d371371 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/WeightCalculatorWebFilter.java +++ b/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 { @@ -164,38 +165,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 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 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); + } } } @@ -219,18 +222,20 @@ public class WeightCalculatorWebFilter double r = this.random.nextDouble(); - List ranges = config.ranges; + synchronized (config) { + List 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; + } } } } diff --git a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/WeightCalculatorWebFilterConcurrentTests.java b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/WeightCalculatorWebFilterConcurrentTests.java new file mode 100644 index 00000000..25f882a2 --- /dev/null +++ b/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 { + + } + +} diff --git a/spring-cloud-gateway-core/src/test/resources/application-weights-concurrent.yml b/spring-cloud-gateway-core/src/test/resources/application-weights-concurrent.yml new file mode 100644 index 00000000..cd6aa428 --- /dev/null +++ b/spring-cloud-gateway-core/src/test/resources/application-weights-concurrent.yml @@ -0,0 +1,4 @@ +test: + concurrent: + execution: + timeInSeconds: 30 From b4eb30045aa7082ac4a4e1976869a19c25697f3a Mon Sep 17 00:00:00 2001 From: Spencer Gibb Date: Wed, 11 Dec 2019 16:43:23 -0500 Subject: [PATCH 2/2] Lowers concurrent test time to 10s --- .../src/test/resources/application-weights-concurrent.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spring-cloud-gateway-core/src/test/resources/application-weights-concurrent.yml b/spring-cloud-gateway-core/src/test/resources/application-weights-concurrent.yml index cd6aa428..add12137 100644 --- a/spring-cloud-gateway-core/src/test/resources/application-weights-concurrent.yml +++ b/spring-cloud-gateway-core/src/test/resources/application-weights-concurrent.yml @@ -1,4 +1,4 @@ test: concurrent: execution: - timeInSeconds: 30 + timeInSeconds: 10