mirror of
https://gricad-gitlab.univ-grenoble-alpes.fr/tienyoun/presentations.git
synced 2026-09-26 13:31:01 +00:00
241 lines
5.8 KiB
Markdown
241 lines
5.8 KiB
Markdown
# Les barres de progression dans Celery
|
|
<img src="./assets/progressbar.png">
|
|
|
|
- **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
|
|
|
|
<div data-mermaid>
|
|
<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>
|
|
</div>
|
|
|
|
|
|
|
|
|
|
<div data-mermaid data-auto-animate>
|
|
<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>
|
|
</div>
|
|
|
|
|
|
|
|
|
|
|
|
<div data-mermaid >
|
|
<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>
|
|
</div>
|
|
|
|
---
|
|
|
|
## 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
|
|
- celery.events.EventDispatcher 📢
|
|
- celery.events.EventReceiver 👂
|
|
|
|
|
|
|
|
|
|
### Coté Worker
|
|
<!-- .slide: data-auto-animate -->
|
|
|
|
```python[]
|
|
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)
|
|
|
|
```
|
|
<!-- .element.code: data-id="code" -->
|
|
> On envoie un évènement `ptf-task-progress` quand on met à jour l'état de la tâche
|
|
|
|
|
|
|
|
### Coté Worker
|
|
<!-- .slide: data-auto-animate -->
|
|
```python[13]
|
|
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)
|
|
|
|
```
|
|
<!-- .element: data-id="code" -->
|
|
> Point important: on peut faire avancer la progression de n'importe quelle tâche tant qu'on a son ID!
|
|
|
|
|
|
|
|
### Coté Worker
|
|
|
|
```python
|
|
@shared_task(name="crawler.tasks.toto", base=PtfTask, bind=True)
|
|
def print_toto(self, toto = "toto"):
|
|
meta = {"current": 0, "total": len(toto)}
|
|
# Initialize progress bar at 0
|
|
self.update_state(
|
|
meta=meta,
|
|
state=states.STARTED,
|
|
)
|
|
|
|
for index, letter in enumerate(toto):
|
|
# Do work
|
|
print(letter)
|
|
|
|
# Update progress bar
|
|
meta["current"] = index
|
|
self.update_state(
|
|
meta=meta,
|
|
state=states.STARTED,
|
|
)
|
|
```
|
|
|
|
---
|
|
|
|
### Coté Django
|
|
|
|
```python
|
|
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`
|
|
|
|
|
|
|
|
### Coté Django
|
|
Django 4.2 Workaround
|
|
<img src="./assets/disconnections.png">
|