fix: cap concurrent shadow planning
This commit is contained in:
@@ -5,6 +5,7 @@ import org.slf4j.LoggerFactory;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.Objects;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
@Component
|
||||
public class ShadowObservationGateway {
|
||||
@@ -12,6 +13,7 @@ public class ShadowObservationGateway {
|
||||
|
||||
private final ShadowObservationProperties properties;
|
||||
private final ShadowPlanningEngine engine;
|
||||
private final AtomicBoolean planning = new AtomicBoolean();
|
||||
|
||||
public ShadowObservationGateway(ShadowObservationProperties properties, ShadowPlanningEngine engine) {
|
||||
this.properties = Objects.requireNonNull(properties, "properties");
|
||||
@@ -22,11 +24,16 @@ public class ShadowObservationGateway {
|
||||
if (!properties.isEnabled() || engine == null) {
|
||||
return InboundShadowObservation.empty();
|
||||
}
|
||||
if (!planning.compareAndSet(false, true)) {
|
||||
return InboundShadowObservation.empty();
|
||||
}
|
||||
try {
|
||||
return engine.planInbound(taskNo, trayNo);
|
||||
} catch (RuntimeException exception) {
|
||||
LOG.warn("Shadow inbound planning failed open for task {}: {}", taskNo, exception.getMessage());
|
||||
return InboundShadowObservation.empty();
|
||||
} finally {
|
||||
planning.set(false);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -34,12 +41,17 @@ public class ShadowObservationGateway {
|
||||
if (!properties.isEnabled() || engine == null) {
|
||||
return OutboundShadowObservation.empty();
|
||||
}
|
||||
if (!planning.compareAndSet(false, true)) {
|
||||
return OutboundShadowObservation.empty();
|
||||
}
|
||||
try {
|
||||
return engine.planProjectOutbound(stockoutDetailId);
|
||||
} catch (RuntimeException exception) {
|
||||
LOG.warn("Shadow outbound planning failed open for detail {}: {}",
|
||||
stockoutDetailId, exception.getMessage());
|
||||
return OutboundShadowObservation.empty();
|
||||
} finally {
|
||||
planning.set(false);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -47,12 +59,17 @@ public class ShadowObservationGateway {
|
||||
if (!properties.isEnabled() || engine == null) {
|
||||
return OutboundShadowObservation.empty();
|
||||
}
|
||||
if (!planning.compareAndSet(false, true)) {
|
||||
return OutboundShadowObservation.empty();
|
||||
}
|
||||
try {
|
||||
return engine.planExactStockOutbound(stockoutDetailId, stockId);
|
||||
} catch (RuntimeException exception) {
|
||||
LOG.warn("Shadow exact-stock planning failed open for detail {} stock {}: {}",
|
||||
stockoutDetailId, stockId, exception.getMessage());
|
||||
return OutboundShadowObservation.empty();
|
||||
} finally {
|
||||
planning.set(false);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -2,6 +2,12 @@ package com.ruoyi.algorithm.shadow;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
@@ -29,11 +35,15 @@ class ShadowObservationGatewayTest {
|
||||
ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class);
|
||||
when(engine.planInbound("TASK-2", "TRAY-2"))
|
||||
.thenThrow(new IllegalStateException("algorithm objects are unavailable"));
|
||||
when(engine.planProjectOutbound("DETAIL-2"))
|
||||
.thenReturn(OutboundShadowObservation.unavailable("W01", "DETAIL-2", "TEST", "completed"));
|
||||
ShadowObservationGateway gateway = new ShadowObservationGateway(properties, engine);
|
||||
|
||||
InboundShadowObservation observation = gateway.planInbound("TASK-2", "TRAY-2");
|
||||
OutboundShadowObservation followingObservation = gateway.planProjectOutbound("DETAIL-2");
|
||||
|
||||
assertThat(observation.observed()).isFalse();
|
||||
assertThat(followingObservation.observed()).isTrue();
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -46,4 +56,45 @@ class ShadowObservationGatewayTest {
|
||||
|
||||
assertThat(observation.observed()).isFalse();
|
||||
}
|
||||
|
||||
@Test
|
||||
void concurrentPlanningIsSkippedInsteadOfAddingDatabaseLoad() throws Exception {
|
||||
ShadowObservationProperties properties = new ShadowObservationProperties();
|
||||
properties.setEnabled(true);
|
||||
ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class);
|
||||
CountDownLatch planningStarted = new CountDownLatch(1);
|
||||
CountDownLatch releasePlanning = new CountDownLatch(1);
|
||||
when(engine.planInbound("TASK-3", "TRAY-3")).thenAnswer(invocation -> {
|
||||
planningStarted.countDown();
|
||||
if (!releasePlanning.await(5, TimeUnit.SECONDS)) {
|
||||
throw new IllegalStateException("test planning was not released");
|
||||
}
|
||||
return InboundShadowObservation.unavailable("W01", "TASK-3", "TEST", "completed");
|
||||
});
|
||||
when(engine.planProjectOutbound("DETAIL-3"))
|
||||
.thenReturn(OutboundShadowObservation.unavailable("W01", "DETAIL-3", "TEST", "completed"));
|
||||
ShadowObservationGateway gateway = new ShadowObservationGateway(properties, engine);
|
||||
ExecutorService executor = Executors.newSingleThreadExecutor();
|
||||
|
||||
try {
|
||||
Future<InboundShadowObservation> first =
|
||||
executor.submit(() -> gateway.planInbound("TASK-3", "TRAY-3"));
|
||||
assertThat(planningStarted.await(5, TimeUnit.SECONDS)).isTrue();
|
||||
|
||||
OutboundShadowObservation concurrent = gateway.planProjectOutbound("DETAIL-3");
|
||||
|
||||
assertThat(concurrent.observed()).isFalse();
|
||||
verify(engine, never()).planProjectOutbound("DETAIL-3");
|
||||
releasePlanning.countDown();
|
||||
assertThat(first.get(5, TimeUnit.SECONDS).observed()).isTrue();
|
||||
|
||||
OutboundShadowObservation afterRelease = gateway.planProjectOutbound("DETAIL-3");
|
||||
|
||||
assertThat(afterRelease.observed()).isTrue();
|
||||
verify(engine).planProjectOutbound("DETAIL-3");
|
||||
} finally {
|
||||
releasePlanning.countDown();
|
||||
executor.shutdownNow();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user