Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def run_and_get_thread_stats(func, *args, **kwargs):
run_with_yappi(func, *args, **kwargs)
return yappi.get_thread_stats()
def run_and_get_thread_stats(func, *args, **kwargs):
run_with_yappi(func, *args, **kwargs)
return yappi.get_thread_stats()
c.start()
c = WorkerThread2()
c.start()
c = IOThread2()
c.start()
yappi.start()
time.sleep(1.0)
yappi.get_func_stats().print_all()
yappi.get_thread_stats().print_all()
def test_start_flags(self):
self.assertEqual(_yappi._get_start_flags(), None)
yappi.start()
def a():
pass
a()
self.assertEqual(_yappi._get_start_flags()["profile_builtins"], 0)
self.assertEqual(_yappi._get_start_flags()["profile_multithread"], 1)
self.assertEqual(len(yappi.get_thread_stats()), 1)
# Start in context 0.
self.context_id = 0
yappi.start()
time.sleep(0.08)
# Switch to context 1.
self.context_id = 1
time.sleep(0.05)
# Switch back to context 0.
self.context_id = 0
time.sleep(0.07)
yappi.stop()
t_stats = yappi.get_thread_stats().sort('id', 'ascending')
self.assertEqual(2, len(t_stats))
self.assertEqual(0, t_stats[0].id)
self.assertEqual(2, t_stats[0].sched_count)
self.assertTrue(0.15 < t_stats[0].ttot < 0.3)
self.assertEqual(1, t_stats[1].id)
self.assertEqual(1, t_stats[1].sched_count)
# Context 1 was first scheduled 0.08 sec after context 0.
self.assertTrue(0.1 < t_stats[1].ttot < 0.2)
recentPath = PreferencesManager().get('shell', 'recent sorted profile stats')
if recentPath is None:
defaultPath = os.path.join(os.path.expanduser('~'), filename)
else:
defaultPath = os.path.join(os.path.split(recentPath)[0], filename)
stats_path, _filter = QFileDialog.getSaveFileName(
self, "Export sorted stats text", defaultPath, "Text files (*.txt)",
options=QFileDialog.Options(QFileDialog.DontUseNativeDialog))
if stats_path:
PreferencesManager().set('shell', 'recent sorted profile stats', stats_path)
# Export the yappi stats to builtin pstats format,
# since pstats provides nicer printing IMHO
stats = yappi.get_thread_stats()
stats.sort(sortby)
with open(stats_path, 'w') as f:
stats.print_all(f)
logger.info("Printed thread stats to file: {}".format(stats_path))
# As a convenience, go ahead and open it.
QDesktopServices.openUrl(QUrl.fromLocalFile(stats_path))
if m.name in ["map", "console"]:
if hasattr(m, 'unload'):
try:
m.unload()
except Exception:
pass
reload(m)
m.init(mpstate)
else:
mpstate.status.exit = True
sys.exit(1)
if opts.profile:
yappi.get_func_stats().print_all()
yappi.get_thread_stats().print_all()
#this loop executes after leaving the above loop and is for cleanup on exit
for (m,pm) in mpstate.modules:
if hasattr(m, 'unload'):
print("Unloading module %s" % m.name)
m.unload()
sys.exit(1)
def injection(thread_stack_info=None, **kwargs):
import threading
import sys
result = []
if thread_stack_info is None:
import inspect
yi = {}
try:
import yappi
if not yappi.is_running():
yappi.start()
for d in yappi.get_thread_stats():
yi[d[2]] = (d[3], d[4])
except:
pass
for t in threading.enumerate():
if not t.name.startswith('__pptop_injection'):
try:
target = '{}.{}'.format(
t._target.__module__, t._target.__qualname__ if hasattr(
t._target, '__qualname__') else t._target.__name__)
except:
target = None
y = yi.get(t.ident)
r = (t.ident, t.daemon, t.name, target, y[0] if y else 0,
y[1] if y else 0)
try:
x = inspect.getframeinfo(sys._current_frames()[t.ident])
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()