docs(javaTest): explain dispatch flow inside methods

This commit is contained in:
zgw
2026-08-04 10:37:20 +08:00
parent 4051e3f446
commit bd5d832543
18 changed files with 96 additions and 0 deletions

View File

@@ -47,6 +47,7 @@ public final class VehicleAssignmentService {
List<Candidate> candidates = new ArrayList<Candidate>();
boolean eligibleWithUnknownSite = false;
// 先按运行状态过滤,再把每台可用车相对任务源点的位置编码为可排序的评分项。
for (VehicleSnapshot vehicle : vehicles) {
if (!isEligible(vehicle)) {
continue;
@@ -60,6 +61,7 @@ public final class VehicleAssignmentService {
}
if (candidates.isEmpty()) {
// 区分“没有可用车”和“车辆快照引用未知站点”,让调用方得到可解释的 HOLD/拒绝原因。
if (eligibleWithUnknownSite) {
return VehicleAssignmentDecision.rejected(
DecisionCode.REJECT_INVALID_SITE,
@@ -72,6 +74,7 @@ public final class VehicleAssignmentService {
);
}
// 排序规则固定为同层优先、楼层距离、y/x 距离、车辆编号,避免历史 agv_id 影响当前选车。
Collections.sort(candidates, Candidate.ORDER);
Candidate selected = candidates.get(0);
return VehicleAssignmentDecision.assigned(
@@ -110,6 +113,7 @@ public final class VehicleAssignmentService {
*/
@Override
public int compare(Candidate left, Candidate right) {
// 逐项比较而非叠加总分,避免权重变化掩盖“同层优先”的硬业务规则。
int result = Integer.compare(
left.sameFloorPenalty,
right.sameFloorPenalty

View File

@@ -35,6 +35,7 @@ public final class GridPathPlanner {
}
Set<String> blocked = new LinkedHashSet<String>(map.getBlockedSites());
// 动态阻塞用于清障等临时场景;起点强制可用,因为车辆已实际占据该节点。
if (dynamicBlocked != null) {
blocked.addAll(dynamicBlocked);
}
@@ -52,8 +53,10 @@ public final class GridPathPlanner {
while (!open.isEmpty()) {
String current = open.removeFirst();
if (current.equals(end)) {
// BFS 首次到达终点时路径边数最少,可直接由前驱表还原。
return buildPath(parent, end);
}
// visited 在入队时标记,防止同一站点被多条等长路线重复扩展。
for (String neighbor : map.getNeighbors(current)) {
if (!blocked.contains(neighbor) && visited.add(neighbor)) {
parent.put(neighbor, current);

View File

@@ -68,6 +68,7 @@ public final class ShadowTrafficCoordinator {
throw new IllegalArgumentException("snapshot map version mismatch");
}
Map<String, TrafficVehicleSnapshot> current = new LinkedHashMap<String, TrafficVehicleSnapshot>();
// 先校验每台车的快照新鲜度;异常快照被隔离,不能参与资源裁决。
for (TrafficVehicleSnapshot vehicle : snapshot.getVehicles()) {
SnapshotFreshnessPolicy.SnapshotHealth health = freshnessPolicy.validate(
previousSnapshots.get(vehicle.getVehicleId()), vehicle, snapshot.getDecisionTime());
@@ -83,6 +84,7 @@ public final class ShadowTrafficCoordinator {
+ candidate.getVehicleId());
}
}
// 输入哈希把地图、资源和候选移动绑定为一次可审计的决策,便于后续重放对比。
String hash = computeInputHash(map, authoritativeLedger, snapshot, candidates);
List<ShadowDecision> decisions = policy.decide(candidates, authoritativeLedger, map);
previousSnapshots.putAll(current);
@@ -108,6 +110,7 @@ public final class ShadowTrafficCoordinator {
.append(snapshot.getMapVersion()).append('|')
.append(snapshot.getPolicyVersion());
List<ResourceKey> capacityKeys = new ArrayList<ResourceKey>(map.getCapacities().keySet());
// 采用稳定排序序列化所有可影响裁决的资源和持有者,避免集合遍历顺序改变哈希。
Collections.sort(capacityKeys);
for (ResourceKey key : capacityKeys) {
canonical.append("|CAP:").append(key.value()).append('=')

View File

@@ -24,12 +24,14 @@ public final class RollingDispatchPolicy {
public List<ShadowDecision> decide(List<ShadowCandidate> input, ResourceLedger authoritative,
EffectiveTrafficMap map) {
List<ShadowCandidate> candidates = new ArrayList<ShadowCandidate>(input);
// 固定排序决定同一轮资源竞争的胜出顺序,优先已承诺移动以避免中途抢占。
Collections.sort(candidates, ORDER);
ResourceLedger simulation = authoritative.copy();
WaitForGraph waits = new WaitForGraph();
List<ShadowDecision> decisions = new ArrayList<ShadowDecision>();
for (ShadowCandidate candidate : candidates) {
if (candidate.isCommitted()) {
// 已承诺车辆仍须在台账副本上复核资源,避免快照与当前资源状态不一致时盲目前进。
ResourceLedger.ProtectionResult committed =
simulation.evaluate(candidate.getVehicleId(), candidate.getResources());
decisions.add(committed.isGranted()
@@ -51,6 +53,7 @@ public final class RollingDispatchPolicy {
result.getConflictingResources(), Collections.<String>emptySet(), null));
continue;
}
// 等待图用于识别 A 等 B、B 等 A 之类的闭环;闭环不能通过继续等待自动解除。
boolean cycle = waits.wouldCreateCycle(candidate.getVehicleId(), result.getBlockingVehicles());
waits.addWait(candidate.getVehicleId(), result.getBlockingVehicles());
if (cycle) {
@@ -58,6 +61,7 @@ public final class RollingDispatchPolicy {
result.getConflictingResources(), result.getBlockingVehicles(), null));
} else if (candidate.getSafeYieldSite() != null
&& map.getSafeYieldSites().contains(candidate.getSafeYieldSite())) {
// 仅在地图明确标记安全让行点时允许 YIELD,避免把车辆引导至可能形成二次堵塞的位置。
decisions.add(decision(candidate, DecisionType.YIELD, "SAFE_YIELD",
result.getConflictingResources(), result.getBlockingVehicles(),
candidate.getSafeYieldSite()));

View File

@@ -37,6 +37,7 @@ public final class ClearanceMovePlanner {
|| reservedResources == null || map == null) {
throw new IllegalArgumentException("clearance planning inputs are required");
}
// 主任务剩余路线、当前车位和节点预约均不能成为清障终点或途经点。
Set<String> excluded = new LinkedHashSet<String>(ownerRemainingRoute);
excluded.addAll(occupiedSites);
for (ResourceKey resource : reservedResources) {
@@ -48,6 +49,7 @@ public final class ClearanceMovePlanner {
GridPathPlanner planner = new GridPathPlanner();
List<Candidate> candidates = new ArrayList<Candidate>();
// 枚举全图而不是只看邻居,使被阻塞车辆能找到最近的可达让路站点。
List<String> siteIds = new ArrayList<String>(map.getSites().keySet());
Collections.sort(siteIds);
for (String target : siteIds) {
@@ -58,6 +60,7 @@ public final class ClearanceMovePlanner {
map.getNeighbors(target).size() <= 1));
}
}
// 先按移动距离最短,再优先放到支路,减少清障车再次挡住主干道的概率。
Collections.sort(candidates, Comparator
.comparingInt((Candidate candidate) -> candidate.route.size())
.thenComparing(candidate -> candidate.branch ? 0 : 1)

View File

@@ -33,6 +33,7 @@ public final class CorridorReservationManager {
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);
@@ -41,6 +42,7 @@ public final class CorridorReservationManager {
"CORRIDOR_DIRECTION_OWNED");
}
}
// 预检成功后一次性登记任务优先级和全部同向边,供后续移动时逐段释放。
ranks.put(taskId, rank);
Set<String> reserved = taskEdges.computeIfAbsent(
taskId, ignored -> new LinkedHashSet<String>());
@@ -61,6 +63,7 @@ public final class CorridorReservationManager {
* @param to 执行该处理所需的 to 输入。
*/
public void releasePassedEdge(String taskId, String from, String to) {
// 车辆已越过该物理边,后车可以立即竞争它;剩余边为空时一并清理任务排序记录。
String edgeId = request(from, to).edgeId;
EdgeReservation reservation = edges.get(edgeId);
if (reservation != null) {
@@ -82,6 +85,7 @@ public final class CorridorReservationManager {
* @param taskId 执行该处理所需的 task id 输入。
*/
public void releaseTask(String taskId) {
// HOLD 重派、车辆故障和正常完成都会走这里,确保未来路线不残留方向占用。
Set<String> reserved = taskEdges.remove(taskId);
if (reserved != null) {
for (String edgeId : new ArrayList<String>(reserved)) {
@@ -117,6 +121,7 @@ public final class CorridorReservationManager {
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));

View File

@@ -42,6 +42,7 @@ public final class HistoricalDispatchSimulationReportWriter {
throw new IllegalArgumentException("output, dataset and result are required");
}
Files.createDirectories(outputDir);
// 摘要回答总体调度效果,两个 CSV 保留逐任务与逐 tick 的可追溯证据。
MAPPER.writeValue(outputDir.resolve("summary.json").toFile(), summary(dataset, result));
writeTaskCsv(outputDir.resolve("task-results.csv"), result);
writeEventCsv(outputDir.resolve("tick-events.csv"), result);
@@ -87,6 +88,7 @@ public final class HistoricalDispatchSimulationReportWriter {
value.put("historicalDuration", timeStats(historicalSamples(result)));
value.put("simulationMinusHistoricalDuration",
timeStats(durationDifferenceSamples(result)));
// 按原因聚合可直接说明等待和协议重试各是由什么业务条件触发。
value.put("holdReasons", aggregateReasons(result, true));
value.put("retryReasons", aggregateReasons(result, false));
value.put("failureReasons", failureReasons(result));
@@ -173,6 +175,7 @@ public final class HistoricalDispatchSimulationReportWriter {
private List<Long> durationDifferenceSamples(HistoricalDispatchSimulationResult result) {
List<Long> values = new ArrayList<Long>();
for (SimulationTaskResult task : result.getTaskResults()) {
// 只比较具备历史完成时间且本次确实完成的任务,避免把失败/等待误写为耗时差异。
Long historical = historicalDuration(task);
Long simulated = task.getCompletionMillis();
if (historical != null && simulated != null) {
@@ -254,6 +257,7 @@ public final class HistoricalDispatchSimulationReportWriter {
"duration_difference_ms", "hold_count", "protocol_attempts", "retry_count",
"hold_reasons", "retry_reasons", "vehicle_changed", "path_changed");
for (SimulationTaskResult task : result.getTaskResults()) {
// 任务行同时保留历史车/路径和新车/路径,供判断中心调度到底改变了什么。
Long historical = historicalDuration(task);
Long simulated = task.getCompletionMillis();
Long scaled = simulated == null ? null : simulated * result.getAcceleration();
@@ -309,6 +313,7 @@ public final class HistoricalDispatchSimulationReportWriter {
* @param values 执行该处理所需的 values 输入。
*/
private void row(StringBuilder target, Object... values) {
// 所有列都经 csv 转义,避免路径、原因中的逗号破坏下游分析工具读取。
for (int i = 0; i < values.length; i++) {
if (i > 0) target.append(',');
target.append(csv(values[i]));

View File

@@ -51,6 +51,7 @@ public final class HistoricalDispatchSimulator {
if (input == null || config == null || faults == null) {
throw new IllegalArgumentException("input, config and fault plan are required");
}
// 收集历史 create_date,以最早任务为零点建立加速后的任务到达时钟。
List<java.time.LocalDateTime> arrivalDates = new ArrayList<java.time.LocalDateTime>();
for (HistoricalSimulationTask task : input.getTasks()) arrivalDates.add(task.getCreatedAt());
if (arrivalDates.isEmpty()) {
@@ -59,6 +60,7 @@ public final class HistoricalDispatchSimulator {
SimulationClock clock = new SimulationClock(arrivalDates, config.getAcceleration());
SimulationTaskQueue queue = new SimulationTaskQueue(config.getAgingIntervalMillis());
Map<String, Execution> executions = new LinkedHashMap<String, Execution>();
// 将每条可调度历史记录登记为等待任务;历史完成状态和 finish_date 不会使任务提前完成。
for (HistoricalSimulationTask task : input.getTasks()) {
long arrival = clock.toSimulationMillis(task.getCreatedAt());
queue.add(task, arrival);
@@ -68,6 +70,7 @@ public final class HistoricalDispatchSimulator {
List<SimulationTickEvent> events = new ArrayList<SimulationTickEvent>();
Map<String, SimulationTaskResult> terminal =
new LinkedHashMap<String, SimulationTaskResult>();
// 输入阶段已识别出的非法任务直接形成 FAILED 结果,避免混入后续调度循环。
for (HistoricalSimulationFailure failure : input.getFailedInputs()) {
terminal.put(failure.getSimulationTaskId(), new SimulationTaskResult(
failure.getSimulationTaskId(), failure.getRecord(),
@@ -82,6 +85,7 @@ public final class HistoricalDispatchSimulator {
Map<String, ClearanceMove> clearanceMoves = new LinkedHashMap<String, ClearanceMove>();
ResourceLedger ledger = new ResourceLedger(input.getTrafficMap().getCapacities());
SimulationInvariantSummary invariants = new SimulationInvariantSummary();
// 初始车位也必须符合资源容量约束;否则把数据问题记录到不变量报告中。
occupyInitialNodes(vehicles, ledger, invariants);
SimulatedPlcGateway plc = new SimulatedPlcGateway(faults);
SimulationArchive archive = new SimulationArchive();
@@ -97,11 +101,13 @@ public final class HistoricalDispatchSimulator {
SimulationProgressWatchdog watchdog = new SimulationProgressWatchdog(
config.getNoProgressLimit(), terminal.size());
while (terminal.size() < expected && tick < config.getTickLimit()) {
// 固定 tick 顺序保证故障、选车、清障和正常移动在相同输入下具有确定性结果。
applyVehicleEvents(tick, now, faults, vehicles, active,
clearanceMoves, queue, executions, reservations, events);
assignReadyTasks(tick, now, input, queue, vehicles, active,
clearanceMoves, executions, terminal, planner, assignment, plc,
reservations, events);
// 本 tick 已通过的边被共享给两类移动,禁止另一车辆同 tick 重复或反向穿越。
Set<String> traversedEdges = new LinkedHashSet<String>();
ClearanceProgress clearanceProgress = advanceClearanceMoves(
tick, now, input, config, vehicles, clearanceMoves,
@@ -113,6 +119,7 @@ public final class HistoricalDispatchSimulator {
events, traversedEdges);
verifyLedger(ledger, invariants);
// 尚未到达的未来任务不应被误判为死锁,因为它只是在等待 create_date 对应的时刻。
boolean awaitingFutureArrival = queue.size() > 0
&& active.isEmpty() && clearanceMoves.isEmpty()
&& queue.readyAt(now).isEmpty();
@@ -135,6 +142,7 @@ public final class HistoricalDispatchSimulator {
+ ", clearance=" + clearanceMoves.keySet());
}
// 对终态按稳定任务标识排序,保证报告、快照和回放比较不受 Map 插入顺序影响。
List<SimulationTaskResult> results = new ArrayList<SimulationTaskResult>(terminal.values());
Collections.sort(results, Comparator.comparing(SimulationTaskResult::getSimulationTaskId));
return new HistoricalDispatchSimulationResult(results, events, invariants,
@@ -208,6 +216,7 @@ public final class HistoricalDispatchSimulator {
for (DeterministicFaultPlan.VehicleEvent event : faults.eventsAt(tick)) {
SimulatedVehicleRuntime vehicle = vehicles.get(event.getVehicleId());
if (vehicle == null) throw new IllegalArgumentException("unknown fault vehicle");
// 故障计划只改变车辆运行态;随后统一由可分配性规则决定是否需要回收任务。
switch (event.getType()) {
case DISCONNECT: vehicle.disconnect(); break;
case CONNECT: vehicle.connect(); break;
@@ -223,6 +232,7 @@ public final class HistoricalDispatchSimulator {
event.getType().name(), vehicle.getCurrentSite(), vehicle.getCurrentSite()));
changed = true;
if (!vehicle.isAssignmentEligible()) {
// 失效车不能继续清障,清障主任务会在下一 tick 重新申请让路。
ClearanceMove clearance = clearanceMoves.remove(vehicle.getVehicleId());
if (clearance != null) {
events.add(event(tick, now, "CLEARANCE_HOLD", clearance.getOwnerTaskId(),
@@ -231,6 +241,7 @@ public final class HistoricalDispatchSimulator {
}
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());
@@ -278,10 +289,13 @@ public final class HistoricalDispatchSimulator {
CorridorReservationManager reservations,
List<SimulationTickEvent> events) {
boolean progress = false;
// 只处理 create_date 已到达的任务;队列已按优先级、等待老化和历史标识完成稳定排序。
List<HistoricalSimulationTask> ready = new ArrayList<HistoricalSimulationTask>(
queue.readyAt(now));
for (HistoricalSimulationTask task : ready) {
if (terminal.containsKey(task.getSimulationTaskId())) continue;
// 服务路径由完整地图重新计算。历史 path_info 不参与规划,空路径表示业务端点永久不可达。
List<String> servicePath = planner.shortestPath(input.getGridMap(),
task.getPhysicalSource(), task.getPhysicalTarget(), Collections.emptySet());
if (servicePath.isEmpty()) {
@@ -292,6 +306,7 @@ public final class HistoricalDispatchSimulator {
continue;
}
// 为每台空闲且健康的车计算到任务源点的接近路径,无法接近源点的车不进入本轮选车。
List<VehicleSnapshot> candidates = new ArrayList<VehicleSnapshot>();
Map<String, List<String>> approaches = new LinkedHashMap<String, List<String>>();
for (SimulatedVehicleRuntime vehicle : vehicles.values()) {
@@ -307,6 +322,7 @@ public final class HistoricalDispatchSimulator {
}
}
if (candidates.isEmpty()) {
// 暂无可分配车不等于失败;只有全车队都无法抵达源点时才关闭为 NO_ROUTE。
if (!fleetCanReach(task, vehicles, input, planner)) {
queue.remove(task.getSimulationTaskId());
terminal.put(task.getSimulationTaskId(), result(executions.get(
@@ -317,6 +333,7 @@ public final class HistoricalDispatchSimulator {
continue;
}
// 调度中心根据当前候选快照重新选车,历史 agv_id 只留在最终结果中作对照。
Site source = input.getGridMap().getSite(task.getPhysicalSource());
DispatchTask dispatchTask = new DispatchTask(task.getSimulationTaskId(),
task.getTaskType(), task.getPhysicalSource(), task.getPhysicalTarget(),
@@ -325,9 +342,13 @@ public final class HistoricalDispatchSimulator {
dispatchTask, candidates, input.getGridMap().getSites());
if (!decision.getVehicle().isPresent()) continue;
String vehicleId = decision.getVehicle().get().getVehicleId();
// 车辆先走到取货/起始点,再沿新规划的服务路径到达任务终点,形成实际执行路线。
List<String> route = new ArrayList<String>(approaches.get(vehicleId));
for (int i = 1; i < servicePath.size(); i++) route.add(servicePath.get(i));
Execution execution = executions.get(task.getSimulationTaskId());
// 提前预约整条路线的走廊方向。冲突只产生 HOLD,不把任务从等待队列移除。
CorridorReservationManager.ReservationDecision reservation = reservations.reserve(
task.getSimulationTaskId(), route,
queue.effectivePriorityFor(task, now),
@@ -338,6 +359,8 @@ public final class HistoricalDispatchSimulator {
reservation.getReason(), route.get(0), route.get(route.size() - 1)));
continue;
}
// 预约成功才正式交接给活动任务;这些字段同时构成“历史方案与新方案”的可审计差异。
if (execution.firstAssignedMillis == null) execution.firstAssignedMillis = now;
execution.lastVehicleId = vehicleId;
execution.plannedPath = new ArrayList<String>(servicePath);
@@ -421,6 +444,7 @@ public final class HistoricalDispatchSimulator {
List<SimulationTickEvent> events,
Set<String> traversedEdges) {
boolean progress = false;
// 按车辆编号遍历活动任务,避免 HashMap 遍历顺序改变同 tick 的资源竞争结果。
List<String> vehicleIds = new ArrayList<String>(active.keySet());
Collections.sort(vehicleIds);
for (String vehicleId : vehicleIds) {
@@ -429,6 +453,7 @@ public final class HistoricalDispatchSimulator {
SimulatedVehicleRuntime vehicle = vehicles.get(vehicleId);
if (!vehicle.isAssignmentEligible()) continue;
if (current.index >= current.route.size() - 1) {
// 路线已经走完时不再申请资源,直接进入归档和 WMS 回调闭环。
complete(tick, now, current, vehicle, active, terminal, plc,
archive, wms, invariants, reservations, events);
progress = true;
@@ -439,6 +464,7 @@ public final class HistoricalDispatchSimulator {
ResourceKey edge = ResourceKey.edge(from, to);
String direction = from + "->" + to;
String reverse = to + "->" + from;
// 同一 tick 禁止重复占边和对向换边,防止两个移动在物理边中相撞。
if (traversedEdges.contains(reverse) || traversedEdges.contains(direction)) {
recordHold(tick, now, current, vehicle, active, queue,
"OPPOSITE_EDGE_WINDOW", reservations, events);
@@ -446,6 +472,7 @@ public final class HistoricalDispatchSimulator {
}
List<ResourceKey> resources = Arrays.asList(
ResourceKey.node(from), edge, ResourceKey.node(to));
// 目标节点被空闲车长期占住时,优先让阻塞车执行一条安全清障路线,而非强行抢占资源。
Set<String> idleBlockers = idleVehiclesAt(to, vehicleId, vehicles,
active, clearanceMoves);
if (!idleBlockers.isEmpty()) {
@@ -458,6 +485,7 @@ public final class HistoricalDispatchSimulator {
reservations, events);
continue;
}
// 将车辆当前位置、意图和剩余路线冻结为影子快照,交通策略不能读取或修改实时对象。
Instant decisionTime = Instant.EPOCH.plusMillis(now);
TrafficVehicleSnapshot frozen = new TrafficVehicleSnapshot(vehicleId,
Long.toString(vehicle.getBootId()), tick + 1, tick + 1,
@@ -478,10 +506,12 @@ public final class HistoricalDispatchSimulator {
Collections.singletonList(candidate));
ShadowDecision decision = decisionEvent.getDecisions().get(0);
if (decision.getType() != DecisionType.GRANT) {
// 策略层拒绝时保留当前任务和预约,只累计 HOLD,等待下一轮优先级老化后再裁决。
recordHold(tick, now, current, vehicle, active, queue,
decision.getReasonCode(), reservations, events);
continue;
}
// 策略许可后仍要通过权威资源台账原子预占起点、边和目标,防止试算与实际状态脱节。
ResourceLedger.ProtectionResult protection = ledger.evaluate(vehicleId, resources);
if (!protection.isGranted()) {
boolean clearancePlanned = planClearanceForIdleBlockers(
@@ -494,10 +524,12 @@ public final class HistoricalDispatchSimulator {
reservations, events);
continue;
}
// 资源已预占后才下发 PLC;只有确认 ACK 才能提交车辆位置变化。
SimulatedPlcGateway.PlcResponse sent = plc.send(
current.execution.task.getSimulationTaskId(), vehicle, from, to);
SimulatedPlcGateway.PlcResponse accepted = plc.accept(sent, vehicle);
if (accepted.getReason() != SimulatedPlcGateway.PlcReason.ACK) {
// 命令未被确认时撤销本步新增占用,但保留当前位置和路线索引以便安全重试。
ledger.release(vehicleId, Arrays.asList(edge, ResourceKey.node(to)));
retry(current.execution, accepted.getReason().name());
events.add(event(tick, now, "PLC_RETRY",
@@ -505,6 +537,7 @@ public final class HistoricalDispatchSimulator {
accepted.getReason().name(), from, to));
if (plc.attemptsFor(current.execution.task.getSimulationTaskId())
% config.getRetryLimit() == 0) {
// 重试额度耗尽后释放未来走廊并回队,避免单车协议异常无限阻塞全局交通。
active.remove(vehicleId);
reservations.releaseTask(current.execution.task.getSimulationTaskId());
queue.requeue(current.execution.task);
@@ -515,6 +548,7 @@ public final class HistoricalDispatchSimulator {
}
continue;
}
// ACK 是物理移动的提交点:更新车位、逐段释放旧节点和边,并撤销已通过的走廊预约。
Site nextSite = input.getGridMap().getSite(to);
vehicle.moveTo(to, nextSite.getFloor());
ledger.release(vehicleId, Arrays.asList(ResourceKey.node(from), edge));
@@ -528,6 +562,7 @@ public final class HistoricalDispatchSimulator {
"ACK", from, to));
progress = true;
if (current.index >= current.route.size() - 1) {
// 最后一条边确认后立即关闭任务,避免下一个 tick 才释放剩余走廊和归档。
complete(tick, now, current, vehicle, active, terminal, plc,
archive, wms, invariants, reservations, events);
}
@@ -564,6 +599,7 @@ public final class HistoricalDispatchSimulator {
Set<String> traversedEdges) {
boolean moved = false;
boolean completed = false;
// 清障同样按稳定车辆顺序推进,避免它与主任务争抢资源时出现非确定性。
List<String> vehicleIds = new ArrayList<String>(clearanceMoves.keySet());
Collections.sort(vehicleIds);
for (String vehicleId : vehicleIds) {
@@ -577,6 +613,7 @@ public final class HistoricalDispatchSimulator {
String from = clearance.currentSite();
String to = clearance.nextSite();
String direction = from + "->" + to;
// 清障不是资源豁免操作,也必须让开本 tick 已使用或反向使用的边。
if (traversedEdges.contains(to + "->" + from)
|| traversedEdges.contains(direction)) {
events.add(event(tick, now, "CLEARANCE_HOLD", clearance.getOwnerTaskId(),
@@ -586,6 +623,7 @@ public final class HistoricalDispatchSimulator {
ResourceKey edge = ResourceKey.edge(from, to);
List<ResourceKey> resources = Arrays.asList(
ResourceKey.node(from), edge, ResourceKey.node(to));
// 使用独立的影子快照裁决清障移动,保证清障遵守正常交通规则而不是直接挪车。
Instant decisionTime = Instant.EPOCH.plusMillis(now);
TrafficVehicleSnapshot frozen = new TrafficVehicleSnapshot(vehicleId,
Long.toString(vehicle.getBootId()), tick + 1, tick + 1,
@@ -613,6 +651,7 @@ public final class HistoricalDispatchSimulator {
vehicleId, "AUTHORITATIVE_RESOURCE_REJECTED", from, to));
continue;
}
// 清障的 PLC 命令同样必须收到当前 bootId 的 ACK;失败时只释放本步预占并等待重试。
clearance.attemptedProtocol();
SimulatedPlcGateway.PlcResponse sent = plc.send(
clearance.clearanceTaskId(), vehicle, from, to);
@@ -624,6 +663,7 @@ public final class HistoricalDispatchSimulator {
accepted.getReason().name(), from, to));
continue;
}
// ACK 后提交清障车位,释放旧资源;让路路线走完后移除清障任务,主任务随后可重新尝试前进。
Site nextSite = input.getGridMap().getSite(to);
vehicle.moveTo(to, nextSite.getFloor());
ledger.release(vehicleId, Arrays.asList(ResourceKey.node(from), edge));
@@ -674,6 +714,7 @@ public final class HistoricalDispatchSimulator {
List<String> blockers = new ArrayList<String>(blockingVehicleIds);
Collections.sort(blockers);
Set<String> occupiedSites = new LinkedHashSet<String>();
// 收集所有当前车位和台账已持有资源,作为清障规划不可穿越的动态障碍。
for (SimulatedVehicleRuntime vehicle : vehicles.values()) {
occupiedSites.add(vehicle.getCurrentSite());
}
@@ -687,12 +728,14 @@ public final class HistoricalDispatchSimulator {
|| active.containsKey(blockerId) || clearanceMoves.containsKey(blockerId)) {
continue;
}
// 规划器会从候选阻塞车当前位置出发,避开主任务剩余路线,选择最近的安全让路目标。
java.util.Optional<ClearanceMove> planned = clearancePlanner.plan(
blockerId, blocker.getCurrentSite(),
owner.execution.task.getSimulationTaskId(),
owner.route.subList(owner.index, owner.route.size()),
occupiedSites, reservedResources, input.getGridMap());
if (planned.isPresent()) {
// 一次只发起一辆车的清障,减少同时让路造成的二次资源冲突。
ClearanceMove clearance = planned.get();
clearanceMoves.put(blockerId, clearance);
events.add(event(tick, now, "CLEARANCE_PLANNED",

View File

@@ -30,6 +30,7 @@ public final class HistoricalGridMapCompiler {
throw new IllegalArgumentException("full map cells are required");
}
List<HistoricalMapCell> passable = new ArrayList<HistoricalMapCell>();
// 仅将地图值 1/2 编入路网;其他单元保持为障碍,不能被新规划绕行穿透。
for (HistoricalMapCell cell : cells) {
if (cell.getValue() == 1 || cell.getValue() == 2) passable.add(cell);
}
@@ -42,6 +43,7 @@ public final class HistoricalGridMapCompiler {
Map<String, HistoricalMapCell> sites = new LinkedHashMap<String, HistoricalMapCell>();
Map<String, HistoricalMapCell> coordinates = new LinkedHashMap<String, HistoricalMapCell>();
for (HistoricalMapCell cell : passable) {
// 站点编号和三维坐标都必须唯一,否则路径规划无法确定车辆实际所在节点。
if (cell.getSiteId().trim().isEmpty() || sites.put(cell.getSiteId(), cell) != null) {
throw new IllegalArgumentException("passable site ids must be unique and nonblank");
}
@@ -54,6 +56,7 @@ public final class HistoricalGridMapCompiler {
Set<TrafficMapCompiler.EdgeDefinition> trafficEdges =
new LinkedHashSet<TrafficMapCompiler.EdgeDefinition>();
// 只检查右侧和上侧邻格即可建立无向连接,避免同一物理边重复定义两次。
for (HistoricalMapCell cell : passable) {
connectIfPresent(gridMap, trafficEdges, cell,
coordinates.get(coordinate(cell.getX() + 1, cell.getY(), cell.getZ())));
@@ -98,6 +101,7 @@ public final class HistoricalGridMapCompiler {
for (String start : ids) {
if (result.containsKey(start)) continue;
component++;
// BFS 扩散到所有可达邻居;同一编号的站点可在调度开始前被判定为可连通。
ArrayDeque<String> queue = new ArrayDeque<String>();
queue.add(start);
result.put(start, component);

View File

@@ -26,6 +26,7 @@ public final class HistoricalSimulationInputFactory {
if (dataset == null) throw new IllegalArgumentException("dataset is required");
HistoricalQueueNormalizer normalizer = new HistoricalQueueNormalizer();
List<HistoricalPath> paths = new ArrayList<HistoricalPath>();
// 规范化只提取历史字段和参考路径质量,不会将历史路径作为本次调度输出。
for (HistoricalQueueRecord record : dataset.getRecords()) {
paths.add(normalizer.normalize(record, dataset.getMapSites()));
}
@@ -41,6 +42,7 @@ public final class HistoricalSimulationInputFactory {
new HistoricalGridMapCompiler().compile(dataset.getFullMapCells());
PhysicalEndpointResolver endpointResolver = new PhysicalEndpointResolver();
int sequence = 0;
// 每条记录独立解析类型和物理端点;非法类型被隔离为失败输入,不影响其他历史任务继续仿真。
for (HistoricalPath path : paths) {
HistoricalQueueRecord record = path.getRecord();
int stableSequence = sequence++;
@@ -54,6 +56,7 @@ public final class HistoricalSimulationInputFactory {
record, compiled.getPassableSites());
if (compiled.getComponentId(endpoints.getSource())
!= compiled.getComponentId(endpoints.getTarget())) {
// 两端地图连通性是调度启动前的硬约束,跨断开区域不能靠等待或选车解决。
throw new IllegalArgumentException("unreachable physical endpoints for " + id);
}
tasks.add(new HistoricalSimulationTask(id, record, type,

View File

@@ -19,6 +19,7 @@ public final class IdempotentWmsCompletionReceiver {
if (taskId == null || taskId.trim().isEmpty()) {
throw new IllegalArgumentException("task id is required");
}
// 传输次数与业务效果分开统计:重复投递可被观测,但 Set 只允许一次完成副作用。
deliveryAttempts++;
return completedTaskIds.add(taskId);
}

View File

@@ -24,6 +24,7 @@ public final class PhysicalEndpointResolver {
if (source == null) {
throw new IllegalArgumentException("unknown physical source " + record.getStartSite());
}
// 物理楼层由地图源站决定;历史 floor 只作为数据质量标志,不篡改真实站点。
int floor = source.getZ();
boolean declaredFloorConsistent = true;
if (record.getFloor() != null && !record.getFloor().trim().isEmpty()) {
@@ -33,6 +34,7 @@ public final class PhysicalEndpointResolver {
declaredFloorConsistent = false;
}
}
// “tsj” 是按楼层展开的逻辑终点,需要映射为该层真实站点后才能参与路径规划。
String targetId = "tsj".equals(record.getEndSite())
? "07050" + floor : record.getEndSite();
HistoricalMapCell target = passableSites.get(targetId);

View File

@@ -116,6 +116,7 @@ public final class SimulatedPlcGateway {
*/
public PlcResponse send(String taskId, SimulatedVehicleRuntime vehicle,
String from, String to) {
// 按任务累计协议尝试次数,使故障计划能精确指定第 N 次发送失败或超时。
int attempt = attempts.containsKey(taskId) ? attempts.get(taskId) + 1 : 1;
attempts.put(taskId, attempt);
PlcReason reason = faults.plcReason(taskId, attempt);
@@ -132,6 +133,7 @@ public final class SimulatedPlcGateway {
*/
public PlcResponse accept(PlcResponse response, SimulatedVehicleRuntime vehicle) {
if (response.getReason() != PlcReason.ACK) return response;
// 车辆重启后 bootId 会变化,旧 ACK 即使迟到也不能再提交路线索引和车位。
if (response.getBootId() != vehicle.getBootId()) {
return new PlcResponse(response.getTaskId(), response.getVehicleId(),
response.getFrom(), response.getTo(), response.getAttempt(),

View File

@@ -48,6 +48,7 @@ public final class SimulatedVehicleRuntime {
* @return 满足全部派车前提时返回 {@code true}。
*/
public boolean isAssignmentEligible() {
// 重启后的未同步状态与断线、故障、低电量一样,都禁止调度中心继续派车。
return connected && !faulted && synchronizedAfterBoot && battery > 30;
}
@@ -64,6 +65,7 @@ public final class SimulatedVehicleRuntime {
* 模拟车辆重启并提升 bootId,强制等待同步以隔离重启前的 PLC 确认。
*/
public void restart() {
// 新 bootId 使重启前滞留的 PLC ACK 可被网关识别并拒绝。
bootId++;
synchronizedAfterBoot = false;
}
@@ -100,6 +102,7 @@ public final class SimulatedVehicleRuntime {
* @return 本次处理产生的结果;调用方据此继续后续流程。
*/
public VehicleSnapshot snapshot() {
// 故障车以不可用健康值导出,让选车服务只看到可用于本轮评分的业务状态。
return new VehicleSnapshot(vehicleId, floor, currentSite, 5,
faulted ? 0 : 1, battery, connected,
connected && !faulted && synchronizedAfterBoot, false);

View File

@@ -60,6 +60,7 @@ public final class SimulationArchive {
* @return 首次归档返回 {@code true};任务已存在时返回 {@code false}。
*/
public boolean complete(String taskId, String vehicleId, long completedAtMillis) {
// 归档记录一旦存在即拒绝覆盖,保留首次真实完成的车辆和仿真时间作为审计事实。
if (records.containsKey(taskId)) return false;
records.put(taskId, new ArchiveRecord(taskId, vehicleId, completedAtMillis));
return true;

View File

@@ -42,10 +42,12 @@ public final class SimulationProgressWatchdog {
throw new IllegalArgumentException("terminal count must be monotonic");
}
if (terminalCount > lastTerminalCount || vehicleMoved || clearanceCompleted) {
// 任务关闭或任何物理位移都证明系统仍在收敛,清零连续无进展计数。
lastTerminalCount = terminalCount;
unchangedEvents = 0;
return false;
}
// 反复 HOLD 而没有关闭、移动或清障完成时,最终把等待环暴露为死锁而非无限循环。
unchangedEvents++;
return unchangedEvents >= unchangedLimit;
}

View File

@@ -36,6 +36,7 @@ public final class SimulationTaskQueue {
if (originalArrivals.containsKey(id)) {
throw new IllegalArgumentException("duplicate task " + id);
}
// 原始到达时刻只记录一次;任务发生 HOLD 重派时必须保留等待资历,老化规则才不会失效。
originalArrivals.put(id, arrivalMillis);
entries.put(id, new Entry(task, arrivalMillis));
}
@@ -47,6 +48,7 @@ public final class SimulationTaskQueue {
public void requeue(HistoricalSimulationTask task) {
Long arrival = originalArrivals.get(task.getSimulationTaskId());
if (arrival == null) throw new IllegalArgumentException("task was never admitted");
// 重新排队不覆盖原始 arrival,确保故障任务恢复后不会不公平地排在新任务之后。
entries.put(task.getSimulationTaskId(), new Entry(task, arrival));
}
@@ -70,10 +72,12 @@ public final class SimulationTaskQueue {
*/
public List<HistoricalSimulationTask> readyAt(final long nowMillis) {
List<Entry> ready = new ArrayList<Entry>();
// create_date 尚未映射到当前仿真时刻的任务留在队列中,不参加本轮选车。
for (Entry entry : entries.values()) {
if (entry.arrivalMillis <= nowMillis) ready.add(entry);
}
Collections.sort(ready, Comparator
// 优先级数字越小越紧急;后续字段只负责消除同优先级下的不确定性。
.comparingInt((Entry entry) -> effectivePriority(entry, nowMillis))
.thenComparingLong(entry -> entry.arrivalMillis)
.thenComparing(entry -> text(entry.task.getRecord().getTaskNo()))
@@ -130,6 +134,7 @@ public final class SimulationTaskQueue {
*/
private int effectivePriority(Entry entry, long nowMillis) {
long waited = Math.max(0L, nowMillis - entry.arrivalMillis);
// 每跨过一个老化区间提高一级优先级,最低为 0,防止反复 HOLD 的任务永久饥饿。
long improvement = waited / agingIntervalMillis;
return (int) Math.max(0L, (long) entry.task.getPriority() - improvement);
}

View File

@@ -34,6 +34,7 @@ public final class ResourceLedger {
public ProtectionResult evaluate(String vehicleId, Collection<ResourceKey> requested) {
LinkedHashSet<ResourceKey> conflicts = new LinkedHashSet<ResourceKey>();
LinkedHashSet<String> blockers = new LinkedHashSet<String>();
// 第一阶段只检查容量,不写入 holders;这样任一资源失败都不会留下半次移动的占用。
for (ResourceKey key : requested) {
if (!capacities.containsKey(key)) {
conflicts.add(key);
@@ -54,6 +55,7 @@ public final class ResourceLedger {
if (!conflicts.isEmpty()) {
return new ProtectionResult(false, false, conflicts, blockers);
}
// 第二阶段一次性提交全部资源,车辆才取得从当前节点进入下一节点的通行保护。
for (ResourceKey key : requested) {
if (!holders.containsKey(key)) {
holders.put(key, new LinkedHashSet<String>());
@@ -70,6 +72,7 @@ public final class ResourceLedger {
* @param resources 执行该处理所需的 resources 输入。
*/
public void release(String vehicleId, Collection<ResourceKey> resources) {
// 释放是幂等的:资源或持有者不存在时不报错,便于处理 PLC 失败后的补偿路径。
for (ResourceKey key : resources) {
Set<String> current = holders.get(key);
if (current != null) {