Added .pid files to ipcluster and ipcontroller and daemon mode.

* The ClusterDir object now creates a /pid sub-dir for storing .pid
  files.
* Both ipcluster and ipcontroller now create .pid files upon starting.
* Both ipcluster and ipcontroller check for the existance of .pid files
  before starting and won't start if another instance is running.
* ipcluster has a daemon mode (--daemon/--nodaemon) will daemonize
  the process and then write the .pid file.
* Added a "ipcluster stop" subcommand which looks for a .pid file
  and kills the process.

IPython/kernel/clusterdir.py

@@ -1,394 +1,458 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The IPython cluster directory
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  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 shutil
 import sys
 
 from twisted.python import log
 
 from IPython.core import release
 from IPython.config.loader import PyFileConfigLoader
 from IPython.core.application import Application
 from IPython.core.component import Component
 from IPython.config.loader import ArgParseConfigLoader, NoConfigDefault
 from IPython.utils.traitlets import Unicode, Bool
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 
 class ClusterDirError(Exception):
     pass
 
 
+class PIDFileError(Exception):
+    pass
+
+
 class ClusterDir(Component):
     """An object to manage the cluster directory and its resources.
 
     The cluster directory is used by :command:`ipcontroller`, 
     :command:`ipcontroller` and :command:`ipcontroller` to manage the
     configuration, logging and security of these applications.
 
     This object knows how to find, create and manage these directories. This
     should be used by any code that want's to handle cluster directories.
     """
 
     security_dir_name = Unicode('security')
     log_dir_name = Unicode('log')
-    security_dir = Unicode()
-    log_dir = Unicode('')
-    location = Unicode('')
+    pid_dir_name = Unicode('pid')
+    security_dir = Unicode(u'')
+    log_dir = Unicode(u'')
+    pid_dir = Unicode(u'')
+    location = Unicode(u'')
 
     def __init__(self, location):
         super(ClusterDir, self).__init__(None)
         self.location = location
 
     def _location_changed(self, name, old, new):
         if not os.path.isdir(new):
             os.makedirs(new, mode=0777)
         else:
             os.chmod(new, 0777)
         self.security_dir = os.path.join(new, self.security_dir_name)
         self.log_dir = os.path.join(new, self.log_dir_name)
+        self.pid_dir = os.path.join(new, self.pid_dir_name)
         self.check_dirs()
 
     def _log_dir_changed(self, name, old, new):
         self.check_log_dir()
 
     def check_log_dir(self):
         if not os.path.isdir(self.log_dir):
             os.mkdir(self.log_dir, 0777)
         else:
             os.chmod(self.log_dir, 0777)
 
     def _security_dir_changed(self, name, old, new):
         self.check_security_dir()
 
     def check_security_dir(self):
         if not os.path.isdir(self.security_dir):
             os.mkdir(self.security_dir, 0700)
         else:
             os.chmod(self.security_dir, 0700)
 
+    def _pid_dir_changed(self, name, old, new):
+        self.check_pid_dir()
+
+    def check_pid_dir(self):
+        if not os.path.isdir(self.pid_dir):
+            os.mkdir(self.pid_dir, 0700)
+        else:
+            os.chmod(self.pid_dir, 0700)
+
     def check_dirs(self):
         self.check_security_dir()
         self.check_log_dir()
+        self.check_pid_dir()
 
     def load_config_file(self, filename):
         """Load a config file from the top level of the cluster dir.
 
         Parameters
         ----------
         filename : unicode or str
             The filename only of the config file that must be located in
             the top-level of the cluster directory.
         """
         loader = PyFileConfigLoader(filename, self.location)
         return loader.load_config()
 
     def copy_config_file(self, config_file, path=None, overwrite=False):
         """Copy a default config file into the active cluster directory.
 
         Default configuration files are kept in :mod:`IPython.config.default`.
         This function moves these from that location to the working cluster
         directory.
         """
         if path is None:
             import IPython.config.default
             path = IPython.config.default.__file__.split(os.path.sep)[:-1]
             path = os.path.sep.join(path)
         src = os.path.join(path, config_file)
         dst = os.path.join(self.location, config_file)
         if not os.path.isfile(dst) or overwrite:
             shutil.copy(src, dst)
 
     def copy_all_config_files(self, path=None, overwrite=False):
         """Copy all config files into the active cluster directory."""
         for f in ['ipcontroller_config.py', 'ipengine_config.py',
                   'ipcluster_config.py']:
             self.copy_config_file(f, path=path, overwrite=overwrite)
 
     @classmethod
     def create_cluster_dir(csl, cluster_dir):
         """Create a new cluster directory given a full path.
 
         Parameters
         ----------
         cluster_dir : str
             The full path to the cluster directory.  If it does exist, it will
             be used.  If not, it will be created.
         """
         return ClusterDir(cluster_dir)
 
     @classmethod
     def create_cluster_dir_by_profile(cls, path, profile='default'):
         """Create a cluster dir by profile name and path.
 
         Parameters
         ----------
         path : str
             The path (directory) to put the cluster directory in.
         profile : str
             The name of the profile.  The name of the cluster directory will
             be "cluster_<profile>".
         """
         if not os.path.isdir(path):
             raise ClusterDirError('Directory not found: %s' % path)
         cluster_dir = os.path.join(path, 'cluster_' + profile)
         return ClusterDir(cluster_dir)
 
     @classmethod
     def find_cluster_dir_by_profile(cls, ipythondir, profile='default'):
         """Find an existing cluster dir by profile name, return its ClusterDir.
 
         This searches through a sequence of paths for a cluster dir.  If it
         is not found, a :class:`ClusterDirError` exception will be raised.
 
         The search path algorithm is:
         1. ``os.getcwd()``
         2. ``ipythondir``
         3. The directories found in the ":" separated 
            :env:`IPCLUSTERDIR_PATH` environment variable.
 
         Parameters
         ----------
         ipythondir : unicode or str
             The IPython directory to use.
         profile : unicode or str
             The name of the profile.  The name of the cluster directory
             will be "cluster_<profile>".
         """
         dirname = 'cluster_' + profile
         cluster_dir_paths = os.environ.get('IPCLUSTERDIR_PATH','')
         if cluster_dir_paths:
             cluster_dir_paths = cluster_dir_paths.split(':')
         else:
             cluster_dir_paths = []
         paths = [os.getcwd(), ipythondir] + cluster_dir_paths
         for p in paths:
             cluster_dir = os.path.join(p, dirname)
             if os.path.isdir(cluster_dir):
                 return ClusterDir(cluster_dir)
         else:
             raise ClusterDirError('Cluster directory not found in paths: %s' % dirname)
 
     @classmethod
     def find_cluster_dir(cls, cluster_dir):
         """Find/create a cluster dir and return its ClusterDir.
 
         This will create the cluster directory if it doesn't exist.
 
         Parameters
         ----------
         cluster_dir : unicode or str
             The path of the cluster directory.  This is expanded using
             :func:`os.path.expandvars` and :func:`os.path.expanduser`.
         """
         cluster_dir = os.path.expandvars(os.path.expanduser(cluster_dir))
         if not os.path.isdir(cluster_dir):
             raise ClusterDirError('Cluster directory not found: %s' % cluster_dir)
         return ClusterDir(cluster_dir)
 
 
 class AppWithClusterDirArgParseConfigLoader(ArgParseConfigLoader):
     """Default command line options for IPython cluster applications."""
 
     def _add_other_arguments(self):
         self.parser.add_argument('-ipythondir', '--ipython-dir', 
             dest='Global.ipythondir',type=str,
             help='Set to override default location of Global.ipythondir.',
             default=NoConfigDefault,
             metavar='Global.ipythondir'
         )
         self.parser.add_argument('-p','-profile', '--profile',
             dest='Global.profile',type=str,
             help='The string name of the profile to be used. This determines '
             'the name of the cluster dir as: cluster_<profile>. The default profile '
             'is named "default".  The cluster directory is resolve this way '
             'if the --cluster-dir option is not used.',
             default=NoConfigDefault,
             metavar='Global.profile'
         )
         self.parser.add_argument('-log_level', '--log-level',
             dest="Global.log_level",type=int,
             help='Set the log level (0,10,20,30,40,50).  Default is 30.',
             default=NoConfigDefault,
             metavar="Global.log_level"
         )
         self.parser.add_argument('-cluster_dir', '--cluster-dir',
             dest='Global.cluster_dir',type=str,
             help='Set the cluster dir. This overrides the logic used by the '
             '--profile option.',
             default=NoConfigDefault,
             metavar='Global.cluster_dir'
         )
         self.parser.add_argument('-clean_logs', '--clean-logs',
             dest='Global.clean_logs', action='store_true',
             help='Delete old log flies before starting.',
             default=NoConfigDefault
         )
         self.parser.add_argument('-noclean_logs', '--no-clean-logs',
             dest='Global.clean_logs', action='store_false',
             help="Don't Delete old log flies before starting.",
             default=NoConfigDefault
         )
 
 class ApplicationWithClusterDir(Application):
     """An application that puts everything into a cluster directory.
 
     Instead of looking for things in the ipythondir, this type of application
     will use its own private directory called the "cluster directory"
     for things like config files, log files, etc.
 
     The cluster directory is resolved as follows:
 
     * If the ``--cluster-dir`` option is given, it is used.
     * If ``--cluster-dir`` is not given, the application directory is 
       resolve using the profile name as ``cluster_<profile>``. The search 
       path for this directory is then i) cwd if it is found there
       and ii) in ipythondir otherwise.
 
     The config file for the application is to be put in the cluster
     dir and named the value of the ``config_file_name`` class attribute.
     """
 
     auto_create_cluster_dir = True
 
     def create_default_config(self):
         super(ApplicationWithClusterDir, self).create_default_config()
         self.default_config.Global.profile = 'default'
         self.default_config.Global.cluster_dir = ''
         self.default_config.Global.log_to_file = False
         self.default_config.Global.clean_logs = False
 
     def create_command_line_config(self):
         """Create and return a command line config loader."""
         return AppWithClusterDirArgParseConfigLoader(
             description=self.description, 
             version=release.version
         )
 
     def find_resources(self):
         """This resolves the cluster directory.
 
         This tries to find the cluster directory and if successful, it will
         have done:
         * Sets ``self.cluster_dir_obj`` to the :class:`ClusterDir` object for 
           the application.
         * Sets ``self.cluster_dir`` attribute of the application and config
           objects.
 
         The algorithm used for this is as follows:
         1. Try ``Global.cluster_dir``.
         2. Try using ``Global.profile``.
         3. If both of these fail and ``self.auto_create_cluster_dir`` is
            ``True``, then create the new cluster dir in the IPython directory.
         4. If all fails, then raise :class:`ClusterDirError`.
         """
 
         try:
             cluster_dir = self.command_line_config.Global.cluster_dir
         except AttributeError:
             cluster_dir = self.default_config.Global.cluster_dir
         cluster_dir = os.path.expandvars(os.path.expanduser(cluster_dir))
         try:
             self.cluster_dir_obj = ClusterDir.find_cluster_dir(cluster_dir)
         except ClusterDirError:
             pass
         else:
             self.log.info('Using existing cluster dir: %s' % \
                 self.cluster_dir_obj.location
             )
             self.finish_cluster_dir()
             return
 
         try:
             self.profile = self.command_line_config.Global.profile
         except AttributeError:
             self.profile = self.default_config.Global.profile
         try:
             self.cluster_dir_obj = ClusterDir.find_cluster_dir_by_profile(
                 self.ipythondir, self.profile)
         except ClusterDirError:
             pass
         else:
             self.log.info('Using existing cluster dir: %s' % \
                 self.cluster_dir_obj.location
             )
             self.finish_cluster_dir()
             return
 
         if self.auto_create_cluster_dir:
             self.cluster_dir_obj = ClusterDir.create_cluster_dir_by_profile(
                 self.ipythondir, self.profile
             )
             self.log.info('Creating new cluster dir: %s' % \
                 self.cluster_dir_obj.location
             )
             self.finish_cluster_dir()
         else:
             raise ClusterDirError('Could not find a valid cluster directory.')
 
     def finish_cluster_dir(self):
         # Set the cluster directory
         self.cluster_dir = self.cluster_dir_obj.location
         
         # These have to be set because they could be different from the one
         # that we just computed.  Because command line has the highest
         # priority, this will always end up in the master_config.
         self.default_config.Global.cluster_dir = self.cluster_dir
         self.command_line_config.Global.cluster_dir = self.cluster_dir
 
         # Set the search path to the cluster directory
         self.config_file_paths = (self.cluster_dir,)
 
     def find_config_file_name(self):
         """Find the config file name for this application."""
         # For this type of Application it should be set as a class attribute.
         if not hasattr(self, 'config_file_name'):
             self.log.critical("No config filename found")
 
     def find_config_file_paths(self):
         # Set the search path to the cluster directory
         self.config_file_paths = (self.cluster_dir,)
 
     def pre_construct(self):
         # The log and security dirs were set earlier, but here we put them
         # into the config and log them.
         config = self.master_config
         sdir = self.cluster_dir_obj.security_dir
         self.security_dir = config.Global.security_dir = sdir
         ldir = self.cluster_dir_obj.log_dir
         self.log_dir = config.Global.log_dir = ldir
+        pdir = self.cluster_dir_obj.pid_dir
+        self.pid_dir = config.Global.pid_dir = pdir
         self.log.info("Cluster directory set to: %s" % self.cluster_dir)
 
     def start_logging(self):
         # Remove old log files
         if self.master_config.Global.clean_logs:
             log_dir = self.master_config.Global.log_dir
             for f in os.listdir(log_dir):
                 if f.startswith(self.name + '-') and f.endswith('.log'):
                     os.remove(os.path.join(log_dir, f))
         # Start logging to the new log file
         if self.master_config.Global.log_to_file:
             log_filename = self.name + '-' + str(os.getpid()) + '.log'
             logfile = os.path.join(self.log_dir, log_filename)
             open_log_file = open(logfile, 'w')
         else:
             open_log_file = sys.stdout
         log.startLogging(open_log_file)
+
+    def write_pid_file(self):
+        """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.pid_dir, self.name + '.pid')
+        if os.path.isfile(pid_file):
+            pid = self.get_pid_from_file()
+            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`.
+        """
+        pid_file = os.path.join(self.pid_dir, self.name + '.pid')
+        if os.path.isfile(pid_file):
+            try:
+                self.log.info("Removing pid file: %s" % pid_file)
+                os.remove(pid_file)
+            except:
+                pass
+
+    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.pid_dir, self.name + '.pid')
+        if os.path.isfile(pid_file):
+            with open(pid_file, 'r') as f:
+                pid = int(f.read().strip())
+                return pid
+        else:
+            raise PIDFileError('pid file not found: %s' % pid_file)
  No newline at end of file

IPython/kernel/ipclusterapp.py

@@ -1,306 +1,378 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The ipcluster application.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  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
 #-----------------------------------------------------------------------------
 
 import logging
 import os
 import signal
 import sys
 
+from twisted.scripts._twistd_unix import daemonize
+
 from IPython.core import release
 from IPython.external import argparse
 from IPython.config.loader import ArgParseConfigLoader, NoConfigDefault
 from IPython.utils.importstring import import_item
 
 from IPython.kernel.clusterdir import (
-    ApplicationWithClusterDir, ClusterDirError
+    ApplicationWithClusterDir, ClusterDirError, PIDFileError
 )
 
 from twisted.internet import reactor, defer
 from twisted.python import log
 
 #-----------------------------------------------------------------------------
 # Code for launchers
 #-----------------------------------------------------------------------------
 
 
 
 #-----------------------------------------------------------------------------
 # The ipcluster application
 #-----------------------------------------------------------------------------
 
 
 class IPClusterCLLoader(ArgParseConfigLoader):
 
     def _add_arguments(self):
         # This has all the common options that all subcommands use
         parent_parser1 = argparse.ArgumentParser(add_help=False)
         parent_parser1.add_argument('-ipythondir', '--ipython-dir', 
             dest='Global.ipythondir',type=str,
             help='Set to override default location of Global.ipythondir.',
             default=NoConfigDefault,
             metavar='Global.ipythondir')
         parent_parser1.add_argument('-log_level', '--log-level',
             dest="Global.log_level",type=int,
             help='Set the log level (0,10,20,30,40,50).  Default is 30.',
             default=NoConfigDefault,
             metavar='Global.log_level')
 
         # This has all the common options that other subcommands use
         parent_parser2 = argparse.ArgumentParser(add_help=False)
         parent_parser2.add_argument('-p','-profile', '--profile',
             dest='Global.profile',type=str,
             default=NoConfigDefault,
             help='The string name of the profile to be used. This determines '
             'the name of the cluster dir as: cluster_<profile>. The default profile '
             'is named "default".  The cluster directory is resolve this way '
             'if the --cluster-dir option is not used.',
             default=NoConfigDefault,
             metavar='Global.profile')
         parent_parser2.add_argument('-cluster_dir', '--cluster-dir',
             dest='Global.cluster_dir',type=str,
             default=NoConfigDefault,
             help='Set the cluster dir. This overrides the logic used by the '
             '--profile option.',
             default=NoConfigDefault,
             metavar='Global.cluster_dir')
         parent_parser2.add_argument('--log-to-file',
             action='store_true', dest='Global.log_to_file', 
             default=NoConfigDefault,
             help='Log to a file in the log directory (default is stdout)'
         )
 
         subparsers = self.parser.add_subparsers(
             dest='Global.subcommand',
             title='ipcluster subcommands',
             description='ipcluster has a variety of subcommands. '
             'The general way of running ipcluster is "ipcluster <cmd> '
             ' [options]""',
             help='For more help, type "ipcluster <cmd> -h"')
 
         parser_list = subparsers.add_parser(
             'list',
             help='List all clusters in cwd and ipythondir.',
             parents=[parent_parser1]
         )
 
         parser_create = subparsers.add_parser(
             'create',
             help='Create a new cluster directory.',
             parents=[parent_parser1, parent_parser2] 
         )
         parser_create.add_argument(
             '--reset-config',
             dest='Global.reset_config', action='store_true',
             default=NoConfigDefault,
             help='Recopy the default config files to the cluster directory. '
             'You will loose any modifications you have made to these files.'
         )
 
         parser_start = subparsers.add_parser(
             'start',
             help='Start a cluster.',
             parents=[parent_parser1, parent_parser2]
         )
         parser_start.add_argument(
             '-n', '--number',
             type=int, dest='Global.n',
             default=NoConfigDefault,
             help='The number of engines to start.',
             metavar='Global.n'
         )
         parser_start.add_argument('-clean_logs', '--clean-logs',
             dest='Global.clean_logs', action='store_true',
             help='Delete old log flies before starting.',
             default=NoConfigDefault
         )
         parser_start.add_argument('-noclean_logs', '--no-clean-logs',
             dest='Global.clean_logs', action='store_false',
             help="Don't delete old log flies before starting.",
             default=NoConfigDefault
         )
+        parser_start.add_argument('--daemon', '-daemon',
+            dest='Global.daemonize', action='store_true',
+            help='Daemonize the ipcluster program. This implies --log-to-file',
+            default=NoConfigDefault
+        )
+        parser_start.add_argument('--nodaemon', '-nodaemon',
+            dest='Global.daemonize', action='store_false',
+            help="Dont't daemonize the ipcluster program.",
+            default=NoConfigDefault
+        )
+
+        parser_start = subparsers.add_parser(
+            'stop',
+            help='Stop a cluster.',
+            parents=[parent_parser1, parent_parser2]
+        )
+        parser_start.add_argument('-sig', '--sig',
+            dest='Global.stop_signal', type=int,
+            help="The signal number to use in stopping the cluster (default=2).",
+            default=NoConfigDefault
+        )
 
 default_config_file_name = 'ipcluster_config.py'
 
 
 class IPClusterApp(ApplicationWithClusterDir):
 
     name = 'ipcluster'
     description = 'Start an IPython cluster (controller and engines).'
     config_file_name = default_config_file_name
     default_log_level = logging.INFO
     auto_create_cluster_dir = False
 
     def create_default_config(self):
         super(IPClusterApp, self).create_default_config()
         self.default_config.Global.controller_launcher = \
             'IPython.kernel.launcher.LocalControllerLauncher'
         self.default_config.Global.engine_launcher = \
             'IPython.kernel.launcher.LocalEngineSetLauncher'
         self.default_config.Global.n = 2
         self.default_config.Global.reset_config = False
         self.default_config.Global.clean_logs = True
+        self.default_config.Global.stop_signal = 2
+        self.default_config.Global.daemonize = False
 
     def create_command_line_config(self):
         """Create and return a command line config loader."""
         return IPClusterCLLoader(
             description=self.description, 
             version=release.version
         )
 
     def find_resources(self):
         subcommand = self.command_line_config.Global.subcommand
         if subcommand=='list':
             self.list_cluster_dirs()
             # Exit immediately because there is nothing left to do.
             self.exit()
         elif subcommand=='create':
             self.auto_create_cluster_dir = True
             super(IPClusterApp, self).find_resources()
-        elif subcommand=='start':
+        elif subcommand=='start' or subcommand=='stop':
             self.auto_create_cluster_dir = False
             try:
                 super(IPClusterApp, self).find_resources()
             except ClusterDirError:
                 raise ClusterDirError(
                     "Could not find a cluster directory. A cluster dir must "
                     "be created before running 'ipcluster start'.  Do "
                     "'ipcluster create -h' or 'ipcluster list -h' for more "
                     "information about creating and listing cluster dirs."
                 )
 
+    def pre_construct(self):
+        super(IPClusterApp, self).pre_construct()
+        config = self.master_config
+        try:
+            daemon = config.Global.daemonize
+            if daemon:
+                config.Global.log_to_file = True
+        except AttributeError:
+            pass
+
     def construct(self):
         config = self.master_config
         if config.Global.subcommand=='list':
             pass
         elif config.Global.subcommand=='create':
             self.log.info('Copying default config files to cluster directory '
             '[overwrite=%r]' % (config.Global.reset_config,))
             self.cluster_dir_obj.copy_all_config_files(overwrite=config.Global.reset_config)
         elif config.Global.subcommand=='start':
             self.start_logging()
             reactor.callWhenRunning(self.start_launchers)
 
     def list_cluster_dirs(self):
         # Find the search paths
         cluster_dir_paths = os.environ.get('IPCLUSTERDIR_PATH','')
         if cluster_dir_paths:
             cluster_dir_paths = cluster_dir_paths.split(':')
         else:
             cluster_dir_paths = []
         try:
             ipythondir = self.command_line_config.Global.ipythondir
         except AttributeError:
             ipythondir = self.default_config.Global.ipythondir
         paths = [os.getcwd(), ipythondir] + \
             cluster_dir_paths
         paths = list(set(paths))
 
         self.log.info('Searching for cluster dirs in paths: %r' % paths)
         for path in paths:
             files = os.listdir(path)
             for f in files:
                 full_path = os.path.join(path, f)
                 if os.path.isdir(full_path) and f.startswith('cluster_'):
                     profile = full_path.split('_')[-1]
                     start_cmd = '"ipcluster start -n 4 -p %s"' % profile
                     print start_cmd + " ==> " + full_path
 
     def start_launchers(self):
         config = self.master_config
 
         # Create the launchers
         el_class = import_item(config.Global.engine_launcher)
         self.engine_launcher = el_class(
             self.cluster_dir, config=config
         )
         cl_class = import_item(config.Global.controller_launcher)
         self.controller_launcher = cl_class(
             self.cluster_dir, config=config
         )
 
         # Setup signals
         signal.signal(signal.SIGINT, self.stop_launchers)
         # signal.signal(signal.SIGKILL, self.stop_launchers)
 
         # Setup the observing of stopping
         d1 = self.controller_launcher.observe_stop()
         d1.addCallback(self.stop_engines)
         d1.addErrback(self.err_and_stop)
         # If this triggers, just let them die
         # d2 = self.engine_launcher.observe_stop()
         
         # Start the controller and engines
         d = self.controller_launcher.start(
             profile=None, cluster_dir=config.Global.cluster_dir
         )
         d.addCallback(lambda _: self.start_engines())
         d.addErrback(self.err_and_stop)
 
     def err_and_stop(self, f):
         log.msg('Unexpected error in ipcluster:')
         log.err(f)
         reactor.stop()
 
     def stop_engines(self, r):
         return self.engine_launcher.stop()
 
     def start_engines(self):
         config = self.master_config
         d = self.engine_launcher.start(
             config.Global.n,
             profile=None, cluster_dir=config.Global.cluster_dir
         )
         return d
 
     def stop_launchers(self, signum, frame):
         log.msg("Stopping cluster")
         d1 = self.engine_launcher.stop()
         d2 = self.controller_launcher.stop()
         # d1.addCallback(lambda _: self.controller_launcher.stop)
         d1.addErrback(self.err_and_stop)
         d2.addErrback(self.err_and_stop)
         reactor.callLater(2.0, reactor.stop)
 
     def start_logging(self):
         # Remove old log files
         if self.master_config.Global.clean_logs:
             log_dir = self.master_config.Global.log_dir
             for f in os.listdir(log_dir):
                 if f.startswith('ipengine' + '-') and f.endswith('.log'):
                     os.remove(os.path.join(log_dir, f))
             for f in os.listdir(log_dir):
                 if f.startswith('ipcontroller' + '-') and f.endswith('.log'):
                     os.remove(os.path.join(log_dir, f))
         super(IPClusterApp, self).start_logging()
 
     def start_app(self):
+        """Start the application, depending on what subcommand is used."""
         config = self.master_config
-        if config.Global.subcommand=='create' or config.Global.subcommand=='list':
+        subcmd = config.Global.subcommand
+        if subcmd=='create' or subcmd=='list':
             return
-        elif config.Global.subcommand=='start':
+        elif subcmd=='start':
+            # First see if the cluster is already running
+            try:
+                pid = self.get_pid_from_file()
+            except:
+                pass
+            else:
+                self.log.critical(
+                    'Cluster is already running with [pid=%s]. '
+                    'use "ipcluster stop" to stop the cluster.' % pid
+                )
+                sys.exit(9)
+            # Now log and daemonize
+            self.log.info('Starting ipcluster with [daemon=%r]' % config.Global.daemonize)
+            if config.Global.daemonize:
+                if os.name=='posix':
+                    os.chdir(config.Global.cluster_dir)
+                    self.log_level = 40
+                    daemonize()
+
+            # Now write the new pid file after our new forked pid is active.
+            self.write_pid_file()
+            reactor.addSystemEventTrigger('during','shutdown', self.remove_pid_file)
             reactor.run()
+        elif subcmd=='stop':
+            try:
+                pid = self.get_pid_from_file()
+            except PIDFileError:
+                self.log.critical(
+                    'Problem reading pid file, cluster is probably not running.'
+                )
+                sys.exit(9)
+            sig = config.Global.stop_signal
+            self.log.info(
+                "Stopping cluster [pid=%r] with [signal=%r]" % (pid, sig)
+            )
+            os.kill(pid, sig)
 
 
 def launch_new_instance():
     """Create and run the IPython cluster."""
     app = IPClusterApp()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()
 

IPython/kernel/ipcontrollerapp.py

@@ -1,254 +1,258 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The IPython controller application.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  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 copy
 import os
 import sys
 
 from twisted.application import service
 from twisted.internet import reactor
 from twisted.python import log
 
 from IPython.config.loader import Config, NoConfigDefault
 
 from IPython.kernel.clusterdir import (
     ApplicationWithClusterDir, 
     AppWithClusterDirArgParseConfigLoader
 )
 
 from IPython.core import release
 
 from IPython.utils.traitlets import Str, Instance
 
 from IPython.kernel import controllerservice
 
 from IPython.kernel.fcutil import FCServiceFactory
 
 #-----------------------------------------------------------------------------
 # Default interfaces
 #-----------------------------------------------------------------------------
 
 
 # The default client interfaces for FCClientServiceFactory.interfaces
 default_client_interfaces = Config()
 default_client_interfaces.Task.interface_chain = [
     'IPython.kernel.task.ITaskController',
     'IPython.kernel.taskfc.IFCTaskController'
 ]
 
 default_client_interfaces.Task.furl_file = 'ipcontroller-tc.furl'
 
 default_client_interfaces.MultiEngine.interface_chain = [
     'IPython.kernel.multiengine.IMultiEngine',
     'IPython.kernel.multienginefc.IFCSynchronousMultiEngine'
 ]
 
 default_client_interfaces.MultiEngine.furl_file = 'ipcontroller-mec.furl'
 
 # Make this a dict we can pass to Config.__init__ for the default
 default_client_interfaces = dict(copy.deepcopy(default_client_interfaces.items()))
 
 
 
 # The default engine interfaces for FCEngineServiceFactory.interfaces
 default_engine_interfaces = Config()
 default_engine_interfaces.Default.interface_chain = [
     'IPython.kernel.enginefc.IFCControllerBase'
 ]
 
 default_engine_interfaces.Default.furl_file = 'ipcontroller-engine.furl'
 
 # Make this a dict we can pass to Config.__init__ for the default
 default_engine_interfaces = dict(copy.deepcopy(default_engine_interfaces.items()))
 
 
 #-----------------------------------------------------------------------------
 # Service factories
 #-----------------------------------------------------------------------------
 
 
 class FCClientServiceFactory(FCServiceFactory):
     """A Foolscap implementation of the client services."""
 
     cert_file = Str('ipcontroller-client.pem', config=True)
     interfaces = Instance(klass=Config, kw=default_client_interfaces,
                           allow_none=False, config=True)
 
 
 class FCEngineServiceFactory(FCServiceFactory):
     """A Foolscap implementation of the engine services."""
 
     cert_file = Str('ipcontroller-engine.pem', config=True)
     interfaces = Instance(klass=dict, kw=default_engine_interfaces,
                           allow_none=False, config=True)
 
 
 #-----------------------------------------------------------------------------
 # The main application
 #-----------------------------------------------------------------------------
 
 
 cl_args = (
     # Client config
     (('--client-ip',), dict(
         type=str, dest='FCClientServiceFactory.ip', default=NoConfigDefault,
         help='The IP address or hostname the controller will listen on for '
         'client connections.',
         metavar='FCClientServiceFactory.ip')
     ),
     (('--client-port',), dict(
         type=int, dest='FCClientServiceFactory.port', default=NoConfigDefault,
         help='The port the controller will listen on for client connections. '
         'The default is to use 0, which will autoselect an open port.',
         metavar='FCClientServiceFactory.port')
     ),
     (('--client-location',), dict(
         type=str, dest='FCClientServiceFactory.location', default=NoConfigDefault,
         help='The hostname or IP that clients should connect to. This does '
         'not control which interface the controller listens on. Instead, this '
         'determines the hostname/IP that is listed in the FURL, which is how '
         'clients know where to connect. Useful if the controller is listening '
         'on multiple interfaces.',
         metavar='FCClientServiceFactory.location')
     ),
     # Engine config
     (('--engine-ip',), dict(
         type=str, dest='FCEngineServiceFactory.ip', default=NoConfigDefault,
         help='The IP address or hostname the controller will listen on for '
         'engine connections.',
         metavar='FCEngineServiceFactory.ip')
     ),
     (('--engine-port',), dict(
         type=int, dest='FCEngineServiceFactory.port', default=NoConfigDefault,
         help='The port the controller will listen on for engine connections. '
         'The default is to use 0, which will autoselect an open port.',
         metavar='FCEngineServiceFactory.port')
     ),
     (('--engine-location',), dict(
         type=str, dest='FCEngineServiceFactory.location', default=NoConfigDefault,
         help='The hostname or IP that engines should connect to. This does '
         'not control which interface the controller listens on. Instead, this '
         'determines the hostname/IP that is listed in the FURL, which is how '
         'engines know where to connect. Useful if the controller is listening '
         'on multiple interfaces.',
         metavar='FCEngineServiceFactory.location')
     ),
     # Global config
     (('--log-to-file',), dict(
         action='store_true', dest='Global.log_to_file', default=NoConfigDefault,
         help='Log to a file in the log directory (default is stdout)')
     ),
     (('-r','--reuse-furls'), dict(
         action='store_true', dest='Global.reuse_furls', default=NoConfigDefault,
         help='Try to reuse all FURL files. If this is not set all FURL files '
         'are deleted before the controller starts. This must be set if '
         'specific ports are specified by --engine-port or --client-port.')
     ),
     (('-ns','--no-security'), dict(
         action='store_false', dest='Global.secure', default=NoConfigDefault,
         help='Turn off SSL encryption for all connections.')
     )
 )
 
 
 class IPControllerAppCLConfigLoader(AppWithClusterDirArgParseConfigLoader):
 
     arguments = cl_args
 
 
 default_config_file_name = 'ipcontroller_config.py'
 
 
 class IPControllerApp(ApplicationWithClusterDir):
 
     name = 'ipcontroller'
     description = 'Start the IPython controller for parallel computing.'
     config_file_name = default_config_file_name
     auto_create_cluster_dir = True
 
     def create_default_config(self):
         super(IPControllerApp, self).create_default_config()
         self.default_config.Global.reuse_furls = False
         self.default_config.Global.secure = True
         self.default_config.Global.import_statements = []
         self.default_config.Global.clean_logs = True
 
     def create_command_line_config(self):
         """Create and return a command line config loader."""
         return IPControllerAppCLConfigLoader(
             description=self.description, 
             version=release.version
         )
 
     def post_load_command_line_config(self):
         # Now setup reuse_furls
         c = self.command_line_config
         if hasattr(c.Global, 'reuse_furls'):
             c.FCClientServiceFactory.reuse_furls = c.Global.reuse_furls
             c.FCEngineServiceFactory.reuse_furls = c.Global.reuse_furls
             del c.Global.reuse_furls
         if hasattr(c.Global, 'secure'):
             c.FCClientServiceFactory.secure = c.Global.secure
             c.FCEngineServiceFactory.secure = c.Global.secure
             del c.Global.secure
 
     def construct(self):
         # I am a little hesitant to put these into InteractiveShell itself.
         # But that might be the place for them
         sys.path.insert(0, '')
 
         self.start_logging()
         self.import_statements()
-        
+
         # Create the service hierarchy
         self.main_service = service.MultiService()
         # The controller service
         controller_service = controllerservice.ControllerService()
         controller_service.setServiceParent(self.main_service)
         # The client tub and all its refereceables
         csfactory = FCClientServiceFactory(self.master_config, controller_service)
         client_service = csfactory.create()
         client_service.setServiceParent(self.main_service)
         # The engine tub
         esfactory = FCEngineServiceFactory(self.master_config, controller_service)
         engine_service = esfactory.create()
         engine_service.setServiceParent(self.main_service)
 
     def import_statements(self):
         statements = self.master_config.Global.import_statements
         for s in statements:
             try:
                 log.msg("Executing statement: '%s'" % s)
                 exec s in globals(), locals()
             except:
                 log.msg("Error running statement: %s" % s)
 
     def start_app(self):
         # Start the controller service and set things running
         self.main_service.startService()
+        self.write_pid_file()
+        reactor.addSystemEventTrigger('during','shutdown', self.remove_pid_file)
         reactor.run()
 
 
 def launch_new_instance():
     """Create and run the IPython controller"""
     app = IPControllerApp()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()