From cdbb59ca652a0f122b9ac9b05c657e6941e138a5 Mon Sep 17 00:00:00 2001 From: zgw <13623675548@163.com> Date: Thu, 30 Jul 2026 13:44:42 +0800 Subject: [PATCH] feat: decide shadow traffic safely --- .../wcs/shadow/decision/DecisionType.java | 5 + .../decision/RollingDispatchPolicy.java | 66 +++++++++++ .../wcs/shadow/decision/ShadowCandidate.java | 56 ++++++++++ .../wcs/shadow/decision/ShadowDecision.java | 39 +++++++ .../wcs/shadow/decision/WaitForGraph.java | 57 ++++++++++ .../wcs/shadow/RollingDispatchPolicyTest.java | 105 ++++++++++++++++++ 6 files changed, 328 insertions(+) create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/decision/DecisionType.java create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/decision/RollingDispatchPolicy.java create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/decision/ShadowCandidate.java create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/decision/ShadowDecision.java create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/decision/WaitForGraph.java create mode 100644 javaTest/src/test/java/com/ft546/wcs/shadow/RollingDispatchPolicyTest.java diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/decision/DecisionType.java b/javaTest/src/main/java/com/ft546/wcs/shadow/decision/DecisionType.java new file mode 100644 index 0000000..9ceda56 --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/decision/DecisionType.java @@ -0,0 +1,5 @@ +package com.ft546.wcs.shadow.decision; + +public enum DecisionType { + GRANT, HOLD, YIELD +} diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/decision/RollingDispatchPolicy.java b/javaTest/src/main/java/com/ft546/wcs/shadow/decision/RollingDispatchPolicy.java new file mode 100644 index 0000000..eeef186 --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/decision/RollingDispatchPolicy.java @@ -0,0 +1,66 @@ +package com.ft546.wcs.shadow.decision; + +import com.ft546.wcs.shadow.map.EffectiveTrafficMap; +import com.ft546.wcs.shadow.resource.ResourceKey; +import com.ft546.wcs.shadow.resource.ResourceLedger; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Set; + +public final class RollingDispatchPolicy { + public List decide(List input, ResourceLedger authoritative, + EffectiveTrafficMap map) { + List candidates = new ArrayList(input); + Collections.sort(candidates, ORDER); + ResourceLedger simulation = authoritative.copy(); + WaitForGraph waits = new WaitForGraph(); + List decisions = new ArrayList(); + for (ShadowCandidate candidate : candidates) { + if (candidate.isCommitted()) { + decisions.add(decision(candidate, DecisionType.GRANT, "COMMITTED_CONTINUATION", + Collections.emptySet(), Collections.emptySet(), null)); + continue; + } + ResourceLedger.ProtectionResult result = + simulation.evaluate(candidate.getVehicleId(), candidate.getResources()); + if (result.isGranted()) { + decisions.add(decision(candidate, DecisionType.GRANT, "RESOURCES_AVAILABLE", + Collections.emptySet(), Collections.emptySet(), null)); + continue; + } + boolean cycle = waits.wouldCreateCycle(candidate.getVehicleId(), result.getBlockingVehicles()); + waits.addWait(candidate.getVehicleId(), result.getBlockingVehicles()); + if (cycle) { + decisions.add(decision(candidate, DecisionType.HOLD, "WAIT_CYCLE", + result.getConflictingResources(), result.getBlockingVehicles(), null)); + } else if (candidate.getSafeYieldSite() != null + && map.getSafeYieldSites().contains(candidate.getSafeYieldSite())) { + decisions.add(decision(candidate, DecisionType.YIELD, "SAFE_YIELD", + result.getConflictingResources(), result.getBlockingVehicles(), + candidate.getSafeYieldSite())); + } else { + decisions.add(decision(candidate, DecisionType.HOLD, "NO_SAFE_YIELD_SITE", + result.getConflictingResources(), result.getBlockingVehicles(), null)); + } + } + return Collections.unmodifiableList(decisions); + } + + private ShadowDecision decision(ShadowCandidate candidate, DecisionType type, String reason, + Set resources, Set blockers, String yieldSite) { + return new ShadowDecision(candidate.getTaskId(), candidate.getVehicleId(), type, reason, + resources, blockers, yieldSite); + } + + private static final Comparator ORDER = + Comparator.comparing(ShadowCandidate::isCommitted).reversed() + .thenComparing(ShadowCandidate::isEmergency, Comparator.reverseOrder()) + .thenComparing(ShadowCandidate::getBusinessPriority, Comparator.reverseOrder()) + .thenComparing(ShadowCandidate::getWaitAge, Comparator.reverseOrder()) + .thenComparingInt(ShadowCandidate::getClearanceCost) + .thenComparing(ShadowCandidate::getTaskId) + .thenComparing(ShadowCandidate::getVehicleId); +} diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/decision/ShadowCandidate.java b/javaTest/src/main/java/com/ft546/wcs/shadow/decision/ShadowCandidate.java new file mode 100644 index 0000000..c295be8 --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/decision/ShadowCandidate.java @@ -0,0 +1,56 @@ +package com.ft546.wcs.shadow.decision; + +import com.ft546.wcs.shadow.resource.ResourceKey; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; + +public final class ShadowCandidate { + private final String taskId; + private final String vehicleId; + private final boolean committed; + private final boolean emergency; + private final int businessPriority; + private final long waitAge; + private final int clearanceCost; + private final List resources; + private final String safeYieldSite; + + private ShadowCandidate(String taskId, String vehicleId, boolean committed, boolean emergency, + int businessPriority, long waitAge, int clearanceCost, + Collection resources, String safeYieldSite) { + this.taskId = taskId; + this.vehicleId = vehicleId; + this.committed = committed; + this.emergency = emergency; + this.businessPriority = businessPriority; + this.waitAge = waitAge; + this.clearanceCost = clearanceCost; + this.resources = Collections.unmodifiableList(new ArrayList(resources)); + this.safeYieldSite = safeYieldSite; + } + + public static ShadowCandidate waiting(String taskId, String vehicleId, boolean emergency, + int businessPriority, long waitAge, int clearanceCost, + Collection resources, String safeYieldSite) { + return new ShadowCandidate(taskId, vehicleId, false, emergency, businessPriority, + waitAge, clearanceCost, resources, safeYieldSite); + } + + public static ShadowCandidate committed(String taskId, String vehicleId, + Collection resources) { + return new ShadowCandidate(taskId, vehicleId, true, false, 0, 0, 0, resources, null); + } + + public String getTaskId() { return taskId; } + public String getVehicleId() { return vehicleId; } + public boolean isCommitted() { return committed; } + public boolean isEmergency() { return emergency; } + public int getBusinessPriority() { return businessPriority; } + public long getWaitAge() { return waitAge; } + public int getClearanceCost() { return clearanceCost; } + public List getResources() { return resources; } + public String getSafeYieldSite() { return safeYieldSite; } +} diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/decision/ShadowDecision.java b/javaTest/src/main/java/com/ft546/wcs/shadow/decision/ShadowDecision.java new file mode 100644 index 0000000..c4c22ae --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/decision/ShadowDecision.java @@ -0,0 +1,39 @@ +package com.ft546.wcs.shadow.decision; + +import com.ft546.wcs.shadow.resource.ResourceKey; + +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.Set; + +public final class ShadowDecision { + private final String taskId; + private final String vehicleId; + private final DecisionType type; + private final String reasonCode; + private final Set conflictingResources; + private final Set blockingVehicles; + private final String yieldSite; + + public ShadowDecision(String taskId, String vehicleId, DecisionType type, String reasonCode, + Set conflictingResources, Set blockingVehicles, + String yieldSite) { + this.taskId = taskId; + this.vehicleId = vehicleId; + this.type = type; + this.reasonCode = reasonCode; + this.conflictingResources = Collections.unmodifiableSet( + new LinkedHashSet(conflictingResources)); + this.blockingVehicles = Collections.unmodifiableSet( + new LinkedHashSet(blockingVehicles)); + this.yieldSite = yieldSite; + } + + public String getTaskId() { return taskId; } + public String getVehicleId() { return vehicleId; } + public DecisionType getType() { return type; } + public String getReasonCode() { return reasonCode; } + public Set getConflictingResources() { return conflictingResources; } + public Set getBlockingVehicles() { return blockingVehicles; } + public String getYieldSite() { return yieldSite; } +} diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/decision/WaitForGraph.java b/javaTest/src/main/java/com/ft546/wcs/shadow/decision/WaitForGraph.java new file mode 100644 index 0000000..0950ba2 --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/decision/WaitForGraph.java @@ -0,0 +1,57 @@ +package com.ft546.wcs.shadow.decision; + +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Set; + +public final class WaitForGraph { + private final Map> edges = new LinkedHashMap>(); + + public boolean wouldCreateCycle(String waiter, Collection holders) { + WaitForGraph copy = copy(); + copy.put(waiter, holders); + return !copy.isAcyclic(); + } + + public boolean addWait(String waiter, Collection holders) { + put(waiter, holders); + return !isAcyclic(); + } + + public boolean isAcyclic() { + Set visiting = new LinkedHashSet(); + Set visited = new LinkedHashSet(); + for (String node : edges.keySet()) { + if (hasCycle(node, visiting, visited)) return false; + } + return true; + } + + private boolean hasCycle(String node, Set visiting, Set visited) { + if (visiting.contains(node)) return true; + if (visited.contains(node)) return false; + visiting.add(node); + for (String target : edges.containsKey(node) ? edges.get(node) : Collections.emptySet()) { + if (hasCycle(target, visiting, visited)) return true; + } + visiting.remove(node); + visited.add(node); + return false; + } + + private void put(String waiter, Collection holders) { + if (!edges.containsKey(waiter)) edges.put(waiter, new LinkedHashSet()); + edges.get(waiter).addAll(holders); + } + + private WaitForGraph copy() { + WaitForGraph copy = new WaitForGraph(); + for (Map.Entry> entry : edges.entrySet()) { + copy.edges.put(entry.getKey(), new LinkedHashSet(entry.getValue())); + } + return copy; + } +} diff --git a/javaTest/src/test/java/com/ft546/wcs/shadow/RollingDispatchPolicyTest.java b/javaTest/src/test/java/com/ft546/wcs/shadow/RollingDispatchPolicyTest.java new file mode 100644 index 0000000..e0e7913 --- /dev/null +++ b/javaTest/src/test/java/com/ft546/wcs/shadow/RollingDispatchPolicyTest.java @@ -0,0 +1,105 @@ +package com.ft546.wcs.shadow; + +import com.ft546.wcs.shadow.decision.DecisionType; +import com.ft546.wcs.shadow.decision.RollingDispatchPolicy; +import com.ft546.wcs.shadow.decision.ShadowCandidate; +import com.ft546.wcs.shadow.decision.ShadowDecision; +import com.ft546.wcs.shadow.decision.WaitForGraph; +import com.ft546.wcs.shadow.map.EffectiveTrafficMap; +import com.ft546.wcs.shadow.map.TrafficMapCompiler; +import com.ft546.wcs.shadow.resource.ResourceKey; +import com.ft546.wcs.shadow.resource.ResourceLedger; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; + +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 RollingDispatchPolicyTest { + + @Test + void sortsDeterministicallyAndAllowsNonConflictingParallelGrants() { + EffectiveTrafficMap map = map(); + List candidates = Arrays.asList( + candidate("T2", "V2", 5, 10, ResourceKey.node("C"), null), + candidate("T1", "V1", 5, 20, ResourceKey.node("A"), null)); + + List decisions = + new RollingDispatchPolicy().decide(candidates, new ResourceLedger(map.getCapacities()), map); + + assertEquals("V1", decisions.get(0).getVehicleId()); + assertEquals("V2", decisions.get(1).getVehicleId()); + assertEquals(DecisionType.GRANT, decisions.get(0).getType()); + assertEquals(DecisionType.GRANT, decisions.get(1).getType()); + } + + @Test + void holdsConflictsAndUsesOnlyPublishedYieldSite() { + EffectiveTrafficMap map = map(); + ResourceLedger ledger = new ResourceLedger(map.getCapacities()); + ledger.evaluate("BLOCKER", Collections.singleton(ResourceKey.node("B"))); + ShadowCandidate yield = candidate("T1", "V1", 5, 10, ResourceKey.node("B"), "C"); + ShadowCandidate unsafe = candidate("T2", "V2", 4, 10, ResourceKey.node("B"), "X"); + + List decisions = + new RollingDispatchPolicy().decide(Arrays.asList(yield, unsafe), ledger, map); + + assertEquals(DecisionType.YIELD, decisions.get(0).getType()); + assertEquals("C", decisions.get(0).getYieldSite()); + assertEquals(DecisionType.HOLD, decisions.get(1).getType()); + assertEquals("NO_SAFE_YIELD_SITE", decisions.get(1).getReasonCode()); + assertEquals(Collections.singleton("BLOCKER"), decisions.get(1).getBlockingVehicles()); + } + + @Test + void rejectsWaitCycleAndNeverRevokesCommittedSegment() { + EffectiveTrafficMap map = map(); + ResourceLedger ledger = new ResourceLedger(map.getCapacities()); + ledger.evaluate("V2", Collections.singleton(ResourceKey.node("A"))); + ledger.evaluate("V1", Collections.singleton(ResourceKey.node("B"))); + ShadowCandidate first = candidate("T1", "V1", 5, 20, ResourceKey.node("A"), null); + ShadowCandidate second = candidate("T2", "V2", 5, 10, ResourceKey.node("B"), null); + ShadowCandidate committed = ShadowCandidate.committed( + "T0", "V0", Collections.singleton(ResourceKey.node("A"))); + + List decisions = + new RollingDispatchPolicy().decide(Arrays.asList(first, second, committed), ledger, map); + + assertEquals(DecisionType.GRANT, decisions.get(0).getType()); + assertEquals("COMMITTED_CONTINUATION", decisions.get(0).getReasonCode()); + assertEquals("WAIT_CYCLE", decisions.get(2).getReasonCode()); + } + + @Test + void detectsCycleBeforeAddingWaitEdges() { + WaitForGraph graph = new WaitForGraph(); + assertFalse(graph.addWait("V1", Collections.singleton("V2"))); + assertTrue(graph.wouldCreateCycle("V2", Collections.singleton("V1"))); + assertTrue(graph.addWait("V2", Collections.singleton("V1"))); + assertFalse(graph.isAcyclic()); + } + + private ShadowCandidate candidate(String task, String vehicle, int priority, long age, + ResourceKey resource, String yieldSite) { + return ShadowCandidate.waiting(task, vehicle, false, priority, age, 1, + Collections.singleton(resource), yieldSite); + } + + private EffectiveTrafficMap map() { + LinkedHashSet nodes = new LinkedHashSet(Arrays.asList("A", "B", "C")); + LinkedHashSet edges = + new LinkedHashSet(); + edges.add(new TrafficMapCompiler.EdgeDefinition("A", "B")); + edges.add(new TrafficMapCompiler.EdgeDefinition("B", "C")); + return new TrafficMapCompiler().compile(new TrafficMapCompiler.MapDefinition( + "FT546", 2L, "ZONE-1", true, "sha", "sha", nodes, edges, + new LinkedHashMap(), Collections.singleton("A"), + Collections.singleton("C"))); + } +}