mirror of
https://gricad-gitlab.univ-grenoble-alpes.fr/tienyoun/presentations.git
synced 2026-09-26 13:31:01 +00:00
5.7 KiB
5.7 KiB
Les barres de progression dans Celery
- Intro
- Communication Worker->Django
- Communication Django->Client
- Accès concurrent
Intro
Liste des courses
- Progression en temps réel
- Barre de progression propre à chaque tâche instanciée
Contraintes
- Tâches possiblement longues
- Éviter le polling (fonctionner sur évènements)
- Pas de résidus en cas de plantage
Situation initiale
- Fonctionnalité non existante dans Celery
- Solutions sur le net : progression par comptage de micro-tâches
- Une bibliothèque existante sur le net : celery-progress
Complexité supplémentaire
<script type="text/mermaid">
%%{init: {'theme': 'light', 'themeVariables': { 'darkMode': false }}}%%
sequenceDiagram
actor User
participant Django
participant Broker@{ "type" : "database" }
participant Worker
participant Postgres@{ "type" : "database" }
User->>Django: Click button
Django->>Broker: Send message : schedule task
Broker->>Worker: Acquire task
activate Worker
Worker->>Postgres: Save results
deactivate Worker
User ->>Django: Query task result
Django->>Postgres: Query task result
Postgres-->>Django: task result
Django-->>User: task result
</script>
<script type="text/mermaid">
sequenceDiagram
actor User
participant Django
participant Broker@{ "type" : "database" }
participant Worker
participant Postgres@{ "type" : "database" }
User->>Django: Click button
Django->>Broker: Send message : schedule task
Broker->>Worker: Acquire task
activate Worker
rect rgb(255, 224, 224)
loop
Worker->>Worker: Work
Worker->>Broker: Progress event
Broker->>Django: Progress event
Django->>User: Progress event
end
end
Worker->>Postgres: Save results
deactivate Worker
User ->>Django: Query task result
Django->>Postgres: Query task result
Postgres-->>Django: task result
Django-->>User: task result
</script>
<script type="text/mermaid">
sequenceDiagram
actor User
participant Django
participant Worker
participant Postgres@{ "type" : "database" }
User->>Django: Click button
Django->>()Worker: Send message : schedule task
activate Worker
rect rgb(255, 224, 224)
loop
Worker->>Worker: Work
Worker->>()Django: Progress event
Django->>User: Progress event
end
end
Worker->>Postgres: Save results
deactivate Worker
User ->>Django: Query task result
Django->>Postgres: Query task result
Postgres-->>Django: task result
Django-->>User: task result
</script>
Communication worker -> server
- Coté Worker
- Coté Django
- Accès concurrent
Coté Worker
https://docs.celeryq.dev/en/main/reference/celery.events.html
The worker has the ability to send a message whenever some event happens. These events are then captured by tools like Flower, and celery events to monitor the cluster.
Coté Worker
class PtfTask(AbortableTask):
_progress_dispatcher = None
@property
def progress_dispatcher(self):
# Singleton
...
def update_state(self, task_id, state, meta, **kwargs):
self.progress_dispatcher.send(
"ptf-task-progress",
data=meta,
uuid=task_id or self.request.id,
)
return super().update_state(task_id, state, meta, **kwargs)
On envoie un évènement
ptf-task-progressquand on met à jour l'état de la tâche
Coté Worker
class PtfTask(AbortableTask):
_progress_dispatcher = None
@property
def progress_dispatcher(self):
# Singleton
...
def update_state(self, task_id, state, meta, **kwargs):
self.progress_dispatcher.send(
"ptf-task-progress",
data=meta,
uuid=task_id or self.request.id,
)
return super().update_state(task_id, state, meta, **kwargs)
Point important: on peut faire avancer la progression de n'importe quelle tâche tant qu'on a son ID!
Coté Worker
@shared_task(name="crawler.tasks.toto", base=PtfTask, bind=True)
def print_toto(self, toto = "toto"):
# Initialize progress bar at 0
self.update_state(
meta={"current": index, "total": len(toto)},
state=states.STARTED,
)
for index, letter in enumerate(toto):
# Do work
print(letter)
# Update progress bar
self.update_state(
meta={"current": index, "total": len(toto)},
state=states.STARTED,
)
Coté Django
from celery import Celery, current_app
def send_sse_message(event):
...
handlers = {
"ptf-task-progress": send_sse_message,
}
receiver = current_app.events.Receiver(
channel=current_app.connection_for_read(), app=current_app, handlers=handlers
)
On envoie un message au client quand on reçoit un évènement
ptf-task-progress
