baseapp.py
241 lines
| 8.2 KiB
| text/x-python
|
PythonLexer
|
r3604 | #!/usr/bin/env python | ||
# encoding: utf-8 | ||||
""" | ||||
|
r4018 | The Base Application class for IPython.parallel apps | ||
Authors: | ||||
* Brian Granger | ||||
* Min RK | ||||
|
r3604 | """ | ||
#----------------------------------------------------------------------------- | ||||
|
r4018 | # Copyright (C) 2008-2011 The IPython Development Team | ||
|
r3604 | # | ||
# Distributed under the terms of the BSD License. The full license is in | ||||
# the file COPYING, distributed as part of this software. | ||||
#----------------------------------------------------------------------------- | ||||
#----------------------------------------------------------------------------- | ||||
# Imports | ||||
#----------------------------------------------------------------------------- | ||||
from __future__ import with_statement | ||||
import os | ||||
|
r3631 | import logging | ||
import re | ||||
|
r3604 | import sys | ||
|
r3846 | from subprocess import Popen, PIPE | ||
|
r3604 | from IPython.core import release | ||
|
r3992 | from IPython.core.crashhandler import CrashHandler | ||
|
r4023 | from IPython.core.application import ( | ||
|
r3992 | BaseIPythonApplication, | ||
base_aliases as base_ip_aliases, | ||||
base_flags as base_ip_flags | ||||
|
r3604 | ) | ||
|
r3992 | from IPython.utils.path import expand_path | ||
|
r3991 | from IPython.utils.traitlets import Unicode, Bool, Instance, Dict, List | ||
|
r3604 | |||
#----------------------------------------------------------------------------- | ||||
# Module errors | ||||
#----------------------------------------------------------------------------- | ||||
class PIDFileError(Exception): | ||||
pass | ||||
#----------------------------------------------------------------------------- | ||||
# Crash handler for this application | ||||
#----------------------------------------------------------------------------- | ||||
|
r3992 | class ParallelCrashHandler(CrashHandler): | ||
|
r3604 | """sys.excepthook for IPython itself, leaves a detailed report on disk.""" | ||
def __init__(self, app): | ||||
|
r3985 | contact_name = release.authors['Min'][0] | ||
contact_email = release.authors['Min'][1] | ||||
|
r3604 | bug_tracker = 'http://github.com/ipython/ipython/issues' | ||
|
r3992 | super(ParallelCrashHandler,self).__init__( | ||
|
r3604 | app, contact_name, contact_email, bug_tracker | ||
) | ||||
#----------------------------------------------------------------------------- | ||||
# Main application | ||||
#----------------------------------------------------------------------------- | ||||
|
r3992 | base_aliases = {} | ||
base_aliases.update(base_ip_aliases) | ||||
base_aliases.update({ | ||||
|
r4214 | 'profile-dir' : 'ProfileDir.location', | ||
'work-dir' : 'BaseParallelApplication.work_dir', | ||||
'log-to-file' : 'BaseParallelApplication.log_to_file', | ||||
'clean-logs' : 'BaseParallelApplication.clean_logs', | ||||
'log-url' : 'BaseParallelApplication.log_url', | ||||
|
r3992 | }) | ||
|
r3985 | |||
base_flags = { | ||||
|
r3994 | 'log-to-file' : ( | ||
{'BaseParallelApplication' : {'log_to_file' : True}}, | ||||
"send log output to a file" | ||||
) | ||||
|
r3985 | } | ||
|
r3992 | base_flags.update(base_ip_flags) | ||
|
r3604 | |||
|
r3992 | class BaseParallelApplication(BaseIPythonApplication): | ||
"""The base Application for IPython.parallel apps | ||||
Principle extensions to BaseIPyythonApplication: | ||||
* work_dir | ||||
* remote logging via pyzmq | ||||
* IOLoop instance | ||||
|
r3604 | """ | ||
|
r3992 | crash_handler_class = ParallelCrashHandler | ||
|
r3986 | |||
def _log_level_default(self): | ||||
# temporarily override default_log_level to INFO | ||||
return logging.INFO | ||||
|
r3985 | |||
work_dir = Unicode(os.getcwdu(), config=True, | ||||
help='Set the working dir for the process.' | ||||
) | ||||
def _work_dir_changed(self, name, old, new): | ||||
self.work_dir = unicode(expand_path(new)) | ||||
log_to_file = Bool(config=True, | ||||
help="whether to log to a file") | ||||
|
r3991 | clean_logs = Bool(False, config=True, | ||
|
r3985 | help="whether to cleanup old logfiles before starting") | ||
|
r3604 | |||
|
r3991 | log_url = Unicode('', config=True, | ||
|
r3988 | help="The ZMQ URL of the iplogger to aggregate logging.") | ||
|
r3604 | |||
|
r3992 | def _config_files_default(self): | ||
return ['ipcontroller_config.py', 'ipengine_config.py', 'ipcluster_config.py'] | ||||
|
r3986 | |||
loop = Instance('zmq.eventloop.ioloop.IOLoop') | ||||
def _loop_default(self): | ||||
from zmq.eventloop.ioloop import IOLoop | ||||
return IOLoop.instance() | ||||
|
r3985 | |||
aliases = Dict(base_aliases) | ||||
flags = Dict(base_flags) | ||||
|
r3986 | |||
def initialize(self, argv=None): | ||||
"""initialize the app""" | ||||
|
r3992 | super(BaseParallelApplication, self).initialize(argv) | ||
|
r3986 | self.to_work_dir() | ||
|
r3989 | self.reinit_logging() | ||
|
r3986 | |||
|
r3604 | def to_work_dir(self): | ||
|
r3985 | wd = self.work_dir | ||
if unicode(wd) != os.getcwdu(): | ||||
|
r3604 | os.chdir(wd) | ||
self.log.info("Changing to working dir: %s" % wd) | ||||
|
r3989 | # This is the working dir by now. | ||
sys.path.insert(0, '') | ||||
|
r3604 | |||
|
r3986 | def reinit_logging(self): | ||
# Remove old log files | ||||
|
r3992 | log_dir = self.profile_dir.log_dir | ||
|
r3986 | if self.clean_logs: | ||
for f in os.listdir(log_dir): | ||||
if re.match(r'%s-\d+\.(log|err|out)'%self.name,f): | ||||
os.remove(os.path.join(log_dir, f)) | ||||
if self.log_to_file: | ||||
# Start logging to the new log file | ||||
log_filename = self.name + u'-' + str(os.getpid()) + u'.log' | ||||
logfile = os.path.join(log_dir, log_filename) | ||||
open_log_file = open(logfile, 'w') | ||||
else: | ||||
open_log_file = None | ||||
if open_log_file is not None: | ||||
self.log.removeHandler(self._log_handler) | ||||
self._log_handler = logging.StreamHandler(open_log_file) | ||||
self._log_formatter = logging.Formatter("[%(name)s] %(message)s") | ||||
self._log_handler.setFormatter(self._log_formatter) | ||||
self.log.addHandler(self._log_handler) | ||||
|
r3604 | |||
def write_pid_file(self, overwrite=False): | ||||
"""Create a .pid file in the pid_dir with my pid. | ||||
This must be called after pre_construct, which sets `self.pid_dir`. | ||||
This raises :exc:`PIDFileError` if the pid file exists already. | ||||
""" | ||||
|
r3992 | pid_file = os.path.join(self.profile_dir.pid_dir, self.name + u'.pid') | ||
|
r3604 | if os.path.isfile(pid_file): | ||
pid = self.get_pid_from_file() | ||||
if not overwrite: | ||||
raise PIDFileError( | ||||
'The pid file [%s] already exists. \nThis could mean that this ' | ||||
'server is already running with [pid=%s].' % (pid_file, pid) | ||||
) | ||||
with open(pid_file, 'w') as f: | ||||
self.log.info("Creating pid file: %s" % pid_file) | ||||
f.write(repr(os.getpid())+'\n') | ||||
def remove_pid_file(self): | ||||
"""Remove the pid file. | ||||
This should be called at shutdown by registering a callback with | ||||
:func:`reactor.addSystemEventTrigger`. This needs to return | ||||
``None``. | ||||
""" | ||||
|
r3992 | pid_file = os.path.join(self.profile_dir.pid_dir, self.name + u'.pid') | ||
|
r3604 | if os.path.isfile(pid_file): | ||
try: | ||||
self.log.info("Removing pid file: %s" % pid_file) | ||||
os.remove(pid_file) | ||||
except: | ||||
self.log.warn("Error removing the pid file: %s" % pid_file) | ||||
def get_pid_from_file(self): | ||||
"""Get the pid from the pid file. | ||||
If the pid file doesn't exist a :exc:`PIDFileError` is raised. | ||||
""" | ||||
|
r3992 | pid_file = os.path.join(self.profile_dir.pid_dir, self.name + u'.pid') | ||
|
r3604 | if os.path.isfile(pid_file): | ||
with open(pid_file, 'r') as f: | ||||
|
r4032 | s = f.read().strip() | ||
try: | ||||
pid = int(s) | ||||
except: | ||||
raise PIDFileError("invalid pid file: %s (contents: %r)"%(pid_file, s)) | ||||
|
r3604 | return pid | ||
else: | ||||
raise PIDFileError('pid file not found: %s' % pid_file) | ||||
|
r3846 | |||
def check_pid(self, pid): | ||||
if os.name == 'nt': | ||||
try: | ||||
import ctypes | ||||
# returns 0 if no such process (of ours) exists | ||||
# positive int otherwise | ||||
p = ctypes.windll.kernel32.OpenProcess(1,0,pid) | ||||
except Exception: | ||||
self.log.warn( | ||||
"Could not determine whether pid %i is running via `OpenProcess`. " | ||||
" Making the likely assumption that it is."%pid | ||||
) | ||||
return True | ||||
return bool(p) | ||||
else: | ||||
try: | ||||
p = Popen(['ps','x'], stdout=PIPE, stderr=PIPE) | ||||
output,_ = p.communicate() | ||||
except OSError: | ||||
self.log.warn( | ||||
"Could not determine whether pid %i is running via `ps x`. " | ||||
" Making the likely assumption that it is."%pid | ||||
) | ||||
return True | ||||
pids = map(int, re.findall(r'^\W*\d+', output, re.MULTILINE)) | ||||
return pid in pids | ||||