swh.scheduler.celery_backend.runner module

swh.scheduler.celery_backend.runner.run_ready_tasks(backend: swh.scheduler.interface.SchedulerInterface, app, task_types: List[Dict] = [], with_priority: bool = False) → List[Dict]

Schedule tasks ready to be scheduled.

This lookups any tasks per task type and mass schedules those accordingly (send messages to rabbitmq and mark as scheduled equivalent tasks in the scheduler backend).

If tasks (per task type) with priority exist, they will get redirected to dedicated high priority queue (standard queue name prefixed with save_code_now:).

Parameters

• backend – scheduler backend to interact with (read/update tasks)

• app (App) – Celery application to send tasks to

• task_types – The list of task types dict to iterate over. By default, empty. When empty, the full list of task types referenced in the scheduler will be used.

• with_priority – If True, only tasks with priority set will be fetched and scheduled. By default, False.

Returns

A list of dictionaries:

{
  'task': the scheduler's task id,
  'backend_id': Celery's task id,
  'scheduler': utcnow()
}

The result can be used to block-wait for the tasks’ results:

backend_tasks = run_ready_tasks(self.scheduler, app)
for task in backend_tasks:
    AsyncResult(id=task['backend_id']).get()

swh.scheduler.celery_backend.runner.main()