111 lines
5.1 KiB
Java
111 lines
5.1 KiB
Java
package com.ruoyi.shadow.kernel;
|
|
|
|
import java.util.ArrayList;
|
|
import java.util.Collections;
|
|
import java.util.LinkedHashMap;
|
|
import java.util.List;
|
|
import java.util.Map;
|
|
import java.util.concurrent.CancellationException;
|
|
import java.util.concurrent.TimeUnit;
|
|
import java.util.function.LongSupplier;
|
|
|
|
/** One bounded diagnostic snapshot per poll, including polls with no task evidence. */
|
|
final class ShadowCycleDiagnostics {
|
|
private final LongSupplier ticker;
|
|
private final long started;
|
|
private long phaseStarted;
|
|
private String phase = "QUEUED";
|
|
private final Map<String, Long> phaseNanos = new LinkedHashMap<String, Long>();
|
|
private Map<String, Object> completed;
|
|
private boolean cancelled;
|
|
private Map<String, Object> dispatch = Collections.emptyMap();
|
|
|
|
ShadowCycleDiagnostics() { this(System::nanoTime); }
|
|
|
|
ShadowCycleDiagnostics(LongSupplier ticker) {
|
|
this.ticker = ticker;
|
|
this.started = ticker.getAsLong();
|
|
this.phaseStarted = started;
|
|
}
|
|
|
|
synchronized void dispatch(com.ruoyi.shadow.facts.ShadowFactBatch batch,
|
|
com.ruoyi.wcs.application.dispatch.DispatchSnapshot snapshot,
|
|
com.ruoyi.wcs.application.dispatch.DispatchCycleResult result) {
|
|
if (completed != null || cancelled) return;
|
|
Map<String, Object> values = new LinkedHashMap<String, Object>();
|
|
values.put("eligiblePhases", snapshot.getReadySteps().size());
|
|
values.put("granted", result.getGranted().size());
|
|
values.put("held", result.getHeld().size());
|
|
values.put("yielded", result.getYielded().size());
|
|
Map<String, Integer> observations = new LinkedHashMap<String, Integer>();
|
|
for (com.ruoyi.shadow.facts.ShadowQueueFact queue : batch.getQueues()) {
|
|
if (!queue.isActive()) continue;
|
|
String reason = com.ruoyi.shadow.facts.ShadowQueueDispatchEligibility.observationReason(queue, batch);
|
|
if (reason != null) observations.merge(reason, 1, Integer::sum);
|
|
}
|
|
values.put("observations", Collections.unmodifiableMap(observations));
|
|
int safe = 0;
|
|
for (String id : snapshot.getTopology().getNodeIds()) {
|
|
com.ruoyi.wcs.domain.topology.TopologyNode node = snapshot.getTopology().node(id);
|
|
if (node.isSafeWait() || node.isSafeYield()) safe++;
|
|
}
|
|
values.put("validatedSafePointCount", safe);
|
|
values.put("safePointStatus", safe == 0 ? "UNCONFIGURED" : "TOPOLOGY_VALIDATED");
|
|
List<Map<String, Object>> holds = new ArrayList<Map<String, Object>>();
|
|
for (com.ruoyi.wcs.application.dispatch.HeldDispatch hold : result.getHeld()) {
|
|
if (holds.size() == 20) break;
|
|
Map<String, Object> row = new LinkedHashMap<String, Object>();
|
|
row.put("taskId", hold.getOrderId()); row.put("vehicleId", hold.getVehicleId());
|
|
row.put("reasonCode", hold.getReasonCode());
|
|
List<String> blockers = new ArrayList<String>(hold.getBlockingVehicleIds());
|
|
row.put("blockingVehicleIds", Collections.unmodifiableList(blockers.subList(0, Math.min(20, blockers.size()))));
|
|
holds.add(Collections.unmodifiableMap(row));
|
|
}
|
|
values.put("holdSamples", Collections.unmodifiableList(holds));
|
|
dispatch = Collections.unmodifiableMap(values);
|
|
}
|
|
|
|
synchronized void enter(String next) {
|
|
if (cancelled || Thread.currentThread().isInterrupted()) {
|
|
throw new CancellationException("SHADOW_CYCLE_CANCELLED");
|
|
}
|
|
if (completed != null) return;
|
|
long now = ticker.getAsLong();
|
|
accumulate(now);
|
|
phase = next;
|
|
phaseStarted = now;
|
|
}
|
|
|
|
synchronized Map<String, Object> finish(String outcome, Thread worker) {
|
|
if (completed != null) return completed;
|
|
long now = ticker.getAsLong();
|
|
accumulate(now);
|
|
Map<String, Object> snapshot = new LinkedHashMap<String, Object>();
|
|
snapshot.put("outcome", outcome);
|
|
snapshot.put("phase", phase);
|
|
snapshot.put("elapsedMillis", TimeUnit.NANOSECONDS.toMillis(now - started));
|
|
Map<String, Long> timings = new LinkedHashMap<String, Long>();
|
|
phaseNanos.forEach((key, value) -> timings.put(key, TimeUnit.NANOSECONDS.toMillis(value)));
|
|
snapshot.put("phaseMillis", Collections.unmodifiableMap(timings));
|
|
snapshot.put("dispatch", dispatch);
|
|
if (worker != null) {
|
|
snapshot.put("workerState", worker.getState().name());
|
|
List<String> stack = new ArrayList<String>();
|
|
StackTraceElement[] frames = worker.getStackTrace();
|
|
for (int i = 0; i < Math.min(16, frames.length); i++) stack.add(frames[i].toString());
|
|
snapshot.put("workerStack", Collections.unmodifiableList(stack));
|
|
}
|
|
completed = Collections.unmodifiableMap(snapshot);
|
|
return completed;
|
|
}
|
|
|
|
synchronized Map<String, Object> cancel(String outcome, Thread worker) {
|
|
cancelled = true;
|
|
return finish(outcome, worker);
|
|
}
|
|
|
|
private void accumulate(long now) {
|
|
phaseNanos.merge(phase, now - phaseStarted, Long::sum);
|
|
}
|
|
}
|