From 1ede0c7fc934af9ce0ec60f386da3425e92ea88c Mon Sep 17 00:00:00 2001 From: zgw <13623675548@163.com> Date: Fri, 31 Jul 2026 14:14:30 +0800 Subject: [PATCH] feat(javaTest): arbitrate corridor direction ownership --- .../CorridorReservationManager.java | 155 ++++++++++++++++++ .../HistoricalDispatchSimulator.java | 59 +++---- .../CorridorReservationManagerTest.java | 43 +++++ .../HistoricalDispatchSimulatorTest.java | 3 +- 4 files changed, 227 insertions(+), 33 deletions(-) create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/CorridorReservationManager.java create mode 100644 javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/CorridorReservationManagerTest.java diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/CorridorReservationManager.java b/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/CorridorReservationManager.java new file mode 100644 index 0000000..bfa439c --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/CorridorReservationManager.java @@ -0,0 +1,155 @@ +package com.ft546.wcs.shadow.history.simulation; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +public final class CorridorReservationManager { + private final Map edges = + new LinkedHashMap(); + private final Map> taskEdges = + new LinkedHashMap>(); + private final Map ranks = new LinkedHashMap(); + + public ReservationDecision reserve(String taskId, List route, + int effectivePriority, long arrivalMillis) { + if (taskId == null || taskId.trim().isEmpty() || route == null || route.isEmpty()) { + throw new IllegalArgumentException("task id and route are required"); + } + Rank rank = new Rank(taskId, effectivePriority, arrivalMillis); + for (int i = 1; i < route.size(); i++) { + EdgeRequest request = request(route.get(i - 1), route.get(i)); + EdgeReservation existing = edges.get(request.edgeId); + if (existing != null && existing.direction != request.direction) { + return ReservationDecision.denied(bestOwner(existing.holders), + "CORRIDOR_DIRECTION_OWNED"); + } + } + ranks.put(taskId, rank); + Set reserved = taskEdges.computeIfAbsent( + taskId, ignored -> new LinkedHashSet()); + for (int i = 1; i < route.size(); i++) { + EdgeRequest request = request(route.get(i - 1), route.get(i)); + EdgeReservation reservation = edges.computeIfAbsent( + request.edgeId, ignored -> new EdgeReservation(request.direction)); + reservation.holders.add(taskId); + reserved.add(request.edgeId); + } + return ReservationDecision.granted(); + } + + public void releasePassedEdge(String taskId, String from, String to) { + String edgeId = request(from, to).edgeId; + EdgeReservation reservation = edges.get(edgeId); + if (reservation != null) { + reservation.holders.remove(taskId); + if (reservation.holders.isEmpty()) edges.remove(edgeId); + } + Set reserved = taskEdges.get(taskId); + if (reserved != null) { + reserved.remove(edgeId); + if (reserved.isEmpty()) { + taskEdges.remove(taskId); + ranks.remove(taskId); + } + } + } + + public void releaseTask(String taskId) { + Set reserved = taskEdges.remove(taskId); + if (reserved != null) { + for (String edgeId : new ArrayList(reserved)) { + EdgeReservation reservation = edges.get(edgeId); + if (reservation == null) continue; + reservation.holders.remove(taskId); + if (reservation.holders.isEmpty()) edges.remove(edgeId); + } + } + ranks.remove(taskId); + } + + public String ownerFor(String left, String right) { + EdgeReservation reservation = edges.get(request(left, right).edgeId); + return reservation == null ? null : bestOwner(reservation.holders); + } + + private String bestOwner(Set holders) { + List candidates = new ArrayList(); + for (String holder : holders) { + Rank rank = ranks.get(holder); + if (rank != null) candidates.add(rank); + } + if (candidates.isEmpty()) return null; + Collections.sort(candidates, Comparator.comparingInt((Rank value) -> value.priority) + .thenComparingLong(value -> value.arrivalMillis) + .thenComparing(value -> value.taskId)); + return candidates.get(0).taskId; + } + + private EdgeRequest request(String from, String to) { + if (from == null || to == null || from.equals(to)) { + throw new IllegalArgumentException("corridor edge endpoints must differ"); + } + boolean canonical = from.compareTo(to) <= 0; + return new EdgeRequest(canonical ? from + "|" + to : to + "|" + from, + canonical ? 1 : -1); + } + + public static final class ReservationDecision { + private final boolean granted; + private final String ownerTaskId; + private final String reason; + + private ReservationDecision(boolean granted, String ownerTaskId, String reason) { + this.granted = granted; + this.ownerTaskId = ownerTaskId; + this.reason = reason; + } + + private static ReservationDecision granted() { + return new ReservationDecision(true, null, "GRANTED"); + } + + private static ReservationDecision denied(String ownerTaskId, String reason) { + return new ReservationDecision(false, ownerTaskId, reason); + } + + public boolean isGranted() { return granted; } + public String getOwnerTaskId() { return ownerTaskId; } + public String getReason() { return reason; } + } + + private static final class EdgeReservation { + private final int direction; + private final Set holders = new LinkedHashSet(); + + private EdgeReservation(int direction) { this.direction = direction; } + } + + private static final class EdgeRequest { + private final String edgeId; + private final int direction; + + private EdgeRequest(String edgeId, int direction) { + this.edgeId = edgeId; + this.direction = direction; + } + } + + private static final class Rank { + private final String taskId; + private final int priority; + private final long arrivalMillis; + + private Rank(String taskId, int priority, long arrivalMillis) { + this.taskId = taskId; + this.priority = priority; + this.arrivalMillis = arrivalMillis; + } + } +} diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/HistoricalDispatchSimulator.java b/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/HistoricalDispatchSimulator.java index 69bcdd2..9d6c47d 100644 --- a/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/HistoricalDispatchSimulator.java +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/HistoricalDispatchSimulator.java @@ -74,6 +74,7 @@ public final class HistoricalDispatchSimulator { IdempotentWmsCompletionReceiver wms = new IdempotentWmsCompletionReceiver(); GridPathPlanner planner = new GridPathPlanner(); VehicleAssignmentService assignment = new VehicleAssignmentService(); + CorridorReservationManager reservations = new CorridorReservationManager(); long now = 0L; long tick = 0L; @@ -82,13 +83,13 @@ public final class HistoricalDispatchSimulator { config.getNoProgressLimit(), terminal.size()); while (terminal.size() < expected && tick < config.getTickLimit()) { applyVehicleEvents(tick, now, faults, vehicles, active, - queue, executions, events); + queue, executions, reservations, events); assignReadyTasks(tick, now, input, queue, vehicles, active, - executions, terminal, planner, assignment, plc, events); + executions, terminal, planner, assignment, plc, reservations, events); Set traversedEdges = new LinkedHashSet(); advanceActiveTasks(tick, now, input, config, queue, vehicles, active, executions, terminal, planner, ledger, plc, archive, wms, - invariants, events, traversedEdges); + invariants, reservations, events, traversedEdges); verifyLedger(ledger, invariants); if (watchdog.observe(terminal.size())) { @@ -139,6 +140,7 @@ public final class HistoricalDispatchSimulator { Map active, SimulationTaskQueue queue, Map executions, + CorridorReservationManager reservations, List events) { boolean changed = false; for (DeterministicFaultPlan.VehicleEvent event : faults.eventsAt(tick)) { @@ -161,6 +163,7 @@ public final class HistoricalDispatchSimulator { if (!vehicle.isAssignmentEligible()) { ActiveTask stopped = active.remove(vehicle.getVehicleId()); if (stopped != null) { + reservations.releaseTask(stopped.execution.task.getSimulationTaskId()); queue.requeue(stopped.execution.task); hold(stopped.execution, event.getType().name()); events.add(event(tick, now, "HOLD", stopped.execution.task.getSimulationTaskId(), @@ -181,6 +184,7 @@ public final class HistoricalDispatchSimulator { GridPathPlanner planner, VehicleAssignmentService assignment, SimulatedPlcGateway plc, + CorridorReservationManager reservations, List events) { boolean progress = false; List ready = new ArrayList( @@ -233,10 +237,14 @@ public final class HistoricalDispatchSimulator { List route = new ArrayList(approaches.get(vehicleId)); for (int i = 1; i < servicePath.size(); i++) route.add(servicePath.get(i)); Execution execution = executions.get(task.getSimulationTaskId()); - if (hasOppositeRouteConflict(route, active.values())) { - hold(execution, "OPPOSITE_ROUTE_CONFLICT"); + CorridorReservationManager.ReservationDecision reservation = reservations.reserve( + task.getSimulationTaskId(), route, + queue.effectivePriorityFor(task, now), + queue.arrivalMillis(task.getSimulationTaskId())); + if (!reservation.isGranted()) { + hold(execution, reservation.getReason()); events.add(event(tick, now, "HOLD", task.getSimulationTaskId(), vehicleId, - "OPPOSITE_ROUTE_CONFLICT", route.get(0), route.get(route.size() - 1))); + reservation.getReason(), route.get(0), route.get(route.size() - 1))); continue; } if (execution.firstAssignedMillis == null) execution.firstAssignedMillis = now; @@ -252,27 +260,6 @@ public final class HistoricalDispatchSimulator { return progress; } - private boolean hasOppositeRouteConflict(List proposed, - Iterable activeTasks) { - Set directions = directedEdges(proposed, 0); - for (ActiveTask activeTask : activeTasks) { - Set activeDirections = directedEdges(activeTask.route, activeTask.index); - for (String direction : directions) { - String[] pair = direction.split("->", -1); - if (activeDirections.contains(pair[1] + "->" + pair[0])) return true; - } - } - return false; - } - - private Set directedEdges(List path, int startIndex) { - Set edges = new LinkedHashSet(); - for (int i = Math.max(1, startIndex + 1); i < path.size(); i++) { - edges.add(path.get(i - 1) + "->" + path.get(i)); - } - return edges; - } - private boolean fleetCanReach(HistoricalSimulationTask task, Map vehicles, HistoricalSimulationInput input, @@ -298,6 +285,7 @@ public final class HistoricalDispatchSimulator { SimulationArchive archive, IdempotentWmsCompletionReceiver wms, SimulationInvariantSummary invariants, + CorridorReservationManager reservations, List events, Set traversedEdges) { boolean progress = false; @@ -310,7 +298,7 @@ public final class HistoricalDispatchSimulator { if (!vehicle.isAssignmentEligible()) continue; if (current.index >= current.route.size() - 1) { complete(tick, now, current, vehicle, active, terminal, plc, - archive, wms, invariants, events); + archive, wms, invariants, reservations, events); progress = true; continue; } @@ -321,7 +309,7 @@ public final class HistoricalDispatchSimulator { String reverse = to + "->" + from; if (traversedEdges.contains(reverse) || traversedEdges.contains(direction)) { recordHold(tick, now, current, vehicle, active, queue, - "OPPOSITE_EDGE_WINDOW", events); + "OPPOSITE_EDGE_WINDOW", reservations, events); continue; } List resources = Arrays.asList( @@ -347,14 +335,14 @@ public final class HistoricalDispatchSimulator { ShadowDecision decision = decisionEvent.getDecisions().get(0); if (decision.getType() != DecisionType.GRANT) { recordHold(tick, now, current, vehicle, active, queue, - decision.getReasonCode(), events); + decision.getReasonCode(), reservations, events); continue; } ResourceLedger.ProtectionResult protection = ledger.evaluate(vehicleId, resources); if (!protection.isGranted()) { invariants.capacityViolation(); recordHold(tick, now, current, vehicle, active, queue, - "AUTHORITATIVE_RESOURCE_REJECTED", events); + "AUTHORITATIVE_RESOURCE_REJECTED", reservations, events); continue; } SimulatedPlcGateway.PlcResponse sent = plc.send( @@ -369,6 +357,7 @@ public final class HistoricalDispatchSimulator { if (plc.attemptsFor(current.execution.task.getSimulationTaskId()) % config.getRetryLimit() == 0) { active.remove(vehicleId); + reservations.releaseTask(current.execution.task.getSimulationTaskId()); queue.requeue(current.execution.task); hold(current.execution, "RETRY_EXHAUSTED_REDISPATCH"); events.add(event(tick, now, "HOLD", @@ -381,6 +370,8 @@ public final class HistoricalDispatchSimulator { Site nextSite = input.getGridMap().getSite(to); vehicle.moveTo(to, nextSite.getFloor()); ledger.release(vehicleId, Arrays.asList(ResourceKey.node(from), edge)); + reservations.releasePassedEdge( + current.execution.task.getSimulationTaskId(), from, to); traversedEdges.add(direction); current.index++; current.holdStreak = 0; @@ -390,7 +381,7 @@ public final class HistoricalDispatchSimulator { progress = true; if (current.index >= current.route.size() - 1) { complete(tick, now, current, vehicle, active, terminal, plc, - archive, wms, invariants, events); + archive, wms, invariants, reservations, events); } } return progress; @@ -406,6 +397,7 @@ public final class HistoricalDispatchSimulator { SimulatedVehicleRuntime vehicle, Map active, SimulationTaskQueue queue, String reason, + CorridorReservationManager reservations, List events) { hold(current.execution, reason); current.holdStreak++; @@ -413,6 +405,7 @@ public final class HistoricalDispatchSimulator { vehicle.getVehicleId(), reason, vehicle.getCurrentSite(), null)); if (current.holdStreak >= HOLD_REDISPATCH_THRESHOLD) { active.remove(vehicle.getVehicleId()); + reservations.releaseTask(current.execution.task.getSimulationTaskId()); queue.requeue(current.execution.task); events.add(event(tick, now, "REDISPATCH", current.execution.task.getSimulationTaskId(), vehicle.getVehicleId(), @@ -428,8 +421,10 @@ public final class HistoricalDispatchSimulator { SimulationArchive archive, IdempotentWmsCompletionReceiver wms, SimulationInvariantSummary invariants, + CorridorReservationManager reservations, List events) { String taskId = activeTask.execution.task.getSimulationTaskId(); + reservations.releaseTask(taskId); if (!archive.complete(taskId, vehicle.getVehicleId(), now)) { invariants.duplicateArchiveAttempt(); } diff --git a/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/CorridorReservationManagerTest.java b/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/CorridorReservationManagerTest.java new file mode 100644 index 0000000..61ce292 --- /dev/null +++ b/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/CorridorReservationManagerTest.java @@ -0,0 +1,43 @@ +package com.ft546.wcs.shadow.history.simulation; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; + +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 CorridorReservationManagerTest { + @Test + void keepsOneDirectionUntilTheOwnerLeavesTheSharedCorridor() { + CorridorReservationManager manager = new CorridorReservationManager(); + + assertTrue(manager.reserve("HIGH", Arrays.asList("A", "B", "C"), 1, 20L) + .isGranted()); + CorridorReservationManager.ReservationDecision denied = manager.reserve( + "LOW", Arrays.asList("C", "B", "A"), 5, 10L); + + assertFalse(denied.isGranted()); + assertEquals("HIGH", denied.getOwnerTaskId()); + assertEquals("HIGH", manager.ownerFor("B", "C")); + manager.releasePassedEdge("HIGH", "A", "B"); + assertFalse(manager.reserve("LOW", Arrays.asList("C", "B", "A"), 0, 10L) + .isGranted()); + manager.releasePassedEdge("HIGH", "B", "C"); + manager.releaseTask("HIGH"); + assertTrue(manager.reserve("LOW", Arrays.asList("C", "B", "A"), 0, 10L) + .isGranted()); + } + + @Test + void admitsSameDirectionWhilePhysicalResourcesRemainCapacityOne() { + CorridorReservationManager manager = new CorridorReservationManager(); + + assertTrue(manager.reserve("FIRST", Arrays.asList("A", "B", "C"), 1, 0L) + .isGranted()); + assertTrue(manager.reserve("FOLLOWER", Arrays.asList("A", "B", "C"), 2, 1L) + .isGranted()); + assertEquals("FIRST", manager.ownerFor("A", "B")); + } +} diff --git a/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/HistoricalDispatchSimulatorTest.java b/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/HistoricalDispatchSimulatorTest.java index f2987ab..d57c680 100644 --- a/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/HistoricalDispatchSimulatorTest.java +++ b/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/HistoricalDispatchSimulatorTest.java @@ -53,7 +53,8 @@ class HistoricalDispatchSimulatorTest { assertEquals(0, result.getWaitingCount()); assertTrue(result.getEvents().stream().anyMatch(event -> "HOLD".equals(event.getType()) - && "OPPOSITE_ROUTE_CONFLICT".equals(event.getReason()))); + && "CORRIDOR_DIRECTION_OWNED".equals(event.getReason()))); + assertEquals(0, result.getInvariants().totalSafetyViolations()); } @Test