feat: add accelerated starvation-safe simulation queue

This commit is contained in:
zgw
2026-07-31 11:21:44 +08:00
parent 78e74daea1
commit 892ea3fce6
5 changed files with 252 additions and 0 deletions

View File

@@ -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<LocalDateTime, Long> arrivals;
public SimulationClock(List<LocalDateTime> 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<LocalDateTime> ordered = new ArrayList<LocalDateTime>(
new LinkedHashSet<LocalDateTime>(sourceArrivals));
if (ordered.contains(null)) throw new IllegalArgumentException("arrival must not be null");
Collections.sort(ordered);
LinkedHashMap<LocalDateTime, Long> converted =
new LinkedHashMap<LocalDateTime, Long>();
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; }
}

View File

@@ -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; }
}

View File

@@ -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<String, Entry> entries = new LinkedHashMap<String, Entry>();
private final Map<String, Long> originalArrivals = new LinkedHashMap<String, Long>();
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<HistoricalSimulationTask> readyAt(final long nowMillis) {
List<Entry> ready = new ArrayList<Entry>();
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<HistoricalSimulationTask> result = new ArrayList<HistoricalSimulationTask>();
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;
}
}
}

View File

@@ -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));
}
}

View File

@@ -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);
}
}