Add discrete-time simulation engine and classical scheduling baselines

.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

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.

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.

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

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",
+]

src/adaptivequeue/__main__.py

@@ -0,0 +1,3 @@
+from adaptivequeue.cli import main
+
+main()

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}")

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),
+    )

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)

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