examples/parallel/iopubwatcher.py

"""A script for watching all traffic on the IOPub channel (stdout/stderr/pyerr) of engines.

This connects to the default cluster, or you can pass the path to your ipcontroller-client.json

Try running this script, and then running a few jobs that print (and call sys.stdout.flush),
and you will see the print statements as they arrive, notably not waiting for the results
to finish.

You can use the zeromq SUBSCRIBE mechanism to only receive information from specific engines,
and easily filter by message type.

Authors
-------
* MinRK
"""

import os
import sys
import json
import zmq

from IPython.kernel.zmq.session import Session
from IPython.parallel.util import disambiguate_url
from IPython.utils.py3compat import str_to_bytes
from IPython.utils.path import get_security_file

def main(connection_file):
    """watch iopub channel, and print messages"""
    
    ctx = zmq.Context.instance()
    
    with open(connection_file) as f:
        cfg = json.loads(f.read())
    
    location = cfg['location']
    reg_url = cfg['url']
    session = Session(key=str_to_bytes(cfg['exec_key']))
    
    query = ctx.socket(zmq.DEALER)
    query.connect(disambiguate_url(cfg['url'], location))
    session.send(query, "connection_request")
    idents,msg = session.recv(query, mode=0)
    c = msg['content']
    iopub_url = disambiguate_url(c['iopub'], location)
    sub = ctx.socket(zmq.SUB)
    # This will subscribe to all messages:
    sub.setsockopt(zmq.SUBSCRIBE, b'')
    # replace with b'' with b'engine.1.stdout' to subscribe only to engine 1's stdout
    # 0MQ subscriptions are simple 'foo*' matches, so 'engine.1.' subscribes
    # to everything from engine 1, but there is no way to subscribe to
    # just stdout from everyone.
    # multiple calls to subscribe will add subscriptions, e.g. to subscribe to
    # engine 1's stderr and engine 2's stdout:
    # sub.setsockopt(zmq.SUBSCRIBE, b'engine.1.stderr')
    # sub.setsockopt(zmq.SUBSCRIBE, b'engine.2.stdout')
    sub.connect(iopub_url)
    while True:
        try:
            idents,msg = session.recv(sub, mode=0)
        except KeyboardInterrupt:
            return
        # ident always length 1 here
        topic = idents[0]
        if msg['msg_type'] == 'stream':
            # stdout/stderr
            # stream names are in msg['content']['name'], if you want to handle
            # them differently
            print("%s: %s" % (topic, msg['content']['data']))
        elif msg['msg_type'] == 'pyerr':
            # Python traceback
            c = msg['content']
            print(topic + ':')
            for line in c['traceback']:
                # indent lines
                print('    ' + line)

if __name__ == '__main__':
    if len(sys.argv) > 1:
        cf = sys.argv[1]
    else:
        # This gets the security file for the default profile:
        cf = get_security_file('ipcontroller-client.json')
    main(cf)