def runReactorWithLogging(config, oldstdout, oldstderr):
from twisted.internet import reactor
try:
if config['profile']:
if not config['nothotshot']:
runWithHotshot(reactor, config)
else:
runWithProfiler(reactor, config)
elif config['debug']:
sys.stdout = oldstdout
sys.stderr = oldstderr
if runtime.platformType == 'posix':
signal.signal(signal.SIGUSR2, lambda *args: pdb.set_trace())
signal.signal(signal.SIGINT, lambda *args: pdb.set_trace())
fixPdb()
pdb.runcall(reactor.run)
else:
reactor.run()
except:
if config['nodaemon']:
file = oldstdout
else:
file = open("TWISTD-CRASH.log",'a')
traceback.print_exc(file=file)
file.flush()
python类runcall()的实例源码
def pdb_run(func):
"""
This decorator inserts a pdb session on top of the call-stack into a
function.
This can be used like so:
.. code-block:: python
@pdb_run
def some_function_to_debug(...):
"""
@wraps(func)
def wrapper(*args, **kw):
pdb.runcall(func, *args, **kw)
return wrapper
def test_pdb_issue4201(self):
test_src = textwrap.dedent("""\
def f():
pass
import pdb
pdb.runcall(f)
""")
with temp_dir() as d:
script_name = make_script(d, 'script', test_src)
p = spawn_python(script_name)
p.stdin.write('l\n')
data = kill_python(p)
self.assertIn(script_name, data)
zip_name, run_name = make_zip_script(d, "test_zip",
script_name, '__main__.py')
p = spawn_python(zip_name)
p.stdin.write('l\n')
data = kill_python(p)
self.assertIn(run_name, data)
test_zipimport_support.py 文件源码
项目：python2-tracer
作者: extremecoders-re
项目源码
文件源码
阅读 22
收藏 0
点赞 0
评论 0
def test_pdb_issue4201(self):
test_src = textwrap.dedent("""\
def f():
pass
import pdb
pdb.runcall(f)
""")
with temp_dir() as d:
script_name = make_script(d, 'script', test_src)
p = spawn_python(script_name)
p.stdin.write('l\n')
data = kill_python(p)
self.assertIn(script_name, data)
zip_name, run_name = make_zip_script(d, "test_zip",
script_name, '__main__.py')
p = spawn_python(zip_name)
p.stdin.write('l\n')
data = kill_python(p)
self.assertIn(run_name, data)
def runReactorWithLogging(config, oldstdout, oldstderr):
from twisted.internet import reactor
try:
if config['profile']:
if not config['nothotshot']:
runWithHotshot(reactor, config)
else:
runWithProfiler(reactor, config)
elif config['debug']:
sys.stdout = oldstdout
sys.stderr = oldstderr
if runtime.platformType == 'posix':
signal.signal(signal.SIGUSR2, lambda *args: pdb.set_trace())
signal.signal(signal.SIGINT, lambda *args: pdb.set_trace())
fixPdb()
pdb.runcall(reactor.run)
else:
reactor.run()
except:
if config['nodaemon']:
file = oldstdout
else:
file = open("TWISTD-CRASH.log",'a')
traceback.print_exc(file=file)
file.flush()
def test_pdb_issue4201(self):
test_src = textwrap.dedent("""\
def f():
pass
import pdb
pdb.runcall(f)
""")
with temp_dir() as d:
script_name = make_script(d, 'script', test_src)
p = spawn_python(script_name)
p.stdin.write('l\n')
data = kill_python(p)
self.assertIn(script_name, data)
zip_name, run_name = make_zip_script(d, "test_zip",
script_name, '__main__.py')
p = spawn_python(zip_name)
p.stdin.write('l\n')
data = kill_python(p)
self.assertIn(run_name, data)
def test_pdb_issue4201(self):
test_src = textwrap.dedent("""\
def f():
pass
import pdb
pdb.runcall(f)
""")
with temp_dir() as d:
script_name = make_script(d, 'script', test_src)
p = spawn_python(script_name)
p.stdin.write('l\n')
data = kill_python(p)
self.assertIn(script_name, data)
zip_name, run_name = make_zip_script(d, "test_zip",
script_name, '__main__.py')
p = spawn_python(zip_name)
p.stdin.write('l\n')
data = kill_python(p)
self.assertIn(run_name, data)
def run(self, reactor):
"""
Run reactor under the standard profiler.
"""
try:
import profile
except ImportError as e:
self._reportImportError("profile", e)
p = profile.Profile()
p.runcall(reactor.run)
if self.saveStats:
p.dump_stats(self.profileOutput)
else:
tmp, sys.stdout = sys.stdout, open(self.profileOutput, 'a')
try:
p.print_stats()
finally:
sys.stdout, tmp = tmp, sys.stdout
tmp.close()
def run(self, reactor):
"""
Run reactor under the cProfile profiler.
"""
try:
import cProfile
import pstats
except ImportError as e:
self._reportImportError("cProfile", e)
p = cProfile.Profile()
p.runcall(reactor.run)
if self.saveStats:
p.dump_stats(self.profileOutput)
else:
with open(self.profileOutput, 'w') as stream:
s = pstats.Stats(p, stream=stream)
s.strip_dirs()
s.sort_stats(-1)
s.print_stats()
def runWithProfiler(reactor, config):
"""Run reactor under standard profiler."""
try:
import profile
except ImportError, e:
s = "Failed to import module profile: %s" % e
s += """
This is most likely caused by your operating system not including
profile.py due to it being non-free. Either do not use the option
--profile, or install profile.py; your operating system vendor
may provide it in a separate package.
"""
traceback.print_exc(file=log.logfile)
log.msg(s)
log.deferr()
sys.exit('\n' + s + '\n')
p = profile.Profile()
p.runcall(reactor.run)
if config['savestats']:
p.dump_stats(config['profile'])
else:
# XXX - omfg python sucks
tmp, sys.stdout = sys.stdout, open(config['profile'], 'a')
p.print_stats()
sys.stdout, tmp = tmp, sys.stdout
tmp.close()
def runWithHotshot(reactor, config):
"""Run reactor under hotshot profiler."""
try:
import hotshot.stats
except ImportError, e:
s = "Failed to import module hotshot: %s" % e
s += """
This is most likely caused by your operating system not including
profile.py due to it being non-free. Either do not use the option
--profile, or install profile.py; your operating system vendor
may provide it in a separate package.
"""
traceback.print_exc(file=log.logfile)
log.msg(s)
log.deferr()
sys.exit('\n' + s + '\n')
# this writes stats straight out
p = hotshot.Profile(config["profile"])
p.runcall(reactor.run)
if config["savestats"]:
# stats are automatically written to file, nothing to do
return
else:
s = hotshot.stats.load(config["profile"])
s.strip_dirs()
s.sort_stats(-1)
tmp, sys.stdout = sys.stdout, open(config['profile'], 'w')
s.print_stats()
sys.stdout, tmp = tmp, sys.stdout
tmp.close()
def runWithProfiler(reactor, config):
"""Run reactor under standard profiler."""
try:
import profile
except ImportError, e:
s = "Failed to import module profile: %s" % e
s += """
This is most likely caused by your operating system not including
profile.py due to it being non-free. Either do not use the option
--profile, or install profile.py; your operating system vendor
may provide it in a separate package.
"""
traceback.print_exc(file=log.logfile)
log.msg(s)
log.deferr()
sys.exit('\n' + s + '\n')
p = profile.Profile()
p.runcall(reactor.run)
if config['savestats']:
p.dump_stats(config['profile'])
else:
# XXX - omfg python sucks
tmp, sys.stdout = sys.stdout, open(config['profile'], 'a')
p.print_stats()
sys.stdout, tmp = tmp, sys.stdout
tmp.close()
def runWithHotshot(reactor, config):
"""Run reactor under hotshot profiler."""
try:
import hotshot.stats
except ImportError, e:
s = "Failed to import module hotshot: %s" % e
s += """
This is most likely caused by your operating system not including
profile.py due to it being non-free. Either do not use the option
--profile, or install profile.py; your operating system vendor
may provide it in a separate package.
"""
traceback.print_exc(file=log.logfile)
log.msg(s)
log.deferr()
sys.exit('\n' + s + '\n')
# this writes stats straight out
p = hotshot.Profile(config["profile"])
p.runcall(reactor.run)
if config["savestats"]:
# stats are automatically written to file, nothing to do
return
else:
s = hotshot.stats.load(config["profile"])
s.strip_dirs()
s.sort_stats(-1)
tmp, sys.stdout = sys.stdout, open(config['profile'], 'w')
s.print_stats()
sys.stdout, tmp = tmp, sys.stdout
tmp.close()
def really_start(main=None):
"""Initializes flag values, and calls main with non-flag arguments.
Only non-flag arguments are passed to main(). The return value of main() is
used as the exit status.
Args:
main: Main function to run with the list of non-flag arguments, or None
so that sys.modules['__main__'].main is to be used.
"""
argv = RegisterAndParseFlagsWithUsage()
if main is None:
main = sys.modules['__main__'].main
try:
if FLAGS.run_with_pdb:
sys.exit(pdb.runcall(main, argv))
else:
if FLAGS.run_with_profiling or FLAGS.profile_file:
# Avoid import overhead since most apps (including performance-sensitive
# ones) won't be run with profiling.
import atexit
if FLAGS.use_cprofile_for_profiling:
import cProfile as profile
else:
import profile
profiler = profile.Profile()
if FLAGS.profile_file:
atexit.register(profiler.dump_stats, FLAGS.profile_file)
else:
atexit.register(profiler.print_stats)
retval = profiler.runcall(main, argv)
sys.exit(retval)
else:
sys.exit(main(argv))
except UsageError, error:
usage(shorthelp=1, detailed_error=error, exitcode=error.exitcode)
except:
if FLAGS.pdb_post_mortem:
traceback.print_exc()
pdb.post_mortem()
raise
def really_start(main=None):
"""Initializes flag values, and calls main with non-flag arguments.
Only non-flag arguments are passed to main(). The return value of main() is
used as the exit status.
Args:
main: Main function to run with the list of non-flag arguments, or None
so that sys.modules['__main__'].main is to be used.
"""
argv = RegisterAndParseFlagsWithUsage()
if main is None:
main = sys.modules['__main__'].main
try:
if FLAGS.run_with_pdb:
sys.exit(pdb.runcall(main, argv))
else:
if FLAGS.run_with_profiling or FLAGS.profile_file:
# Avoid import overhead since most apps (including performance-sensitive
# ones) won't be run with profiling.
import atexit
if FLAGS.use_cprofile_for_profiling:
import cProfile as profile
else:
import profile
profiler = profile.Profile()
if FLAGS.profile_file:
atexit.register(profiler.dump_stats, FLAGS.profile_file)
else:
atexit.register(profiler.print_stats)
retval = profiler.runcall(main, argv)
sys.exit(retval)
else:
sys.exit(main(argv))
except UsageError, error:
usage(shorthelp=1, detailed_error=error, exitcode=error.exitcode)
except:
if FLAGS.pdb_post_mortem:
traceback.print_exc()
pdb.post_mortem()
raise
def runReactorWithLogging(config, oldstdout, oldstderr, profiler=None,
reactor=None):
"""
Start the reactor, using profiling if specified by the configuration, and
log any error happening in the process.
@param config: configuration of the twistd application.
@type config: L{ServerOptions}
@param oldstdout: initial value of C{sys.stdout}.
@type oldstdout: C{file}
@param oldstderr: initial value of C{sys.stderr}.
@type oldstderr: C{file}
@param profiler: object used to run the reactor with profiling.
@type profiler: L{AppProfiler}
@param reactor: The reactor to use. If L{None}, the global reactor will
be used.
"""
if reactor is None:
from twisted.internet import reactor
try:
if config['profile']:
if profiler is not None:
profiler.run(reactor)
elif config['debug']:
sys.stdout = oldstdout
sys.stderr = oldstderr
if runtime.platformType == 'posix':
signal.signal(signal.SIGUSR2, lambda *args: pdb.set_trace())
signal.signal(signal.SIGINT, lambda *args: pdb.set_trace())
fixPdb()
pdb.runcall(reactor.run)
else:
reactor.run()
except:
close = False
if config['nodaemon']:
file = oldstdout
else:
file = open("TWISTD-CRASH.log", "a")
close = True
try:
traceback.print_exc(file=file)
file.flush()
finally:
if close:
file.close()
