|
|
# encoding: utf-8
|
|
|
"""
|
|
|
The Base Application class for IPython.parallel apps
|
|
|
|
|
|
Authors:
|
|
|
|
|
|
* Brian Granger
|
|
|
* Min RK
|
|
|
|
|
|
"""
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
|
# Copyright (C) 2008-2011 The IPython Development Team
|
|
|
#
|
|
|
# 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
|
|
|
import logging
|
|
|
import re
|
|
|
import sys
|
|
|
|
|
|
from subprocess import Popen, PIPE
|
|
|
|
|
|
from IPython.config.application import catch_config_error, LevelFormatter
|
|
|
from IPython.core import release
|
|
|
from IPython.core.crashhandler import CrashHandler
|
|
|
from IPython.core.application import (
|
|
|
BaseIPythonApplication,
|
|
|
base_aliases as base_ip_aliases,
|
|
|
base_flags as base_ip_flags
|
|
|
)
|
|
|
from IPython.utils.path import expand_path
|
|
|
|
|
|
from IPython.utils.traitlets import Unicode, Bool, Instance, Dict, List
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
|
# Module errors
|
|
|
#-----------------------------------------------------------------------------
|
|
|
|
|
|
class PIDFileError(Exception):
|
|
|
pass
|
|
|
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
|
# Crash handler for this application
|
|
|
#-----------------------------------------------------------------------------
|
|
|
|
|
|
class ParallelCrashHandler(CrashHandler):
|
|
|
"""sys.excepthook for IPython itself, leaves a detailed report on disk."""
|
|
|
|
|
|
def __init__(self, app):
|
|
|
contact_name = release.authors['Min'][0]
|
|
|
contact_email = release.author_email
|
|
|
bug_tracker = 'https://github.com/ipython/ipython/issues'
|
|
|
super(ParallelCrashHandler,self).__init__(
|
|
|
app, contact_name, contact_email, bug_tracker
|
|
|
)
|
|
|
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
|
# Main application
|
|
|
#-----------------------------------------------------------------------------
|
|
|
base_aliases = {}
|
|
|
base_aliases.update(base_ip_aliases)
|
|
|
base_aliases.update({
|
|
|
'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',
|
|
|
'cluster-id' : 'BaseParallelApplication.cluster_id',
|
|
|
})
|
|
|
|
|
|
base_flags = {
|
|
|
'log-to-file' : (
|
|
|
{'BaseParallelApplication' : {'log_to_file' : True}},
|
|
|
"send log output to a file"
|
|
|
)
|
|
|
}
|
|
|
base_flags.update(base_ip_flags)
|
|
|
|
|
|
class BaseParallelApplication(BaseIPythonApplication):
|
|
|
"""The base Application for IPython.parallel apps
|
|
|
|
|
|
Principle extensions to BaseIPyythonApplication:
|
|
|
|
|
|
* work_dir
|
|
|
* remote logging via pyzmq
|
|
|
* IOLoop instance
|
|
|
"""
|
|
|
|
|
|
crash_handler_class = ParallelCrashHandler
|
|
|
|
|
|
def _log_level_default(self):
|
|
|
# temporarily override default_log_level to INFO
|
|
|
return logging.INFO
|
|
|
|
|
|
def _log_format_default(self):
|
|
|
"""override default log format to include time"""
|
|
|
return u"%(asctime)s.%(msecs).03d [%(name)s]%(highlevel)s %(message)s"
|
|
|
|
|
|
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")
|
|
|
|
|
|
clean_logs = Bool(False, config=True,
|
|
|
help="whether to cleanup old logfiles before starting")
|
|
|
|
|
|
log_url = Unicode('', config=True,
|
|
|
help="The ZMQ URL of the iplogger to aggregate logging.")
|
|
|
|
|
|
cluster_id = Unicode('', config=True,
|
|
|
help="""String id to add to runtime files, to prevent name collisions when
|
|
|
using multiple clusters with a single profile simultaneously.
|
|
|
|
|
|
When set, files will be named like: 'ipcontroller-<cluster_id>-engine.json'
|
|
|
|
|
|
Since this is text inserted into filenames, typical recommendations apply:
|
|
|
Simple character strings are ideal, and spaces are not recommended (but should
|
|
|
generally work).
|
|
|
"""
|
|
|
)
|
|
|
def _cluster_id_changed(self, name, old, new):
|
|
|
self.name = self.__class__.name
|
|
|
if new:
|
|
|
self.name += '-%s'%new
|
|
|
|
|
|
def _config_files_default(self):
|
|
|
return ['ipcontroller_config.py', 'ipengine_config.py', 'ipcluster_config.py']
|
|
|
|
|
|
loop = Instance('zmq.eventloop.ioloop.IOLoop')
|
|
|
def _loop_default(self):
|
|
|
from zmq.eventloop.ioloop import IOLoop
|
|
|
return IOLoop.instance()
|
|
|
|
|
|
aliases = Dict(base_aliases)
|
|
|
flags = Dict(base_flags)
|
|
|
|
|
|
@catch_config_error
|
|
|
def initialize(self, argv=None):
|
|
|
"""initialize the app"""
|
|
|
super(BaseParallelApplication, self).initialize(argv)
|
|
|
self.to_work_dir()
|
|
|
self.reinit_logging()
|
|
|
|
|
|
def to_work_dir(self):
|
|
|
wd = self.work_dir
|
|
|
if unicode(wd) != os.getcwdu():
|
|
|
os.chdir(wd)
|
|
|
self.log.info("Changing to working dir: %s" % wd)
|
|
|
# This is the working dir by now.
|
|
|
sys.path.insert(0, '')
|
|
|
|
|
|
def reinit_logging(self):
|
|
|
# Remove old log files
|
|
|
log_dir = self.profile_dir.log_dir
|
|
|
if self.clean_logs:
|
|
|
for f in os.listdir(log_dir):
|
|
|
if re.match(r'%s-\d+\.(log|err|out)' % self.name, f):
|
|
|
try:
|
|
|
os.remove(os.path.join(log_dir, f))
|
|
|
except (OSError, IOError):
|
|
|
# probably just conflict from sibling process
|
|
|
# already removing it
|
|
|
pass
|
|
|
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:
|
|
|
while self.log.handlers:
|
|
|
self.log.removeHandler(self.log.handlers[0])
|
|
|
self._log_handler = logging.StreamHandler(open_log_file)
|
|
|
self.log.addHandler(self._log_handler)
|
|
|
else:
|
|
|
self._log_handler = self.log.handlers[0]
|
|
|
# Add timestamps to log format:
|
|
|
self._log_formatter = LevelFormatter(self.log_format,
|
|
|
datefmt=self.log_datefmt)
|
|
|
self._log_handler.setFormatter(self._log_formatter)
|
|
|
# do not propagate log messages to root logger
|
|
|
# ipcluster app will sometimes print duplicate messages during shutdown
|
|
|
# if this is 1 (default):
|
|
|
self.log.propagate = False
|
|
|
|
|
|
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.
|
|
|
"""
|
|
|
pid_file = os.path.join(self.profile_dir.pid_dir, self.name + u'.pid')
|
|
|
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``.
|
|
|
"""
|
|
|
pid_file = os.path.join(self.profile_dir.pid_dir, self.name + u'.pid')
|
|
|
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.
|
|
|
"""
|
|
|
pid_file = os.path.join(self.profile_dir.pid_dir, self.name + u'.pid')
|
|
|
if os.path.isfile(pid_file):
|
|
|
with open(pid_file, 'r') as f:
|
|
|
s = f.read().strip()
|
|
|
try:
|
|
|
pid = int(s)
|
|
|
except:
|
|
|
raise PIDFileError("invalid pid file: %s (contents: %r)"%(pid_file, s))
|
|
|
return pid
|
|
|
else:
|
|
|
raise PIDFileError('pid file not found: %s' % pid_file)
|
|
|
|
|
|
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
|
|
|
|
