Surveillance

Enregistrez les files et lisez les messages en vol

olympipe.enable_monitoring(inspect_messages=False) enregistre, pour tout le processus, les files créées ensuite : ce sont des MonitorableQueue, rangées dans pipeline.queues sous leur identifiant. Avec inspect_messages=True, chaque file garde une copie des payloads en vol, que peek_at, peek_all et drain lisent. L'appel est idempotent et pose aussi OLYMPIPE_MONITORING=1. Un appel ultérieur avec inspect_messages=False n'éteint pas l'inspection.

Signature

olympipe.enable_monitoring(inspect_messages: bool = False) -> None
Pipeline(datas, *, monitoring: bool | None = None, inspect_messages: bool | None = None)
# pipeline.queues: dict[str, MonitorableQueue]
MonitorableQueue.peek_at(i) -> item
MonitorableQueue.peek_all() -> list
MonitorableQueue.drain() -> list
MonitorableQueue.pause() / resume()
MonitorableQueue.snapshot() -> dict

python

import time
import olympipe
from olympipe import Pipeline

# Avant enable_monitoring : l'inspection globale est encore éteinte.
# monitoring=True enregistre la file ; sans inspect_messages, pas de copie des payloads.
registered = Pipeline([1], monitoring=True)
queue = next(iter(registered.queues.values()))
queue.put("hidden")
try:
    queue.peek_at(0)
except IndexError as exc:
    print(exc)
print(queue.snapshot()["put_count"], queue.snapshot()["size"])
registered.cleanup()

# inspect_messages=True garde une copie des payloads en vol.
held = Pipeline([0], monitoring=True, inspect_messages=True)
queue = next(iter(held.queues.values()))
queue.put("a")
queue.put("b")
print(queue.peek_at(0), queue.peek_all())
print(queue.drain())
held.cleanup()

# Pour tout le processus. Le second appel n'éteint pas l'inspection.
olympipe.enable_monitoring(inspect_messages=True)
olympipe.enable_monitoring(inspect_messages=False)

out = Pipeline([1, 2, 3]).task(lambda x: x + 10)
# queues : identifiant -> file. La source est enregistrée d'abord, puis la file de l'étape.
output = list(out.queues.values())[-1]
output.pause()
out.start()
deadline = time.time() + 30
while time.time() < deadline and len(output.peek_all()) < 3:
    time.sleep(0.05)
print(output.peek_all())
output.resume()
print(out.wait_for_result())

Sortie

No message at index 0 (size=0)
1 1
a ['a', 'b']
['a', 'b']
[11, 12, 13]
[11, 12, 13]

Comment ça fonctionne

Les mêmes interrupteurs sont des arguments du constructeur, hérités par chaque étape ajoutée ensuite. None suit enable_monitoring() et OLYMPIPE_MONITORING. Sans inspection, la file reste en mode rapide : snapshot() donne size, max_size, paused, put_count, get_count et errors, mais le miroir est vide et peek_at lève IndexError. Le choix est recopié sur la file et voyage avec elle dans les workers, y compris en spawn. pause() bloque les get() jusqu'à resume(). À appeler avant de construire le pipeline.

Quand l'appeler, et comment s'en retirer

Les files déjà créées ne sont pas enregistrées rétroactivement. Les étapes ajoutées ensuite sur un pipeline dont monitoring vaut None suivent l'interrupteur. monitoring=False retire tout le graphe du registre, étapes suivantes comprises, même après enable_monitoring(). inspect_messages=False explicite sur le constructeur laisse la file enregistrée mais sans miroir.

python

import olympipe
from olympipe import Pipeline

early = Pipeline([1, 2, 3])
olympipe.enable_monitoring(inspect_messages=True)
print(len(early.queues))                    # file déjà créée : pas enregistrée
later = early.task(lambda x: x + 1)         # cette étape suit l'interrupteur
print(len(later.queues))
print(sorted(later.wait_for_result()))

plain = Pipeline([1, 2], monitoring=False).task(lambda x: x * 2)
print(len(plain.queues))
print(sorted(plain.wait_for_result()))

opt = Pipeline([5], monitoring=True, inspect_messages=False)
q = next(iter(opt.queues.values()))
q.put(5)
try:
    q.peek_at(0)
except IndexError as exc:
    print(exc)
opt.cleanup()

Sortie

0
1
[2, 3, 4]
0
[2, 4]
No message at index 0 (size=0)

Précédent Démarrer l'alimentation