Scheduler

mlumina.scheduler

Отдельный FastAPI-сервис — приоритетная очередь задач поверх Ray Jobs API. Не заменяет ray.tune/ray.train — решает, когда и в каком порядке отправлять уже готовые job-скрипты в Ray-кластер, с учётом заявленных ресурсов и приоритета.

В Docker-стеке (docker-compose.yml) это сервис scheduler: собирается из корневого Dockerfile, зависит от head, слушает порт 8264, точка входа — python /home/ray/src/mlumina/scheduler/main.py.

import mlumina.scheduler.main имеет побочные эффекты: на уровне модуля создаётся Scheduler() (что сразу пытается подключиться к Ray-кластеру через JobSubmissionClient и ray.init()) и вызывается scheduler.start() (блокирующий запуск uvicorn на 0.0.0.0:8264). Импортировать этот модуль просто "для класса" нельзя — он запустит сервер.

Как это работает

• Scheduler.__init__ подключается к Ray dashboard (RAY_DASHBOARD_ADDRESS, по умолчанию http://head:8265) через JobSubmissionClient и к самому кластеру (RAY_CLUSTER_ADDRESS, по умолчанию ray://head:10001) через ray.init().
• Задачи хранятся в priority_queue: {0: [...], 1: [...], ..., 9: [...]} (9 — наивысший приоритет).
• Фоновый поток (_main, цикл раз в sleep_time=1 сек, под threading.Condition) на каждой итерации:
  1. проверяет статус запущенных задач через ray_client.get_job_status (завершённые/упавшие — убирает из running_tasks, логирует, для FAILED дополнительно подтягивает get_job_logs);
  2. отменяет задачи, превысившие свой timeout;
  3. пытается запустить из очереди задачи с наивысшим непустым приоритетом, если под них хватает незанятых ресурсов кластера (_check_resources_by_allocation — сравнивает уже выделенное текущим задачам с ray.cluster_resources(); учитываются cpu/gpu/memory, GPU-квота не проверяется на приём в очередь методом _add_task_to_queue, но участвует в допуске к запуску);
  4. пишет снепшот использования ресурсов в src/mlumina/scheduler/data/resource_history.jsonl (файл пересоздаётся пустым при каждом старте сервиса).
• Задача с cpu == 0 and gpu == 0 and memory == 0 отклоняется сразу ("task must request at least one resource"), как и задача, запрашивающая больше ресурсов, чем есть во всём кластере (ray.cluster_resources()).

HTTP API (порт 8264)

TaskData (тело POST /add_task)

class TaskData(BaseModel):
    name: str
    entrypoint: str        # команда запуска, например "python ray_wrapper/examples/fcnn/tune.py"
    runtime_env: dict       # ray runtime_env: working_dir, pip, env_vars и т.д.
    priority: int            # 0..9, больше — выше приоритет
    cpu: int                 # >= 0
    gpu: int                 # >= 0
    memory: int               # >= 0, в ГБ
    timeout: int              # >= 0, секунды; 0 = без таймаута

Пример вызова:

import requests

response = requests.post(
    "http://127.0.0.1:8264/add_task",
    json={
        "name": "my_job",
        "entrypoint": "python examples/scheduler/test.py",
        "runtime_env": {"working_dir": "/home/ray/src/mlumina"},
        "priority": 1,
        "cpu": 2,
        "gpu": 0,
        "memory": 4,
        "timeout": 300,
    },
    timeout=60,
)
print(response.status_code, response.json())

entrypoint выполняется в рабочей директории runtime_env["working_dir"] — это путь внутри контейнера, где запущен scheduler (в docker-compose это /home/ray/src, с примонтированным ./src), а не на машине, откуда отправляется HTTP-запрос. Больше примеров вызова — src/mlumina/ray_wrapper/examples/*/schedule_*.py и examples/scheduler/test.py (тривиальная тестовая задача-заглушка).

Идентификатор задачи (Task.id) генерируется как mlumina-{name}-priority-{priority}-{cpu}cpu-{gpu}gpu-{memory}memory-{uuid4} и используется и как ключ внутри планировщика, и как submission_id в Ray Jobs API — по нему же нужно передавать task_id в /change_task_priority и /cancel_task.