diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml new file mode 100644 index 0000000..abdee7d --- /dev/null +++ b/.github/workflows/checks.yml @@ -0,0 +1,20 @@ +name: Checks + +on: [push, pull_request] + +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.10", "3.12"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - run: python -m pip install -e '.[dev]' + - run: python -m pytest + - run: ruff check . + - run: ruff format --check . + - run: mypy diff --git a/README.md b/README.md index 2b639da..9438a31 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,38 @@ -# adaptivequeue -A lightweight framework for reinforcement learning-based adaptive job scheduling. +# AdaptiveQueue + +A lightweight Python project for experimenting with adaptive background job +scheduling. The first milestone provides a discrete-time, single-worker simulator +and FIFO, SPT, EDF, Priority, and seeded Random baselines. + +Requires Python 3.10 or newer. There are no runtime dependencies. + +```sh +python -m venv .venv +source .venv/bin/activate +python -m pip install -e '.[dev]' +``` + +Run an episode with 60 jobs: + +```sh +adaptivequeue --workload mixed --policy edf --seed 42 +``` + +The CLI reports priority-weighted scheduling cost (lower is better), deadline miss +rate, average and 95th-percentile waiting time, and average flow time. Workloads +are `stable`, `bursty`, or `mixed`; policies are `fifo`, `spt`, `edf`, `priority`, +or `random`. Use the same seed and workload settings to compare policies fairly. + +```sh +python -m adaptivequeue --workload bursty --policy spt --seed 7 \ + --jobs 60 --arrival-rate 0.3 --deadline-factor 3 +python -m pytest +ruff check . +ruff format --check . +mypy +``` + +See [simulation rules and parameters](docs/simulation.md) or run +`adaptivequeue --help`. The Python API also exposes per-job execution records. +Reinforcement learning is planned for a later milestone; the current scope is +simulation and conventional scheduling baselines. diff --git a/docs/simulation.md b/docs/simulation.md new file mode 100644 index 0000000..9c424ff --- /dev/null +++ b/docs/simulation.md @@ -0,0 +1,88 @@ +# Simulation + +## Rules + +Time starts at zero and uses integer units. A job specifies a unique string +`job_id`, nonnegative `arrival_time`, positive `duration`, nonnegative absolute +`deadline`, positive `priority_weight`, and descriptive `job_type`. An already +unachievable deadline is allowed. Specifications are immutable, so one workload +can be reused across policies. + +One worker runs one job to completion without preemption. Whenever it is idle, +all jobs with arrival time at or before the current time enter the waiting queue. +The engine calls `policy(waiting_jobs, current_time)` with an immutable snapshot +of that queue and the current integer simulation time. The policy must select a +waiting job and cannot observe future arrivals. Jobs arriving during execution +wait until completion; arrivals exactly at completion participate in the next +decision. The engine skips to completion or the next arrival rather than stepping +through ticks where no scheduling decision is possible. It runs until every job +finishes, including late jobs, without deliberately idling when work is available. + +## Policies + +| CLI name | Selects | +| --- | --- | +| `fifo` | Earliest arrival | +| `spt` | Shortest processing duration | +| `edf` | Earliest absolute deadline | +| `priority` | Highest priority weight | +| `random` | Uniformly random waiting job, using a local seeded RNG | + +These baselines accept `current_time` but do not use it in their selection rules. +The first four break ties by earliest arrival, then lexicographically smallest +job ID. Random's queue is populated in arrival/ID order, making seeded results +independent of input ordering. Recreate a Random policy with the same seed for +each replay. Workload generation and policy selection use separate RNGs and do +not modify Python's global random state. + +## Workloads + +Every generator takes an explicit seed and an optional `WorkloadConfig`. + +| Parameter | Default | Meaning | +| --- | --- | --- | +| `num_jobs` / `--jobs` | 60 | Jobs in an episode; zero gives an empty episode | +| `arrival_rate` / `--arrival-rate` | 0.28 | Target mean jobs per time unit | +| `deadline_factor` / `--deadline-factor` | 4.0 | Relative deadline budget; larger is looser | +| `duration_range` (Python API) | (1, 6) | Inclusive integer duration bounds | +| `burst_size` / `--burst-size` | 6 | Jobs per burst, with a smaller final burst if needed | +| `burst_window` / `--burst-window` | 3 | Burst offsets range from 0 to window minus 1 | + +`stable` uses independent exponential interarrival gaps with a constant rate and +the first arrival at zero. Cumulative arrival times are rounded down to integers, +so several jobs can arrive together. `bursty` places batches within short windows; +burst starts have exponential gaps with mean `burst_size / arrival_rate`. Bursts +can overlap. `mixed` uses stable arrivals with interactive jobs (weight 4, deadline +multiplier 0.5), reports (weight 2, multiplier 1), and maintenance jobs (weight 1, +multiplier 2), chosen with equal probability. + +Durations are uniform within the configured range. Stable and bursty workloads +choose weights uniformly from 1, 2, and 4, and descriptive types from email, +image, and backup. All deadline budgets are +`max(1, ceil(duration * deadline_factor * type_multiplier * U(0.75, 1.25)))`; +the absolute deadline is arrival plus budget. Stable and bursty type multipliers +are 1. Increasing arrival rate increases contention. The default rate gives +nominal utilization `0.28 * 3.5 = 0.98`; finite episodes vary by seed. + +## Records and evaluation + +`simulate(jobs, policy)` returns `SimulationResult`. Its `jobs` tuple holds +`JobResult` records in execution order, including each job, start time, completion +time, and derived waiting time, flow time, and tardiness: + +- Waiting time = start minus arrival. +- Flow time = completion minus arrival. +- Tardiness = max(0, completion minus absolute deadline). +- Primary cost `J = mean(priority_weight * (tardiness + 0.1 * flow_time))`. +- Deadline miss rate = fraction of jobs with positive tardiness. Completion on + the deadline counts as on time. + +Lower primary cost is better. The result also reports average waiting time, +95th-percentile waiting time, and average flow time. The percentile linearly +interpolates sorted waiting times at index `0.95 * (n - 1)` (zero-based), and +equals the lone value for one job. Every aggregate metric is zero for an empty +episode. Cost is averaged over jobs, without normalizing by total priority weight. + +Compare policies on identical generated jobs and settings. Seeds control both +workloads and Random decisions independently; several episode seeds give a more +useful comparison than a single run. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..9d3b9a2 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,36 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "adaptivequeue" +version = "0.1.0" +description = "A discrete-time job scheduling simulator and baseline policies" +readme = "README.md" +requires-python = ">=3.10" +license = {file = "LICENSE"} +dependencies = [] + +[project.optional-dependencies] +dev = ["pytest>=8", "ruff>=0.9", "mypy>=1.14"] + +[project.scripts] +adaptivequeue = "adaptivequeue.cli:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] + +[tool.ruff] +target-version = "py310" +line-length = 100 + +[tool.ruff.lint] +select = ["E", "F", "I"] + +[tool.mypy] +files = ["src/adaptivequeue"] +python_version = "3.10" +strict = true diff --git a/src/adaptivequeue/__init__.py b/src/adaptivequeue/__init__.py new file mode 100644 index 0000000..0b735f9 --- /dev/null +++ b/src/adaptivequeue/__init__.py @@ -0,0 +1,20 @@ +"""Discrete-time simulation and conventional job scheduling baselines.""" + +from adaptivequeue.metrics import Metrics +from adaptivequeue.models import Job, JobResult +from adaptivequeue.policies import make_policy +from adaptivequeue.simulation import SimulationResult, simulate +from adaptivequeue.workloads import WorkloadConfig, bursty_arrivals, mixed_workload, stable_arrivals + +__all__ = [ + "Job", + "JobResult", + "Metrics", + "SimulationResult", + "WorkloadConfig", + "bursty_arrivals", + "make_policy", + "mixed_workload", + "simulate", + "stable_arrivals", +] diff --git a/src/adaptivequeue/__main__.py b/src/adaptivequeue/__main__.py new file mode 100644 index 0000000..841ab14 --- /dev/null +++ b/src/adaptivequeue/__main__.py @@ -0,0 +1,3 @@ +from adaptivequeue.cli import main + +main() diff --git a/src/adaptivequeue/cli.py b/src/adaptivequeue/cli.py new file mode 100644 index 0000000..d94a6f9 --- /dev/null +++ b/src/adaptivequeue/cli.py @@ -0,0 +1,47 @@ +"""Command-line entry point for one reproducible scheduling episode.""" + +import argparse + +from adaptivequeue.policies import POLICY_NAMES, make_policy +from adaptivequeue.simulation import simulate +from adaptivequeue.workloads import WORKLOADS, WorkloadConfig + + +def main() -> None: + parser = argparse.ArgumentParser(description="Run an AdaptiveQueue scheduling episode.") + parser.add_argument("--workload", choices=WORKLOADS, default="stable") + parser.add_argument("--policy", choices=POLICY_NAMES, default="fifo") + parser.add_argument("--seed", type=int, default=42) + parser.add_argument("--jobs", type=int, default=60) + parser.add_argument("--arrival-rate", type=float, default=0.28, help="Mean jobs per time unit") + parser.add_argument( + "--deadline-factor", type=float, default=4.0, help="Larger means looser deadlines" + ) + parser.add_argument("--burst-size", type=int, default=6) + parser.add_argument( + "--burst-window", type=int, default=3, help="Burst arrival window in time units" + ) + args = parser.parse_args() + + try: + config = WorkloadConfig( + num_jobs=args.jobs, + arrival_rate=args.arrival_rate, + deadline_factor=args.deadline_factor, + burst_size=args.burst_size, + burst_window=args.burst_window, + ) + except ValueError as error: + parser.error(str(error)) + + jobs = WORKLOADS[args.workload](args.seed, config) + result = simulate(jobs, make_policy(args.policy, seed=args.seed)) + metrics = result.metrics + print( + f"Workload: {args.workload} | Policy: {args.policy} | Seed: {args.seed} | Jobs: {len(jobs)}" + ) + print(f"Primary cost J (lower is better): {metrics.primary_cost:.3f}") + print(f"Deadline miss rate: {metrics.deadline_miss_rate:.1%}") + print(f"Average waiting time: {metrics.average_waiting_time:.3f}") + print(f"95th-percentile waiting time: {metrics.p95_waiting_time:.3f}") + print(f"Average flow time: {metrics.average_flow_time:.3f}") diff --git a/src/adaptivequeue/metrics.py b/src/adaptivequeue/metrics.py new file mode 100644 index 0000000..f87e39b --- /dev/null +++ b/src/adaptivequeue/metrics.py @@ -0,0 +1,37 @@ +"""Episode metrics; the waiting-time percentile uses linear interpolation.""" + +from dataclasses import dataclass +from math import ceil, floor +from statistics import fmean +from typing import Sequence + +from adaptivequeue.models import JobResult + + +@dataclass(frozen=True) +class Metrics: + primary_cost: float + deadline_miss_rate: float + average_waiting_time: float + p95_waiting_time: float + average_flow_time: float + + +def calculate_metrics(results: Sequence[JobResult]) -> Metrics: + if not results: + return Metrics(0.0, 0.0, 0.0, 0.0, 0.0) + + waiting = sorted(result.waiting_time for result in results) + position = 0.95 * (len(waiting) - 1) + lower, upper = floor(position), ceil(position) + p95 = waiting[lower] + (waiting[upper] - waiting[lower]) * (position - lower) + return Metrics( + primary_cost=fmean( + result.job.priority_weight * (result.tardiness + 0.1 * result.flow_time) + for result in results + ), + deadline_miss_rate=fmean(result.tardiness > 0 for result in results), + average_waiting_time=fmean(waiting), + p95_waiting_time=p95, + average_flow_time=fmean(result.flow_time for result in results), + ) diff --git a/src/adaptivequeue/models.py b/src/adaptivequeue/models.py new file mode 100644 index 0000000..eb5b577 --- /dev/null +++ b/src/adaptivequeue/models.py @@ -0,0 +1,45 @@ +"""Immutable job specifications and execution records.""" + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class Job: + job_id: str + arrival_time: int + duration: int + deadline: int + priority_weight: int + job_type: str + + def __post_init__(self) -> None: + if not self.job_id or not self.job_type: + raise ValueError("job_id and job_type must be nonempty") + if any( + not isinstance(value, int) + for value in (self.arrival_time, self.duration, self.deadline, self.priority_weight) + ): + raise ValueError("job times, duration, and priority_weight must be integers") + if self.arrival_time < 0 or self.deadline < 0: + raise ValueError("arrival_time and deadline must be nonnegative") + if self.duration <= 0 or self.priority_weight <= 0: + raise ValueError("duration and priority_weight must be positive") + + +@dataclass(frozen=True) +class JobResult: + job: Job + start_time: int + completion_time: int + + @property + def waiting_time(self) -> int: + return self.start_time - self.job.arrival_time + + @property + def flow_time(self) -> int: + return self.completion_time - self.job.arrival_time + + @property + def tardiness(self) -> int: + return max(0, self.completion_time - self.job.deadline) diff --git a/src/adaptivequeue/policies.py b/src/adaptivequeue/policies.py new file mode 100644 index 0000000..eb83e07 --- /dev/null +++ b/src/adaptivequeue/policies.py @@ -0,0 +1,43 @@ +"""Policies receive the waiting jobs and current simulation time.""" + +import random +from typing import Callable, Sequence + +from adaptivequeue.models import Job + +Policy = Callable[[Sequence[Job], int], Job] + + +def fifo(waiting: Sequence[Job], current_time: int) -> Job: + return min(waiting, key=lambda job: (job.arrival_time, job.job_id)) + + +def spt(waiting: Sequence[Job], current_time: int) -> Job: + return min(waiting, key=lambda job: (job.duration, job.arrival_time, job.job_id)) + + +def edf(waiting: Sequence[Job], current_time: int) -> Job: + return min(waiting, key=lambda job: (job.deadline, job.arrival_time, job.job_id)) + + +def priority(waiting: Sequence[Job], current_time: int) -> Job: + return min(waiting, key=lambda job: (-job.priority_weight, job.arrival_time, job.job_id)) + + +POLICIES: dict[str, Policy] = {"fifo": fifo, "spt": spt, "edf": edf, "priority": priority} +POLICY_NAMES = (*POLICIES, "random") + + +def make_policy(name: str, *, seed: int) -> Policy: + """Create a policy; recreate it with the same seed to replay a random run.""" + if name == "random": + rng = random.Random(seed) + + def random_policy(waiting: Sequence[Job], current_time: int) -> Job: + return rng.choice(waiting) + + return random_policy + try: + return POLICIES[name] + except KeyError: + raise ValueError(f"unknown policy: {name}") from None diff --git a/src/adaptivequeue/simulation.py b/src/adaptivequeue/simulation.py new file mode 100644 index 0000000..f373489 --- /dev/null +++ b/src/adaptivequeue/simulation.py @@ -0,0 +1,44 @@ +"""A single-worker, non-preemptive simulation on integer time boundaries.""" + +from dataclasses import dataclass +from typing import Iterable + +from adaptivequeue.metrics import Metrics, calculate_metrics +from adaptivequeue.models import Job, JobResult +from adaptivequeue.policies import Policy + + +@dataclass(frozen=True) +class SimulationResult: + jobs: tuple[JobResult, ...] + metrics: Metrics + + +def simulate(jobs: Iterable[Job], policy: Policy) -> SimulationResult: + """Run until every job finishes, returning records in execution order.""" + arrivals = sorted(jobs, key=lambda job: (job.arrival_time, job.job_id)) + if len({job.job_id for job in arrivals}) != len(arrivals): + raise ValueError("job IDs must be unique within an episode") + + waiting: list[Job] = [] + completed: list[JobResult] = [] + time = 0 + next_arrival = 0 + + while waiting or next_arrival < len(arrivals): + if not waiting: + time = max(time, arrivals[next_arrival].arrival_time) + while next_arrival < len(arrivals) and arrivals[next_arrival].arrival_time <= time: + waiting.append(arrivals[next_arrival]) + next_arrival += 1 + + selected = policy(tuple(waiting), time) + if selected not in waiting: + raise ValueError("policy must select a job from the waiting queue") + waiting.remove(selected) + completion_time = time + selected.duration + completed.append(JobResult(selected, time, completion_time)) + time = completion_time + + records = tuple(completed) + return SimulationResult(records, calculate_metrics(records)) diff --git a/src/adaptivequeue/workloads.py b/src/adaptivequeue/workloads.py new file mode 100644 index 0000000..c0b421d --- /dev/null +++ b/src/adaptivequeue/workloads.py @@ -0,0 +1,90 @@ +"""Seeded synthetic workloads with configurable load and deadline budgets.""" + +import random +from dataclasses import dataclass +from math import ceil, isfinite +from typing import Callable + +from adaptivequeue.models import Job + + +@dataclass(frozen=True) +class WorkloadConfig: + num_jobs: int = 60 + arrival_rate: float = 0.28 + deadline_factor: float = 4.0 + duration_range: tuple[int, int] = (1, 6) + burst_size: int = 6 + burst_window: int = 3 + + def __post_init__(self) -> None: + if self.num_jobs < 0: + raise ValueError("num_jobs must be nonnegative") + if not isfinite(self.arrival_rate) or self.arrival_rate <= 0: + raise ValueError("arrival_rate must be finite and positive") + if not isfinite(self.deadline_factor) or self.deadline_factor <= 0: + raise ValueError("deadline_factor must be finite and positive") + low, high = self.duration_range + if low < 1 or high < low: + raise ValueError("duration_range must satisfy 1 <= minimum <= maximum") + if self.burst_size < 1 or self.burst_window < 1: + raise ValueError("burst_size and burst_window must be positive") + + +def _steady_arrivals(rng: random.Random, config: WorkloadConfig) -> list[int]: + time = 0.0 + arrivals: list[int] = [] + for _ in range(config.num_jobs): + arrivals.append(int(time)) + time += rng.expovariate(config.arrival_rate) + return arrivals + + +def _make_jobs( + arrivals: list[int], rng: random.Random, config: WorkloadConfig, *, mixed: bool = False +) -> list[Job]: + jobs: list[Job] = [] + for index, arrival in enumerate(sorted(arrivals)): + duration = rng.randint(*config.duration_range) + if mixed: + job_type, weight, factor = rng.choice( + (("interactive", 4, 0.5), ("report", 2, 1.0), ("maintenance", 1, 2.0)) + ) + else: + job_type = rng.choice(("email", "image", "backup")) + weight = rng.choice((1, 2, 4)) + factor = 1.0 + budget = max(1, ceil(duration * config.deadline_factor * factor * rng.uniform(0.75, 1.25))) + jobs.append(Job(f"job-{index:03d}", arrival, duration, arrival + budget, weight, job_type)) + return jobs + + +def stable_arrivals(seed: int, config: WorkloadConfig = WorkloadConfig()) -> list[Job]: + """Constant-rate arrivals, rounded down to integer time units.""" + rng = random.Random(seed) + return _make_jobs(_steady_arrivals(rng, config), rng, config) + + +def bursty_arrivals(seed: int, config: WorkloadConfig = WorkloadConfig()) -> list[Job]: + """Batches arriving within short windows, separated by random gaps.""" + rng = random.Random(seed) + time = 0.0 + arrivals: list[int] = [] + for offset in range(0, config.num_jobs, config.burst_size): + size = min(config.burst_size, config.num_jobs - offset) + arrivals.extend(int(time) + rng.randrange(config.burst_window) for _ in range(size)) + time += rng.expovariate(config.arrival_rate / config.burst_size) + return _make_jobs(arrivals, rng, config) + + +def mixed_workload(seed: int, config: WorkloadConfig = WorkloadConfig()) -> list[Job]: + """Constant-rate arrivals with tight/high-weight and loose/low-weight jobs.""" + rng = random.Random(seed) + return _make_jobs(_steady_arrivals(rng, config), rng, config, mixed=True) + + +WORKLOADS: dict[str, Callable[[int, WorkloadConfig], list[Job]]] = { + "stable": stable_arrivals, + "bursty": bursty_arrivals, + "mixed": mixed_workload, +} diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..e3a19b5 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,43 @@ +import subprocess +import sys + + +def test_module_cli_displays_metrics() -> None: + result = subprocess.run( + [ + sys.executable, + "-m", + "adaptivequeue", + "--workload", + "mixed", + "--policy", + "edf", + "--seed", + "7", + "--jobs", + "12", + ], + check=True, + capture_output=True, + text=True, + ) + assert "Workload: mixed | Policy: edf | Seed: 7 | Jobs: 12" in result.stdout + for label in ( + "Primary cost J", + "Deadline miss rate", + "Average waiting time", + "95th-percentile waiting time", + "Average flow time", + ): + assert label in result.stdout + + +def test_invalid_cli_configuration_has_a_useful_error() -> None: + result = subprocess.run( + [sys.executable, "-m", "adaptivequeue", "--arrival-rate", "0"], + capture_output=True, + text=True, + ) + assert result.returncode == 2 + assert "arrival_rate must be finite and positive" in result.stderr + assert "Traceback" not in result.stderr diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..8601c81 --- /dev/null +++ b/tests/test_metrics.py @@ -0,0 +1,44 @@ +import pytest + +from adaptivequeue import Job, make_policy, simulate + + +def test_job_times_and_metrics_match_hand_calculation() -> None: + jobs = [ + Job("a", 0, 3, 3, 1, "test"), + Job("b", 1, 2, 4, 4, "test"), + Job("c", 5, 1, 5, 2, "test"), + ] + result = simulate(jobs, make_policy("fifo", seed=0)) + assert [(r.start_time, r.completion_time) for r in result.jobs] == [(0, 3), (3, 5), (5, 6)] + assert [r.waiting_time for r in result.jobs] == [0, 2, 0] + assert [r.flow_time for r in result.jobs] == [3, 4, 1] + assert [r.tardiness for r in result.jobs] == [0, 1, 1] + + metrics = result.metrics + # Per-job costs: 0.3, 4 * (1 + 0.4), 2 * (1 + 0.1). + assert metrics.primary_cost == pytest.approx(2.7) + assert metrics.deadline_miss_rate == pytest.approx(2 / 3) + assert metrics.average_waiting_time == pytest.approx(2 / 3) + assert metrics.p95_waiting_time == pytest.approx(1.8) + assert metrics.average_flow_time == pytest.approx(8 / 3) + + +def test_empty_episode_has_zero_metrics_and_never_calls_policy() -> None: + def fail_if_called(waiting: object, current_time: int) -> Job: + pytest.fail("an empty episode must not request a scheduling decision") + + result = simulate([], fail_if_called) + assert result.jobs == () + assert result.metrics.primary_cost == 0 + assert result.metrics.deadline_miss_rate == 0 + assert result.metrics.average_waiting_time == 0 + assert result.metrics.p95_waiting_time == 0 + assert result.metrics.average_flow_time == 0 + + +def test_single_job_finishing_on_deadline_is_not_a_miss() -> None: + result = simulate([Job("a", 10, 2, 12, 4, "test")], make_policy("fifo", seed=0)) + assert result.metrics.deadline_miss_rate == 0 + assert result.metrics.p95_waiting_time == 0 + assert result.metrics.primary_cost == pytest.approx(0.8) diff --git a/tests/test_policies.py b/tests/test_policies.py new file mode 100644 index 0000000..d31d2ec --- /dev/null +++ b/tests/test_policies.py @@ -0,0 +1,51 @@ +import pytest + +from adaptivequeue import Job, make_policy, simulate +from adaptivequeue.policies import POLICIES + + +@pytest.mark.parametrize( + ("name", "expected"), + [ + ("fifo", ["a", "b", "c", "d"]), + ("spt", ["c", "b", "d", "a"]), + ("edf", ["c", "d", "a", "b"]), + ("priority", ["b", "d", "c", "a"]), + ], +) +def test_scheduling_order(name: str, expected: list[str]) -> None: + jobs = [ + Job("d", 0, 2, 4, 4, "test"), + Job("b", 0, 2, 9, 4, "test"), + Job("a", 0, 4, 7, 1, "test"), + Job("c", 0, 1, 3, 2, "test"), + ] + result = simulate(jobs, make_policy(name, seed=1)) + assert [record.job.job_id for record in result.jobs] == expected + + +@pytest.mark.parametrize("name", POLICIES) +def test_ties_use_arrival_then_job_id(name: str) -> None: + jobs = [ + Job("b", 1, 2, 9, 2, "test"), + Job("a", 1, 2, 9, 2, "test"), + Job("z", 0, 2, 9, 2, "test"), + ] + policy = make_policy(name, seed=1) + assert policy(jobs, 1).job_id == "z" + assert policy(jobs[:2], 1).job_id == "a" + + +def test_random_policy_replays_seeded_choices() -> None: + jobs = [Job(str(i), 0, 1, 10, 1, "test") for i in range(8)] + first = simulate(jobs, make_policy("random", seed=42)) + replay = simulate(reversed(jobs), make_policy("random", seed=42)) + other = simulate(jobs, make_policy("random", seed=7)) + assert first == replay + assert [record.job.job_id for record in first.jobs] == ["1", "0", "7", "4", "3", "2", "5", "6"] + assert first.jobs != other.jobs + + +def test_unknown_policy_is_rejected() -> None: + with pytest.raises(ValueError, match="unknown policy"): + make_policy("missing", seed=0) diff --git a/tests/test_simulation.py b/tests/test_simulation.py new file mode 100644 index 0000000..d277f18 --- /dev/null +++ b/tests/test_simulation.py @@ -0,0 +1,71 @@ +from typing import Sequence + +import pytest + +from adaptivequeue import Job, make_policy, simulate +from adaptivequeue.policies import spt + + +def test_arrivals_do_not_preempt_and_idle_time_is_skipped() -> None: + jobs = [ + Job("late", 20, 1, 25, 1, "test"), + Job("boundary", 8, 2, 10, 2, "test"), + Job("short", 3, 1, 4, 4, "test"), + Job("long", 2, 6, 20, 1, "test"), + ] + result = simulate(jobs, make_policy("spt", seed=0)) + assert [(r.job.job_id, r.start_time, r.completion_time) for r in result.jobs] == [ + ("long", 2, 8), + ("short", 8, 9), + ("boundary", 9, 11), + ("late", 20, 21), + ] + assert len(result.jobs) == len(jobs) + assert all(r.completion_time - r.start_time == r.job.duration for r in result.jobs) + + +def test_policy_sees_available_jobs_and_current_time_at_each_decision() -> None: + jobs = [ + Job("running", 0, 5, 20, 1, "test"), + Job("waiting", 1, 4, 20, 1, "test"), + Job("at-completion", 5, 1, 20, 1, "test"), + Job("future", 30, 1, 40, 1, "test"), + ] + observations = [] + + def observe(waiting: Sequence[Job], current_time: int) -> Job: + assert isinstance(waiting, tuple) + observations.append((current_time, tuple(job.job_id for job in waiting))) + return spt(waiting, current_time) + + result = simulate(jobs, observe) + assert observations == [ + (0, ("running",)), + (5, ("waiting", "at-completion")), + (6, ("waiting",)), + (30, ("future",)), + ] + assert [r.start_time for r in result.jobs] == [0, 5, 6, 30] + + +def test_duplicate_ids_are_rejected() -> None: + job = Job("duplicate", 0, 1, 5, 1, "test") + with pytest.raises(ValueError, match="unique"): + simulate([job, job], make_policy("fifo", seed=0)) + + +def test_policy_cannot_select_a_future_job() -> None: + jobs = [Job("available", 0, 2, 10, 1, "test"), Job("future", 3, 1, 10, 1, "test")] + with pytest.raises(ValueError, match="waiting queue"): + simulate(jobs, lambda waiting, current_time: jobs[1]) + + +@pytest.mark.parametrize( + ("arrival", "duration", "deadline", "weight"), + [(-1, 1, 5, 1), (0, 0, 5, 1), (0, 1, -1, 1), (0, 1, 5, 0), (0.5, 1, 5, 1)], +) +def test_invalid_job_values_are_rejected( + arrival: int, duration: int, deadline: int, weight: int +) -> None: + with pytest.raises(ValueError): + Job("invalid", arrival, duration, deadline, weight, "test") diff --git a/tests/test_workloads.py b/tests/test_workloads.py new file mode 100644 index 0000000..a19ddd4 --- /dev/null +++ b/tests/test_workloads.py @@ -0,0 +1,69 @@ +import random +from dataclasses import replace + +import pytest + +from adaptivequeue import WorkloadConfig, make_policy, simulate +from adaptivequeue.workloads import WORKLOADS + + +@pytest.mark.parametrize("name", WORKLOADS) +def test_workload_and_episode_are_reproducible_without_global_rng_changes(name: str) -> None: + generate = WORKLOADS[name] + config = WorkloadConfig() + global_state = random.getstate() + jobs = generate(42, config) + assert jobs == generate(42, config) + assert jobs != generate(7, config) + assert simulate(jobs, make_policy("random", seed=42)) == simulate( + generate(42, config), make_policy("random", seed=42) + ) + assert random.getstate() == global_state + assert len(jobs) == len({job.job_id for job in jobs}) == 60 + assert all(1 <= job.duration <= 6 and job.priority_weight in (1, 2, 4) for job in jobs) + assert all(job.deadline > job.arrival_time and job.job_type for job in jobs) + + +@pytest.mark.parametrize("name", WORKLOADS) +def test_workload_configuration_controls_size_load_and_deadlines(name: str) -> None: + generate = WORKLOADS[name] + config = WorkloadConfig(num_jobs=13, duration_range=(2, 2), burst_size=4) + jobs = generate(42, config) + faster = generate(42, replace(config, arrival_rate=1.0)) + looser = generate(42, replace(config, deadline_factor=8.0)) + assert len(jobs) == 13 + assert all(job.duration == 2 for job in jobs) + assert max(job.arrival_time for job in faster) < max(job.arrival_time for job in jobs) + assert [job.arrival_time for job in jobs] == [job.arrival_time for job in looser] + assert all(a.deadline < b.deadline for a, b in zip(jobs, looser)) + assert generate(42, WorkloadConfig(num_jobs=0)) == [] + + +def test_burst_windows_and_mixed_job_classes() -> None: + config = WorkloadConfig(num_jobs=6, burst_size=6, burst_window=3) + burst = WORKLOADS["bursty"](42, config) + assert all(0 <= job.arrival_time < 3 for job in burst) + mixed = WORKLOADS["mixed"](42, WorkloadConfig()) + assert {(job.job_type, job.priority_weight) for job in mixed} == { + ("interactive", 4), + ("report", 2), + ("maintenance", 1), + } + + +@pytest.mark.parametrize( + "changes", + [ + {"num_jobs": -1}, + {"arrival_rate": 0}, + {"arrival_rate": float("nan")}, + {"deadline_factor": -1}, + {"deadline_factor": float("inf")}, + {"duration_range": (3, 2)}, + {"burst_size": 0}, + {"burst_window": 0}, + ], +) +def test_invalid_configuration_is_rejected(changes: dict) -> None: + with pytest.raises(ValueError): + WorkloadConfig(**changes)