Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return self._profiler.annotate(*args, **kw)
def dump_stats(self, *args, **kw):
warn('deprecated', DeprecationWarning)
return self._profiler.dump_stats(*args, **kw)
def print_stats(self, *args, **kw):
warn('deprecated', DeprecationWarning)
return self._profiler.print_stats(*args, **kw)
def iterSource(self, *args, **kw):
warn('deprecated', DeprecationWarning)
return self._profiler.iterSource(*args, **kw)
# BBB
StatisticalThread = StatisticThread
# profile/cProfile-like API (no sort parameter !)
def _run(threads, verbose, func_name, filename, *args, **kw):
if threads:
klass = ThreadProfile
else:
klass = Profile
prof = klass(verbose=verbose)
try:
try:
getattr(prof, func_name)(*args, **kw)
except SystemExit:
pass
finally:
if filename is None:
prof.print_stats()
def __call__(self, period=.001, single=True, group=None, name=None):
"""
Instanciate StatisticThread.
>>> s_profile = StatisticProfile()
>>> with s_profile(single=False):
>>> # Code to profile
Is equivalent to:
>>> s_profile = StatisticProfile()
>>> s_thread = StatisticThread(profiler=s_profile, single=False)
>>> with s_thread:
>>> # Code to profile
"""
return StatisticThread(
profiler=self, period=period, single=single, group=group,
name=name,
)
args.append(options.script)
args.extend(options.argv)
runner_method_kw = {
'module': options.module,
'argv': args,
}
runner_method_id = 'runmodule'
if options.format is None:
if _isCallgrindName(options.out):
options.format = 'callgrind'
else:
options.format = 'text'
relative_path = options.format == 'callgrind' and options.zipfile
if options.statistic:
prof = StatisticalProfile()
runner = StatisticalThread(
profiler=prof,
period=options.statistic,
single=not options.threads,
)
else:
if options.threads:
klass = ThreadProfile
else:
klass = Profile
prof = runner = klass(verbose=options.verbose)
try:
getattr(runner, runner_method_id)(**runner_method_kw)
finally:
if options.out == '-':
out = EncodeOrReplaceWriter(sys.stdout)
close = lambda: None
def benchmark():
start = time()
stop = start + 1
for index, _ in enumerate(F()):
now = time()
if now > stop:
break
return index / (now - start)
raw = benchmark()
with pprofile.Profile():
single = benchmark()
with pprofile.ThreadProfile():
threaded = benchmark()
with pprofile.StatisticThread():
statistic = benchmark()
for caption, value in (
('single', single),
('threaded', threaded),
('statistic', statistic),
):
print('%s speed: %.2f%%' % (caption, value * 100 / raw))
The smaller, the more profiling overhead, but the faster results
become meaningful.
The larger, the less profiling overhead, but requires long profiling
session to get meaningful results.
single (bool)
Profile only the thread which created this instance.
group, name
See Python's threading.Thread API.
"""
if profiler is None:
profiler = StatisticProfile()
if single:
self._test = lambda x, ident=threading.current_thread().ident: ident == x
else:
self._test = None
super(StatisticThread, self).__init__(
group=group,
name=name,
)
self._stop_event = threading.Event()
self._period = period
self._profiler = profiler
profiler.total_time = 0
self.daemon = True
self.clean_exit = False
def start(self):
self.clean_exit = False
self._can_run = True
self._start_time = time()
super(StatisticThread, self).start()