磁盘调度器给 I/O 排好队，让磁头移动更省

(
  : "${DEVICE:?Set DEVICE to the reviewed whole-device name}"
  case "$DEVICE" in
    ''|[!A-Za-z]*|*[!A-Za-z0-9_-]*|ALL) printf 'Invalid device name\n' >&2; exit 2 ;;
  esac
  queue="/sys/block/$DEVICE/queue"
  [ -d "$queue" ] || { printf 'No such whole-device queue\n' >&2; exit 1; }
  for field in rotational scheduler; do
    [ -r "$queue/$field" ] || { printf 'Missing: %s\n' "$field" >&2; exit 1; }
    value=$(cat "$queue/$field") || exit 1
    printf '%s=%s\n' "$field" "$value"
  done
  for field in read_expire fifo_batch; do
    if [ -r "$queue/iosched/$field" ]; then
      value=$(cat "$queue/iosched/$field") || exit 1
      printf '%s=%s\n' "$field" "$value"
    else
      printf '%s: unavailable for the active scheduler\n' "$field"
    fi
  done
  LC_ALL=C iostat -x -y -t "$DEVICE" 2 5
)

"""Unexecuted proposal: temporary policy change on an exclusively controlled test device."""
import json
import re
import signal
import subprocess
import sys
from pathlib import Path


def device_name(value):
    if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_-]*", value) or value == "ALL":
        raise ValueError("use one named whole device, not a path, option, number or ALL")
    return value


def scheduler_state(path):
    tokens = path.read_text().split()
    active = [s[1:-1] for s in tokens if re.fullmatch(r"\[[A-Za-z0-9_-]+\]", s)]
    if len(active) != 1 or not tokens:
        raise ValueError("scheduler state must have exactly one bracketed choice")
    if any(not re.fullmatch(r"(?:[A-Za-z0-9_-]+|\[[A-Za-z0-9_-]+\])", s) for s in tokens):
        raise ValueError("unexpected scheduler state")
    available = [s.strip("[]") for s in tokens]
    if len(set(available)) != len(available):
        raise ValueError("duplicate scheduler choices")
    return active[0], available


def exercise(path, device, expected, requested, run=subprocess.run, emit=print):
    original, available = scheduler_state(path)
    if original != expected or requested not in available or requested == original:
        raise ValueError("current/requested scheduler does not match the reviewed plan")
    emit(json.dumps({"device": device, "original": original, "requested": requested}), flush=True)
    primary_error = restore_error = None
    attempted = False
    try:
        if scheduler_state(path) != (original, available):
            raise ValueError("scheduler state changed before the test")
        attempted = True  # even a failed write may have changed state
        path.write_text(requested + "\n")
        if scheduler_state(path)[0] != requested:
            raise RuntimeError("requested scheduler was not activated")
        result = run(["iostat", "-x", "-y", "-t", device, "2", "5"], check=False)
        if result.returncode != 0:
            raise RuntimeError(f"iostat exited with status {result.returncode}")
    except BaseException as error:
        primary_error = f"{type(error).__name__}: {error}"
    finally:
        if attempted:
            try:
                if original not in scheduler_state(path)[1]:
                    raise RuntimeError("original scheduler is no longer available")
                path.write_text(original + "\n")
                if scheduler_state(path)[0] != original:
                    raise RuntimeError("original scheduler was not restored")
            except BaseException as error:
                restore_error = f"{type(error).__name__}: {error}"
    emit(json.dumps({"primary_error": primary_error, "restore_error": restore_error,
                     "restored": attempted and restore_error is None}), flush=True)
    # Report both errors; restoration failure never hides the primary failure.
    return (1 if primary_error else 0) + (2 if restore_error else 0)


def main(args):
    if len(args) != 4:
        raise ValueError("usage: scheduler_exercise.py DEVICE EXPECTED_CURRENT REQUESTED DISPOSABLE:DEVICE")
    device, expected, requested, acknowledgement = args
    device = device_name(device)
    if acknowledgement != "DISPOSABLE:" + device:
        raise ValueError("explicit disposable-device acknowledgement required")
    path = Path("/sys/block") / device / "queue/scheduler"
    return exercise(path, device, expected, requested)


if __name__ == "__main__":
    def interrupted(number, frame):
        raise InterruptedError(f"received signal {number}; attempting restoration")
    signal.signal(signal.SIGINT, interrupted)
    signal.signal(signal.SIGTERM, interrupted)
    try:
        sys.exit(main(sys.argv[1:]))
    except (OSError, ValueError) as error:
        sys.exit(f"Scheduler test refused: {error}")

python3 scheduler_exercise.py "${DEVICE:?}" "${EXPECTED_CURRENT:?}" \
  "${REQUESTED:?}" "${ACK_DEVICE:?}"