Mercurial > hg
view contrib/catapipe.py @ 39408:c7f213d7f4d2
py3: add new passing test to whitelist found by buildbot
Differential Revision: https://phab.mercurial-scm.org/D4451
author | Pulkit Goyal <pulkit@yandex-team.ru> |
---|---|
date | Mon, 03 Sep 2018 13:56:53 +0300 |
parents | 9a81f126f9fa |
children | e9706686451b |
line wrap: on
line source
#!/usr/bin/env python3 # # Copyright 2018 Google LLC. # # This software may be used and distributed according to the terms of the # GNU General Public License version 2 or any later version. """Tool read primitive events from a pipe to produce a catapult trace. For now the event stream supports START $SESSIONID ... and END $SESSIONID ... events. Everything after the SESSIONID (which must not contain spaces) is used as a label for the event. Events are timestamped as of when they arrive in this process and are then used to produce catapult traces that can be loaded in Chrome's about:tracing utility. It's important that the event stream *into* this process stay simple, because we have to emit it from the shell scripts produced by run-tests.py. Typically you'll want to place the path to the named pipe in the HGCATAPULTSERVERPIPE environment variable, which both run-tests and hg understand. """ from __future__ import absolute_import, print_function import argparse import datetime import json import os _TYPEMAP = { 'START': 'B', 'END': 'E', } _threadmap = {} def main(): parser = argparse.ArgumentParser() parser.add_argument('pipe', type=str, nargs=1, help='Path of named pipe to create and listen on.') parser.add_argument('output', default='trace.json', type=str, nargs='?', help='Path of named pipe to create and listen on.') parser.add_argument('--debug', default=False, action='store_true', help='Print useful debug messages') args = parser.parse_args() fn = args.pipe[0] os.mkfifo(fn) try: with open(fn) as f, open(args.output, 'w') as out: out.write('[\n') start = datetime.datetime.now() while True: ev = f.readline().strip() if not ev: continue now = datetime.datetime.now() if args.debug: print(ev) verb, session, label = ev.split(' ', 2) if session not in _threadmap: _threadmap[session] = len(_threadmap) pid = _threadmap[session] ts_micros = (now - start).total_seconds() * 1000000 out.write(json.dumps( { "name": label, "cat": "misc", "ph": _TYPEMAP[verb], "ts": ts_micros, "pid": pid, "tid": 1, "args": {} })) out.write(',\n') finally: os.unlink(fn) if __name__ == '__main__': main()