From 407184b9d57e5b5db15c01216a818e0327065ff0 Mon Sep 17 00:00:00 2001 From: zgw <13623675548@163.com> Date: Thu, 6 Aug 2026 17:47:27 +0800 Subject: [PATCH] feat(javaTest): reroute after repeated traffic hold --- .../CorridorReservationManager.java | 80 +++++- .../HistoricalDispatchSimulator.java | 269 +++++++++++++++--- .../simulation/SimulationTaskResult.java | 9 +- .../simulation/TrafficHoldTracker.java | 107 +++++++ .../CorridorReservationManagerTest.java | 22 ++ .../HistoricalDispatchSimulatorTest.java | 65 +++++ .../simulation/TrafficHoldTrackerTest.java | 48 ++++ 7 files changed, 549 insertions(+), 51 deletions(-) create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/TrafficHoldTracker.java create mode 100644 javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/TrafficHoldTrackerTest.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 index b55ca1f..db09bcf 100644 --- 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 @@ -14,6 +14,8 @@ public final class CorridorReservationManager { new LinkedHashMap(); private final Map> taskEdges = new LinkedHashMap>(); + private final Map> taskEdgeUses = + new LinkedHashMap>(); private final Map ranks = new LinkedHashMap(); /** @@ -33,11 +35,12 @@ public final class CorridorReservationManager { throw new IllegalArgumentException("task id and route are required"); } Rank rank = new Rank(taskId, effectivePriority, arrivalMillis); + List requests = routeRequests(route); // 先做全路线预检：任一边已有相反方向持有者即整体拒绝，避免只预约半条路线。 - for (int i = 1; i < route.size(); i++) { - EdgeRequest request = request(route.get(i - 1), route.get(i)); + for (EdgeRequest request : requests) { EdgeReservation existing = edges.get(request.edgeId); - if (existing != null && existing.direction != request.direction) { + if (existing != null && (existing.direction != request.direction + || existing.direction == 0)) { return ReservationDecision.denied(bestOwner(existing.holders), "CORRIDOR_DIRECTION_OWNED"); } @@ -46,13 +49,13 @@ public final class CorridorReservationManager { 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)); + for (EdgeRequest request : requests) { EdgeReservation reservation = edges.computeIfAbsent( request.edgeId, ignored -> new EdgeReservation(request.direction)); reservation.holders.add(taskId); reserved.add(request.edgeId); } + taskEdgeUses.put(taskId, edgeUseCounts(route)); return ReservationDecision.granted(); } @@ -76,17 +79,17 @@ public final class CorridorReservationManager { throw new IllegalArgumentException("task id and new route are required"); } - List requests = new ArrayList(); - for (int i = 1; i < newRoute.size(); i++) { - EdgeRequest request = request(newRoute.get(i - 1), newRoute.get(i)); - requests.add(request); + List requests = routeRequests(newRoute); + for (EdgeRequest request : requests) { EdgeReservation existing = edges.get(request.edgeId); - if (existing == null || existing.direction == request.direction) continue; + if (existing == null) continue; // 当前任务旧预约可以被翻转；其他任务仍持有旧方向时，新路线不能提交。 Set otherHolders = new LinkedHashSet(existing.holders); otherHolders.remove(taskId); - if (!otherHolders.isEmpty()) { + if (!otherHolders.isEmpty() + && (existing.direction != request.direction + || existing.direction == 0 || request.direction == 0)) { return ReservationDecision.denied(bestOwner(otherHolders), "CORRIDOR_DIRECTION_OWNED"); } @@ -120,6 +123,7 @@ public final class CorridorReservationManager { } else { taskEdges.put(taskId, replacementEdges); } + taskEdgeUses.put(taskId, edgeUseCounts(newRoute)); return ReservationDecision.granted(); } @@ -141,9 +145,9 @@ public final class CorridorReservationManager { List holders = new ArrayList(reservation.holders); Collections.sort(holders); result.add(new CorridorReservationSnapshot( - reservation.direction > 0 ? left : right, - reservation.direction > 0 ? right : left, - holders)); + reservation.direction >= 0 ? left : right, + reservation.direction >= 0 ? right : left, + holders, reservation.direction == 0)); } return Collections.unmodifiableList(result); } @@ -157,6 +161,17 @@ public final class CorridorReservationManager { public void releasePassedEdge(String taskId, String from, String to) { // 车辆已越过该物理边，后车可以立即竞争它；剩余边为空时一并清理任务排序记录。 String edgeId = request(from, to).edgeId; + Map uses = taskEdgeUses.get(taskId); + Integer remainingUses = uses == null ? null : uses.get(edgeId); + if (remainingUses != null && remainingUses > 1) { + // 路线后面还会再次通过此边时继续保持预约，防止其他车辆插入任务的回程窗口。 + uses.put(edgeId, remainingUses - 1); + return; + } + if (uses != null) { + uses.remove(edgeId); + if (uses.isEmpty()) taskEdgeUses.remove(taskId); + } EdgeReservation reservation = edges.get(edgeId); if (reservation != null) { reservation.holders.remove(taskId); @@ -179,6 +194,7 @@ public final class CorridorReservationManager { public void releaseTask(String taskId) { // HOLD 重派、车辆故障和正常完成都会走这里，确保未来路线不残留方向占用。 Set reserved = taskEdges.remove(taskId); + taskEdgeUses.remove(taskId); if (reserved != null) { for (String edgeId : new ArrayList(reserved)) { EdgeReservation reservation = edges.get(edgeId); @@ -235,6 +251,34 @@ public final class CorridorReservationManager { canonical ? 1 : -1); } + /** + * 把路线折叠为每条物理边的一次方向请求；同一任务往返经过时用方向 0 表示双向独占。 + */ + private List routeRequests(List route) { + Map requests = new LinkedHashMap(); + for (int index = 1; index < route.size(); index++) { + EdgeRequest next = request(route.get(index - 1), route.get(index)); + EdgeRequest existing = requests.get(next.edgeId); + if (existing == null) { + requests.put(next.edgeId, next); + } else if (existing.direction != next.direction) { + requests.put(next.edgeId, new EdgeRequest(next.edgeId, 0)); + } + } + return new ArrayList(requests.values()); + } + + /** 统计任务未来通过每条边的次数，确保最后一次通过后才逐段释放预约。 */ + private Map edgeUseCounts(List route) { + Map counts = new LinkedHashMap(); + for (int index = 1; index < route.size(); index++) { + String edgeId = request(route.get(index - 1), route.get(index)).edgeId; + Integer count = counts.get(edgeId); + counts.put(edgeId, count == null ? 1 : count + 1); + } + return counts; + } + public static final class ReservationDecision { private final boolean granted; private final String ownerTaskId; @@ -304,12 +348,15 @@ public final class CorridorReservationManager { private final String fromSite; private final String toSite; private final List taskIds; + private final boolean bidirectionalExclusive; private CorridorReservationSnapshot(String fromSite, String toSite, - List taskIds) { + List taskIds, + boolean bidirectionalExclusive) { this.fromSite = fromSite; this.toSite = toSite; this.taskIds = Collections.unmodifiableList(new ArrayList(taskIds)); + this.bidirectionalExclusive = bidirectionalExclusive; } /** 返回预约方向的起点。 */ @@ -320,6 +367,9 @@ public final class CorridorReservationManager { /** 返回共享该方向预约的任务编号，按编号稳定排序。 */ public List getTaskIds() { return taskIds; } + + /** 返回是否因同一任务后续需要反向通过而对这条边双向独占。 */ + public boolean isBidirectionalExclusive() { return bidirectionalExclusive; } } private static final class EdgeReservation { 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 d7dc926..5304919 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 @@ -5,11 +5,12 @@ import com.ft546.wcs.demo.center.VehicleAssignmentService; import com.ft546.wcs.demo.model.DispatchTask; import com.ft546.wcs.demo.model.Site; import com.ft546.wcs.demo.model.VehicleSnapshot; -import com.ft546.wcs.demo.traffic.GridPathPlanner; +import com.ft546.wcs.demo.traffic.RouteCandidate; import com.ft546.wcs.demo.traffic.RouteCandidatePlanner; import com.ft546.wcs.demo.traffic.RouteCandidateSet; import com.ft546.wcs.demo.traffic.RoutePlanDecision; import com.ft546.wcs.demo.traffic.RoutePlanningPolicy; +import com.ft546.wcs.demo.traffic.RouteQuality; import com.ft546.wcs.demo.traffic.RouteQualityEvaluator; import com.ft546.wcs.demo.traffic.TrafficRouteSnapshot; import com.ft546.wcs.shadow.application.ShadowTrafficCoordinator; @@ -36,7 +37,7 @@ import java.util.Set; public final class HistoricalDispatchSimulator { private static final long STEP_MILLIS = 1_000L; - private static final int HOLD_REDISPATCH_THRESHOLD = 10; + private static final int TRAFFIC_HOLD_REROUTE_THRESHOLD = 3; /** * 按中心调度的完整业务链路回放历史任务，而不是复用历史车辆或 {@code path_info}。 @@ -84,7 +85,7 @@ public final class HistoricalDispatchSimulator { Collections.emptyList(), Collections.emptyList(), 0L, null, null, 0, 0, 0, Collections.emptyMap(), Collections.emptyMap(), - Collections.emptyList())); + Collections.emptyList(), 0)); } Map vehicles = createRuntimes(input); @@ -97,7 +98,6 @@ public final class HistoricalDispatchSimulator { SimulatedPlcGateway plc = new SimulatedPlcGateway(faults); SimulationArchive archive = new SimulationArchive(); IdempotentWmsCompletionReceiver wms = new IdempotentWmsCompletionReceiver(); - GridPathPlanner planner = new GridPathPlanner(); RouteCandidatePlanner routePlanner = new RouteCandidatePlanner(); RouteQualityEvaluator routeEvaluator = new RouteQualityEvaluator(); VehicleAssignmentService assignment = new VehicleAssignmentService(); @@ -123,9 +123,9 @@ public final class HistoricalDispatchSimulator { ledger, plc, events, traversedEdges); boolean businessProgress = advanceActiveTasks( tick, now, input, config, queue, vehicles, - active, executions, terminal, planner, ledger, plc, archive, wms, - invariants, reservations, clearanceMoves, clearancePlanner, - events, traversedEdges); + active, executions, terminal, routePlanner, routeEvaluator, + ledger, plc, archive, wms, invariants, reservations, + clearanceMoves, clearancePlanner, events, traversedEdges); verifyLedger(ledger, invariants); // 尚未到达的未来任务不应被误判为死锁，因为它只是在等待 create_date 对应的时刻。 @@ -397,7 +397,8 @@ public final class HistoricalDispatchSimulator { execution.planningEvidence.add(new RoutePlanningEvidence( RoutePlanningEvidence.PlanningStage.SERVICE, execution.pathVersion, now, false, serviceDecision)); - active.put(vehicleId, new ActiveTask(execution, route)); + active.put(vehicleId, new ActiveTask(execution, route, + approachDecision.getSelectedPath().size() - 1)); queue.remove(task.getSimulationTaskId()); events.add(event(tick, now, "ASSIGN", task.getSimulationTaskId(), vehicleId, decision.getReason(), route.get(0), route.get(route.size() - 1))); @@ -413,7 +414,8 @@ public final class HistoricalDispatchSimulator { * @param task 执行该处理所需的 task 输入。 * @param vehicles 执行该处理所需的 vehicles 输入。 * @param input 执行该处理所需的 input 输入。 - * @param planner 执行该处理所需的 planner 输入。 + * @param routePlanner 执行该处理所需的 route planner 输入。 + * @param routeEvaluator 执行该处理所需的 route evaluator 输入。 * @return 存在通往源点的车队路径时返回 {@code true}。 */ private boolean fleetCanReach(HistoricalSimulationTask task, @@ -458,8 +460,14 @@ public final class HistoricalDispatchSimulator { for (CorridorReservationManager.CorridorReservationSnapshot reservation : reservations.snapshot()) { for (String taskId : reservation.getTaskIds()) { - snapshot.reserveDirection(reservation.getFromSite(), - reservation.getToSite(), taskId); + if (reservation.isBidirectionalExclusive()) { + // 往返任务在该边两个方向都可能再次通行，对其他任务表现为不可兼容的独占边。 + snapshot.reserveEdge(reservation.getFromSite(), + reservation.getToSite(), taskId); + } else { + snapshot.reserveDirection(reservation.getFromSite(), + reservation.getToSite(), taskId); + } } } return snapshot.build(); @@ -502,7 +510,8 @@ public final class HistoricalDispatchSimulator { Map active, Map executions, Map terminal, - GridPathPlanner planner, + RouteCandidatePlanner routePlanner, + RouteQualityEvaluator routeEvaluator, ResourceLedger ledger, SimulatedPlcGateway plc, SimulationArchive archive, @@ -536,8 +545,10 @@ public final class HistoricalDispatchSimulator { String reverse = to + "->" + from; // 同一 tick 禁止重复占边和对向换边，防止两个移动在物理边中相撞。 if (traversedEdges.contains(reverse) || traversedEdges.contains(direction)) { - recordHold(tick, now, current, vehicle, active, queue, - "OPPOSITE_EDGE_WINDOW", reservations, events); + progress |= recordTrafficHold(tick, now, current, vehicle, queue, + "OPPOSITE_EDGE_WINDOW", edge.value(), input, config, + routePlanner, routeEvaluator, vehicles, active, ledger, + reservations, events); continue; } List resources = Arrays.asList( @@ -549,10 +560,11 @@ public final class HistoricalDispatchSimulator { boolean clearancePlanned = planClearanceForIdleBlockers( tick, now, current, vehicles, active, clearanceMoves, clearancePlanner, ledger, input, idleBlockers, events); - recordHold(tick, now, current, vehicle, active, queue, + progress |= recordTrafficHold(tick, now, current, vehicle, queue, clearancePlanned ? "IDLE_VEHICLE_CLEARANCE" : "NO_CLEARANCE_SITE", - reservations, events); + ResourceKey.node(to).value(), input, config, routePlanner, + routeEvaluator, vehicles, active, ledger, reservations, events); continue; } // 将车辆当前位置、意图和剩余路线冻结为影子快照，交通策略不能读取或修改实时对象。 @@ -577,8 +589,11 @@ public final class HistoricalDispatchSimulator { ShadowDecision decision = decisionEvent.getDecisions().get(0); if (decision.getType() != DecisionType.GRANT) { // 策略层拒绝时保留当前任务和预约，只累计 HOLD，等待下一轮优先级老化后再裁决。 - recordHold(tick, now, current, vehicle, active, queue, - decision.getReasonCode(), reservations, events); + progress |= recordTrafficHold(tick, now, current, vehicle, queue, + decision.getReasonCode(), firstResource( + decision.getConflictingResources(), edge.value()), + input, config, routePlanner, routeEvaluator, vehicles, + active, ledger, reservations, events); continue; } // 策略许可后仍要通过权威资源台账原子预占起点、边和目标，防止试算与实际状态脱节。 @@ -588,10 +603,12 @@ public final class HistoricalDispatchSimulator { tick, now, current, vehicles, active, clearanceMoves, clearancePlanner, ledger, input, protection.getBlockingVehicles(), events); - recordHold(tick, now, current, vehicle, active, queue, + progress |= recordTrafficHold(tick, now, current, vehicle, queue, clearancePlanned ? "IDLE_VEHICLE_CLEARANCE" : "AUTHORITATIVE_RESOURCE_REJECTED", - reservations, events); + firstResource(protection.getConflictingResources(), edge.value()), + input, config, routePlanner, routeEvaluator, vehicles, + active, ledger, reservations, events); continue; } // 资源已预占后才下发 PLC；只有确认 ACK 才能提交车辆位置变化。 @@ -626,7 +643,7 @@ public final class HistoricalDispatchSimulator { current.execution.task.getSimulationTaskId(), from, to); traversedEdges.add(direction); current.index++; - current.holdStreak = 0; + current.moved(to); events.add(event(tick, now, "MOVE", current.execution.task.getSimulationTaskId(), vehicleId, "ACK", from, to)); @@ -873,24 +890,170 @@ public final class HistoricalDispatchSimulator { * @param reservations 执行该处理所需的 reservations 输入。 * @param events 执行该处理所需的 events 输入。 */ - private void recordHold(long tick, long now, ActiveTask current, + private boolean recordTrafficHold(long tick, long now, ActiveTask current, SimulatedVehicleRuntime vehicle, - Map active, SimulationTaskQueue queue, String reason, + String blockingResource, + HistoricalSimulationInput input, + SimulationConfig config, + RouteCandidatePlanner routePlanner, + RouteQualityEvaluator routeEvaluator, + Map vehicles, + Map active, + ResourceLedger ledger, CorridorReservationManager reservations, List events) { hold(current.execution, reason); - current.holdStreak++; + TrafficHoldTracker.Observation observation = current.trafficHolds.record( + current.execution.task.getSimulationTaskId(), vehicle.getVehicleId(), + blockingResource, reason); events.add(event(tick, now, "HOLD", current.execution.task.getSimulationTaskId(), 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(), - reason, vehicle.getCurrentSite(), null)); + if (!observation.isRerouteTriggered()) return false; + + events.add(event(tick, now, "REROUTE_TRIGGER", + current.execution.task.getSimulationTaskId(), vehicle.getVehicleId(), + reason + ";resource=" + blockingResource + ";count=" + + observation.getConsecutiveCount(), + vehicle.getCurrentSite(), null)); + return attemptReroute(tick, now, current, vehicle, queue, input, config, + routePlanner, routeEvaluator, vehicles, active, ledger, + reservations, events); + } + + /** + * 在第三次相同交通 HOLD 时，从车辆当前位置生成并试换严格更优的剩余路线。 + */ + private boolean attemptReroute(long tick, long now, ActiveTask current, + SimulatedVehicleRuntime vehicle, + SimulationTaskQueue queue, + HistoricalSimulationInput input, + SimulationConfig config, + RouteCandidatePlanner routePlanner, + RouteQualityEvaluator routeEvaluator, + Map vehicles, + Map active, + ResourceLedger ledger, + CorridorReservationManager reservations, + List events) { + String taskId = current.execution.task.getSimulationTaskId(); + if (!current.movedSinceReroute) { + events.add(event(tick, now, "REROUTE_REJECTED", taskId, + vehicle.getVehicleId(), "MOVE_REQUIRED_AFTER_LAST_REROUTE", + vehicle.getCurrentSite(), null)); + return false; } + + TrafficRouteSnapshot snapshot = trafficSnapshot(vehicles, active, ledger, reservations); + List oldRemaining = new ArrayList( + current.route.subList(current.index, current.route.size())); + List newEvidence = new ArrayList(); + List newRemaining; + RoutePlanDecision serviceDecision; + int nextVersion = current.execution.pathVersion + 1; + + // 未到业务源点时同时重新评价接近段和完整服务段；进入服务段后只规划当前位置到终点。 + if (!current.hasReachedServiceSource()) { + RouteCandidateSet approachCandidates = routePlanner.plan(input.getGridMap(), + vehicle.getCurrentSite(), current.execution.task.getPhysicalSource(), + Collections.emptySet(), RoutePlanningPolicy.defaultPolicy()); + RouteCandidateSet serviceCandidates = routePlanner.plan(input.getGridMap(), + current.execution.task.getPhysicalSource(), + current.execution.task.getPhysicalTarget(), + Collections.emptySet(), RoutePlanningPolicy.defaultPolicy()); + if (approachCandidates.getCandidates().isEmpty() + || serviceCandidates.getCandidates().isEmpty()) { + return rejectReroute(tick, now, current, vehicle, + "NO_BOUNDED_CANDIDATE", events); + } + RoutePlanDecision approachDecision = routeEvaluator.select(input.getGridMap(), + approachCandidates, snapshot, taskId, vehicle.getVehicleId()); + serviceDecision = routeEvaluator.select(input.getGridMap(), serviceCandidates, + snapshot, taskId, vehicle.getVehicleId()); + newRemaining = new ArrayList(approachDecision.getSelectedPath()); + for (int index = 1; index < serviceDecision.getSelectedPath().size(); index++) { + newRemaining.add(serviceDecision.getSelectedPath().get(index)); + } + newEvidence.add(new RoutePlanningEvidence( + RoutePlanningEvidence.PlanningStage.APPROACH, nextVersion, + now, true, approachDecision)); + } else { + RouteCandidateSet serviceCandidates = routePlanner.plan(input.getGridMap(), + vehicle.getCurrentSite(), current.execution.task.getPhysicalTarget(), + Collections.emptySet(), RoutePlanningPolicy.defaultPolicy()); + if (serviceCandidates.getCandidates().isEmpty()) { + return rejectReroute(tick, now, current, vehicle, + "NO_BOUNDED_CANDIDATE", events); + } + serviceDecision = routeEvaluator.select(input.getGridMap(), serviceCandidates, + snapshot, taskId, vehicle.getVehicleId()); + newRemaining = new ArrayList(serviceDecision.getSelectedPath()); + } + newEvidence.add(new RoutePlanningEvidence( + RoutePlanningEvidence.PlanningStage.SERVICE, nextVersion, + now, true, serviceDecision)); + + RouteQuality oldQuality = routeEvaluator.evaluate(input.getGridMap(), + new RouteCandidate(oldRemaining, + routePlanner.pathDistance(input.getGridMap(), oldRemaining)), + snapshot, taskId, vehicle.getVehicleId()); + RouteQuality newQuality = routeEvaluator.evaluate(input.getGridMap(), + new RouteCandidate(newRemaining, + routePlanner.pathDistance(input.getGridMap(), newRemaining)), + snapshot, taskId, vehicle.getVehicleId()); + if (newRemaining.equals(oldRemaining) || newQuality.compareTo(oldQuality) >= 0) { + return rejectReroute(tick, now, current, vehicle, + "NO_STRICTLY_BETTER_ROUTE", events); + } + + CorridorReservationManager.ReservationDecision replacement = + reservations.replaceRoute(taskId, newRemaining, + queuePriority(current.execution, now, config), + current.execution.arrivalMillis); + if (!replacement.isGranted()) { + return rejectReroute(tick, now, current, vehicle, + "RESERVATION_REPLACEMENT_FAILED:" + replacement.getReason(), events); + } + + // 预约先成功，才提交路线和证据；这条顺序保证失败时旧路线、旧索引和旧证据原样保留。 + boolean beforeSource = !current.hasReachedServiceSource(); + current.replaceRemainingRoute(newRemaining, beforeSource + ? newEvidence.get(0).getDecision().getSelectedPath().size() - 1 : 0); + if (beforeSource) { + current.execution.plannedPath = new ArrayList( + serviceDecision.getSelectedPath()); + } else { + List fullServicePath = new ArrayList(current.traversedServicePath); + for (int index = 1; index < serviceDecision.getSelectedPath().size(); index++) { + fullServicePath.add(serviceDecision.getSelectedPath().get(index)); + } + current.execution.plannedPath = fullServicePath; + } + current.execution.pathVersion = nextVersion; + current.execution.rerouteCount++; + current.execution.planningEvidence.addAll(newEvidence); + events.add(event(tick, now, "REROUTE", taskId, vehicle.getVehicleId(), + "STRICTLY_BETTER_TRAFFIC_QUALITY", oldRemaining.get(0), + newRemaining.get(newRemaining.size() - 1))); + return true; + } + + /** 记录未换路原因并保持现有路线和预约。 */ + private boolean rejectReroute(long tick, long now, ActiveTask current, + SimulatedVehicleRuntime vehicle, String reason, + List events) { + events.add(event(tick, now, "REROUTE_REJECTED", + current.execution.task.getSimulationTaskId(), vehicle.getVehicleId(), + reason, vehicle.getCurrentSite(), null)); + return false; + } + + /** 从资源集合稳定选出阻塞标识；策略未返回明细时使用本步物理边。 */ + private String firstResource(Set resources, String fallback) { + if (resources == null || resources.isEmpty()) return fallback; + List ordered = new ArrayList(resources); + Collections.sort(ordered); + return ordered.get(0).value(); } /** @@ -991,7 +1154,7 @@ public final class HistoricalDispatchSimulator { execution.holdCount, plc.attemptsFor(execution.task.getSimulationTaskId()), execution.retryCount, execution.holdReasons, execution.retryReasons, - execution.planningEvidence); + execution.planningEvidence, execution.rerouteCount); } /** @@ -1078,6 +1241,7 @@ public final class HistoricalDispatchSimulator { private final List planningEvidence = new ArrayList(); private int pathVersion; + private int rerouteCount; /** * 创建 {@code Execution} 实例并初始化其内部状态。 @@ -1094,9 +1258,13 @@ public final class HistoricalDispatchSimulator { private static final class ActiveTask { private final Execution execution; - private final List route; + private List route; private int index; - private int holdStreak; + private int serviceStartIndex; + private final List traversedServicePath = new ArrayList(); + private final TrafficHoldTracker trafficHolds = + new TrafficHoldTracker(TRAFFIC_HOLD_REROUTE_THRESHOLD); + private boolean movedSinceReroute = true; /** * 创建 {@code ActiveTask} 实例并初始化其内部状态。 @@ -1104,10 +1272,41 @@ public final class HistoricalDispatchSimulator { *

该方法属于 {@code ActiveTask}，用于保持当前模块的职责边界清晰。 * @param execution 执行该处理所需的 execution 输入。 * @param route 执行该处理所需的 route 输入。 + * @param serviceStartIndex 服务段源点在完整执行路线中的下标。 */ - private ActiveTask(Execution execution, List route) { + private ActiveTask(Execution execution, List route, int serviceStartIndex) { this.execution = execution; this.route = Collections.unmodifiableList(new ArrayList(route)); + this.serviceStartIndex = serviceStartIndex; + if (serviceStartIndex == 0) traversedServicePath.add(route.get(0)); + } + + /** 返回车辆是否已经到达业务源点并开始执行服务段。 */ + private boolean hasReachedServiceSource() { + return index >= serviceStartIndex || !traversedServicePath.isEmpty(); + } + + /** 成功换路后从当前位置重新建立剩余路线索引，并启用移动一次的防抖门禁。 */ + private void replaceRemainingRoute(List replacement, + int replacementServiceStartIndex) { + route = Collections.unmodifiableList(new ArrayList(replacement)); + index = 0; + serviceStartIndex = replacementServiceStartIndex; + movedSinceReroute = false; + } + + /** ACK 移动后更新服务段实际前缀、清空旧 HOLD，并解除换路防抖门禁。 */ + private void moved(String toSite) { + if (index >= serviceStartIndex && traversedServicePath.isEmpty()) { + traversedServicePath.add(execution.task.getPhysicalSource()); + } + if (!traversedServicePath.isEmpty() + && !traversedServicePath.get(traversedServicePath.size() - 1) + .equals(toSite)) { + traversedServicePath.add(toSite); + } + trafficHolds.clearAfterMovement(); + movedSinceReroute = true; } } diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationTaskResult.java b/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationTaskResult.java index 74ca692..c6e4cab 100644 --- a/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationTaskResult.java +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationTaskResult.java @@ -25,6 +25,7 @@ public final class SimulationTaskResult { private final Map holdReasons; private final Map retryReasons; private final List planningEvidence; + private final int rerouteCount; /** * 创建 {@code SimulationTaskResult} 实例并初始化其内部状态。 @@ -46,6 +47,7 @@ public final class SimulationTaskResult { * @param holdReasons 执行该处理所需的 hold reasons 输入。 * @param retryReasons 执行该处理所需的 retry reasons 输入。 * @param planningEvidence 每个实际采用路径版本的接近段和服务段决策证据。 + * @param rerouteCount 运行中成功替换剩余路线的次数。 */ SimulationTaskResult(String simulationTaskId, HistoricalQueueRecord record, SimulationTaskStatus status, String reason, @@ -55,7 +57,8 @@ public final class SimulationTaskResult { int holdCount, int protocolAttempts, int retryCount, Map holdReasons, Map retryReasons, - List planningEvidence) { + List planningEvidence, + int rerouteCount) { this.simulationTaskId = simulationTaskId; this.record = record; this.status = status; @@ -72,6 +75,7 @@ public final class SimulationTaskResult { this.holdReasons = immutableMap(holdReasons); this.retryReasons = immutableMap(retryReasons); this.planningEvidence = immutable(planningEvidence); + this.rerouteCount = rerouteCount; } /** @@ -182,6 +186,9 @@ public final class SimulationTaskResult { /** 返回任务每个路径版本的接近段和服务段中心规划证据。 */ public List getPlanningEvidence() { return planningEvidence; } + + /** 返回运行中因连续交通 HOLD 成功切换路线的次数。 */ + public int getRerouteCount() { return rerouteCount; } /** * 返回当前对象保存的“waiting millis”信息。 * diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/TrafficHoldTracker.java b/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/TrafficHoldTracker.java new file mode 100644 index 0000000..cc081d6 --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/TrafficHoldTracker.java @@ -0,0 +1,107 @@ +package com.ft546.wcs.shadow.history.simulation; + +import java.util.Objects; + +/** 只跟踪同一交通资源连续阻塞次数，不接收 PLC 或车辆健康事件。 */ +public final class TrafficHoldTracker { + private final int rerouteThreshold; + private HoldKey currentKey; + private int consecutiveCount; + + /** + * 创建连续交通 HOLD 跟踪器。 + * + * @param rerouteThreshold 第几次完全相同的交通阻塞触发一次重规划 + */ + public TrafficHoldTracker(int rerouteThreshold) { + if (rerouteThreshold <= 0) { + throw new IllegalArgumentException("reroute threshold must be positive"); + } + this.rerouteThreshold = rerouteThreshold; + } + + /** + * 登记一次交通 HOLD，并判断本次是否恰好到达重规划阈值。 + * + *

任务、车辆、阻塞资源或原因任一变化都代表交通条件已经变化，新序列从 1 开始。阈值只在恰好 + * 到达时触发一次；若没有更优路线，同一未变化阻塞不会每个 tick 重复做昂贵规划。 + */ + public Observation record(String taskId, String vehicleId, + String blockingResourceId, String holdReason) { + HoldKey next = new HoldKey(required(taskId), required(vehicleId), + required(blockingResourceId), required(holdReason)); + if (!next.equals(currentKey)) { + currentKey = next; + consecutiveCount = 1; + } else { + consecutiveCount++; + } + return new Observation(consecutiveCount, + consecutiveCount == rerouteThreshold, + blockingResourceId, holdReason); + } + + /** 车辆成功移动后清除旧阻塞序列，下一次 HOLD 必须重新累计三次。 */ + public void clearAfterMovement() { + currentKey = null; + consecutiveCount = 0; + } + + /** 拒绝空标识，保证事件中每次触发都能追溯到具体任务、车辆和资源。 */ + private String required(String value) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException("traffic hold key parts are required"); + } + return value; + } + + /** 一次 HOLD 累计后的可审计结果。 */ + public static final class Observation { + private final int consecutiveCount; + private final boolean rerouteTriggered; + private final String blockingResourceId; + private final String holdReason; + + private Observation(int consecutiveCount, boolean rerouteTriggered, + String blockingResourceId, String holdReason) { + this.consecutiveCount = consecutiveCount; + this.rerouteTriggered = rerouteTriggered; + this.blockingResourceId = blockingResourceId; + this.holdReason = holdReason; + } + + public int getConsecutiveCount() { return consecutiveCount; } + public boolean isRerouteTriggered() { return rerouteTriggered; } + public String getBlockingResourceId() { return blockingResourceId; } + public String getHoldReason() { return holdReason; } + } + + /** 完整标识“相同阻塞”的业务键。 */ + private static final class HoldKey { + private final String taskId; + private final String vehicleId; + private final String resourceId; + private final String reason; + + private HoldKey(String taskId, String vehicleId, + String resourceId, String reason) { + this.taskId = taskId; + this.vehicleId = vehicleId; + this.resourceId = resourceId; + this.reason = reason; + } + + @Override + public boolean equals(Object other) { + if (!(other instanceof HoldKey)) return false; + HoldKey that = (HoldKey) other; + return taskId.equals(that.taskId) && vehicleId.equals(that.vehicleId) + && resourceId.equals(that.resourceId) && reason.equals(that.reason); + } + + @Override + public int hashCode() { + return Objects.hash(taskId, vehicleId, resourceId, reason); + } + } +} 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 index 6f3075f..fd58587 100644 --- 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 @@ -91,6 +91,28 @@ class CorridorReservationManagerTest { assertEquals("TASK-1", manager.ownerFor("A", "D")); } + @Test + void routeThatUsesOneCorridorBothWaysKeepsItExclusiveToThatTask() { + CorridorReservationManager manager = new CorridorReservationManager(); + + assertTrue(manager.reserve("TURNAROUND", + Arrays.asList("B", "A", "B", "C"), 1, 0L).isGranted()); + CorridorReservationManager.ReservationDecision sameDirectionOtherTask = + manager.reserve("OTHER", Arrays.asList("D", "B", "A"), 2, 1L); + + assertFalse(sameDirectionOtherTask.isGranted()); + assertEquals("TURNAROUND", sameDirectionOtherTask.getOwnerTaskId()); + assertTrue(manager.snapshot().stream().anyMatch(reservation -> + reservation.isBidirectionalExclusive() + && reservation.getTaskIds().contains("TURNAROUND"))); + manager.releasePassedEdge("TURNAROUND", "B", "A"); + assertFalse(manager.reserve("WAITING", Arrays.asList("D", "B", "A"), + 2, 2L).isGranted()); + manager.releasePassedEdge("TURNAROUND", "A", "B"); + assertTrue(manager.reserve("AFTER-TURN", Arrays.asList("D", "B", "A"), + 2, 3L).isGranted()); + } + private Set edgesFor(CorridorReservationManager manager, String taskId) { Set edges = new LinkedHashSet(); for (CorridorReservationManager.CorridorReservationSnapshot reservation 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 44c49d5..47a5931 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 @@ -106,6 +106,33 @@ class HistoricalDispatchSimulatorTest { assertEquals(0, result.getFailedCount()); } + @Test + void reroutesOnThirdIdenticalTrafficHoldAfterConflictAppearsDuringExecution() { + DeterministicFaultPlan faults = DeterministicFaultPlan.builder() + .vehicleEvent(1L, "1", + DeterministicFaultPlan.VehicleEventType.DISCONNECT) + .build(); + + HistoricalDispatchSimulationResult result = new HistoricalDispatchSimulator().simulate( + runtimeRerouteInput(), new SimulationConfig(100, 30_000L, 3, 100, 300), + faults); + + SimulationTaskResult task = result.getTaskResults().stream() + .filter(value -> "REROUTE".equals(value.getSimulationTaskId())) + .findFirst().orElseThrow(AssertionError::new); + assertEquals(Arrays.asList("A", "C", "D"), task.getPlannedPath()); + assertEquals(1, task.getRerouteCount()); + assertTrue(result.getEvents().stream().anyMatch(event -> + "REROUTE_TRIGGER".equals(event.getType()))); + assertTrue(result.getEvents().stream().anyMatch(event -> + "REROUTE".equals(event.getType()))); + assertEquals(2, task.getPlanningEvidence().stream() + .map(RoutePlanningEvidence::getPathVersion) + .distinct().count()); + assertEquals(2, result.getCompletedCount()); + assertEquals(0, result.getInvariants().totalSafetyViolations()); + } + /** * 验证“moves idle blockers node by node and completes both business directions”场景，确保相关业务规则和断言成立。 * @@ -274,6 +301,44 @@ class HistoricalDispatchSimulatorTest { traffic, vehicles); } + private HistoricalSimulationInput runtimeRerouteInput() { + GridMap grid = new GridMap(); + grid.addSite(new Site("A", 1, 0, 0, "TEST")); + grid.addSite(new Site("B", 1, 5, 0, "TEST")); + grid.addSite(new Site("C", 1, 5, 1, "TEST")); + grid.addSite(new Site("D", 1, 10, 0, "TEST")); + grid.addSite(new Site("E", 1, 5, -1, "TEST")); + grid.connect("A", "B"); + grid.connect("B", "D"); + grid.connect("A", "C"); + grid.connect("C", "D"); + grid.connect("E", "B"); + + LinkedHashSet nodes = new LinkedHashSet( + Arrays.asList("A", "B", "C", "D", "E")); + LinkedHashSet edges = + new LinkedHashSet(); + edges.add(new TrafficMapCompiler.EdgeDefinition("A", "B")); + edges.add(new TrafficMapCompiler.EdgeDefinition("B", "D")); + edges.add(new TrafficMapCompiler.EdgeDefinition("A", "C")); + edges.add(new TrafficMapCompiler.EdgeDefinition("C", "D")); + edges.add(new TrafficMapCompiler.EdgeDefinition("E", "B")); + EffectiveTrafficMap traffic = new TrafficMapCompiler().compile( + new TrafficMapCompiler.MapDefinition("REROUTE", 1L, "TEST", true, + "test", "test", nodes, edges, Collections.emptyMap(), + Collections.emptySet(), Collections.emptySet())); + List tasks = Arrays.asList( + customTask("REROUTE", "A", "D", 1, 0, + Arrays.asList("A", "B", "D")), + customTask("BLOCKER", "E", "B", 2, 0, + Arrays.asList("E", "B"))); + List vehicles = Arrays.asList( + new VehicleSnapshot("1", 1, "E", 5, 1, 100, true, true, false), + new VehicleSnapshot("2", 1, "A", 5, 1, 100, true, true, false)); + return new HistoricalSimulationInput(tasks, Collections.emptyList(), grid, + traffic, vehicles); + } + /** * 验证“blocked three vehicle corridor input”场景，确保相关业务规则和断言成立。 * diff --git a/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/TrafficHoldTrackerTest.java b/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/TrafficHoldTrackerTest.java new file mode 100644 index 0000000..896f44a --- /dev/null +++ b/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/TrafficHoldTrackerTest.java @@ -0,0 +1,48 @@ +package com.ft546.wcs.shadow.history.simulation; + +import org.junit.jupiter.api.Test; + +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 TrafficHoldTrackerTest { + + @Test + void triggersOnlyOnThirdConsecutiveIdenticalTrafficBlock() { + TrafficHoldTracker tracker = new TrafficHoldTracker(3); + + TrafficHoldTracker.Observation first = tracker.record( + "TASK", "AGV", "NODE:B", "NODE_OCCUPIED"); + TrafficHoldTracker.Observation second = tracker.record( + "TASK", "AGV", "NODE:B", "NODE_OCCUPIED"); + TrafficHoldTracker.Observation third = tracker.record( + "TASK", "AGV", "NODE:B", "NODE_OCCUPIED"); + TrafficHoldTracker.Observation fourth = tracker.record( + "TASK", "AGV", "NODE:B", "NODE_OCCUPIED"); + + assertEquals(1, first.getConsecutiveCount()); + assertFalse(first.isRerouteTriggered()); + assertEquals(2, second.getConsecutiveCount()); + assertFalse(second.isRerouteTriggered()); + assertEquals(3, third.getConsecutiveCount()); + assertTrue(third.isRerouteTriggered()); + assertEquals(4, fourth.getConsecutiveCount()); + assertFalse(fourth.isRerouteTriggered()); + } + + @Test + void changedResourceReasonOrMovementStartsANewSequence() { + TrafficHoldTracker tracker = new TrafficHoldTracker(3); + tracker.record("TASK", "AGV", "NODE:B", "NODE_OCCUPIED"); + tracker.record("TASK", "AGV", "NODE:B", "NODE_OCCUPIED"); + + assertEquals(1, tracker.record( + "TASK", "AGV", "EDGE:B:C", "EDGE_RESERVED").getConsecutiveCount()); + assertEquals(1, tracker.record( + "TASK", "AGV", "EDGE:B:C", "OPPOSITE_DIRECTION").getConsecutiveCount()); + tracker.clearAfterMovement(); + assertEquals(1, tracker.record( + "TASK", "AGV", "EDGE:B:C", "OPPOSITE_DIRECTION").getConsecutiveCount()); + } +}