Compression scheduler: poll_running_jobs sleeps ~0.5s per in-flight job, delaying dispatch of new jobs

[goynam]

Summary

The compression scheduler's main loop is single threaded and alternates between poll_running_jobs() and search_and_schedule_new_tasks(). poll_running_jobs() calls ResultHandle.get_result() serially for every in-flight job. For any job whose tasks are not complete yet, the call sleeps 0.5s (the celery default poll interval) before returning None, so one scheduler cycle takes roughly 0.5s x N for N in-flight jobs, and new PENDING jobs are not dispatched until the current cycle finishes.

On our production deployment with 150-190 jobs in flight, scheduler cycles take 75-95s and dispatch latency (creation_time -> start_time) is 140-330s. Jobs sit undispatched for 2-5 minutes while the celery queue is empty and workers are idle. The delay scales linearly with the number of concurrent jobs, so deployments with many datasets are hit hardest.

Environment

• clp-package 0.12.0 lineage, CLP_S storage engine
• CeleryTaskManager with Redis result backend, RabbitMQ broker
• celery 5.6.3, kombu 5.6.2, redis-py 6.4.0
• ~50 datasets ingesting concurrently, 300 celery workers
• jobs_poll_delay: 0.1, max_concurrent_tasks_per_job: 0

Root cause

components/job-orchestration/job_orchestration/scheduler/compress/task_manager/celery_task_manager.py:

def get_result(self, timeout: float = 0.1) -> list[CompressionTaskResult] | None:
    try:
        results = self._celery_result.get(timeout=timeout)
        return [CompressionTaskResult.model_validate(res) for res in results]
    except celery.exceptions.TimeoutError:
        return None

timeout=0.1 does not bound the wall time. With celery 5.6.3 and the Redis result backend the call path is:

GroupResult.get(timeout=0.1) -> ResultSet.join_native (Redis has supports_native_join = True) -> iter_native -> AsyncBackendMixin._wait_for_pending -> BaseResultConsumer._wait_for_pending -> Redis on_wait_for_pending -> result._iter_meta() -> BaseKeyValueStoreBackend.get_many(ids, timeout=0.1, interval=0.5, max_iterations=1).

In celery/backends/base.py::get_many(), when any task in the group is not READY after the mget:

if timeout and iterations * interval >= timeout:   # 0 * 0.5 >= 0.1 is False on the first pass
    raise TimeoutError(f'Operation timed out ({timeout})')
if on_interval:
    on_interval()
time.sleep(interval)  # don't busy loop.
iterations += 1
if max_iterations and iterations >= max_iterations:
    break

The timeout check compares iterations * interval rather than elapsed time and runs before the sleep with iterations == 0, so it never fires on the first pass; the 0.5s sleep always happens. interval is the default from ResultSet.get() and is never overridden by the caller. After get_many returns, Drainer.drain_events_until raises socket.timeout, which _wait_for_pending converts to the celery.exceptions.TimeoutError that get_result() catches. Total cost per incomplete job is the 0.5s sleep plus one Redis mget plus a short pubsub drain, which matches the ~0.504s spacing in our scheduler logs.

compression_scheduler.py::poll_running_jobs() then does this serially:

for job_id, job in scheduled_jobs.items():
    returned_results = job.result_handle.get_result()
    if returned_results is None:
        continue

Cycle cost is about 0.5s per incomplete job. search_and_schedule_new_tasks() only runs between sweeps, so dispatch latency for a new job is 1-3 full cycles.

Evidence

• Scheduler INFO logs show job completion processing spaced at almost exactly the poll interval: 05:21:43.446 -> 05:21:43.950 -> 05:21:45.214 -> 05:21:45.719 -> 05:21:46.224 (~0.50s per still-running job in between).
• compression_jobs rows show TIMESTAMPDIFF(SECOND, creation_time, start_time) of 141/234/328s, stepping in ~95s increments, i.e. quantized by scheduler cycle length (190 in-flight jobs x 0.5s).
• The RabbitMQ compression queue is empty and consumers are idle during these waits, so this is not worker capacity. No scheduler DB statement exceeded 1s in sampled processlist output.

Impact

• Adds 2-5 minutes of ingestion latency to every job, regardless of job size.
• Completion bookkeeping (SUCCEEDED status, duration) lags by the same amount, which skews job latency metrics.
• The effect compounds during backlog catch-up: more jobs in flight means longer cycles means longer dispatch waits.
• Configurations that produce smaller, more frequent jobs increase N and make the cycle proportionally worse.

Suggested fix

Check readiness before collecting results:

def get_result(self, timeout: float = 0.1) -> list[CompressionTaskResult] | None:
    if not self._celery_result.ready():
        return None
    results = self._celery_result.get(timeout=timeout)
    return [CompressionTaskResult.model_validate(res) for res in results]

GroupResult.ready() is a plain backend read with no sleep (sub-ms per task on Redis). The poll sweep over ~190 jobs drops from ~95s to milliseconds and dispatch latency collapses to roughly jobs_poll_delay. The get() after a successful ready() returns immediately. Passing interval=0.01 to .get() would also work, but ready() avoids raising and catching a TimeoutError per job on every cycle.

Running polling and dispatch on separate threads would remove the coupling entirely, but it needs a second DB connection (DbContext wraps a connection that is not thread safe), locking around the shared scheduled_jobs dict, and changes to the SIGTERM drain logic. With the ready() check the sweep is milliseconds, so I don't think that's warranted.

Note that when max_concurrent_tasks_per_job > 0, _dispatch_next_task_batch() is gated on the same sweep, so intra-job batch turnaround inherits the same latency. The ready() check fixes that path as well.

Repro

1. Run enough concurrent compression jobs that 100+ are in RUNNING simultaneously (many datasets or long-running jobs).
2. Observe creation_time -> start_time on newly inserted jobs grow linearly with the in-flight job count (~0.5s each) despite an empty broker queue and idle workers.
3. Add the ready() check; dispatch latency drops to ~jobs_poll_delay.

[gibber9809]

Good catch, we actually fixed a very similar issue on the search side in #1899 (note this comment explains why that PR is written the way it is).

One note on your proposed fix: even when ready() returns true get() almost always goes through one polling cycle before returning the result, seemingly because the mechanism to record the state of a task is separate from the mechanism for making a task's result available, so there tends to be a period of time where the task is "ready", but the result isn't yet available.

The fix we used in the search path was more like:

if not result.ready():
  return None
return result.get(interval=0.005, timeout=10)

So that when get() does end up polling, the polling interval is small at least. We should be able to adapt your proposed solution to also use interval= in the get() to avoid that pitfall.

[goynam]

Hey @gibber9809

Need more feedback, Can we do this :
Dispatch and Poll run on the same scheduler thread, so a slow poll pass stalls dispatch — we've seen multi-minute dispatch delays in production just because many jobs finished at once. But polling celery for results is redundant: workers already write task status to compression_tasks. The scheduler could just check the DB — if all tasks of a job are terminal, mark the job done. That makes poll cost one indexed query no matter how many jobs are in flight, and it fully decouples dispatch from poll — they could even run as separate processes. Gaps to close: per-task error detail needs a DB column (today it only exists in the celery result), and we'd need a staleness rule for tasks whose worker died before writing a final status.

[gibber9809]

I think I'd prefer just trying to speed up the current approach with ready()/get(interval=,timeout=) for now, mostly because:

1. Celery will not garbage collect results in the result backend until we call get() or forget() on a task result, so if we try to manage task/job status only through the DB I think the result backend might start slowly filling up its storage.
2. We're getting close to integrating Spider to handle compression (at least for the compression paths you use as I understand it), so I don't want to significantly change this flow or make it more complicated as we're trying to replace it if we can avoid it.

If the simpler ready()/get(interval=,timeout=) approach ends up not being enough, and Spider still isn't ready, then I think we can look at bigger changes to the flow in the compression scheduler to give us more performance while we wait for Spider.