QProcess event queue

borisruna

Hello ! I am trying to parse the output of traceroute and display it on a QTableWidget.
I am doing this by using QProcess ,but I found out that while processing the stdout of traceroute some content is missing . I guess this is because while parsing, QProcess emits signals (readyReadStandardOutput) that are getting missed by my main application.
As an example part of a normal output of traceroute would be this:
https://paste.pound-python.org/show/q8F1sgny2thXCanE1rd1/ and this is what I am getting: https://paste.pound-python.org/show/Slcu3EmcaTpTNJJkFcCJ/
and my code breaks.
Is there a way I can assure that all signals get processed from my main thread ? Also do signals of QProcess stack in some kind of event queue ? I am using python with pyqt4.
Any help would be much appreciated . Thank you in advance !

SGaist

Hi and welcome to devnet,

How are you doing the parsing of the data you received ?

borisruna

I am using a modified version of this https://pypi.python.org/pypi/trparse . I have modified the lib cause it was not capable of parsing real time data (one line at a time) .
Appreciate your feedback .

borisruna

And this is the code I am using :

    def init_process(self):
        self.process.readyReadStandardOutput.connect(self.parse_n_populate)
    def parse_n_populate(self):
        trace_output = str(self.process.readAllStandardOutput())
        hop = tr.get_hop(trace_output)
        print trace_output
        
        if hop:
            print hop
            cursor = self.tbl.textCursor()
            cursor.movePosition(cursor.End)
            cursor.insertText(str(hop))
            self.tbl.ensureCursorVisible()

print trace_output works as intended , i.e. this prints the output of traceroute correctly. print hop though misses allot of the output!

kshegunov

@borisruna said in QProcess event queue:

I guess this is because while parsing, QProcess emits signals (readyReadStandardOutput) that are getting missed by my main application.

QProcess parses nothing. That's your job, its is to provide you access to the stdout of the child process. I don't see how signals will get missed.

Is there a way I can assure that all signals get processed from my main thread?

Why do you have more than one thread to begin with? But to answer your question, yes, if the receiver is in the main thread, then the slots will be invoked in the main thread.

Also do signals of QProcess stack in some kind of event queue?

The slots those signals are connected to (if across threads) are put in an event queue. If you're not sending things across different threads signal-slot calls are immediate.

If you will, show the implementation of get_hop too. My guess is it's not parsing things properly, thus you get holes.

borisruna

Only the trparse module parses of course , not the QProcess. However I tested the get_hop method without qt and it is working as intended . The modified trparse module is this :

import re

RE_HEADER = re.compile(r'traceroute\sto\s(\S+)\s+\((?:(\d+\.\d+\.\d+\.\d+)|([0-9a-fA-F:]+))\)')
RE_HOP = re.compile(r'^\s*(\d+)\s+(?:\[AS(\d+)\]\s+)?([\s\S]+?)$', re.M)

RE_PROBE_NAME = re.compile(r'^([a-zA-z0-9.-]+)$|^(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})$|^([0-9a-fA-F:]+)$')
RE_PROBE_IP = re.compile(r'\((?:(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})|([0-9a-fA-F:]+))\)+')
RE_PROBE_RTT = re.compile(r'^(\d+(?:\.?\d+)?)$')
RE_PROBE_ANNOTATION = re.compile(r'^(!\w*)$')
RE_PROBE_TIMEOUT = re.compile(r'(\*)')

tr_data = """
traceroute to edgecastcdn.net (72.21.81.13), 30 hops max, 38 byte packets
1  *  *
2  *  *
3  *  *
4  10.251.11.32 (10.251.11.32)  3574.616 ms  0.153 ms
5  10.251.10.2 (10.251.10.2)  465.821 ms  2500.031 ms
6  172.18.68.206 (172.18.68.206)  170.197 ms  78.979 ms
7  172.18.59.165 (172.18.59.165)  151.123 ms  525.177 ms
8  172.18.59.170 (172.18.59.170)  150.909 ms  172.18.59.174 (172.18.59.174)  62.591 ms
9  172.18.75.5 (172.18.75.5)  123.078 ms  68.847 ms
10  12.91.11.5 (12.91.11.5)  79.834 ms  556.366 ms
11  cr2.ptdor.ip.att.net (12.123.157.98)  245.606 ms  83.038 ms
12  cr81.st0wa.ip.att.net (12.122.5.197)  80.078 ms  96.588 ms
13  gar1.omhne.ip.att.net (12.122.82.17)  363.800 ms  12.122.111.9 (12.122.111.9)  72.113 ms
14  206.111.7.89.ptr.us.xo.net (206.111.7.89)  188.965 ms  270.203 ms
15  xe-0-6-0-5.r04.sttlwa01.us.ce.gin.ntt.net (129.250.196.230)  706.390 ms  ae-6.r21.sttlwa01.us.bb.gin.ntt.net (129.250.5.44)  118.042 ms
16  xe-9-3-2-0.co1-96c-1b.ntwk.msn.net (207.46.47.85)  675.110 ms  72.21.81.13 (72.21.81.13)  82.306 ms
"""


class Hop(object):
    """
    Abstraction of a hop in a traceroute.
    """

    def __init__(self, idx, asn=None):
        self.idx = idx  # Hop count, starting at 1
        self.asn = asn  # Autonomous System number
        self.probes = []  # Series of Probe instances

    def add_probe(self, probe):
        """Adds a Probe instance to this hop's results."""
        if self.probes:
            # Increment probe.num as soon as there are more than
            # one probes in the Hop
            probe.num = len(self.probes) + 1
            probe_last = self.probes[-1]
            if not probe.ip:
                probe.ip = probe_last.ip
                probe.name = probe_last.name

        self.probes.append(probe)

    def __str__(self):
        text = "{:>3d} ".format(self.idx)
        if self.asn:
            text += "[AS{:>5d}] ".format(self.asn)
        text_len = len(text)
        for n, probe in enumerate(self.probes):
            text_probe = str(probe)
            if n:
                text += (text_len * " ") + text_probe
            else:
                text += text_probe
        text += "\n"
        return text


class Probe(object):
    """
    Abstraction of a probe in a traceroute.
    """

    def __init__(self, name=None, ip=None, rtt=None, anno=''):
        # Default num is 1 . This will increment from Hop.add_probe()
        # as soon as hops are added to the probe.
        self.num = 1
        self.name = name
        self.ip = ip
        self.rtt = rtt  # RTT in ms
        self.anno = anno  # Annotation, such as !H, !N, !X, etc

    def __str__(self):
        if self.rtt:
            text = "{:s} ({:s}) {:1.3f} ms {:s}\n".format(self.name, self.ip, self.rtt, self.anno)
        else:
            text = "*\n"
        return text


class ParseError(Exception):
    pass


def get_hop(line):
    matches_hop = RE_HOP.findall(line)
    for match_hop in matches_hop:
        idx = int(match_hop[0])
        if match_hop[1]:
            asn = int(match_hop[1])
        else:
            asn = None
        hop = Hop(idx, asn)
        probes_data = match_hop[2].split()
        # Get rid of 'ms': <name> | <(IP)> | <rtt> | '*'
        probes_data = filter(lambda s: s.lower() != 'ms', probes_data)

        i = 0
        while i < len(probes_data):
            # For each hop parse probes
            name = None
            ip = None
            rtt = None
            anno = ''

            # RTT check comes first because RE_PROBE_NAME can confuse rtt with an IP as name
            # The regex RE_PROBE_NAME can be improved
            if RE_PROBE_RTT.match(probes_data[i]):
                # Matched rtt, so name and IP have been parsed before
                rtt = float(probes_data[i])
                i += 1
            elif RE_PROBE_NAME.match(probes_data[i]):
                # Matched a name, so next elements are IP and rtt
                name = probes_data[i]
                ip = probes_data[i + 1].strip('()')
                rtt = float(probes_data[i + 2])
                i += 3
            elif RE_PROBE_TIMEOUT.match(probes_data[i]):
                # Its a timeout, so maybe name and IP have been parsed before
                # or maybe not. But it's Hop job to deal with it
                rtt = None
                i += 1
            else:
                ext = "i: %d\nprobes_data: %s\nname: %s\nip: %s\nrtt: %s\nanno: %s" % (
                i, probes_data, name, ip, rtt, anno)
                raise ParseError("Parse error \n%s" % ext)
            # Check for annotation
            try:
                if RE_PROBE_ANNOTATION.match(probes_data[i]):
                    anno = probes_data[i]
                    i += 1
            except IndexError:
                pass

            probe = Probe(name, ip, rtt, anno)
            hop.add_probe(probe)
        # Get hop
        return hop


def _print_output(hop):
    print "Hop : %s" % hop.idx, "\n"
    for prob in hop.probes:
        print "\tProbe %d : \t Destination:%s , Ip: %s , Rtt: %s" % (prob.num, prob.name, prob.ip, prob.rtt), "\n"


# TESTS:
def _feed_multi(data):
    data_split = data.split("\n")
    for line in data_split:
        if line != '':
            get_hop(line)


def _feed_single(line):
    get_hop(line)

# feed_single('4  10.251.11.32 (10.251.11.32)  3574.616 ms  0.153 ms')
# feed_multi(tr_data)

borisruna

This is the pyqt part:

from PyQt4 import QtGui, QtCore
import sys
import time
from libraries import trparse_rlt as tr


class MainApp(QtGui.QMainWindow):
    def __init__(self):
        super(MainApp, self).__init__()
        self.setupUI()

    def setupUI(self):
        self.setEnabled(False)
        self.init_elements()
        self.create_layouts()
        self.set_central_widget()
        self.setWindowTitle("traceroute")
        self.loaded()
        self.show()

    def set_central_widget(self):
        central_widget = QtGui.QWidget()
        central_widget.setLayout(self.main_layout)
        self.setCentralWidget(central_widget)

    def create_layouts(self):
        h1box = QtGui.QHBoxLayout()
        h1box.addWidget(self.cmbBox)
        h1box.addWidget(self.btn_trace)

        h2box = QtGui.QHBoxLayout()
        h2box.addWidget(self.tbl)

        self.main_layout = QtGui.QVBoxLayout()
        self.main_layout.addLayout(h1box)
        self.main_layout.addLayout(h2box)

    def init_elements(self):
        self.init_process()
        self.init_cmbBox()
        self.init_table()
        self.init_btn()

    def init_process(self):
        self.process = QtCore.QProcess(self)
        self.process.readyReadStandardOutput.connect(self.parse_n_populate)
        self.process.started.connect(lambda: self.btn_trace.setEnabled(False))
        self.process.finished.connect(lambda: self.btn_trace.setEnabled(True))



    def init_cmbBox(self):
        self.cmbBox = QtGui.QComboBox()

    def init_table(self):
        self.tbl = QtGui.QTextEdit()
        self.tbl.setMinimumWidth(800)

    def init_btn(self):
        self.btn_trace = QtGui.QPushButton("Trace")
        self.btn_trace.setMaximumWidth(self.btn_trace.sizeHint().width())
        self.btn_trace.clicked.connect(self.call_program)

    def call_program(self):
        self.process.start('traceroute www.google.com')

    def loaded(self):
        self.setEnabled(True)

    def parse_n_populate(self):
        trace_output = str(self.process.readAllStandardOutput())
        hop = tr.get_hop(trace_output)
        print "traceoutput is :", trace_output, "\nhop is : ", hop
        if hop:
            cursor = self.tbl.textCursor()
            cursor.movePosition(cursor.End)
            cursor.insertText(str(hop))
            self.tbl.ensureCursorVisible()


def main():
    app = QtGui.QApplication(sys.argv)
    mainapp = MainApp()
    sys.exit(app.exec_())


if __name__ == '__main__':
    main()

borisruna

Tried with :

        trace_output = str(self.process.readLine())

which seems to work better , but traceroute output stops sooner with no apparent reason .

Traceroute from terminal gives this : http://pastebin.com/u77Gww40

with the above implementation with readLine() : http://pastebin.com/CVEqvEPC ,i.e. traceroute stops at hop 23.

SGaist

Hi,

Unless you put that in a loop, readLine will only read one line each time. One thing you can do is split the content of trace_output on the carriage return and parse each element after that.

borisruna

@SGaist said in QProcess event queue:

Hi,

Unless you put that in a loop, readLine will only read one line each time. One thing you can do is split the content of trace_output on the carriage return and parse each element after that.

Sry didn't get notified about your reply. I have managed to solve this using subprocess module on a different thread. Would like to see it working with QProcess too though. Did you find any bug on my code?