feat(wcs): prevent traffic deadlocks
This commit is contained in:
@@ -5,6 +5,7 @@ import java.util.Collections;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
public final class EffectiveTopology {
|
||||
private final String warehouseCode;
|
||||
@@ -52,6 +53,10 @@ public final class EffectiveTopology {
|
||||
return nodes.get(nodeId);
|
||||
}
|
||||
|
||||
public Set<String> getNodeIds() {
|
||||
return nodes.keySet();
|
||||
}
|
||||
|
||||
public List<String> neighbors(String nodeId) {
|
||||
List<String> result = neighbors.get(nodeId);
|
||||
return result == null ? Collections.<String>emptyList() : result;
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
package com.ruoyi.wcs.domain.traffic;
|
||||
|
||||
import com.ruoyi.wcs.domain.shared.DomainRuleViolation;
|
||||
import com.ruoyi.wcs.domain.topology.EffectiveTopology;
|
||||
import com.ruoyi.wcs.domain.topology.TopologyNode;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
import java.util.Set;
|
||||
import java.util.TreeSet;
|
||||
|
||||
public final class DeadlockGuard {
|
||||
private final WaitForGraph waitForGraph;
|
||||
private final RoutePlanner routePlanner;
|
||||
|
||||
public DeadlockGuard(WaitForGraph waitForGraph, RoutePlanner routePlanner) {
|
||||
this.waitForGraph = Objects.requireNonNull(waitForGraph, "waitForGraph");
|
||||
this.routePlanner = Objects.requireNonNull(routePlanner, "routePlanner");
|
||||
}
|
||||
|
||||
public TrafficDecision decide(TrafficCandidate candidate,
|
||||
ResourceLedger ledger,
|
||||
EffectiveTopology topology) {
|
||||
Objects.requireNonNull(candidate, "candidate");
|
||||
Objects.requireNonNull(ledger, "ledger");
|
||||
Objects.requireNonNull(topology, "topology");
|
||||
ReservationRequest request = candidate.getReservationRequest();
|
||||
Set<String> blockers = new TreeSet<String>();
|
||||
for (String resourceKey : request.getResourceKeys()) {
|
||||
if (!ledger.containsResource(resourceKey)) {
|
||||
return TrafficDecision.hold("UNKNOWN_RESOURCE",
|
||||
Collections.<String>emptySet());
|
||||
}
|
||||
if (ledger.availableCapacity(resourceKey) <= 0) {
|
||||
blockers.addAll(ledger.holders(resourceKey));
|
||||
}
|
||||
}
|
||||
blockers.remove(candidate.getVehicleId());
|
||||
|
||||
if (blockers.isEmpty()) {
|
||||
waitForGraph.clearWaits(candidate.getVehicleId());
|
||||
ReservationDecision reservation = ledger.reserve(request);
|
||||
if (!reservation.isGranted()) {
|
||||
return TrafficDecision.hold(reservation.getReasonCode(),
|
||||
Collections.<String>emptySet());
|
||||
}
|
||||
return TrafficDecision.grant(reservation);
|
||||
}
|
||||
|
||||
waitForGraph.clearWaits(candidate.getVehicleId());
|
||||
if (waitForGraph.wouldCreateCycle(candidate.getVehicleId(), blockers)) {
|
||||
return TrafficDecision.hold("WAIT_CYCLE", blockers);
|
||||
}
|
||||
waitForGraph.replaceWaits(candidate.getVehicleId(), blockers);
|
||||
|
||||
RoutePlan yieldRoute = findYieldRoute(candidate, ledger, topology);
|
||||
if (yieldRoute == null) {
|
||||
return TrafficDecision.hold("NO_SAFE_YIELD_POINT", blockers);
|
||||
}
|
||||
return TrafficDecision.yield("SAFE_YIELD_POINT", blockers,
|
||||
yieldRoute.getNodeIds().get(yieldRoute.getNodeIds().size() - 1),
|
||||
yieldRoute);
|
||||
}
|
||||
|
||||
private RoutePlan findYieldRoute(TrafficCandidate candidate,
|
||||
ResourceLedger ledger,
|
||||
EffectiveTopology topology) {
|
||||
Set<String> blockedResources = new LinkedHashSet<String>();
|
||||
for (String resourceKey : ledger.resourceKeys()) {
|
||||
if (ledger.availableCapacity(resourceKey) <= 0) {
|
||||
blockedResources.add(resourceKey);
|
||||
}
|
||||
}
|
||||
List<String> nodeIds = new ArrayList<String>(topology.getNodeIds());
|
||||
Collections.sort(nodeIds);
|
||||
for (String nodeId : nodeIds) {
|
||||
TopologyNode node = topology.node(nodeId);
|
||||
if ((!node.isSafeWait() && !node.isSafeYield())
|
||||
|| nodeId.equals(candidate.getCurrentNodeId())) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
return routePlanner.plan(topology, candidate.getCurrentNodeId(),
|
||||
nodeId, blockedResources);
|
||||
} catch (DomainRuleViolation unavailable) {
|
||||
// Try the next published safe point.
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
@@ -178,6 +178,11 @@ public final class ResourceLedger {
|
||||
return capacities.containsKey(resourceKey);
|
||||
}
|
||||
|
||||
public synchronized Set<String> resourceKeys() {
|
||||
return Collections.unmodifiableSet(
|
||||
new TreeSet<String>(capacities.keySet()));
|
||||
}
|
||||
|
||||
public synchronized ResourceLedger copy() {
|
||||
return new ResourceLedger(this);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
package com.ruoyi.wcs.domain.traffic;
|
||||
|
||||
import java.util.Objects;
|
||||
|
||||
public final class TrafficCandidate {
|
||||
private final ReservationRequest reservationRequest;
|
||||
private final String currentNodeId;
|
||||
private final RoutePlan businessRoute;
|
||||
|
||||
public TrafficCandidate(ReservationRequest reservationRequest,
|
||||
String currentNodeId, RoutePlan businessRoute) {
|
||||
this.reservationRequest = Objects.requireNonNull(
|
||||
reservationRequest, "reservationRequest");
|
||||
if (currentNodeId == null || currentNodeId.trim().isEmpty()) {
|
||||
throw new IllegalArgumentException("currentNodeId must not be blank");
|
||||
}
|
||||
this.currentNodeId = currentNodeId.trim();
|
||||
this.businessRoute = Objects.requireNonNull(businessRoute, "businessRoute");
|
||||
if (!businessRoute.getNodeIds().contains(this.currentNodeId)) {
|
||||
throw new IllegalArgumentException("business route does not contain current node");
|
||||
}
|
||||
if (!businessRoute.getResourceKeys().containsAll(
|
||||
reservationRequest.getResourceKeys())) {
|
||||
throw new IllegalArgumentException(
|
||||
"reservation resources must belong to the business route");
|
||||
}
|
||||
}
|
||||
|
||||
public ReservationRequest getReservationRequest() {
|
||||
return reservationRequest;
|
||||
}
|
||||
|
||||
public String getVehicleId() {
|
||||
return reservationRequest.getVehicleId();
|
||||
}
|
||||
|
||||
public String getCurrentNodeId() {
|
||||
return currentNodeId;
|
||||
}
|
||||
|
||||
public RoutePlan getBusinessRoute() {
|
||||
return businessRoute;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
package com.ruoyi.wcs.domain.traffic;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.Set;
|
||||
|
||||
public final class TrafficDecision {
|
||||
private final TrafficDecisionType type;
|
||||
private final String reasonCode;
|
||||
private final ReservationDecision reservationDecision;
|
||||
private final Set<String> blockingVehicleIds;
|
||||
private final String yieldNodeId;
|
||||
private final RoutePlan yieldRoute;
|
||||
|
||||
private TrafficDecision(TrafficDecisionType type, String reasonCode,
|
||||
ReservationDecision reservationDecision,
|
||||
Set<String> blockingVehicleIds,
|
||||
String yieldNodeId, RoutePlan yieldRoute) {
|
||||
this.type = type;
|
||||
this.reasonCode = reasonCode;
|
||||
this.reservationDecision = reservationDecision;
|
||||
this.blockingVehicleIds = Collections.unmodifiableSet(
|
||||
new LinkedHashSet<String>(blockingVehicleIds));
|
||||
this.yieldNodeId = yieldNodeId;
|
||||
this.yieldRoute = yieldRoute;
|
||||
}
|
||||
|
||||
public static TrafficDecision grant(ReservationDecision reservationDecision) {
|
||||
return new TrafficDecision(TrafficDecisionType.GRANT, "RESOURCES_GRANTED",
|
||||
reservationDecision, Collections.<String>emptySet(), null, null);
|
||||
}
|
||||
|
||||
public static TrafficDecision hold(String reasonCode, Set<String> blockers) {
|
||||
return new TrafficDecision(TrafficDecisionType.HOLD, reasonCode, null,
|
||||
blockers, null, null);
|
||||
}
|
||||
|
||||
public static TrafficDecision yield(String reasonCode, Set<String> blockers,
|
||||
String yieldNodeId, RoutePlan yieldRoute) {
|
||||
return new TrafficDecision(TrafficDecisionType.YIELD, reasonCode, null,
|
||||
blockers, yieldNodeId, yieldRoute);
|
||||
}
|
||||
|
||||
public TrafficDecisionType getType() {
|
||||
return type;
|
||||
}
|
||||
|
||||
public String getReasonCode() {
|
||||
return reasonCode;
|
||||
}
|
||||
|
||||
public ReservationDecision getReservationDecision() {
|
||||
return reservationDecision;
|
||||
}
|
||||
|
||||
public Set<String> getBlockingVehicleIds() {
|
||||
return blockingVehicleIds;
|
||||
}
|
||||
|
||||
public String getYieldNodeId() {
|
||||
return yieldNodeId;
|
||||
}
|
||||
|
||||
public RoutePlan getYieldRoute() {
|
||||
return yieldRoute;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package com.ruoyi.wcs.domain.traffic;
|
||||
|
||||
public enum TrafficDecisionType {
|
||||
GRANT,
|
||||
HOLD,
|
||||
YIELD
|
||||
}
|
||||
@@ -0,0 +1,112 @@
|
||||
package com.ruoyi.wcs.domain.traffic;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.TreeSet;
|
||||
|
||||
public final class WaitForGraph {
|
||||
private final Map<String, Set<String>> edges =
|
||||
new LinkedHashMap<String, Set<String>>();
|
||||
|
||||
public synchronized void addWait(String waiter, String holder) {
|
||||
String normalizedWaiter = requireText(waiter, "waiter");
|
||||
String normalizedHolder = requireText(holder, "holder");
|
||||
edges.computeIfAbsent(normalizedWaiter,
|
||||
ignored -> new TreeSet<String>()).add(normalizedHolder);
|
||||
}
|
||||
|
||||
public synchronized boolean wouldCreateCycle(String waiter, Set<String> holders) {
|
||||
String normalizedWaiter = requireText(waiter, "waiter");
|
||||
if (holders == null) {
|
||||
throw new IllegalArgumentException("holders must not be null");
|
||||
}
|
||||
Map<String, Set<String>> candidate = copyEdges();
|
||||
Set<String> outgoing = candidate.computeIfAbsent(normalizedWaiter,
|
||||
ignored -> new TreeSet<String>());
|
||||
for (String holder : holders) {
|
||||
outgoing.add(requireText(holder, "holder"));
|
||||
}
|
||||
return hasCycle(candidate);
|
||||
}
|
||||
|
||||
public synchronized void replaceWaits(String waiter, Set<String> holders) {
|
||||
String normalizedWaiter = requireText(waiter, "waiter");
|
||||
if (holders == null || holders.isEmpty()) {
|
||||
edges.remove(normalizedWaiter);
|
||||
return;
|
||||
}
|
||||
Set<String> copied = new TreeSet<String>();
|
||||
for (String holder : holders) {
|
||||
copied.add(requireText(holder, "holder"));
|
||||
}
|
||||
edges.put(normalizedWaiter, copied);
|
||||
}
|
||||
|
||||
public synchronized void clearWaits(String waiter) {
|
||||
edges.remove(requireText(waiter, "waiter"));
|
||||
}
|
||||
|
||||
public synchronized Set<String> holdersWaitedOnBy(String waiter) {
|
||||
Set<String> holders = edges.get(requireText(waiter, "waiter"));
|
||||
if (holders == null) {
|
||||
return Collections.emptySet();
|
||||
}
|
||||
return Collections.unmodifiableSet(new LinkedHashSet<String>(holders));
|
||||
}
|
||||
|
||||
private Map<String, Set<String>> copyEdges() {
|
||||
Map<String, Set<String>> copy = new LinkedHashMap<String, Set<String>>();
|
||||
for (Map.Entry<String, Set<String>> entry : edges.entrySet()) {
|
||||
copy.put(entry.getKey(), new TreeSet<String>(entry.getValue()));
|
||||
}
|
||||
return copy;
|
||||
}
|
||||
|
||||
private static boolean hasCycle(Map<String, Set<String>> graph) {
|
||||
Set<String> allNodes = new TreeSet<String>();
|
||||
allNodes.addAll(graph.keySet());
|
||||
for (Set<String> holders : graph.values()) {
|
||||
allNodes.addAll(holders);
|
||||
}
|
||||
Set<String> visited = new LinkedHashSet<String>();
|
||||
Set<String> visiting = new LinkedHashSet<String>();
|
||||
for (String node : allNodes) {
|
||||
if (visit(node, graph, visited, visiting)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private static boolean visit(String node, Map<String, Set<String>> graph,
|
||||
Set<String> visited, Set<String> visiting) {
|
||||
if (visiting.contains(node)) {
|
||||
return true;
|
||||
}
|
||||
if (visited.contains(node)) {
|
||||
return false;
|
||||
}
|
||||
visiting.add(node);
|
||||
Set<String> next = graph.get(node);
|
||||
if (next != null) {
|
||||
for (String holder : new TreeSet<String>(next)) {
|
||||
if (visit(holder, graph, visited, visiting)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
visiting.remove(node);
|
||||
visited.add(node);
|
||||
return false;
|
||||
}
|
||||
|
||||
private static String requireText(String value, String field) {
|
||||
if (value == null || value.trim().isEmpty()) {
|
||||
throw new IllegalArgumentException(field + " must not be blank");
|
||||
}
|
||||
return value.trim();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,126 @@
|
||||
package com.ruoyi.wcs.domain.traffic;
|
||||
|
||||
import com.ruoyi.wcs.domain.topology.EffectiveTopology;
|
||||
import com.ruoyi.wcs.domain.topology.ResourceDefinition;
|
||||
import com.ruoyi.wcs.domain.topology.TopologyCompiler;
|
||||
import com.ruoyi.wcs.domain.topology.TopologyDefinition;
|
||||
import com.ruoyi.wcs.domain.topology.TopologyEdge;
|
||||
import com.ruoyi.wcs.domain.topology.TopologyNode;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
class DeadlockGuardTest {
|
||||
|
||||
@Test
|
||||
void rejectsGrantThatClosesAWaitCycle() {
|
||||
WaitForGraph graph = new WaitForGraph();
|
||||
graph.addWait("V1", "V2");
|
||||
graph.addWait("V2", "V3");
|
||||
|
||||
assertTrue(graph.wouldCreateCycle("V3", Collections.singleton("V1")));
|
||||
assertFalse(graph.wouldCreateCycle("V4", Collections.singleton("V1")));
|
||||
}
|
||||
|
||||
@Test
|
||||
void doesNotInventYieldPointWhenTopologyHasNone() {
|
||||
EffectiveTopology topology = topology(false);
|
||||
ResourceLedger ledger = new ResourceLedger(topology);
|
||||
ledger.reserve(request("HOLDER", "V1", "NODE:B"));
|
||||
DeadlockGuard guard = guard(new WaitForGraph());
|
||||
|
||||
TrafficDecision decision = guard.decide(
|
||||
candidate("CANDIDATE", "V2", topology), ledger, topology);
|
||||
|
||||
assertEquals(TrafficDecisionType.HOLD, decision.getType());
|
||||
assertEquals("NO_SAFE_YIELD_POINT", decision.getReasonCode());
|
||||
assertEquals(Collections.singleton("V1"), decision.getBlockingVehicleIds());
|
||||
}
|
||||
|
||||
@Test
|
||||
void yieldsOnlyToPublishedSafePointOverAnUnoccupiedRoute() {
|
||||
EffectiveTopology topology = topology(true);
|
||||
ResourceLedger ledger = new ResourceLedger(topology);
|
||||
ledger.reserve(request("HOLDER", "V1", "NODE:B"));
|
||||
|
||||
TrafficDecision decision = guard(new WaitForGraph()).decide(
|
||||
candidate("CANDIDATE", "V2", topology), ledger, topology);
|
||||
|
||||
assertEquals(TrafficDecisionType.YIELD, decision.getType());
|
||||
assertEquals("SAFE_YIELD_POINT", decision.getReasonCode());
|
||||
assertEquals("C", decision.getYieldNodeId());
|
||||
assertEquals(Arrays.asList("A", "C"), decision.getYieldRoute().getNodeIds());
|
||||
}
|
||||
|
||||
@Test
|
||||
void holdsInsteadOfAddingAnEdgeThatWouldCreateCycle() {
|
||||
EffectiveTopology topology = topology(true);
|
||||
ResourceLedger ledger = new ResourceLedger(topology);
|
||||
ledger.reserve(request("HOLDER", "V1", "NODE:B"));
|
||||
WaitForGraph graph = new WaitForGraph();
|
||||
graph.addWait("V1", "V2");
|
||||
|
||||
TrafficDecision decision = guard(graph).decide(
|
||||
candidate("CANDIDATE", "V2", topology), ledger, topology);
|
||||
|
||||
assertEquals(TrafficDecisionType.HOLD, decision.getType());
|
||||
assertEquals("WAIT_CYCLE", decision.getReasonCode());
|
||||
assertFalse(graph.holdersWaitedOnBy("V2").contains("V1"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void grantsAndReservesWhenAllResourcesAreAvailable() {
|
||||
EffectiveTopology topology = topology(false);
|
||||
ResourceLedger ledger = new ResourceLedger(topology);
|
||||
TrafficCandidate candidate = new TrafficCandidate(
|
||||
request("CANDIDATE", "V2", "NODE:B"), "A",
|
||||
new AStarRoutePlanner().plan(topology, "A", "B",
|
||||
Collections.<String>emptySet()));
|
||||
|
||||
TrafficDecision decision = guard(new WaitForGraph())
|
||||
.decide(candidate, ledger, topology);
|
||||
|
||||
assertEquals(TrafficDecisionType.GRANT, decision.getType());
|
||||
assertTrue(decision.getReservationDecision().isGranted());
|
||||
assertEquals(1, ledger.usage("NODE:B"));
|
||||
}
|
||||
|
||||
private static DeadlockGuard guard(WaitForGraph graph) {
|
||||
return new DeadlockGuard(graph, new AStarRoutePlanner());
|
||||
}
|
||||
|
||||
private static TrafficCandidate candidate(String reservationId,
|
||||
String vehicleId,
|
||||
EffectiveTopology topology) {
|
||||
return new TrafficCandidate(request(reservationId, vehicleId, "NODE:B"),
|
||||
"A", new AStarRoutePlanner().plan(topology, "A", "B",
|
||||
Collections.<String>emptySet()));
|
||||
}
|
||||
|
||||
private static ReservationRequest request(String reservationId,
|
||||
String vehicleId,
|
||||
String resourceKey) {
|
||||
return new ReservationRequest(reservationId, "O-" + reservationId,
|
||||
"S-" + reservationId, vehicleId,
|
||||
Collections.singletonList(resourceKey));
|
||||
}
|
||||
|
||||
private static EffectiveTopology topology(boolean safeYieldPoint) {
|
||||
TopologyDefinition definition = new TopologyDefinition("FT546", 7L, true,
|
||||
Arrays.asList(
|
||||
new TopologyNode("A", "F1", 0, 0, 1, false, false),
|
||||
new TopologyNode("B", "F1", 1, 0, 1, false, false),
|
||||
new TopologyNode("C", "F1", 0, 1, 1, false, safeYieldPoint)),
|
||||
Arrays.asList(
|
||||
new TopologyEdge("A", "B", 1, 1),
|
||||
new TopologyEdge("A", "C", 1, 1)),
|
||||
Collections.<ResourceDefinition>emptyList(),
|
||||
Collections.<String>emptyList(), Collections.<String>emptyList());
|
||||
return new TopologyCompiler().compile(definition);
|
||||
}
|
||||
}
|
||||
@@ -95,6 +95,30 @@ class ResourceLedgerTest {
|
||||
assertTrue(ledger.reserve(request("R2", "V2", "NODE:A")).isGranted());
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsUnknownResourceWithoutChangingKnownCapacity() {
|
||||
ResourceLedger ledger = ledger("NODE:A", 1);
|
||||
|
||||
ReservationDecision decision = ledger.reserve(
|
||||
request("R1", "V1", "NODE:UNKNOWN"));
|
||||
|
||||
assertFalse(decision.isGranted());
|
||||
assertEquals("UNKNOWN_RESOURCE", decision.getReasonCode());
|
||||
assertEquals(0, ledger.usage("NODE:A"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void unresolvedRecoveryKeepsOccupiedResourcesProtected() {
|
||||
ResourceLedger ledger = ledger("NODE:A", 1);
|
||||
ReservationDecision decision = ledger.reserve(request("R1", "V1", "NODE:A"));
|
||||
ledger.occupy("R1", decision.getFencingToken());
|
||||
ledger.markRecoveryUnresolved("R1");
|
||||
|
||||
assertThrows(DomainRuleViolation.class, () -> ledger.release(
|
||||
"R1", "PLC-SNAPSHOT-1", decision.getFencingToken()));
|
||||
assertEquals(1, ledger.usage("NODE:A"));
|
||||
}
|
||||
|
||||
private static ResourceLedger ledger(String resourceKey, int capacity) {
|
||||
return new ResourceLedger(Collections.singletonMap(resourceKey, capacity));
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user