Monitoring

olympipe.enable_monitoring(inspect_messages=False) registers, for the whole process, the queues created afterwards: they are MonitorableQueues, stored in pipeline.queues under their id. With inspect_messages=True, each queue keeps a copy of the payloads in flight, which peek_at, peek_all and drain read. The call is idempotent and also sets OLYMPIPE_MONITORING=1. A later call with inspect_messages=False does not turn inspection off.

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

# Before enable_monitoring: process-wide inspection is still off.
# monitoring=True registers the queue; without inspect_messages there is no payload copy.
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 keeps a copy of in-flight payloads.
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()

# Process-wide. The second call does not turn inspection off.
olympipe.enable_monitoring(inspect_messages=True)
olympipe.enable_monitoring(inspect_messages=False)

out = Pipeline([1, 2, 3]).task(lambda x: x + 10)
# queues: id -> queue. The source is registered first, then the stage's queue.
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())

Output

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

How it works

The same switches are constructor arguments, inherited by every stage added afterwards. None follows enable_monitoring() and OLYMPIPE_MONITORING. Without inspection the queue stays on the fast path: snapshot() reports size, max_size, paused, put_count, get_count and errors, but the mirror is empty and peek_at raises IndexError. The choice is copied onto the queue and travels with it into workers, including under spawn. pause() blocks get() until resume(). Call it before building the pipeline.

When to call it, and how to opt out

Queues already created are not registered retroactively. Stages added afterwards on a pipeline whose monitoring is None follow the switch. monitoring=False keeps the whole graph out of the registry, including later stages, even after enable_monitoring(). An explicit inspect_messages=False on the constructor leaves the queue registered but without a mirror.

python

import olympipe
from olympipe import Pipeline

early = Pipeline([1, 2, 3])
olympipe.enable_monitoring(inspect_messages=True)
print(len(early.queues))                    # queue already created: not registered
later = early.task(lambda x: x + 1)         # this stage follows the switch
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()

Output

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

Previous Starting the feed