Repository navigation
Description
Steps to reproduce
- Have runs with many job submissions (retries, replicas, redeployments).
- Call
POST /api/runs/list.
Actual behaviour
On dstack Sky, slow /api/runs/list requests take 0.5-2.2 s. Most of it is loading jobs: the first jobs query takes 0.6-1.8 s and a second one ~100 ms, while the runs query takes 15-100 ms.
To return the latest job_submissions_limit submissions per job, the server numbers every submission of the listed runs with row_number() OVER (PARTITION BY run_id, replica_num, job_num ORDER BY submission_num DESC), selecting all columns, and keeps those numbered up to N:
dstack/src/dstack/_internal/server/services/runs/__init__.py
Lines 428 to 476 in 5e3bbf9
| async def _list_latest_job_models_per_job( | |
| session: AsyncSession, | |
| run_ids: List[uuid.UUID], | |
| limit_per_job: int, | |
| include_probes: bool, | |
| only_with_termination_reason: bool = False, | |
| ) -> List[JobModel]: | |
| """ | |
| List up to N newest submissions for each job. | |
| A newer submission has a higher submission_num. The SQL window applies the | |
| per-job limit in the database instead of loading all retries and slicing them | |
| in Python. | |
| """ | |
| row_number = ( | |
| func.row_number() | |
| .over( | |
| partition_by=(JobModel.run_id, JobModel.replica_num, JobModel.job_num), | |
| order_by=JobModel.submission_num.desc(), | |
| ) | |
| .label("row_number") | |
| ) | |
| filters = [JobModel.run_id.in_(run_ids)] | |
| if only_with_termination_reason: | |
| filters.append(JobModel.termination_reason.isnot(None)) | |
| jobs_sq = ( | |
| select( | |
| JobModel, | |
| row_number, | |
| ) | |
| .where(*filters) | |
| .subquery() | |
| ) | |
| job_alias = aliased(JobModel, jobs_sq) | |
| options = [] | |
| if include_probes: | |
| options.append(joinedload(job_alias.probes)) | |
| res = await session.execute( | |
| select(job_alias) | |
| .where(jobs_sq.c.row_number <= limit_per_job) | |
| .options(*options) | |
| .order_by( | |
| job_alias.run_id, | |
| job_alias.replica_num, | |
| job_alias.job_num, | |
| job_alias.submission_num, | |
| ) | |
| ) | |
| return list(res.unique().scalars().all()) |
So the database sorts every submission with all its columns, including job_spec_data, job_provisioning_data and job_runtime_data, to return one or a few per job. It does this twice, the second time for the latest submission with a termination reason:
dstack/src/dstack/_internal/server/services/runs/__init__.py
Lines 403 to 416 in 5e3bbf9
| requested_jobs = await _list_latest_job_models_per_job( | |
| session=session, | |
| run_ids=run_ids, | |
| limit_per_job=max(job_submissions_limit, 1), | |
| include_probes=include_probes, | |
| ) | |
| # Also load rows needed to preserve run.status_message, e.g. `retrying`. | |
| status_message_jobs = await _list_latest_job_models_per_job( | |
| session=session, | |
| run_ids=run_ids, | |
| limit_per_job=1, | |
| include_probes=False, | |
| only_with_termination_reason=True, | |
| ) |
The UI, dstack ps and Client.runs.list() pass job_submissions_limit=1. Callers that omit it, such as other API clients and CLI versions from before #3986, get 10:
dstack/src/dstack/_internal/server/routers/runs.py
Lines 73 to 75 in 5e3bbf9
| job_submissions_limit = body.job_submissions_limit | |
| if job_submissions_limit is None: | |
| job_submissions_limit = MAX_JOB_SUBMISSIONS_LIMIT |
Expected behaviour
The cost of loading jobs depends on the number of submissions returned, not on how many each job has accumulated.
dstack version
0.22.7
Server logs
n/aAdditional information
Proposed fix: one query that finds each job's latest submission numbers and fetches only the submissions within the last N:
SELECT jobs.*
FROM jobs
JOIN (
SELECT run_id, replica_num, job_num,
max(submission_num) AS latest,
max(CASE WHEN termination_reason IS NOT NULL THEN submission_num END) AS latest_terminated
FROM jobs
WHERE run_id IN (:run_ids)
GROUP BY run_id, replica_num, job_num
) l
ON jobs.run_id = l.run_id AND jobs.replica_num = l.replica_num AND jobs.job_num = l.job_num
WHERE jobs.submission_num > l.latest - :n
OR jobs.submission_num = l.latest_terminatedThe subquery groups four narrow columns and needs no sort, and full rows are read only for the returned submissions. It replaces both queries, and runs on SQLite and Postgres. An index on (run_id, replica_num, job_num, submission_num) would let both parts read only what they need.
submission_num > latest - N is exact for N=1, which is what the UI and CLI use. For N>1 it can return fewer than N submissions, because submission numbers have gaps where no-capacity submissions were deleted:
dstack/src/dstack/_internal/server/background/pipeline_tasks/runs/common.py
Lines 114 to 140 in 5e3bbf9
| async def delete_superseded_no_capacity_job_submissions( | |
| session: AsyncSession, | |
| run_id: uuid.UUID, | |
| new_job_models: list[JobModel], | |
| ) -> None: | |
| """ | |
| Delete previous job submissions that failed to start due to no capacity without | |
| ever provisioning. Such submissions are created each time a run is retrying | |
| on no capacity and carry no information beyond the events. | |
| The direct predecessor of each new submission is kept so that the run's | |
| `status_message` can still show `retrying` while the new submission is in flight. | |
| """ | |
| for job_model in new_job_models: | |
| if job_model.submission_num < 2: | |
| continue | |
| await session.execute( | |
| delete(JobModel).where( | |
| JobModel.run_id == run_id, | |
| JobModel.replica_num == job_model.replica_num, | |
| JobModel.job_num == job_model.job_num, | |
| JobModel.submission_num < job_model.submission_num - 1, | |
| JobModel.status.in_(JobStatus.finished_statuses()), | |
| JobModel.termination_reason | |
| == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, | |
| JobModel.job_provisioning_data.is_(None), | |
| ) | |
| ) |
That fits the API, which promises at most job_submissions_limit latest submissions per job, and only drops submissions older than deleted ones.
Getting exactly N per job needs per-job ranking, and every way to do it is either costly or complex: a window function sorting all submissions (today), a window over narrow columns joined back to jobs, a correlated count per submission, a LATERAL join with LIMIT N (Postgres-only, so SQLite needs another code path), or loading narrow rows for every submission into Python.
Activity