Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
tlocal._tag = 0
yappi.set_tag_callback(tag_cbk)
yappi.start()
ts = []
for i in range(_TCOUNT):
t = threading.Thread(target=a, args=(i + 1, ))
ts.append(t)
for t in ts:
t.start()
for t in ts:
t.join()
yappi.stop()
traces = yappi.get_func_stats()
t1 = '''
..p/yappi/tests/utils.py:134 burn_io 20 0.000638 2.004059 0.100203
'''
self.assert_traces_almost_equal(t1, traces)
traces = yappi.get_func_stats(filter={'tag': 3})
t1 = '''
..p/yappi/tests/utils.py:134 burn_io 1 0.000038 0.100446 0.100446
'''
self.assert_traces_almost_equal(t1, traces)
def run_with_yappi(func, *args, **kwargs):
yappi.start()
func(*args, **kwargs)
yappi.stop()
outfile = "callgrind.mitmdump-{}-c{}".format(clock_type, concurrency)
a = ApacheBenchThread(concurrency)
a.start()
if profiler == "yappi":
yappi.set_clock_type(clock_type)
yappi.start(builtins=True)
print("Start mitmdump...")
mitmdump(["-k", "-q", "-S", "1024example"])
print("mitmdump stopped.")
print("Save profile information...")
if profiler == "yappi":
yappi.stop()
stats = yappi.get_func_stats()
stats.save(outfile, type='callgrind')
print("Done.")
def test_recursive_coroutine(self):
@asyncio.coroutine
def a(n):
if n <= 0:
return
yield from async_sleep(0.1)
burn_cpu(0.1)
yield from a(n - 1)
yield from a(n - 2)
yappi.set_clock_type("cpu")
yappi.start()
asyncio.get_event_loop().run_until_complete(a(3))
yappi.stop()
r1 = '''
..p/yappi/tests/test_asyncio.py:11 a 9/1 0.000124 0.400667 0.044519
../yappi/tests/utils.py:126 burn_cpu 4 0.000000 0.400099 0.100025
async_sleep 4 0.000000 0.000444 0.000111
'''
stats = yappi.get_func_stats()
self.assert_traces_almost_equal(r1, stats)
def main():
try:
if conf.ENABLE_PROFILING:
yappi.start()
launcher.main(sys.argv[1:])
yappi.stop()
else:
launcher.main(sys.argv[1:])
except KeyboardInterrupt:
if conf.ENABLE_PROFILING:
yappi.stop()
print('Yappi result (func stats) ======================')
yappi.get_func_stats().print_all()
print('Yappi result (thread stats) ======================')
yappi.get_thread_stats().print_all()
def _capture_profile(fname=''):
if not fname:
yappi.set_clock_type('cpu')
# We need to set context to greenlet to profile greenlets
# https://bitbucket.org/sumerc/yappi/pull-requests/3
yappi.set_context_id_callback(
lambda: id(greenlet.getcurrent()))
yappi.set_context_name_callback(
lambda: greenlet.getcurrent().__class__.__name__)
yappi.start()
else:
yappi.stop()
stats = yappi.get_func_stats()
# User should provide filename. This file with a suffix .prof
# will be created in temp directory.
try:
stats_file = os.path.join(tempfile.gettempdir(), fname + '.prof')
stats.save(stats_file, "pstat")
except Exception as e:
print("Error while saving the trace stats ", str(e))
finally:
yappi.clear_stats()
def stop_profiler(self):
"""
Stop yappi and write the stats to the output directory.
Return the path of the yappi statistics file.
"""
if not self.profiler_running:
raise RuntimeError("Profiler is not running")
if not HAS_YAPPI:
raise RuntimeError("Yappi cannot be found. Plase install the yappi library using your preferred package "
"manager and restart Tribler afterwards.")
yappi.stop()
yappi_stats = yappi.get_func_stats()
yappi_stats.sort("tsub")
log_dir = os.path.join(self.session.config.get_state_dir(), 'logs')
file_path = os.path.join(log_dir, 'yappi_%s.stats' % self.profiler_start_time)
# Make the log directory if it does not exist
if not os.path.exists(log_dir):
os.makedirs(log_dir)
yappi_stats.save(file_path, type='callgrind')
yappi.clear_stats()
self.profiler_running = False
return file_path
if __name__ == "__main__":
import yappi
from LightInstallation import main
yappi.start()
main(['','config/10kConfig.xml'])
yappi.stop()
yappi.print_stats()