Added better documentation to command line programs.

IPython/kernel/ipclusterapp.py

@@ -1,459 +1,471 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
 
 if os.name=='posix':
     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, PIDFileError
 )
 
 from twisted.internet import reactor, defer
 from twisted.python import log, failure
 
 
 #-----------------------------------------------------------------------------
 # The ipcluster application
 #-----------------------------------------------------------------------------
 
 
 # Exit codes for ipcluster
 
 # This will be the exit code if the ipcluster appears to be running because
 # a .pid file exists
 ALREADY_STARTED = 10
 
 # This will be the exit code if ipcluster stop is run, but there is not .pid
 # file to be found.
 ALREADY_STOPPED = 11
 
 
 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('--ipython-dir', 
             dest='Global.ipython_dir',type=unicode,
             help='Set to override default location of Global.ipython_dir.',
             default=NoConfigDefault,
             metavar='Global.ipython_dir')
         parent_parser1.add_argument('--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',
             dest='Global.profile',type=unicode,
             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',
             dest='Global.cluster_dir',type=unicode,
             help='Set the cluster dir. This overrides the logic used by the '
             '--profile option.',
             default=NoConfigDefault,
             metavar='Global.cluster_dir'),
         parent_parser2.add_argument('--work-dir',
             dest='Global.work_dir',type=unicode,
             help='Set the working dir for the process.',
             default=NoConfigDefault,
             metavar='Global.work_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 ipython_dir.',
             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',
             dest='Global.clean_logs', action='store_true',
             help='Delete old log flies before starting.',
             default=NoConfigDefault
         )
         parser_start.add_argument('--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',
             dest='Global.daemonize', action='store_true',
             help='Daemonize the ipcluster program. This implies --log-to-file',
             default=NoConfigDefault
         )
         parser_start.add_argument('--no-daemon',
             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('--signal',
             dest='Global.signal', type=int,
             help="The signal number to use in stopping the cluster (default=2).",
             metavar="Global.signal",
             default=NoConfigDefault
         )
 
 
 default_config_file_name = u'ipcluster_config.py'
 
 
+_description = """Start an IPython cluster for parallel computing.\n\n
+
+An IPython cluster consists of 1 controller and 1 or more engines.
+This command automates the startup of these processes using a wide
+range of startup methods (SSH, local processes, PBS, mpiexec,
+Windows HPC Server 2008). To start a cluster with 4 engines on your
+local host simply do "ipcluster start -n 4". For more complex usage 
+you will typically do "ipcluster create -p mycluster", then edit
+configuration files, followed by "ipcluster start -p mycluster -n 4".
+"""
+
+
 class IPClusterApp(ApplicationWithClusterDir):
 
     name = u'ipcluster'
-    description = 'Start an IPython cluster (controller and engines).'
+    description = _description
     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.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' or subcommand=='stop':
             self.auto_create_cluster_dir = True
             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 list_cluster_dirs(self):
         # Find the search paths
         cluster_dir_paths = os.environ.get('IPCLUSTER_DIR_PATH','')
         if cluster_dir_paths:
             cluster_dir_paths = cluster_dir_paths.split(':')
         else:
             cluster_dir_paths = []
         try:
             ipython_dir = self.command_line_config.Global.ipython_dir
         except AttributeError:
             ipython_dir = self.default_config.Global.ipython_dir
         paths = [os.getcwd(), ipython_dir] + \
             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 -p %s -n 4' % profile
                     print start_cmd + " ==> " + full_path
 
     def pre_construct(self):
         # IPClusterApp.pre_construct() is where we cd to the working directory.
         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
         subcmd = config.Global.subcommand
         reset = config.Global.reset_config
         if subcmd == 'list':
             return
         if subcmd == 'create':
             self.log.info('Copying default config files to cluster directory '
             '[overwrite=%r]' % (reset,))
             self.cluster_dir_obj.copy_all_config_files(overwrite=reset)
         if subcmd =='start':
             self.cluster_dir_obj.copy_all_config_files(overwrite=False)
             self.start_logging()
             reactor.callWhenRunning(self.start_launchers)
 
     def start_launchers(self):
         config = self.master_config
 
         # Create the launchers. In both bases, we set the work_dir of
         # the launcher to the cluster_dir. This is where the launcher's
         # subprocesses will be launched. It is not where the controller
         # and engine will be launched.
         el_class = import_item(config.Global.engine_launcher)
         self.engine_launcher = el_class(
             work_dir=self.cluster_dir, config=config
         )
         cl_class = import_item(config.Global.controller_launcher)
         self.controller_launcher = cl_class(
             work_dir=self.cluster_dir, config=config
         )
 
         # Setup signals
         signal.signal(signal.SIGINT, self.sigint_handler)
 
         # Setup the observing of stopping. If the controller dies, shut
         # everything down as that will be completely fatal for the engines.
         d1 = self.controller_launcher.observe_stop()
         d1.addCallback(self.stop_launchers)
         # But, we don't monitor the stopping of engines. An engine dying
         # is just fine and in principle a user could start a new engine.
         # Also, if we did monitor engine stopping, it is difficult to
         # know what to do when only some engines die. Currently, the
         # observing of engine stopping is inconsistent. Some launchers
         # might trigger on a single engine stopping, other wait until
         # all stop.  TODO: think more about how to handle this.
         
         # Start the controller and engines
         self._stopping = False  # Make sure stop_launchers is not called 2x.
         d = self.start_controller()
         d.addCallback(self.start_engines)
         d.addCallback(self.startup_message)
         # If the controller or engines fail to start, stop everything
         d.addErrback(self.stop_launchers)
         return d
 
     def startup_message(self, r=None):
         log.msg("IPython cluster: started")
         return r
         
     def start_controller(self, r=None):
         # log.msg("In start_controller")
         config = self.master_config
         d = self.controller_launcher.start(
             cluster_dir=config.Global.cluster_dir
         )
         return d
             
     def start_engines(self, r=None):
         # log.msg("In start_engines")
         config = self.master_config
         d = self.engine_launcher.start(
             config.Global.n,
             cluster_dir=config.Global.cluster_dir
         )
         return d
 
     def stop_controller(self, r=None):
         # log.msg("In stop_controller")
         if self.controller_launcher.running:
             d = self.controller_launcher.stop()
             d.addErrback(self.log_err)
             return d
         else:
             return defer.succeed(None)
 
     def stop_engines(self, r=None):
         # log.msg("In stop_engines")
         if self.engine_launcher.running:
             d = self.engine_launcher.stop()
             d.addErrback(self.log_err)
             return d
         else:
             return defer.succeed(None)
 
     def log_err(self, f):
         log.msg(f.getTraceback())
         return None
         
     def stop_launchers(self, r=None):
         if not self._stopping:
             self._stopping = True
             if isinstance(r, failure.Failure):
                 log.msg('Unexpected error in ipcluster:')
                 log.msg(r.getTraceback())
             log.msg("IPython cluster: stopping")
             d= self.stop_engines()
             d2 = self.stop_controller()
             # Wait a few seconds to let things shut down.
             reactor.callLater(4.0, reactor.stop)
 
     def sigint_handler(self, signum, frame):
         self.stop_launchers()
         
     def start_logging(self):
         # Remove old log files of the controller and engine
         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' + '-'):
                     if f.endswith('.log') or f.endswith('.out') or f.endswith('.err'):
                         os.remove(os.path.join(log_dir, f))
                 if f.startswith('ipcontroller' + '-'):
                     if f.endswith('.log') or f.endswith('.out') or f.endswith('.err'):
                         os.remove(os.path.join(log_dir, f))
         # This will remote old log files for ipcluster itself
         super(IPClusterApp, self).start_logging()
 
     def start_app(self):
         """Start the application, depending on what subcommand is used."""
         subcmd = self.master_config.Global.subcommand
         if subcmd=='create' or subcmd=='list':
             return
         elif subcmd=='start':
             self.start_app_start()
         elif subcmd=='stop':
             self.start_app_stop()
 
     def start_app_start(self):
         """Start the app for the start subcommand."""
         config = self.master_config
         # First see if the cluster is already running
         try:
             pid = self.get_pid_from_file()
         except PIDFileError:
             pass
         else:
             self.log.critical(
                 'Cluster is already running with [pid=%s]. '
                 'use "ipcluster stop" to stop the cluster.' % pid
             )
             # Here I exit with a unusual exit status that other processes
             # can watch for to learn how I existed.
             self.exit(ALREADY_STARTED)
 
         # Now log and daemonize
         self.log.info(
             'Starting ipcluster with [daemon=%r]' % config.Global.daemonize
         )
         # TODO: Get daemonize working on Windows or as a Windows Server.
         if config.Global.daemonize:
             if os.name=='posix':
                 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()
 
     def start_app_stop(self):
         """Start the app for the stop subcommand."""
         config = self.master_config
         try:
             pid = self.get_pid_from_file()
         except PIDFileError:
             self.log.critical(
                 'Problem reading pid file, cluster is probably not running.'
             )
             # Here I exit with a unusual exit status that other processes
             # can watch for to learn how I existed.
             self.exit(ALREADY_STOPPED)
         else:
             if os.name=='posix':
                 sig = config.Global.signal
                 self.log.info(
                     "Stopping cluster [pid=%r] with [signal=%r]" % (pid, sig)
                 )
                 os.kill(pid, sig)
             elif os.name=='nt':
                 # As of right now, we don't support daemonize on Windows, so
                 # stop will not do anything. Minimally, it should clean up the
                 # old .pid files.
                 self.remove_pid_file()
 
 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,265 +1,275 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, Unicode
 
 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 = u'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 = u'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 = Unicode(u'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 = Unicode(u'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.')
     ),
     (('--no-secure',), dict(
         action='store_false', dest='Global.secure', default=NoConfigDefault,
         help='Turn off SSL encryption for all connections.')
     ),
     (('--secure',), dict(
         action='store_true', dest='Global.secure', default=NoConfigDefault,
         help='Turn off SSL encryption for all connections.')
     )
 )
 
 
 class IPControllerAppCLConfigLoader(AppWithClusterDirArgParseConfigLoader):
 
     arguments = cl_args
 
 
+_description = """Start the IPython controller for parallel computing.
+
+The IPython controller provides a gateway between the IPython engines and
+clients. The controller needs to be started before the engines and can be
+configured using command line options or using a cluster directory. Cluster
+directories contain config, log and security files and are usually located in
+your .ipython directory and named as "cluster_<profile>". See the --profile
+and --cluster-dir options for details.
+"""
+
 default_config_file_name = u'ipcontroller_config.py'
 
 
 class IPControllerApp(ApplicationWithClusterDir):
 
     name = u'ipcontroller'
-    description = 'Start the IPython controller for parallel computing.'
+    description = _description
     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):
         # This is the working dir by now.
         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.
         self.main_service.startService()
         # Write the .pid file overwriting old ones. This allow multiple
         # controllers to clober each other. But Windows is not cleaning
         # these up properly.
         self.write_pid_file(overwrite=True)
         # Add a trigger to delete the .pid file upon shutting down.
         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()
 

IPython/kernel/ipengineapp.py

@@ -1,237 +1,248 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
 #-----------------------------------------------------------------------------
 
 import os
 import sys
 
 from twisted.application import service
 from twisted.internet import reactor
 from twisted.python import log
 
 from IPython.config.loader import NoConfigDefault
 
 from IPython.kernel.clusterdir import (
     ApplicationWithClusterDir, 
     AppWithClusterDirArgParseConfigLoader
 )
 from IPython.core import release
 
 from IPython.utils.importstring import import_item
 
 from IPython.kernel.engineservice import EngineService
 from IPython.kernel.fcutil import Tub
 from IPython.kernel.engineconnector import EngineConnector
 
 #-----------------------------------------------------------------------------
 # The main application
 #-----------------------------------------------------------------------------
 
 
 cl_args = (
     # Controller config
     (('--furl-file',), dict(
         type=unicode, dest='Global.furl_file', default=NoConfigDefault,
         help='The full location of the file containing the FURL of the '
         'controller. If this is not given, the FURL file must be in the '
         'security directory of the cluster directory.  This location is '
         'resolved using the --profile and --app-dir options.',
         metavar='Global.furl_file')
     ),
     # MPI
     (('--mpi',), dict(
         type=str, dest='MPI.use', default=NoConfigDefault,
         help='How to enable MPI (mpi4py, pytrilinos, or empty string to disable).',
         metavar='MPI.use')
     ),
     # 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)')
     )
 )
 
 
 class IPEngineAppCLConfigLoader(AppWithClusterDirArgParseConfigLoader):
 
     arguments = cl_args
 
 
 mpi4py_init = """from mpi4py import MPI as mpi
 mpi.size = mpi.COMM_WORLD.Get_size()
 mpi.rank = mpi.COMM_WORLD.Get_rank()
 """
 
 pytrilinos_init = """from PyTrilinos import Epetra
 class SimpleStruct:
 pass
 mpi = SimpleStruct()
 mpi.rank = 0
 mpi.size = 0
 """
 
 
 default_config_file_name = u'ipengine_config.py'
 
 
+_description = """Start an IPython engine for parallel computing.\n\n
+
+IPython engines run in parallel and perform computations on behalf of a client
+and controller. A controller needs to be started before the engines. The
+engine can be configured using command line options or using a cluster
+directory. Cluster directories contain config, log and security files and are
+usually located in your .ipython directory and named as "cluster_<profile>".
+See the --profile and --cluster-dir options for details.
+"""
+
+
 class IPEngineApp(ApplicationWithClusterDir):
 
     name = u'ipengine'
-    description = 'Start the IPython engine for parallel computing.'
+    description = _description
     config_file_name = default_config_file_name
     auto_create_cluster_dir = True
 
     def create_default_config(self):
         super(IPEngineApp, self).create_default_config()
 
         # The engine should not clean logs as we don't want to remove the
         # active log files of other running engines.
         self.default_config.Global.clean_logs = False
 
         # Global config attributes
         self.default_config.Global.exec_lines = []
         self.default_config.Global.shell_class = 'IPython.kernel.core.interpreter.Interpreter'
 
         # Configuration related to the controller
         # This must match the filename (path not included) that the controller
         # used for the FURL file.
         self.default_config.Global.furl_file_name = u'ipcontroller-engine.furl'
         # If given, this is the actual location of the controller's FURL file.
         # If not, this is computed using the profile, app_dir and furl_file_name
         self.default_config.Global.furl_file = u''
 
         # The max number of connection attemps and the initial delay between
         # those attemps.
         self.default_config.Global.connect_delay = 0.1
         self.default_config.Global.connect_max_tries = 15
 
         # MPI related config attributes
         self.default_config.MPI.use = ''
         self.default_config.MPI.mpi4py = mpi4py_init
         self.default_config.MPI.pytrilinos = pytrilinos_init
 
     def create_command_line_config(self):
         """Create and return a command line config loader."""
         return IPEngineAppCLConfigLoader(
             description=self.description, 
             version=release.version
         )
 
     def post_load_command_line_config(self):
         pass
 
     def pre_construct(self):
         super(IPEngineApp, self).pre_construct()
         self.find_cont_furl_file()
 
     def find_cont_furl_file(self):
         """Set the furl file.
 
         Here we don't try to actually see if it exists for is valid as that
         is hadled by the connection logic.
         """
         config = self.master_config
         # Find the actual controller FURL file
         if not config.Global.furl_file:
             try_this = os.path.join(
                 config.Global.cluster_dir, 
                 config.Global.security_dir,
                 config.Global.furl_file_name
             )
             config.Global.furl_file = try_this
 
     def construct(self):
         # This is the working dir by now.
         sys.path.insert(0, '')
 
         self.start_mpi()
         self.start_logging()
 
         # Create the underlying shell class and EngineService
         shell_class = import_item(self.master_config.Global.shell_class)
         self.engine_service = EngineService(shell_class, mpi=mpi)
 
         self.exec_lines()
 
         # Create the service hierarchy
         self.main_service = service.MultiService()
         self.engine_service.setServiceParent(self.main_service)
         self.tub_service = Tub()
         self.tub_service.setServiceParent(self.main_service)
         # This needs to be called before the connection is initiated
         self.main_service.startService()
 
         # This initiates the connection to the controller and calls
         # register_engine to tell the controller we are ready to do work
         self.engine_connector = EngineConnector(self.tub_service)
 
         log.msg("Using furl file: %s" % self.master_config.Global.furl_file)
 
         reactor.callWhenRunning(self.call_connect)
 
     def call_connect(self):
         d = self.engine_connector.connect_to_controller(
             self.engine_service, 
             self.master_config.Global.furl_file,
             self.master_config.Global.connect_delay,
             self.master_config.Global.connect_max_tries
         )
 
         def handle_error(f):
             log.msg('Error connecting to controller. This usually means that '
             'i) the controller was not started, ii) a firewall was blocking '
             'the engine from connecting to the controller or iii) the engine '
             ' was not pointed at the right FURL file:')
             log.msg(f.getErrorMessage())
             reactor.callLater(0.1, reactor.stop)
 
         d.addErrback(handle_error)
 
     def start_mpi(self):
         global mpi
         mpikey = self.master_config.MPI.use
         mpi_import_statement = self.master_config.MPI.get(mpikey, None)
         if mpi_import_statement is not None:
             try:
                 self.log.info("Initializing MPI:")
                 self.log.info(mpi_import_statement)
                 exec mpi_import_statement in globals()
             except:
                 mpi = None
         else:
             mpi = None
 
     def exec_lines(self):
         for line in self.master_config.Global.exec_lines:
             try:
                 log.msg("Executing statement: '%s'" % line)
                 self.engine_service.execute(line)
             except:
                 log.msg("Error executing statement: %s" % line)
 
     def start_app(self):
         reactor.run()
 
 
 def launch_new_instance():
     """Create and run the IPython controller"""
     app = IPEngineApp()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()