Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 16 additions & 4 deletions README.md
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
# 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.
scheduling. Includes a discrete-time, single-worker simulator, FIFO, SPT, EDF,
Priority, and seeded Random baselines, and a tabular Q-Learning meta-scheduler.

Requires Python 3.10 or newer. There are no runtime dependencies.

Expand All @@ -23,6 +23,18 @@ 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.

Train a meta-scheduler that chooses FIFO, SPT, EDF, or Priority at each decision,
then evaluate its frozen Q-table on held-out workload seeds:

```sh
adaptivequeue train --workload mixed --episodes 1000 --seed 42 --output policy.json
adaptivequeue evaluate --workload mixed --episodes 20 --model policy.json
```

Evaluation defaults to seeds following the saved training range, with exploration
disabled. Use `--seed` to choose a different disjoint range. Both commands accept
the workload options below; `train --help` lists learning and exploration settings.

```sh
python -m adaptivequeue --workload bursty --policy spt --seed 7 \
--jobs 60 --arrival-rate 0.3 --deadline-factor 3
Expand All @@ -34,5 +46,5 @@ 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.
The current scope is simulation, classical baselines, and CPU-only tabular
Q-Learning; no additional runtime dependencies are needed.
58 changes: 58 additions & 0 deletions docs/simulation.md
Original file line number Diff line number Diff line change
Expand Up @@ -86,3 +86,61 @@ episode. Cost is averaged over jobs, without normalizing by total priority weigh
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.

## Tabular Q-Learning meta-scheduler

`Simulation(jobs)` exposes `waiting_jobs`, `current_time`, and `done` at each
decision. Its `step(policy)` executes exactly one job, returns its `JobResult`,
and advances to the next ready queue (or termination). `simulate()` uses this
same engine. Only the engine holds future arrivals. `qlearning.step_action`
accepts an action index and returns the completed job, reward, next encoded state,
and `done`. Terminal states are represented by `None`, including empty episodes.

The state tuple has 96 possible values: three zero-based bucket indices and a
boolean, computed only from currently waiting jobs and the current time:

| Feature | Buckets in order |
| --- | --- |
| Queue length | 1; 2–3; 4–7; 8+ |
| Minimum slack | overdue: <0; urgent: 0–2; moderate: 3–6; relaxed: 7+ |
| Short-job proportion | low: <1/3; medium: 1/3 to <2/3; high: >=2/3 |
| High-priority job waiting | true if any weight is >=4, otherwise false |

A short job has duration <=2. Each job's slack is
`deadline - current_time - duration`: its remaining deadline budget after allowing
for processing. Waiting jobs have not started, so their remaining processing time
is their full duration. Negative slack means even immediate execution would miss
the deadline; it does not require the absolute deadline to have passed.

Actions 0, 1, 2, and 3 invoke FIFO, SPT, EDF, and Priority respectively, preserving
their tie-breaking rules. The chosen baseline selects the next job. Reward is
`-priority_weight * (tardiness + 0.1 * flow_time)` for that completed job. The
episode return is `-n * J`; for a fixed job count, maximizing undiscounted return
minimizes the existing primary cost.

The update is `Q(s,a) += alpha * (reward + gamma * max_a' Q(s',a') - Q(s,a))`,
omitting the bootstrap term for terminal transitions. Defaults are alpha=0.1,
gamma=1.0, and epsilon-greedy exploration with
`epsilon(episode) = max(0.05, 1.0 * 0.995**episode)` (episode numbering starts at
zero). Discounting, when configured, applies per scheduled job, not per time unit.
Q-values start at zero. Greedy ties choose the lowest action index, so unseen
states fall back to FIFO. Exploration chooses uniformly among all four actions.

`qlearning.train()` generates workload seeds `seed` through `seed + episodes - 1`;
action selection uses a separate local RNG. RL seeds must be nonnegative.
`evaluate()` disables exploration and all Q-table updates, including insertion of
unseen states. It defaults to the next workload seed after training and rejects
overlapping training/evaluation seed ranges. All five metrics are computed over
pooled jobs from evaluation episodes; p95 is not an average of episode percentiles.
For classical comparisons, run baselines on those same evaluation seeds and settings.

`QLearningAgent.save()` writes readable, versioned JSON containing action order,
state thresholds, reward definition, learning parameters, training workload and
seed range, and Q-values. `load()` preserves greedy predictions and rejects
incompatible formulations. The format supports evaluation, not exact continuation
of an interrupted training RNG stream. A loaded agent's `policy` method also works
with `simulate(jobs, agent.policy)`.

The coarse state aliases different queues and omits future arrivals, so it is not
a full Markov description. Training does not guarantee improvement over a baseline;
evaluate on several held-out seeds and retrain when workload characteristics change.
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "adaptivequeue"
version = "0.1.0"
description = "A discrete-time job scheduling simulator and baseline policies"
description = "A job scheduling simulator with classical and tabular Q-Learning policies"
readme = "README.md"
requires-python = ">=3.10"
license = {file = "LICENSE"}
Expand Down
5 changes: 3 additions & 2 deletions src/adaptivequeue/__init__.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,16 @@
"""Discrete-time simulation and conventional job scheduling baselines."""
"""Discrete-time simulation, scheduling baselines, and a tabular meta-scheduler."""

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.simulation import Simulation, SimulationResult, simulate
from adaptivequeue.workloads import WorkloadConfig, bursty_arrivals, mixed_workload, stable_arrivals

__all__ = [
"Job",
"JobResult",
"Metrics",
"Simulation",
"SimulationResult",
"WorkloadConfig",
"bursty_arrivals",
Expand Down
98 changes: 79 additions & 19 deletions src/adaptivequeue/cli.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,20 @@
"""Command-line entry point for one reproducible scheduling episode."""
"""Baseline episodes and tabular Q-Learning training/evaluation."""

import argparse
from pathlib import Path

from adaptivequeue.metrics import Metrics
from adaptivequeue.policies import POLICY_NAMES, make_policy
from adaptivequeue.qlearning import LearningConfig, QLearningAgent, evaluate, train
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)
def _add_workload_arguments(
parser: argparse.ArgumentParser, *, workload: str = "stable", seed: int | None = 42
) -> None:
parser.add_argument("--workload", choices=WORKLOADS, default=workload)
parser.add_argument("--seed", type=int, default=seed)
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(
Expand All @@ -21,6 +24,36 @@ def main() -> None:
parser.add_argument(
"--burst-window", type=int, default=3, help="Burst arrival window in time units"
)


def _print_metrics(metrics: Metrics) -> None:
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}")


def main() -> None:
parser = argparse.ArgumentParser(description="Run AdaptiveQueue scheduling experiments.")
_add_workload_arguments(parser)
parser.add_argument("--policy", choices=POLICY_NAMES, default="fifo")
commands = parser.add_subparsers(dest="command")

training = commands.add_parser("train", help="Train and save a Q-Learning meta-scheduler")
_add_workload_arguments(training, workload="mixed")
training.add_argument("--episodes", type=int, default=1000)
training.add_argument("--learning-rate", type=float, default=0.1)
training.add_argument("--gamma", type=float, default=1.0)
training.add_argument("--epsilon-start", type=float, default=1.0)
training.add_argument("--epsilon-end", type=float, default=0.05)
training.add_argument("--epsilon-decay", type=float, default=0.995)
training.add_argument("--output", type=Path, required=True)

evaluation = commands.add_parser("evaluate", help="Evaluate a frozen Q-table on held-out seeds")
_add_workload_arguments(evaluation, workload="mixed", seed=None)
evaluation.add_argument("--episodes", type=int, default=20)
evaluation.add_argument("--model", type=Path, required=True)
args = parser.parse_args()

try:
Expand All @@ -31,17 +64,44 @@ def main() -> None:
burst_size=args.burst_size,
burst_window=args.burst_window,
)
except ValueError as error:
if args.command == "train":
learning = LearningConfig(
args.learning_rate,
args.gamma,
args.epsilon_start,
args.epsilon_end,
args.epsilon_decay,
)
agent = train(
workload=args.workload,
episodes=args.episodes,
seed=args.seed,
workload_config=config,
learning_config=learning,
)
agent.save(args.output)
print(
f"Trained {args.episodes} episodes; "
f"{len(agent.q_table)} states; saved {args.output}"
)
elif args.command == "evaluate":
agent = QLearningAgent.load(args.model)
metrics = evaluate(
agent,
workload=args.workload,
episodes=args.episodes,
seed=args.seed,
workload_config=config,
)
print(f"Workload: {args.workload} | Policy: qlearning | Episodes: {args.episodes}")
_print_metrics(metrics)
else:
jobs = WORKLOADS[args.workload](args.seed, config)
result = simulate(jobs, make_policy(args.policy, seed=args.seed))
print(
f"Workload: {args.workload} | Policy: {args.policy} | "
f"Seed: {args.seed} | Jobs: {len(jobs)}"
)
_print_metrics(result.metrics)
except (OSError, ValueError, KeyError, TypeError) 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}")
5 changes: 1 addition & 4 deletions src/adaptivequeue/metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,10 +26,7 @@ def calculate_metrics(results: Sequence[JobResult]) -> Metrics:
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
),
primary_cost=fmean(result.cost for result in results),
deadline_miss_rate=fmean(result.tardiness > 0 for result in results),
average_waiting_time=fmean(waiting),
p95_waiting_time=p95,
Expand Down
4 changes: 4 additions & 0 deletions src/adaptivequeue/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,3 +43,7 @@ def flow_time(self) -> int:
@property
def tardiness(self) -> int:
return max(0, self.completion_time - self.job.deadline)

@property
def cost(self) -> float:
return self.job.priority_weight * (self.tardiness + 0.1 * self.flow_time)
Loading
Loading