From 892ea3fce65434579c07f989046c8d8781dff7d3 Mon Sep 17 00:00:00 2001 From: zgw <13623675548@163.com> Date: Fri, 31 Jul 2026 11:21:44 +0800 Subject: [PATCH] feat: add accelerated starvation-safe simulation queue --- .../history/simulation/SimulationClock.java | 55 +++++++++++++ .../history/simulation/SimulationConfig.java | 35 ++++++++ .../simulation/SimulationTaskQueue.java | 81 +++++++++++++++++++ .../simulation/SimulationClockTest.java | 34 ++++++++ .../simulation/SimulationTaskQueueTest.java | 47 +++++++++++ 5 files changed, 252 insertions(+) create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationClock.java create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationConfig.java create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationTaskQueue.java create mode 100644 javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/SimulationClockTest.java create mode 100644 javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/SimulationTaskQueueTest.java diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationClock.java b/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationClock.java new file mode 100644 index 0000000..6927799 --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationClock.java @@ -0,0 +1,55 @@ +package com.ft546.wcs.shadow.history.simulation; + +import java.time.Duration; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; + +public final class SimulationClock { + private static final long MAX_SOURCE_GAP_MILLIS = 30L * 60L * 1000L; + private final int acceleration; + private final Map arrivals; + + public SimulationClock(List sourceArrivals, int acceleration) { + if (acceleration != 10 && acceleration != 100) { + throw new IllegalArgumentException("acceleration must be 10 or 100"); + } + if (sourceArrivals == null || sourceArrivals.isEmpty()) { + throw new IllegalArgumentException("source arrivals are required"); + } + this.acceleration = acceleration; + List ordered = new ArrayList( + new LinkedHashSet(sourceArrivals)); + if (ordered.contains(null)) throw new IllegalArgumentException("arrival must not be null"); + Collections.sort(ordered); + LinkedHashMap converted = + new LinkedHashMap(); + long simulationMillis = 0L; + LocalDateTime previous = null; + for (LocalDateTime arrival : ordered) { + if (previous != null) { + long sourceGap = Duration.between(previous, arrival).toMillis(); + simulationMillis += Math.min(sourceGap, MAX_SOURCE_GAP_MILLIS) / acceleration; + } + converted.put(arrival, simulationMillis); + previous = arrival; + } + this.arrivals = Collections.unmodifiableMap(converted); + } + + public long toSimulationMillis(LocalDateTime sourceArrival) { + Long value = arrivals.get(sourceArrival); + if (value == null) throw new IllegalArgumentException("unknown source arrival"); + return value; + } + + public long sourceDurationMillis(long simulationDurationMillis) { + return Math.multiplyExact(simulationDurationMillis, acceleration); + } + + public int getAcceleration() { return acceleration; } +} diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationConfig.java b/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationConfig.java new file mode 100644 index 0000000..b5b082a --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationConfig.java @@ -0,0 +1,35 @@ +package com.ft546.wcs.shadow.history.simulation; + +public final class SimulationConfig { + private final int acceleration; + private final long agingIntervalMillis; + private final int retryLimit; + private final int noProgressLimit; + private final int tickLimit; + + public SimulationConfig(int acceleration, long agingIntervalMillis, int retryLimit, + int noProgressLimit, int tickLimit) { + if (acceleration != 10 && acceleration != 100) { + throw new IllegalArgumentException("acceleration must be 10 or 100"); + } + if (agingIntervalMillis <= 0 || retryLimit <= 0 + || noProgressLimit <= 0 || tickLimit <= 0) { + throw new IllegalArgumentException("simulation limits must be positive"); + } + this.acceleration = acceleration; + this.agingIntervalMillis = agingIntervalMillis; + this.retryLimit = retryLimit; + this.noProgressLimit = noProgressLimit; + this.tickLimit = tickLimit; + } + + public static SimulationConfig defaults() { + return new SimulationConfig(100, 30_000L, 3, 500, 20_000); + } + + public int getAcceleration() { return acceleration; } + public long getAgingIntervalMillis() { return agingIntervalMillis; } + public int getRetryLimit() { return retryLimit; } + public int getNoProgressLimit() { return noProgressLimit; } + public int getTickLimit() { return tickLimit; } +} diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationTaskQueue.java b/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationTaskQueue.java new file mode 100644 index 0000000..7521fa8 --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/SimulationTaskQueue.java @@ -0,0 +1,81 @@ +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.List; +import java.util.Map; + +public final class SimulationTaskQueue { + private final long agingIntervalMillis; + private final Map entries = new LinkedHashMap(); + private final Map originalArrivals = new LinkedHashMap(); + + public SimulationTaskQueue(long agingIntervalMillis) { + if (agingIntervalMillis <= 0) { + throw new IllegalArgumentException("aging interval must be positive"); + } + this.agingIntervalMillis = agingIntervalMillis; + } + + public void add(HistoricalSimulationTask task, long arrivalMillis) { + if (task == null || arrivalMillis < 0) { + throw new IllegalArgumentException("task and non-negative arrival are required"); + } + String id = task.getSimulationTaskId(); + if (originalArrivals.containsKey(id)) { + throw new IllegalArgumentException("duplicate task " + id); + } + originalArrivals.put(id, arrivalMillis); + entries.put(id, new Entry(task, arrivalMillis)); + } + + public void requeue(HistoricalSimulationTask task) { + Long arrival = originalArrivals.get(task.getSimulationTaskId()); + if (arrival == null) throw new IllegalArgumentException("task was never admitted"); + entries.put(task.getSimulationTaskId(), new Entry(task, arrival)); + } + + public HistoricalSimulationTask remove(String simulationTaskId) { + Entry entry = entries.remove(simulationTaskId); + return entry == null ? null : entry.task; + } + + public List readyAt(final long nowMillis) { + List ready = new ArrayList(); + 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())) + .thenComparing(entry -> text(entry.task.getRecord().getGuid())) + .thenComparingInt(entry -> entry.task.getArrivalSequence())); + List result = new ArrayList(); + for (Entry entry : ready) result.add(entry.task); + return Collections.unmodifiableList(result); + } + + public boolean isEmpty() { return entries.isEmpty(); } + public int size() { return entries.size(); } + + private int effectivePriority(Entry entry, long nowMillis) { + long waited = Math.max(0L, nowMillis - entry.arrivalMillis); + long improvement = waited / agingIntervalMillis; + return (int) Math.max(0L, (long) entry.task.getPriority() - improvement); + } + + private static String text(String value) { return value == null ? "" : value; } + + private static final class Entry { + private final HistoricalSimulationTask task; + private final long arrivalMillis; + + private Entry(HistoricalSimulationTask task, long arrivalMillis) { + this.task = task; + this.arrivalMillis = arrivalMillis; + } + } +} diff --git a/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/SimulationClockTest.java b/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/SimulationClockTest.java new file mode 100644 index 0000000..9b571ea --- /dev/null +++ b/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/SimulationClockTest.java @@ -0,0 +1,34 @@ +package com.ft546.wcs.shadow.history.simulation; + +import org.junit.jupiter.api.Test; + +import java.time.LocalDateTime; +import java.util.Arrays; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +class SimulationClockTest { + @Test + void scalesRelativeArrivalsAndCapsLongIdleGaps() { + LocalDateTime base = LocalDateTime.of(2026, 3, 1, 10, 0); + LocalDateTime tenSecondsLater = base.plusSeconds(10); + LocalDateTime twoHoursLater = base.plusHours(2); + LocalDateTime afterCappedGap = twoHoursLater.plusSeconds(10); + SimulationClock clock = new SimulationClock(Arrays.asList( + base, tenSecondsLater, twoHoursLater, afterCappedGap), 100); + + assertEquals(0L, clock.toSimulationMillis(base)); + assertEquals(100L, clock.toSimulationMillis(tenSecondsLater)); + assertEquals(18_100L, clock.toSimulationMillis(twoHoursLater)); + assertEquals(18_200L, clock.toSimulationMillis(afterCappedGap)); + assertEquals(1_000L, clock.sourceDurationMillis(10L)); + } + + @Test + void acceptsOnlySupportedAccelerationFactors() { + LocalDateTime base = LocalDateTime.of(2026, 3, 1, 10, 0); + assertThrows(IllegalArgumentException.class, + () -> new SimulationClock(Arrays.asList(base), 25)); + } +} diff --git a/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/SimulationTaskQueueTest.java b/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/SimulationTaskQueueTest.java new file mode 100644 index 0000000..c8d6f87 --- /dev/null +++ b/javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/SimulationTaskQueueTest.java @@ -0,0 +1,47 @@ +package com.ft546.wcs.shadow.history.simulation; + +import com.ft546.wcs.demo.model.DispatchTask; +import com.ft546.wcs.shadow.history.HistoricalQueueRecord; +import org.junit.jupiter.api.Test; + +import java.time.LocalDateTime; +import java.util.Arrays; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +class SimulationTaskQueueTest { + @Test + void ordersByPriorityThenArrivalAndProtectsOldTasksFromStarvation() { + HistoricalSimulationTask low = task("LOW", 3, 0); + HistoricalSimulationTask high = task("HIGH", 2, 1); + SimulationTaskQueue queue = new SimulationTaskQueue(30_000L); + queue.add(low, 0L); + queue.add(high, 20_000L); + + assertEquals(Arrays.asList(high, low), queue.readyAt(20_000L)); + assertEquals(Arrays.asList(low, high), queue.readyAt(90_000L)); + } + + @Test + void requeueKeepsTheOriginalArrivalAndAccumulatedAge() { + HistoricalSimulationTask old = task("OLD", 4, 0); + HistoricalSimulationTask newer = task("NEW", 2, 1); + SimulationTaskQueue queue = new SimulationTaskQueue(30_000L); + queue.add(old, 0L); + queue.add(newer, 60_000L); + queue.remove(old.getSimulationTaskId()); + queue.requeue(old); + + assertEquals(old, queue.readyAt(60_000L).get(0)); + } + + private HistoricalSimulationTask task(String id, int priority, int sequence) { + LocalDateTime created = LocalDateTime.of(2026, 3, 1, 10, 0).plusSeconds(sequence); + HistoricalQueueRecord record = new HistoricalQueueRecord( + id, id, "1", priority, "A", "B", "A,B", created.toString(), + created.plusMinutes(1).toString(), "1"); + return new HistoricalSimulationTask(id, record, DispatchTask.TaskType.INBOUND, + priority, created, created.plusMinutes(1), "A", "B", + Arrays.asList("A", "B"), sequence); + } +}