parallel docs, tests, default config updated to newconfig

IPython/parallel/apps/clusterdir.py

@@ -1,544 +1,544
 #!/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 logging
 import re
 import shutil
 import sys
 
 from subprocess import Popen, PIPE
 
 from IPython.config.loader import PyFileConfigLoader, Config
 from IPython.config.configurable import Configurable
 from IPython.config.application import Application
 from IPython.core.crashhandler import CrashHandler
 from IPython.core.newapplication import BaseIPythonApplication
 from IPython.core import release
 from IPython.utils.path import (
     get_ipython_package_dir,
     get_ipython_dir,
     expand_path
 )
 from IPython.utils.traitlets import Unicode, Bool, Instance, Dict
 
 #-----------------------------------------------------------------------------
 # Module errors
 #-----------------------------------------------------------------------------
 
 class ClusterDirError(Exception):
     pass
 
 
 class PIDFileError(Exception):
     pass
 
 
 #-----------------------------------------------------------------------------
 # Class for managing cluster directories
 #-----------------------------------------------------------------------------
 
 class ClusterDir(Configurable):
     """An object to manage the cluster directory and its resources.
 
     The cluster directory is used by :command:`ipengine`, 
     :command:`ipcontroller` and :command:`ipclsuter` 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')
     pid_dir_name = Unicode('pid')
     security_dir = Unicode(u'')
     log_dir = Unicode(u'')
     pid_dir = Unicode(u'')
 
     auto_create = Bool(False,
         help="""Whether to automatically create the ClusterDirectory if it does
         not exist""")
     overwrite = Bool(False,
         help="""Whether to overwrite existing config files""")
     location = Unicode(u'', config=True,
         help="""Set the cluster dir. This overrides the logic used by the
         `profile` option.""",
         )
     profile = Unicode(u'default', config=True,
         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."""
         )
 
     _location_isset = Bool(False) # flag for detecting multiply set location
     _new_dir = Bool(False) # flag for whether a new dir was created
 
     def __init__(self, **kwargs):
         # make sure auto_create,overwrite are set *before* location
         for name in ('auto_create', 'overwrite'):
             v = kwargs.pop(name, None)
             if v is not None:
                 setattr(self, name, v)
         super(ClusterDir, self).__init__(**kwargs)
         if not self.location:
             self._profile_changed('profile', 'default', self.profile)
 
     def _location_changed(self, name, old, new):
         if self._location_isset:
             raise RuntimeError("Cannot set ClusterDir more than once.")
         self._location_isset = True
         if not os.path.isdir(new):
             if self.auto_create:# or self.config.ClusterDir.auto_create:
                 os.makedirs(new)
                 self._new_dir = True
             else:
                 raise ClusterDirError('Directory not found: %s' % new)
         
         # ensure config files exist:
         self.copy_all_config_files(overwrite=self.overwrite)
         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 _profile_changed(self, name, old, new):
         if self._location_isset:
             raise RuntimeError("ClusterDir already set.  Cannot set by profile.")
         self.location = os.path.join(get_ipython_dir(), 'cluster_'+new)
 
     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)
 
     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)
         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)
         os.chmod(self.pid_dir, 0700)
 
     def check_dirs(self):
         self.check_security_dir()
         self.check_log_dir()
         self.check_pid_dir()
 
     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 [u'ipcontroller_config.py', u'ipengine_config.py',
                   u'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(location=cluster_dir)
 
     @classmethod
     def create_cluster_dir_by_profile(cls, path, profile=u'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, u'cluster_' + profile)
         return ClusterDir(location=cluster_dir)
 
     @classmethod
     def find_cluster_dir_by_profile(cls, ipython_dir, profile=u'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. ``ipython_dir``
         3. The directories found in the ":" separated 
            :env:`IPCLUSTER_DIR_PATH` environment variable.
 
         Parameters
         ----------
         ipython_dir : 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 = u'cluster_' + profile
         cluster_dir_paths = os.environ.get('IPCLUSTER_DIR_PATH','')
         if cluster_dir_paths:
             cluster_dir_paths = cluster_dir_paths.split(':')
         else:
             cluster_dir_paths = []
         paths = [os.getcwd(), ipython_dir] + cluster_dir_paths
         for p in paths:
             cluster_dir = os.path.join(p, dirname)
             if os.path.isdir(cluster_dir):
                 return ClusterDir(location=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:`IPython.utils.genutils.expand_path`.
         """
         cluster_dir = expand_path(cluster_dir)
         if not os.path.isdir(cluster_dir):
             raise ClusterDirError('Cluster directory not found: %s' % cluster_dir)
         return ClusterDir(location=cluster_dir)
 
 
 #-----------------------------------------------------------------------------
 # Crash handler for this application
 #-----------------------------------------------------------------------------
 
 
 _message_template = """\
 Oops, $self.app_name crashed. We do our best to make it stable, but...
 
 A crash report was automatically generated with the following information:
   - A verbatim copy of the crash traceback.
   - Data on your current $self.app_name configuration.
 
 It was left in the file named:
 \t'$self.crash_report_fname'
 If you can email this file to the developers, the information in it will help
 them in understanding and correcting the problem.
 
 You can mail it to: $self.contact_name at $self.contact_email
 with the subject '$self.app_name Crash Report'.
 
 If you want to do it now, the following command will work (under Unix):
 mail -s '$self.app_name Crash Report' $self.contact_email < $self.crash_report_fname
 
 To ensure accurate tracking of this issue, please file a report about it at:
 $self.bug_tracker
 """
 
 class ClusterDirCrashHandler(CrashHandler):
     """sys.excepthook for IPython itself, leaves a detailed report on disk."""
 
     message_template = _message_template
 
     def __init__(self, app):
         contact_name = release.authors['Min'][0]
         contact_email = release.authors['Min'][1]
         bug_tracker = 'http://github.com/ipython/ipython/issues'
         super(ClusterDirCrashHandler,self).__init__(
             app, contact_name, contact_email, bug_tracker
         )
 
 
 #-----------------------------------------------------------------------------
 # Main application
 #-----------------------------------------------------------------------------
 base_aliases = {
     'profile' : "ClusterDir.profile",
     'cluster_dir' : 'ClusterDir.location',
     'auto_create' : 'ClusterDirApplication.auto_create',
     'log_level' : 'ClusterApplication.log_level',
     'work_dir' : 'ClusterApplication.work_dir',
     'log_to_file' : 'ClusterApplication.log_to_file',
     'clean_logs' : 'ClusterApplication.clean_logs',
     'log_url' : 'ClusterApplication.log_url',
 }
 
 base_flags = {
     'debug' : ( {"ClusterApplication" : {"log_level" : logging.DEBUG}}, "set loglevel to DEBUG"),
     'quiet' : ( {"ClusterApplication" : {"log_level" : logging.CRITICAL}}, "set loglevel to CRITICAL (minimal output)"),
     'log-to-file' : ( {"ClusterApplication" : {"log_to_file" : True}}, "redirect log output to a file"),
 }
 for k,v in base_flags.iteritems():
     base_flags[k] = (Config(v[0]),v[1])
 
 class ClusterApplication(BaseIPythonApplication):
     """An application that puts everything into a cluster directory.
 
     Instead of looking for things in the ipython_dir, 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 
+    * 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 ipython_dir 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.
     """
 
     crash_handler_class = ClusterDirCrashHandler
     auto_create_cluster_dir = Bool(True, config=True,
         help="whether to create the cluster_dir if it doesn't exist")
     cluster_dir = Instance(ClusterDir)
     classes = [ClusterDir]
     
     def _log_level_default(self):
         # temporarily override default_log_level to INFO
         return logging.INFO
 
     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, shortname='--clean-logs', config=True,
         help="whether to cleanup old logfiles before starting")
 
     log_url = Unicode('', shortname='--log-url', config=True,
         help="The ZMQ URL of the iplogger to aggregate logging.")
 
     config_file = Unicode(u'', config=True,
         help="""Path to ipcontroller configuration file.  The default is to use
          <appname>_config.py, as found by cluster-dir."""
     )
     
     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)
 
     def init_clusterdir(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:
             self.cluster_dir = ClusterDir(auto_create=self.auto_create_cluster_dir, config=self.config)
         except ClusterDirError as e:
             self.log.fatal("Error initializing cluster dir: %s"%e)
             self.log.fatal("A cluster dir must be created before running this command.")
             self.log.fatal("Do 'ipcluster create -h' or 'ipcluster list -h' for more "
             "information about creating and listing cluster dirs."
             )
             self.exit(1)
             
         if self.cluster_dir._new_dir:
             self.log.info('Creating new cluster dir: %s' % \
                             self.cluster_dir.location)
         else:
             self.log.info('Using existing cluster dir: %s' % \
                             self.cluster_dir.location)
     
     def initialize(self, argv=None):
         """initialize the app"""
         self.init_crash_handler()
         self.parse_command_line(argv)
         cl_config = self.config
         self.init_clusterdir()
         if self.config_file:
             self.load_config_file(self.config_file)
         elif self.default_config_file_name:
             try:
                 self.load_config_file(self.default_config_file_name, 
                                         path=self.cluster_dir.location)
             except IOError:
                 self.log.warn("Warning: Default config file not found")
         # command-line should *override* config file, but command-line is necessary
         # to determine clusterdir, etc.
         self.update_config(cl_config)
         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 load_config_file(self, filename, path=None):
         """Load a .py based config file by filename and path."""
         # use config.application.Application.load_config
         # instead of inflexible core.newapplication.BaseIPythonApplication.load_config
         return Application.load_config_file(self, filename, path=path)
     #
     # def load_default_config_file(self):
     #     """Load a .py based config file by filename and path."""
     #     return BaseIPythonApplication.load_config_file(self)
 
     # disable URL-logging
     def reinit_logging(self):
         # Remove old log files
         log_dir = self.cluster_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):
                     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)
 
     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.cluster_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.cluster_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.cluster_dir.pid_dir, self.name + u'.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)
     
     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

IPython/parallel/apps/ipclusterapp.py

@@ -1,537 +1,542
 #!/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 errno
 import logging
 import os
 import re
 import signal
 
 from subprocess import check_call, CalledProcessError, PIPE
 import zmq
 from zmq.eventloop import ioloop
 
 from IPython.config.application import Application, boolean_flag
 from IPython.config.loader import Config
 from IPython.core.newapplication import BaseIPythonApplication
 from IPython.utils.importstring import import_item
 from IPython.utils.traitlets import Int, Unicode, Bool, CFloat, Dict, List
 
 from IPython.parallel.apps.clusterdir import (
     ClusterApplication, ClusterDirError, ClusterDir,
     PIDFileError,
     base_flags, base_aliases
 )
 
 
 #-----------------------------------------------------------------------------
 # Module level variables
 #-----------------------------------------------------------------------------
 
 
 default_config_file_name = u'ipcluster_config.py'
 
 
-_description = """\
-Start an IPython cluster for parallel computing.\n\n
+_description = """Start an IPython cluster for parallel computing.
 
 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 profile=mycluster', then edit
 configuration files, followed by 'ipcluster start profile=mycluster n=4'.
 """
 
 
 # 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
 
 # This will be the exit code if ipcluster engines is run, but there is not .pid
 # file to be found.
 NO_CLUSTER = 12
 
 
 #-----------------------------------------------------------------------------
 # Main application
 #-----------------------------------------------------------------------------
-start_help = """
+start_help = """Start an IPython cluster for parallel computing
+
 Start an ipython cluster by its profile name or cluster
 directory. Cluster directories contain configuration, log and
 security related files and are named using the convention
 'cluster_<profile>' and should be creating using the 'start'
 subcommand of 'ipcluster'. If your cluster directory is in 
 the cwd or the ipython directory, you can simply refer to it
 using its profile name, 'ipcluster start n=4 profile=<profile>`,
 otherwise use the 'cluster_dir' option.
 """
-stop_help = """
+stop_help = """Stop a running IPython cluster
+
 Stop a running ipython cluster by its profile name or cluster
 directory. Cluster directories are named using the convention
 'cluster_<profile>'. If your cluster directory is in 
 the cwd or the ipython directory, you can simply refer to it
 using its profile name, 'ipcluster stop profile=<profile>`, otherwise
 use the 'cluster_dir' option.
 """
-engines_help = """
+engines_help = """Start engines connected to an existing IPython cluster
+
 Start one or more engines to connect to an existing Cluster
 by profile name or cluster directory.
 Cluster directories contain configuration, log and
 security related files and are named using the convention
 'cluster_<profile>' and should be creating using the 'start'
 subcommand of 'ipcluster'. If your cluster directory is in 
 the cwd or the ipython directory, you can simply refer to it
 using its profile name, 'ipcluster engines n=4 profile=<profile>`,
 otherwise use the 'cluster_dir' option.
 """
-create_help = """
+create_help = """Create an ipcluster profile by name
+
 Create an ipython cluster directory by its profile name or
 cluster directory path. Cluster directories contain
 configuration, log and security related files and are named
 using the convention 'cluster_<profile>'. By default they are
 located in your ipython directory. Once created, you will
 probably need to edit the configuration files in the cluster
 directory to configure your cluster. Most users will create a
 cluster directory by profile name,
 `ipcluster create profile=mycluster`, which will put the directory
 in `<ipython_dir>/cluster_mycluster`.
 """
-list_help = """List all available clusters, by cluster directory, that can
+list_help = """List available cluster profiles
+
+List all available clusters, by cluster directory, that can
 be found in the current working directly or in the ipython
 directory. Cluster directories are named using the convention
 'cluster_<profile>'.
 """
 
 
 class IPClusterList(BaseIPythonApplication):
     name = u'ipcluster-list'
     description = list_help
     
     # empty aliases
     aliases=Dict()
     flags = Dict(base_flags)
     
     def _log_level_default(self):
         return 20
     
     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 = []
         
         ipython_dir = self.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 profile=%s n=4' % profile
                     print start_cmd + " ==> " + full_path
     
     def start(self):
         self.list_cluster_dirs()
 
 create_flags = {}
 create_flags.update(base_flags)
 create_flags.update(boolean_flag('reset', 'IPClusterCreate.reset', 
                 "reset config files to defaults", "leave existing config files"))
 
 class IPClusterCreate(ClusterApplication):
     name = u'ipcluster'
     description = create_help
     auto_create_cluster_dir = Bool(True,
         help="whether to create the cluster_dir if it doesn't exist")
     default_config_file_name = default_config_file_name
     
     reset = Bool(False, config=True,
         help="Whether to reset config files as part of 'create'."
         )
     
     flags = Dict(create_flags)
     
     aliases = Dict(dict(profile='ClusterDir.profile'))
     
     classes = [ClusterDir]
     
     def init_clusterdir(self):
         super(IPClusterCreate, self).init_clusterdir()
         self.log.info('Copying default config files to cluster directory '
         '[overwrite=%r]' % (self.reset,))
         self.cluster_dir.copy_all_config_files(overwrite=self.reset)
     
     def initialize(self, argv=None):
         self.parse_command_line(argv)
         self.init_clusterdir()
     
 stop_aliases = dict(
     signal='IPClusterStop.signal',
     profile='ClusterDir.profile',
     cluster_dir='ClusterDir.location',
 )
 
 class IPClusterStop(ClusterApplication):
     name = u'ipcluster'
     description = stop_help
     auto_create_cluster_dir = Bool(False)
     default_config_file_name = default_config_file_name
     
     signal = Int(signal.SIGINT, config=True,
         help="signal to use for stopping processes.")
         
     aliases = Dict(stop_aliases)
     
     def init_clusterdir(self):
         try:
             super(IPClusterStop, self).init_clusterdir()
         except ClusterDirError as e:
             self.log.fatal("Failed ClusterDir init: %s"%e)
             self.exit(1)
     
     def start(self):
         """Start the app for the stop subcommand."""
         try:
             pid = self.get_pid_from_file()
         except PIDFileError:
             self.log.critical(
                 'Could not read 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.remove_pid_file()
             self.exit(ALREADY_STOPPED)
         
         if not self.check_pid(pid):
             self.log.critical(
                 'Cluster [pid=%r] is not running.' % pid
             )
             self.remove_pid_file()
             # Here I exit with a unusual exit status that other processes
             # can watch for to learn how I existed.
             self.exit(ALREADY_STOPPED)
             
         elif os.name=='posix':
             sig = self.signal
             self.log.info(
                 "Stopping cluster [pid=%r] with [signal=%r]" % (pid, sig)
             )
             try:
                 os.kill(pid, sig)
             except OSError:
                 self.log.error("Stopping cluster failed, assuming already dead.",
                     exc_info=True)
                 self.remove_pid_file()
         elif os.name=='nt':
             try:
                 # kill the whole tree
                 p = check_call(['taskkill', '-pid', str(pid), '-t', '-f'], stdout=PIPE,stderr=PIPE)
             except (CalledProcessError, OSError):
                 self.log.error("Stopping cluster failed, assuming already dead.",
                     exc_info=True)
             self.remove_pid_file()
     
 engine_aliases = {}
 engine_aliases.update(base_aliases)
 engine_aliases.update(dict(
     n='IPClusterEngines.n',
     elauncher = 'IPClusterEngines.engine_launcher_class',
 ))
 class IPClusterEngines(ClusterApplication):
 
     name = u'ipcluster'
     description = engines_help
     usage = None
     default_config_file_name = default_config_file_name
     default_log_level = logging.INFO
     auto_create_cluster_dir = Bool(False)
     classes = List()
     def _classes_default(self):
         from IPython.parallel.apps import launcher
         launchers = launcher.all_launchers
         eslaunchers = [ l for l in launchers if 'EngineSet' in l.__name__]
         return [ClusterDir]+eslaunchers
     
     n = Int(2, config=True,
         help="The number of engines to start.")
 
     engine_launcher_class = Unicode('LocalEngineSetLauncher',
         config=True,
         help="The class for launching a set of Engines."
         )
     daemonize = Bool(False, config=True,
         help='Daemonize the ipcluster program. This implies --log-to-file')
 
     def _daemonize_changed(self, name, old, new):
         if new:
             self.log_to_file = True
 
     aliases = Dict(engine_aliases)
     # flags = Dict(flags)
     _stopping = False
 
     def initialize(self, argv=None):
         super(IPClusterEngines, self).initialize(argv)
         self.init_signal()
         self.init_launchers()
     
     def init_launchers(self):
         self.engine_launcher = self.build_launcher(self.engine_launcher_class)
         self.engine_launcher.on_stop(lambda r: self.loop.stop())
     
     def init_signal(self):
         # Setup signals
         signal.signal(signal.SIGINT, self.sigint_handler)
     
     def build_launcher(self, clsname):
         """import and instantiate a Launcher based on importstring"""
         if '.' not in clsname:
             # not a module, presume it's the raw name in apps.launcher
             clsname = 'IPython.parallel.apps.launcher.'+clsname
         # print repr(clsname)
         klass = import_item(clsname)
 
         launcher = klass(
             work_dir=self.cluster_dir.location, config=self.config, logname=self.log.name
         )
         return launcher
     
     def start_engines(self):
         self.log.info("Starting %i engines"%self.n)
         self.engine_launcher.start(
             self.n,
             cluster_dir=self.cluster_dir.location
         )
 
     def stop_engines(self):
         self.log.info("Stopping Engines...")
         if self.engine_launcher.running:
             d = self.engine_launcher.stop()
             return d
         else:
             return None
 
     def stop_launchers(self, r=None):
         if not self._stopping:
             self._stopping = True
             self.log.error("IPython cluster: stopping")
             self.stop_engines()
             # Wait a few seconds to let things shut down.
             dc = ioloop.DelayedCallback(self.loop.stop, 4000, self.loop)
             dc.start()
 
     def sigint_handler(self, signum, frame):
         self.log.debug("SIGINT received, stopping launchers...")
         self.stop_launchers()
         
     def start_logging(self):
         # Remove old log files of the controller and engine
         if self.clean_logs:
             log_dir = self.cluster_dir.log_dir
             for f in os.listdir(log_dir):
                 if re.match(r'ip(engine|controller)z-\d+\.(log|err|out)',f):
                     os.remove(os.path.join(log_dir, f))
         # This will remove old log files for ipcluster itself
         # super(IPClusterApp, self).start_logging()
 
     def start(self):
         """Start the app for the engines subcommand."""
         self.log.info("IPython cluster: started")
         # First see if the cluster is already running
         
         # Now log and daemonize
         self.log.info(
             'Starting engines with [daemon=%r]' % self.daemonize
         )
         # TODO: Get daemonize working on Windows or as a Windows Server.
         if self.daemonize:
             if os.name=='posix':
                 from twisted.scripts._twistd_unix import daemonize
                 daemonize()
 
         dc = ioloop.DelayedCallback(self.start_engines, 0, self.loop)
         dc.start()
         # Now write the new pid file AFTER our new forked pid is active.
         # self.write_pid_file()
         try:
             self.loop.start()
         except KeyboardInterrupt:
             pass
         except zmq.ZMQError as e:
             if e.errno == errno.EINTR:
                 pass
             else:
                 raise
 
 start_aliases = {}
 start_aliases.update(engine_aliases)
 start_aliases.update(dict(
     delay='IPClusterStart.delay',
     clean_logs='IPClusterStart.clean_logs',
 ))
 
 class IPClusterStart(IPClusterEngines):
 
     name = u'ipcluster'
     description = start_help
     usage = None
     default_config_file_name = default_config_file_name
     default_log_level = logging.INFO
     auto_create_cluster_dir = Bool(True, config=True,
         help="whether to create the cluster_dir if it doesn't exist")
     classes = List()
     def _classes_default(self,):
         from IPython.parallel.apps import launcher
         return [ClusterDir]+launcher.all_launchers
     
     clean_logs = Bool(True, config=True, 
         help="whether to cleanup old logs before starting")
 
     delay = CFloat(1., config=True,
         help="delay (in s) between starting the controller and the engines")
 
     controller_launcher_class = Unicode('LocalControllerLauncher',
         config=True,
         help="The class for launching a Controller."
         )
     reset = Bool(False, config=True,
         help="Whether to reset config files as part of '--create'."
         )
     
     # flags = Dict(flags)
     aliases = Dict(start_aliases)
 
     def init_launchers(self):
         self.controller_launcher = self.build_launcher(self.controller_launcher_class)
         self.engine_launcher = self.build_launcher(self.engine_launcher_class)
         self.controller_launcher.on_stop(self.stop_launchers)
     
     def start_controller(self):
         self.controller_launcher.start(
             cluster_dir=self.cluster_dir.location
         )
         
     def stop_controller(self):
         # self.log.info("In stop_controller")
         if self.controller_launcher and self.controller_launcher.running:
             return self.controller_launcher.stop()
 
     def stop_launchers(self, r=None):
         if not self._stopping:
             self.stop_controller()
             super(IPClusterStart, self).stop_launchers()
 
     def start(self):
         """Start the app for the start subcommand."""
         # First see if the cluster is already running
         try:
             pid = self.get_pid_from_file()
         except PIDFileError:
             pass
         else:
             if self.check_pid(pid):
                 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)
             else:
                 self.remove_pid_file()
                 
 
         # Now log and daemonize
         self.log.info(
             'Starting ipcluster with [daemon=%r]' % self.daemonize
         )
         # TODO: Get daemonize working on Windows or as a Windows Server.
         if self.daemonize:
             if os.name=='posix':
                 from twisted.scripts._twistd_unix import daemonize
                 daemonize()
 
         dc = ioloop.DelayedCallback(self.start_controller, 0, self.loop)
         dc.start()
         dc = ioloop.DelayedCallback(self.start_engines, 1000*self.delay, self.loop)
         dc.start()
         # Now write the new pid file AFTER our new forked pid is active.
         self.write_pid_file()
         try:
             self.loop.start()
         except KeyboardInterrupt:
             pass
         except zmq.ZMQError as e:
             if e.errno == errno.EINTR:
                 pass
             else:
                 raise
         finally:
             self.remove_pid_file()
 
 base='IPython.parallel.apps.ipclusterapp.IPCluster'
 
 class IPClusterApp(Application):
     name = u'ipcluster'
     description = _description
     
     subcommands = {'create' : (base+'Create', create_help),
                 'list' : (base+'List', list_help),
                 'start' : (base+'Start', start_help),
                 'stop' : (base+'Stop', stop_help),
                 'engines' : (base+'Engines', engines_help),
     }
     
     # no aliases or flags for parent App
     aliases = Dict()
     flags = Dict()
     
     def start(self):
         if self.subapp is None:
             print "No subcommand specified! Must specify one of: %s"%(self.subcommands.keys())
             print
             self.print_subcommands()
             self.exit(1)
         else:
             return self.subapp.start()
 
 def launch_new_instance():
     """Create and run the IPython cluster."""
     app = IPClusterApp()
     app.initialize()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()
 

IPython/parallel/apps/ipcontrollerapp.py

@@ -1,403 +1,405
 #!/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 logging
 import socket
 import stat
 import sys
 import uuid
 
 from multiprocessing import Process
 
 import zmq
 from zmq.devices import ProcessMonitoredQueue
 from zmq.log.handlers import PUBHandler
 from zmq.utils import jsonapi as json
 
 from IPython.config.loader import Config
 
 from IPython.parallel import factory
 
 from IPython.parallel.apps.clusterdir import (
     ClusterDir,
     ClusterApplication,
     base_flags
     # ClusterDirConfigLoader
 )
 from IPython.utils.importstring import import_item
 from IPython.utils.traitlets import Instance, Unicode, Bool, List, Dict
 
 # from IPython.parallel.controller.controller import ControllerFactory
 from IPython.parallel.streamsession import StreamSession
 from IPython.parallel.controller.heartmonitor import HeartMonitor
 from IPython.parallel.controller.hub import Hub, HubFactory
 from IPython.parallel.controller.scheduler import TaskScheduler,launch_scheduler
 from IPython.parallel.controller.sqlitedb import SQLiteDB
 
 from IPython.parallel.util import signal_children,disambiguate_ip_address, split_url
 
 # conditional import of MongoDB backend class
 
 try:
     from IPython.parallel.controller.mongodb import MongoDB
 except ImportError:
     maybe_mongo = []
 else:
     maybe_mongo = [MongoDB]
 
 
 #-----------------------------------------------------------------------------
 # Module level variables
 #-----------------------------------------------------------------------------
 
 
 #: The default config file name for this application
 default_config_file_name = u'ipcontroller_config.py'
 
 
 _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.
+your ipython directory and named as "cluster_<profile>". See the `profile`
+and `cluster_dir` options for details.
 """
 
 
 
 
 #-----------------------------------------------------------------------------
 # The main application
 #-----------------------------------------------------------------------------
 flags = {}
 flags.update(base_flags)
 flags.update({
-    'usethreads' : ( {'IPControllerApp' : {'usethreads' : True}},
+    'usethreads' : ( {'IPControllerApp' : {'use_threads' : True}},
                     'Use threads instead of processes for the schedulers'),
-    'sqlitedb' : ({'HubFactory' : {'db_class' : 'IPython.parallel.controller.sqlitedb.SQLiteDB'}},
+    'sqlitedb' : ({'HubFactory' : Config({'db_class' : 'IPython.parallel.controller.sqlitedb.SQLiteDB'})},
                     'use the SQLiteDB backend'),
-    'mongodb' : ({'HubFactory' : {'db_class' : 'IPython.parallel.controller.mongodb.MongoDB'}},
+    'mongodb' : ({'HubFactory' : Config({'db_class' : 'IPython.parallel.controller.mongodb.MongoDB'})},
                     'use the MongoDB backend'),
-    'dictdb' : ({'HubFactory' : {'db_class' : 'IPython.parallel.controller.dictdb.DictDB'}},
+    'dictdb' : ({'HubFactory' : Config({'db_class' : 'IPython.parallel.controller.dictdb.DictDB'})},
                     'use the in-memory DictDB backend'),
+    'reuse' : ({'IPControllerApp' : Config({'reuse_files' : True})},
+                    'reuse existing json connection files')
 })
 
 flags.update()
 
 class IPControllerApp(ClusterApplication):
 
     name = u'ipcontroller'
     description = _description
     # command_line_loader = IPControllerAppConfigLoader
     default_config_file_name = default_config_file_name
     classes = [ClusterDir, StreamSession, HubFactory, TaskScheduler, HeartMonitor, SQLiteDB] + maybe_mongo
     
     auto_create_cluster_dir = Bool(True, config=True,
         help="Whether to create cluster_dir if it exists.")
     reuse_files = Bool(False, config=True,
         help='Whether to reuse existing json connection files [default: False]'
     )
     secure = Bool(True, config=True,
         help='Whether to use exec_keys for extra authentication [default: True]'
     )
     ssh_server = Unicode(u'', config=True,
         help="""ssh url for clients to use when connecting to the Controller
         processes. It should be of the form: [user@]server[:port]. The
         Controller\'s listening addresses must be accessible from the ssh server""",
     )
     location = Unicode(u'', config=True,
         help="""The external IP or domain name of the Controller, used for disambiguating
         engine and client connections.""",
     )
     import_statements = List([], config=True,
         help="import statements to be run at startup.  Necessary in some environments"
     )
 
-    usethreads = Bool(False, config=True,
+    use_threads = Bool(False, config=True,
         help='Use threads instead of processes for the schedulers',
         )
 
     # internal
     children = List()
     mq_class = Unicode('zmq.devices.ProcessMonitoredQueue')
 
-    def _usethreads_changed(self, name, old, new):
+    def _use_threads_changed(self, name, old, new):
         self.mq_class = 'zmq.devices.%sMonitoredQueue'%('Thread' if new else 'Process')
 
     aliases = Dict(dict(
         config = 'IPControllerApp.config_file',
         # file = 'IPControllerApp.url_file',
         log_level = 'IPControllerApp.log_level',
         log_url = 'IPControllerApp.log_url',
         reuse_files = 'IPControllerApp.reuse_files',
         secure = 'IPControllerApp.secure',
         ssh = 'IPControllerApp.ssh_server',
-        usethreads = 'IPControllerApp.usethreads',
+        use_threads = 'IPControllerApp.use_threads',
         import_statements = 'IPControllerApp.import_statements',
         location = 'IPControllerApp.location',
 
         ident = 'StreamSession.session',
         user = 'StreamSession.username',
         exec_key = 'StreamSession.keyfile',
 
         url = 'HubFactory.url',
         ip = 'HubFactory.ip',
         transport = 'HubFactory.transport',
         port = 'HubFactory.regport',
 
         ping = 'HeartMonitor.period',
 
         scheme = 'TaskScheduler.scheme_name',
         hwm = 'TaskScheduler.hwm',
 
 
         profile = "ClusterDir.profile",
         cluster_dir = 'ClusterDir.location',
         
     ))
     flags = Dict(flags)
     
 
     def save_connection_dict(self, fname, cdict):
         """save a connection dict to json file."""
         c = self.config
         url = cdict['url']
         location = cdict['location']
         if not location:
             try:
                 proto,ip,port = split_url(url)
             except AssertionError:
                 pass
             else:
                 location = socket.gethostbyname_ex(socket.gethostname())[2][-1]
             cdict['location'] = location
         fname = os.path.join(self.cluster_dir.security_dir, fname)
         with open(fname, 'w') as f:
             f.write(json.dumps(cdict, indent=2))
         os.chmod(fname, stat.S_IRUSR|stat.S_IWUSR)
     
     def load_config_from_json(self):
         """load config from existing json connector files."""
         c = self.config
         # load from engine config
         with open(os.path.join(self.cluster_dir.security_dir, 'ipcontroller-engine.json')) as f:
             cfg = json.loads(f.read())
         key = c.StreamSession.key = cfg['exec_key']
         xport,addr = cfg['url'].split('://')
         c.HubFactory.engine_transport = xport
         ip,ports = addr.split(':')
         c.HubFactory.engine_ip = ip
         c.HubFactory.regport = int(ports)
         self.location = cfg['location']
         
         # load client config
         with open(os.path.join(self.cluster_dir.security_dir, 'ipcontroller-client.json')) as f:
             cfg = json.loads(f.read())
         assert key == cfg['exec_key'], "exec_key mismatch between engine and client keys"
         xport,addr = cfg['url'].split('://')
         c.HubFactory.client_transport = xport
         ip,ports = addr.split(':')
         c.HubFactory.client_ip = ip
         self.ssh_server = cfg['ssh']
         assert int(ports) == c.HubFactory.regport, "regport mismatch"
     
     def init_hub(self):
         c = self.config
         
         self.do_import_statements()
         reusing = self.reuse_files
         if reusing:
             try:
                 self.load_config_from_json()
             except (AssertionError,IOError):
                 reusing=False
         # check again, because reusing may have failed:
         if reusing:
             pass
         elif self.secure:
             key = str(uuid.uuid4())
             # keyfile = os.path.join(self.cluster_dir.security_dir, self.exec_key)
             # with open(keyfile, 'w') as f:
             #     f.write(key)
             # os.chmod(keyfile, stat.S_IRUSR|stat.S_IWUSR)
             c.StreamSession.key = key
         else:
             key = c.StreamSession.key = ''
         
         try:
             self.factory = HubFactory(config=c, log=self.log)
             # self.start_logging()
             self.factory.init_hub()
         except:
             self.log.error("Couldn't construct the Controller", exc_info=True)
             self.exit(1)
         
         if not reusing:
             # save to new json config files
             f = self.factory
             cdict = {'exec_key' : key,
                     'ssh' : self.ssh_server,
                     'url' : "%s://%s:%s"%(f.client_transport, f.client_ip, f.regport),
                     'location' : self.location
                     }
             self.save_connection_dict('ipcontroller-client.json', cdict)
             edict = cdict
             edict['url']="%s://%s:%s"%((f.client_transport, f.client_ip, f.regport))
             self.save_connection_dict('ipcontroller-engine.json', edict)
 
     #
     def init_schedulers(self):
         children = self.children
         mq = import_item(str(self.mq_class))
         
         hub = self.factory
-        # maybe_inproc = 'inproc://monitor' if self.usethreads else self.monitor_url
+        # maybe_inproc = 'inproc://monitor' if self.use_threads else self.monitor_url
         # IOPub relay (in a Process)
         q = mq(zmq.PUB, zmq.SUB, zmq.PUB, 'N/A','iopub')
         q.bind_in(hub.client_info['iopub'])
         q.bind_out(hub.engine_info['iopub'])
         q.setsockopt_out(zmq.SUBSCRIBE, '')
         q.connect_mon(hub.monitor_url)
         q.daemon=True
         children.append(q)
 
         # Multiplexer Queue (in a Process)
         q = mq(zmq.XREP, zmq.XREP, zmq.PUB, 'in', 'out')
         q.bind_in(hub.client_info['mux'])
         q.setsockopt_in(zmq.IDENTITY, 'mux')
         q.bind_out(hub.engine_info['mux'])
         q.connect_mon(hub.monitor_url)
         q.daemon=True
         children.append(q)
 
         # Control Queue (in a Process)
         q = mq(zmq.XREP, zmq.XREP, zmq.PUB, 'incontrol', 'outcontrol')
         q.bind_in(hub.client_info['control'])
         q.setsockopt_in(zmq.IDENTITY, 'control')
         q.bind_out(hub.engine_info['control'])
         q.connect_mon(hub.monitor_url)
         q.daemon=True
         children.append(q)
         try:
             scheme = self.config.TaskScheduler.scheme_name
         except AttributeError:
             scheme = TaskScheduler.scheme_name.get_default_value()
         # Task Queue (in a Process)
         if scheme == 'pure':
             self.log.warn("task::using pure XREQ Task scheduler")
             q = mq(zmq.XREP, zmq.XREQ, zmq.PUB, 'intask', 'outtask')
             # q.setsockopt_out(zmq.HWM, hub.hwm)
             q.bind_in(hub.client_info['task'][1])
             q.setsockopt_in(zmq.IDENTITY, 'task')
             q.bind_out(hub.engine_info['task'])
             q.connect_mon(hub.monitor_url)
             q.daemon=True
             children.append(q)
         elif scheme == 'none':
             self.log.warn("task::using no Task scheduler")
 
         else:
             self.log.info("task::using Python %s Task scheduler"%scheme)
             sargs = (hub.client_info['task'][1], hub.engine_info['task'],
                                 hub.monitor_url, hub.client_info['notification'])
             kwargs = dict(logname='scheduler', loglevel=self.log_level,
                             log_url = self.log_url, config=dict(self.config))
             q = Process(target=launch_scheduler, args=sargs, kwargs=kwargs)
             q.daemon=True
             children.append(q)
 
     
     def save_urls(self):
         """save the registration urls to files."""
         c = self.config
         
         sec_dir = self.cluster_dir.security_dir
         cf = self.factory
         
         with open(os.path.join(sec_dir, 'ipcontroller-engine.url'), 'w') as f:
             f.write("%s://%s:%s"%(cf.engine_transport, cf.engine_ip, cf.regport))
         
         with open(os.path.join(sec_dir, 'ipcontroller-client.url'), 'w') as f:
             f.write("%s://%s:%s"%(cf.client_transport, cf.client_ip, cf.regport))
         
     
     def do_import_statements(self):
         statements = self.import_statements
         for s in statements:
             try:
                 self.log.msg("Executing statement: '%s'" % s)
                 exec s in globals(), locals()
             except:
                 self.log.msg("Error running statement: %s" % s)
 
     def forward_logging(self):
         if self.log_url:
             self.log.info("Forwarding logging to %s"%self.log_url)
             context = zmq.Context.instance()
             lsock = context.socket(zmq.PUB)
             lsock.connect(self.log_url)
             handler = PUBHandler(lsock)
             self.log.removeHandler(self._log_handler)
             handler.root_topic = 'controller'
             handler.setLevel(self.log_level)
             self.log.addHandler(handler)
             self._log_handler = handler
     # #
     
     def initialize(self, argv=None):
         super(IPControllerApp, self).initialize(argv)
         self.forward_logging()
         self.init_hub()
         self.init_schedulers()
     
     def start(self):
         # Start the subprocesses:
         self.factory.start()
         child_procs = []
         for child in self.children:
             child.start()
             if isinstance(child, ProcessMonitoredQueue):
                 child_procs.append(child.launcher)
             elif isinstance(child, Process):
                 child_procs.append(child)
         if child_procs:
             signal_children(child_procs)
 
         self.write_pid_file(overwrite=True)
 
         try:
             self.factory.loop.start()
         except KeyboardInterrupt:
             self.log.critical("Interrupted, Exiting...\n")
             
 
 
 def launch_new_instance():
     """Create and run the IPython controller"""
     app = IPControllerApp()
     app.initialize()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()

IPython/parallel/apps/ipengineapp.py

@@ -1,277 +1,277
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The IPython engine 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 json
 import os
 import sys
 
 import zmq
 from zmq.eventloop import ioloop
 
 from IPython.parallel.apps.clusterdir import (
     ClusterApplication,
     ClusterDir,
     # ClusterDirConfigLoader
 )
 from IPython.zmq.log import EnginePUBHandler
 
 from IPython.config.configurable import Configurable
 from IPython.parallel.streamsession import StreamSession
 from IPython.parallel.engine.engine import EngineFactory
 from IPython.parallel.engine.streamkernel import Kernel
 from IPython.parallel.util import disambiguate_url
 
 from IPython.utils.importstring import import_item
 from IPython.utils.traitlets import Bool, Unicode, Dict, List
 
 
 #-----------------------------------------------------------------------------
 # Module level variables
 #-----------------------------------------------------------------------------
 
 #: The default config file name for this application
 default_config_file_name = u'ipengine_config.py'
 
-_description = """Start an IPython engine for parallel computing.\n\n
+_description = """Start an IPython engine for parallel computing.
 
 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.
 """
 
 
 #-----------------------------------------------------------------------------
 # MPI configuration
 #-----------------------------------------------------------------------------
 
 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
 """
 
 class MPI(Configurable):
     """Configurable for MPI initialization"""
     use = Unicode('', config=True,
         help='How to enable MPI (mpi4py, pytrilinos, or empty string to disable).'
         )
 
     def _on_use_changed(self, old, new):
         # load default init script if it's not set
         if not self.init_script:
             self.init_script = self.default_inits.get(new, '')
 
     init_script = Unicode('', config=True,
         help="Initialization code for MPI")
 
     default_inits = Dict({'mpi4py' : mpi4py_init, 'pytrilinos':pytrilinos_init},
         config=True)
 
 
 #-----------------------------------------------------------------------------
 # Main application
 #-----------------------------------------------------------------------------
 
 
 class IPEngineApp(ClusterApplication):
 
     app_name = Unicode(u'ipengine')
     description = Unicode(_description)
     default_config_file_name = default_config_file_name
     classes = List([ClusterDir, StreamSession, EngineFactory, Kernel, MPI])
 
     auto_create_cluster_dir = Bool(False,
         help="whether to create the cluster_dir if it doesn't exist")
     
     startup_script = Unicode(u'', config=True,
         help='specify a script to be run at startup')
     startup_command = Unicode('', config=True,
             help='specify a command to be run at startup')
 
     url_file = Unicode(u'', config=True,
         help="""The full location of the file containing the connection information for
         the controller. If this is not given, the file must be in the
         security directory of the cluster directory.  This location is
         resolved using the `profile` or `cluster_dir` options.""",
         )
 
     url_file_name = Unicode(u'ipcontroller-engine.json')
     log_url = Unicode('', config=True,
         help="""The URL for the iploggerapp instance, for forwarding
         logging to a central location.""")
 
     aliases = Dict(dict(
         config = 'IPEngineApp.config_file',
         file = 'IPEngineApp.url_file',
         c = 'IPEngineApp.startup_command',
         s = 'IPEngineApp.startup_script',
 
         ident = 'StreamSession.session',
         user = 'StreamSession.username',
         exec_key = 'StreamSession.keyfile',
 
         url = 'EngineFactory.url',
         ip = 'EngineFactory.ip',
         transport = 'EngineFactory.transport',
         port = 'EngineFactory.regport',
         location = 'EngineFactory.location',
 
         timeout = 'EngineFactory.timeout',
 
         profile = "ClusterDir.profile",
         cluster_dir = 'ClusterDir.location',
 
         mpi = 'MPI.use',
 
         log_level = 'IPEngineApp.log_level',
         log_url = 'IPEngineApp.log_url'
     ))
 
     # def find_key_file(self):
     #     """Set the key 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 key file
     #     if not config.Global.key_file:
     #         try_this = os.path.join(
     #             config.Global.cluster_dir, 
     #             config.Global.security_dir,
     #             config.Global.key_file_name
     #         )
     #         config.Global.key_file = try_this
         
     def find_url_file(self):
         """Set the key 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.config
         # Find the actual controller key file
         if not self.url_file:
             self.url_file = os.path.join(
                 self.cluster_dir.security_dir,
                 self.url_file_name
             )
     def init_engine(self):
         # This is the working dir by now.
         sys.path.insert(0, '')
         config = self.config
         # print config
         self.find_url_file()
 
         # if os.path.exists(config.Global.key_file) and config.Global.secure:
         #     config.SessionFactory.exec_key = config.Global.key_file
         if os.path.exists(self.url_file):
             with open(self.url_file) as f:
                 d = json.loads(f.read())
                 for k,v in d.iteritems():
                     if isinstance(v, unicode):
                         d[k] = v.encode()
             if d['exec_key']:
                 config.StreamSession.key = d['exec_key']
             d['url'] = disambiguate_url(d['url'], d['location'])
             config.EngineFactory.url = d['url']
             config.EngineFactory.location = d['location']
         
         try:
             exec_lines = config.Kernel.exec_lines
         except AttributeError:
             config.Kernel.exec_lines = []
             exec_lines = config.Kernel.exec_lines
         
         if self.startup_script:
             enc = sys.getfilesystemencoding() or 'utf8'
             cmd="execfile(%r)"%self.startup_script.encode(enc)
             exec_lines.append(cmd)
         if self.startup_command:
             exec_lines.append(self.startup_command)
 
         # Create the underlying shell class and Engine
         # shell_class = import_item(self.master_config.Global.shell_class)
         # print self.config
         try:
             self.engine = EngineFactory(config=config, log=self.log)
         except:
             self.log.error("Couldn't start the Engine", exc_info=True)
             self.exit(1)
         
     def forward_logging(self):
         if self.log_url:
             self.log.info("Forwarding logging to %s"%self.log_url)
             context = self.engine.context
             lsock = context.socket(zmq.PUB)
             lsock.connect(self.log_url)
             self.log.removeHandler(self._log_handler)
             handler = EnginePUBHandler(self.engine, lsock)
             handler.setLevel(self.log_level)
             self.log.addHandler(handler)
             self._log_handler = handler
     #
     def init_mpi(self):
         global mpi
         self.mpi = MPI(config=self.config)
 
         mpi_import_statement = self.mpi.init_script
         if mpi_import_statement:
             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 initialize(self, argv=None):
         super(IPEngineApp, self).initialize(argv)
         self.init_mpi()
         self.init_engine()
         self.forward_logging()
     
     def start(self):
         self.engine.start()
         try:
             self.engine.loop.start()
         except KeyboardInterrupt:
             self.log.critical("Engine Interrupted, shutting down...\n")
 
 
 def launch_new_instance():
     """Create and run the IPython engine"""
     app = IPEngineApp()
     app.initialize()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()
 

IPython/parallel/apps/iploggerapp.py

@@ -1,97 +1,97
 #!/usr/bin/env python
 # encoding: utf-8
 """
 A simple IPython logger application
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 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
 #-----------------------------------------------------------------------------
 
 import os
 import sys
 
 import zmq
 
 from IPython.utils.traitlets import Bool, Dict
 
 from IPython.parallel.apps.clusterdir import (
     ClusterApplication,
     ClusterDir,
     base_aliases
 )
 from IPython.parallel.apps.logwatcher import LogWatcher
 
 #-----------------------------------------------------------------------------
 # Module level variables
 #-----------------------------------------------------------------------------
 
 #: The default config file name for this application
 default_config_file_name = u'iplogger_config.py'
 
-_description = """Start an IPython logger for parallel computing.\n\n
+_description = """Start an IPython logger for parallel computing.
 
 IPython controllers and engines (and your own processes) can broadcast log messages
 by registering a `zmq.log.handlers.PUBHandler` with the `logging` module. The
 logger 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.
+See the `profile` and `cluster_dir` options for details.
 """
 
 
 #-----------------------------------------------------------------------------
 # Main application
 #-----------------------------------------------------------------------------
 aliases = {}
 aliases.update(base_aliases)
 aliases.update(dict(url='LogWatcher.url', topics='LogWatcher.topics'))
 
 class IPLoggerApp(ClusterApplication):
 
     name = u'iploggerz'
     description = _description
     default_config_file_name = default_config_file_name
     auto_create_cluster_dir = Bool(False)
     
     classes = [LogWatcher, ClusterDir]
     aliases = Dict(aliases)
 
     def initialize(self, argv=None):
         super(IPLoggerApp, self).initialize(argv)
         self.init_watcher()
     
     def init_watcher(self):
         try:
             self.watcher = LogWatcher(config=self.config, logname=self.log.name)
         except:
             self.log.error("Couldn't start the LogWatcher", exc_info=True)
             self.exit(1)
         self.log.info("Listening for log messages on %r"%self.watcher.url)
         
 
     def start(self):
         self.watcher.start()
         try:
             self.watcher.loop.start()
         except KeyboardInterrupt:
             self.log.critical("Logging Interrupted, shutting down...\n")
 
 
 def launch_new_instance():
     """Create and run the IPython LogWatcher"""
     app = IPLoggerApp()
     app.initialize()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()
 

IPython/parallel/engine/engine.py

@@ -1,166 +1,165
 #!/usr/bin/env python
 """A simple engine that talks to a controller over 0MQ.
 it handles registration, etc. and launches a kernel
 connected to the Controller's Schedulers.
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-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.
 #-----------------------------------------------------------------------------
 
 from __future__ import print_function
 
 import sys
 import time
 
 import zmq
 from zmq.eventloop import ioloop, zmqstream
 
 # internal
 from IPython.utils.traitlets import Instance, Dict, Int, Type, CFloat, Unicode
 # from IPython.utils.localinterfaces import LOCALHOST 
 
 from IPython.parallel.controller.heartmonitor import Heart
 from IPython.parallel.factory import RegistrationFactory
 from IPython.parallel.streamsession import Message
 from IPython.parallel.util import disambiguate_url
 
 from .streamkernel import Kernel
 
 class EngineFactory(RegistrationFactory):
     """IPython engine"""
     
     # configurables:
     out_stream_factory=Type('IPython.zmq.iostream.OutStream', config=True,
         help="""The OutStream for handling stdout/err.
         Typically 'IPython.zmq.iostream.OutStream'""")
     display_hook_factory=Type('IPython.zmq.displayhook.DisplayHook', config=True,
         help="""The class for handling displayhook.
         Typically 'IPython.zmq.displayhook.DisplayHook'""")
     location=Unicode(config=True,
         help="""The location (an IP address) of the controller.  This is
         used for disambiguating URLs, to determine whether
         loopback should be used to connect or the public address.""")
     timeout=CFloat(2,config=True,
         help="""The time (in seconds) to wait for the Controller to respond
         to registration requests before giving up.""")
     
     # not configurable:
     user_ns=Dict()
     id=Int(allow_none=True)
     registrar=Instance('zmq.eventloop.zmqstream.ZMQStream')
     kernel=Instance(Kernel)
     
     
     def __init__(self, **kwargs):
         super(EngineFactory, self).__init__(**kwargs)
         self.ident = self.session.session
         ctx = self.context
         
         reg = ctx.socket(zmq.XREQ)
         reg.setsockopt(zmq.IDENTITY, self.ident)
         reg.connect(self.url)
         self.registrar = zmqstream.ZMQStream(reg, self.loop)
         
     def register(self):
         """send the registration_request"""
         
         self.log.info("registering")
         content = dict(queue=self.ident, heartbeat=self.ident, control=self.ident)
         self.registrar.on_recv(self.complete_registration)
         # print (self.session.key)
         self.session.send(self.registrar, "registration_request",content=content)
     
     def complete_registration(self, msg):
         # print msg
         self._abort_dc.stop()
         ctx = self.context
         loop = self.loop
         identity = self.ident
         
         idents,msg = self.session.feed_identities(msg)
         msg = Message(self.session.unpack_message(msg))
         
         if msg.content.status == 'ok':
             self.id = int(msg.content.id)
             
             # create Shell Streams (MUX, Task, etc.):
             queue_addr = msg.content.mux
             shell_addrs = [ str(queue_addr) ]
             task_addr = msg.content.task
             if task_addr:
                 shell_addrs.append(str(task_addr))
             
             # Uncomment this to go back to two-socket model
             # shell_streams = []
             # for addr in shell_addrs:
             #     stream = zmqstream.ZMQStream(ctx.socket(zmq.XREP), loop)
             #     stream.setsockopt(zmq.IDENTITY, identity)
             #     stream.connect(disambiguate_url(addr, self.location))
             #     shell_streams.append(stream)
             
             # Now use only one shell stream for mux and tasks
             stream = zmqstream.ZMQStream(ctx.socket(zmq.XREP), loop)
             stream.setsockopt(zmq.IDENTITY, identity)
             shell_streams = [stream]
             for addr in shell_addrs:
                 stream.connect(disambiguate_url(addr, self.location))
             # end single stream-socket
             
             # control stream:
             control_addr = str(msg.content.control)
             control_stream = zmqstream.ZMQStream(ctx.socket(zmq.XREP), loop)
             control_stream.setsockopt(zmq.IDENTITY, identity)
             control_stream.connect(disambiguate_url(control_addr, self.location))
             
             # create iopub stream:
             iopub_addr = msg.content.iopub
             iopub_stream = zmqstream.ZMQStream(ctx.socket(zmq.PUB), loop)
             iopub_stream.setsockopt(zmq.IDENTITY, identity)
             iopub_stream.connect(disambiguate_url(iopub_addr, self.location))
             
             # launch heartbeat
             hb_addrs = msg.content.heartbeat
             # print (hb_addrs)
             
             # # Redirect input streams and set a display hook.
             if self.out_stream_factory:
                 sys.stdout = self.out_stream_factory(self.session, iopub_stream, u'stdout')
                 sys.stdout.topic = 'engine.%i.stdout'%self.id
                 sys.stderr = self.out_stream_factory(self.session, iopub_stream, u'stderr')
                 sys.stderr.topic = 'engine.%i.stderr'%self.id
             if self.display_hook_factory:
                 sys.displayhook = self.display_hook_factory(self.session, iopub_stream)
                 sys.displayhook.topic = 'engine.%i.pyout'%self.id
             
             self.kernel = Kernel(config=self.config, int_id=self.id, ident=self.ident, session=self.session, 
                     control_stream=control_stream, shell_streams=shell_streams, iopub_stream=iopub_stream, 
                     loop=loop, user_ns = self.user_ns, log=self.log)
             self.kernel.start()
             hb_addrs = [ disambiguate_url(addr, self.location) for addr in hb_addrs ]
             heart = Heart(*map(str, hb_addrs), heart_id=identity)
-            # ioloop.DelayedCallback(heart.start, 1000, self.loop).start()
             heart.start()
             
             
         else:
             self.log.fatal("Registration Failed: %s"%msg)
             raise Exception("Registration Failed: %s"%msg)
         
         self.log.info("Completed registration with id %i"%self.id)
     
     
     def abort(self):
         self.log.fatal("Registration timed out after %.1f seconds"%self.timeout)
         self.session.send(self.registrar, "unregistration_request", content=dict(id=self.id))
         time.sleep(1)
         sys.exit(255)
     
     def start(self):
         dc = ioloop.DelayedCallback(self.register, 0, self.loop)
         dc.start()
         self._abort_dc = ioloop.DelayedCallback(self.abort, self.timeout*1000, self.loop)
         self._abort_dc.start()
 

IPython/parallel/tests/__init__.py

@@ -1,107 +1,107
 """toplevel setup/teardown for parallel tests."""
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 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
 #-------------------------------------------------------------------------------
 
 import os
 import tempfile
 import time
 from subprocess import Popen
 
 from IPython.utils.path import get_ipython_dir
 from IPython.parallel import Client
 from IPython.parallel.apps.launcher import (LocalProcessLauncher,
                                                   ipengine_cmd_argv,
                                                   ipcontroller_cmd_argv,
                                                   SIGKILL)
 
 # globals
 launchers = []
 blackhole = open(os.devnull, 'w')
 
 # Launcher class
 class TestProcessLauncher(LocalProcessLauncher):
     """subclass LocalProcessLauncher, to prevent extra sockets and threads being created on Windows"""
     def start(self):
         if self.state == 'before':
             self.process = Popen(self.args,
                 stdout=blackhole, stderr=blackhole,
                 env=os.environ,
                 cwd=self.work_dir
             )
             self.notify_start(self.process.pid)
             self.poll = self.process.poll
         else:
             s = 'The process was already started and has state: %r' % self.state
             raise ProcessStateError(s)
 
 # nose setup/teardown
 
 def setup():
     cp = TestProcessLauncher()
     cp.cmd_and_args = ipcontroller_cmd_argv + \
-                ['--profile', 'iptest', '--log-level', '99', '-r']
+                ['profile=iptest', 'log_level=50', '--reuse']
     cp.start()
     launchers.append(cp)
     cluster_dir = os.path.join(get_ipython_dir(), 'cluster_iptest')
     engine_json = os.path.join(cluster_dir, 'security', 'ipcontroller-engine.json')
     client_json = os.path.join(cluster_dir, 'security', 'ipcontroller-client.json')
     tic = time.time()
     while not os.path.exists(engine_json) or not os.path.exists(client_json):
         if cp.poll() is not None:
             print cp.poll()
             raise RuntimeError("The test controller failed to start.")
         elif time.time()-tic > 10:
             raise RuntimeError("Timeout waiting for the test controller to start.")
         time.sleep(0.1)
     add_engines(1)
 
 def add_engines(n=1, profile='iptest'):
     rc = Client(profile=profile)
     base = len(rc)
     eps = []
     for i in range(n):
         ep = TestProcessLauncher()
-        ep.cmd_and_args = ipengine_cmd_argv + ['--profile', profile, '--log-level', '99']
+        ep.cmd_and_args = ipengine_cmd_argv + ['profile=%s'%profile, 'log_level=50']
         ep.start()
         launchers.append(ep)
         eps.append(ep)
     tic = time.time()
     while len(rc) < base+n:
         if any([ ep.poll() is not None for ep in eps ]):
             raise RuntimeError("A test engine failed to start.")
         elif time.time()-tic > 10:
             raise RuntimeError("Timeout waiting for engines to connect.")
         time.sleep(.1)
         rc.spin()
     rc.close()
     return eps
 
 def teardown():
     time.sleep(1)
     while launchers:
         p = launchers.pop()
         if p.poll() is None:
             try:
                 p.stop()
             except Exception, e:
                 print e
                 pass
         if p.poll() is None:
             time.sleep(.25)
         if p.poll() is None:
             try:
                 print 'cleaning up test process...'
                 p.signal(SIGKILL)
             except:
                 print "couldn't shutdown process: ", p
     blackhole.close()
     

IPython/parallel/tests/test_streamsession.py

@@ -1,111 +1,111
 """test building messages with streamsession"""
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 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
 #-------------------------------------------------------------------------------
 
 import os
 import uuid
 import zmq
 
 from zmq.tests import BaseZMQTestCase
 from zmq.eventloop.zmqstream import ZMQStream
 # from IPython.zmq.tests import SessionTestCase
 from IPython.parallel import streamsession as ss
 
 class SessionTestCase(BaseZMQTestCase):
     
     def setUp(self):
         BaseZMQTestCase.setUp(self)
         self.session = ss.StreamSession()
 
 class TestSession(SessionTestCase):
     
     def test_msg(self):
         """message format"""
         msg = self.session.msg('execute')
         thekeys = set('header msg_id parent_header msg_type content'.split())
         s = set(msg.keys())
         self.assertEquals(s, thekeys)
         self.assertTrue(isinstance(msg['content'],dict))
         self.assertTrue(isinstance(msg['header'],dict))
         self.assertTrue(isinstance(msg['parent_header'],dict))
         self.assertEquals(msg['msg_type'], 'execute')
         
             
     
     def test_args(self):
         """initialization arguments for StreamSession"""
         s = self.session
         self.assertTrue(s.pack is ss.default_packer)
         self.assertTrue(s.unpack is ss.default_unpacker)
         self.assertEquals(s.username, os.environ.get('USER', 'username'))
         
-        s = ss.StreamSession(username=None)
+        s = ss.StreamSession()
         self.assertEquals(s.username, os.environ.get('USER', 'username'))
         
-        self.assertRaises(TypeError, ss.StreamSession, packer='hi')
-        self.assertRaises(TypeError, ss.StreamSession, unpacker='hi')
+        self.assertRaises(TypeError, ss.StreamSession, pack='hi')
+        self.assertRaises(TypeError, ss.StreamSession, unpack='hi')
         u = str(uuid.uuid4())
         s = ss.StreamSession(username='carrot', session=u)
         self.assertEquals(s.session, u)
         self.assertEquals(s.username, 'carrot')
         
     def test_tracking(self):
         """test tracking messages"""
         a,b = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
         s = self.session
         stream = ZMQStream(a)
         msg = s.send(a, 'hello', track=False)
         self.assertTrue(msg['tracker'] is None)
         msg = s.send(a, 'hello', track=True)
         self.assertTrue(isinstance(msg['tracker'], zmq.MessageTracker))
         M = zmq.Message(b'hi there', track=True)
         msg = s.send(a, 'hello', buffers=[M], track=True)
         t = msg['tracker']
         self.assertTrue(isinstance(t, zmq.MessageTracker))
         self.assertRaises(zmq.NotDone, t.wait, .1)
         del M
         t.wait(1) # this will raise
         
         
     # def test_rekey(self):
     #     """rekeying dict around json str keys"""
     #     d = {'0': uuid.uuid4(), 0:uuid.uuid4()}
     #     self.assertRaises(KeyError, ss.rekey, d)
     #     
     #     d = {'0': uuid.uuid4(), 1:uuid.uuid4(), 'asdf':uuid.uuid4()}
     #     d2 = {0:d['0'],1:d[1],'asdf':d['asdf']}
     #     rd = ss.rekey(d)
     #     self.assertEquals(d2,rd)
     #     
     #     d = {'1.5':uuid.uuid4(),'1':uuid.uuid4()}
     #     d2 = {1.5:d['1.5'],1:d['1']}
     #     rd = ss.rekey(d)
     #     self.assertEquals(d2,rd)
     #     
     #     d = {'1.0':uuid.uuid4(),'1':uuid.uuid4()}
     #     self.assertRaises(KeyError, ss.rekey, d)
     # 
     def test_unique_msg_ids(self):
         """test that messages receive unique ids"""
         ids = set()
         for i in range(2**12):
             h = self.session.msg_header('test')
             msg_id = h['msg_id']
             self.assertTrue(msg_id not in ids)
             ids.add(msg_id)
     
     def test_feed_identities(self):
         """scrub the front for zmq IDENTITIES"""
         theids = "engine client other".split()
         content = dict(code='whoda',stuff=object())
         themsg = self.session.msg('execute',content=content)
         pmsg = theids

docs/source/parallel/parallel_intro.txt

@@ -1,253 +1,253
 .. _ip1par:
 
 ============================
 Overview and getting started
 ============================
 
 Introduction
 ============
 
 This section gives an overview of IPython's sophisticated and powerful
 architecture for parallel and distributed computing. This architecture
 abstracts out parallelism in a very general way, which enables IPython to
 support many different styles of parallelism including:
 
 * Single program, multiple data (SPMD) parallelism.
 * Multiple program, multiple data (MPMD) parallelism.
 * Message passing using MPI.
 * Task farming.
 * Data parallel.
 * Combinations of these approaches.
 * Custom user defined approaches.
 
 Most importantly, IPython enables all types of parallel applications to
 be developed, executed, debugged and monitored *interactively*. Hence,
 the ``I`` in IPython.  The following are some example usage cases for IPython:
 
 * Quickly parallelize algorithms that are embarrassingly parallel
   using a number of simple approaches.  Many simple things can be
   parallelized interactively in one or two lines of code.
 
 * Steer traditional MPI applications on a supercomputer from an
   IPython session on your laptop.
 
 * Analyze and visualize large datasets (that could be remote and/or
   distributed) interactively using IPython and tools like
   matplotlib/TVTK.
 
 * Develop, test and debug new parallel algorithms 
   (that may use MPI) interactively.
 
 * Tie together multiple MPI jobs running on different systems into
   one giant distributed and parallel system.
 
 * Start a parallel job on your cluster and then have a remote 	
   collaborator connect to it and pull back data into their 
   local IPython session for plotting and analysis.
 
 * Run a set of tasks on a set of CPUs using dynamic load balancing.
 
 Architecture overview
 =====================
 
 The IPython architecture consists of four components:
 
 * The IPython engine.
 * The IPython hub.
 * The IPython schedulers.
 * The controller client.
 
 These components live in the :mod:`IPython.parallel` package and are
 installed with IPython.  They do, however, have additional dependencies
 that must be installed.  For more information, see our
 :ref:`installation documentation <install_index>`.
 
 .. TODO: include zmq in install_index
 
 IPython engine
 ---------------
 
 The IPython engine is a Python instance that takes Python commands over a
 network connection. Eventually, the IPython engine will be a full IPython
 interpreter, but for now, it is a regular Python interpreter. The engine
 can also handle incoming and outgoing Python objects sent over a network
 connection.  When multiple engines are started, parallel and distributed
 computing becomes possible. An important feature of an IPython engine is
 that it blocks while user code is being executed. Read on for how the
 IPython controller solves this problem to expose a clean asynchronous API
 to the user.
 
 IPython controller
 ------------------
 
 The IPython controller processes provide an interface for working with a set of engines.
 At a general level, the controller is a collection of processes to which IPython engines
 and clients can connect. The controller is composed of a :class:`Hub` and a collection of
 :class:`Schedulers`. These Schedulers are typically run in separate processes but on the
 same machine as the Hub, but can be run anywhere from local threads or on remote machines.
 
 The controller also provides a single point of contact for users who wish to
 utilize the engines connected to the controller. There are different ways of
 working with a controller. In IPython, all of these models are implemented via
 the client's :meth:`.View.apply` method, with various arguments, or
 constructing :class:`.View` objects to represent subsets of engines. The two
 primary models for interacting with engines are:
 
 * A **Direct** interface, where engines are addressed explicitly.
 * A **LoadBalanced** interface, where the Scheduler is trusted with assigning work to
   appropriate engines.
 
 Advanced users can readily extend the View models to enable other
 styles of parallelism. 
 
 .. note:: 
 
     A single controller and set of engines can be used with multiple models
     simultaneously. This opens the door for lots of interesting things.
 
 
 The Hub
 *******
 
 The center of an IPython cluster is the Hub. This is the process that keeps
 track of engine connections, schedulers, clients, as well as all task requests and
 results. The primary role of the Hub is to facilitate queries of the cluster state, and
 minimize the necessary information required to establish the many connections involved in
 connecting new clients and engines.
 
 
 Schedulers
 **********
 
 All actions that can be performed on the engine go through a Scheduler. While the engines
 themselves block when user code is run, the schedulers hide that from the user to provide
 a fully asynchronous interface to a set of engines.
 
 
 IPython client and views
 ------------------------
 
 There is one primary object, the :class:`~.parallel.Client`, for connecting to a cluster.
 For each execution model, there is a corresponding :class:`~.parallel.View`. These views
 allow users to interact with a set of engines through the interface. Here are the two default
 views:
 
 * The :class:`DirectView` class for explicit addressing.
 * The :class:`LoadBalancedView` class for destination-agnostic scheduling.
 
 Security
 --------
 
 IPython uses ZeroMQ for networking, which has provided many advantages, but
 one of the setbacks is its utter lack of security [ZeroMQ]_. By default, no IPython
 connections are encrypted, but open ports only listen on localhost. The only
 source of security for IPython is via ssh-tunnel. IPython supports both shell
 (`openssh`) and `paramiko` based tunnels for connections.  There is a key necessary
 to submit requests, but due to the lack of encryption, it does not provide
 significant security if loopback traffic is compromised.
 
 In our architecture, the controller is the only process that listens on
 network ports, and is thus the main point of vulnerability. The standard model
 for secure connections is to designate that the controller listen on
 localhost, and use ssh-tunnels to connect clients and/or
 engines.
 
 To connect and authenticate to the controller an engine or client needs
 some information that the controller has stored in a JSON file.
 Thus, the JSON files need to be copied to a location where
 the clients and engines can find them. Typically, this is the
 :file:`~/.ipython/cluster_default/security` directory on the host where the 
 client/engine is running (which could be a different host than the controller). 
 Once the JSON files are copied over, everything should work fine.
 
 Currently, there are two JSON files that the controller creates:
 
 ipcontroller-engine.json
     This JSON file has the information necessary for an engine to connect
     to a controller.
 
 ipcontroller-client.json
     The client's connection information.  This may not differ from the engine's,
     but since the controller may listen on different ports for clients and
     engines, it is stored separately.
 
 More details of how these JSON files are used are given below.
 
 A detailed description of the security model and its implementation in IPython
 can be found :ref:`here <parallelsecurity>`.
 
 .. warning::
 
     Even at its most secure, the Controller listens on ports on localhost, and
     every time you make a tunnel, you open a localhost port on the connecting
     machine that points to the Controller. If localhost on the Controller's
     machine, or the machine of any client or engine, is untrusted, then your
     Controller is insecure. There is no way around this with ZeroMQ.
 
 
 
 Getting Started
 ===============
 
 To use IPython for parallel computing, you need to start one instance of the
 controller and one or more instances of the engine. Initially, it is best to
 simply start a controller and engines on a single host using the
 :command:`ipcluster` command. To start a controller and 4 engines on your
 localhost, just do::
 
-    $ ipcluster start -n 4
+    $ ipcluster start n=4
 
 More details about starting the IPython controller and engines can be found
 :ref:`here <parallel_process>`
 
 Once you have started the IPython controller and one or more engines, you
 are ready to use the engines to do something useful. To make sure
 everything is working correctly, try the following commands:
 
 .. sourcecode:: ipython
 
 	In [1]: from IPython.parallel import Client
 	
 	In [2]: c = Client()
 	
 	In [4]: c.ids
 	Out[4]: set([0, 1, 2, 3])
 	
 	In [5]: c[:].apply_sync(lambda : "Hello, World")
 	Out[5]: [ 'Hello, World', 'Hello, World', 'Hello, World', 'Hello, World' ]
 
 
 When a client is created with no arguments, the client tries to find the corresponding JSON file
 in the local `~/.ipython/cluster_default/security` directory. Or if you specified a profile,
 you can use that with the Client.  This should cover most cases:
 
 .. sourcecode:: ipython
 
     In [2]: c = Client(profile='myprofile')
 
 If you have put the JSON file in a different location or it has a different name, create the
 client like this:
 
 .. sourcecode:: ipython
 
     In [2]: c = Client('/path/to/my/ipcontroller-client.json')
 
 Remember, a client needs to be able to see the Hub's ports to connect. So if they are on a
 different machine, you may need to use an ssh server to tunnel access to that machine,
 then you would connect to it with:
 
 .. sourcecode:: ipython
 
     In [2]: c = Client(sshserver='myhub.example.com')
 
 Where 'myhub.example.com' is the url or IP address of the machine on
 which the Hub process is running (or another machine that has direct access to the Hub's ports).
 
 The SSH server may already be specified in ipcontroller-client.json, if the controller was
 instructed at its launch time.
 
 You are now ready to learn more about the :ref:`Direct
 <parallel_multiengine>` and :ref:`LoadBalanced <parallel_task>` interfaces to the
 controller.
 
 .. [ZeroMQ] ZeroMQ.  http://www.zeromq.org

docs/source/parallel/parallel_mpi.txt

@@ -1,156 +1,156
 .. _parallelmpi:
 
 =======================
 Using MPI with IPython
 =======================
 
 .. note::
 
     Not adapted to zmq yet
     This is out of date wrt ipcluster in general as well
 
 Often, a parallel algorithm will require moving data between the engines. One
 way of accomplishing this is by doing a pull and then a push using the
 multiengine client. However, this will be slow as all the data has to go
 through the controller to the client and then back through the controller, to
 its final destination.
 
 A much better way of moving data between engines is to use a message passing
 library, such as the Message Passing Interface (MPI) [MPI]_. IPython's
 parallel computing architecture has been designed from the ground up to
 integrate with MPI. This document describes how to use MPI with IPython.
 
 Additional installation requirements
 ====================================
 
 If you want to use MPI with IPython, you will need to install:
 
 * A standard MPI implementation such as OpenMPI [OpenMPI]_ or MPICH.
 * The mpi4py [mpi4py]_ package.
 
 .. note::
 
     The mpi4py package is not a strict requirement. However, you need to
     have *some* way of calling MPI from Python. You also need some way of
     making sure that :func:`MPI_Init` is called when the IPython engines start
     up. There are a number of ways of doing this and a good number of
     associated subtleties. We highly recommend just using mpi4py as it
     takes care of most of these problems. If you want to do something
     different, let us know and we can help you get started.
 
 Starting the engines with MPI enabled
 =====================================
 
 To use code that calls MPI, there are typically two things that MPI requires.
 
 1. The process that wants to call MPI must be started using
    :command:`mpiexec` or a batch system (like PBS) that has MPI support.
 2. Once the process starts, it must call :func:`MPI_Init`.
 
 There are a couple of ways that you can start the IPython engines and get
 these things to happen.
 
 Automatic starting using :command:`mpiexec` and :command:`ipcluster`
 --------------------------------------------------------------------
 
-The easiest approach is to use the `mpiexec` mode of :command:`ipcluster`,
+The easiest approach is to use the `MPIExec` Launchers in :command:`ipcluster`,
 which will first start a controller and then a set of engines using
 :command:`mpiexec`::
 
-    $ ipcluster mpiexec -n 4
+    $ ipcluster start n=4 elauncher=MPIExecEngineSetLauncher
 
 This approach is best as interrupting :command:`ipcluster` will automatically
 stop and clean up the controller and engines.
 
 Manual starting using :command:`mpiexec`
 ----------------------------------------
 
 If you want to start the IPython engines using the :command:`mpiexec`, just
 do::
 
-    $ mpiexec -n 4 ipengine --mpi=mpi4py
+    $ mpiexec n=4 ipengine mpi=mpi4py
 
 This requires that you already have a controller running and that the FURL
 files for the engines are in place. We also have built in support for
 PyTrilinos [PyTrilinos]_, which can be used (assuming is installed) by
 starting the engines with::
 
-    $ mpiexec -n 4 ipengine --mpi=pytrilinos
+    $ mpiexec n=4 ipengine mpi=pytrilinos
 
 Automatic starting using PBS and :command:`ipcluster`
 ------------------------------------------------------
 
 The :command:`ipcluster` command also has built-in integration with PBS. For
 more information on this approach, see our documentation on :ref:`ipcluster
 <parallel_process>`.
 
 Actually using MPI
 ==================
 
 Once the engines are running with MPI enabled, you are ready to go. You can
 now call any code that uses MPI in the IPython engines. And, all of this can
 be done interactively. Here we show a simple example that uses mpi4py
 [mpi4py]_ version 1.1.0 or later.
 
 First, lets define a simply function that uses MPI to calculate the sum of a
 distributed array. Save the following text in a file called :file:`psum.py`:
 
 .. sourcecode:: python
 
     from mpi4py import MPI
     import numpy as np
 
     def psum(a):
         s = np.sum(a)
         rcvBuf = np.array(0.0,'d')
         MPI.COMM_WORLD.Allreduce([s, MPI.DOUBLE],
             [rcvBuf, MPI.DOUBLE],
             op=MPI.SUM)
         return rcvBuf
 
 Now, start an IPython cluster::
 
-    $ ipcluster start -p mpi -n 4
+    $ ipcluster start profile=mpi n=4
     
 .. note::
 
     It is assumed here that the mpi profile has been set up, as described :ref:`here
     <parallel_process>`.
 
 Finally, connect to the cluster and use this function interactively. In this
 case, we create a random array on each engine and sum up all the random arrays
 using our :func:`psum` function:
 
 .. sourcecode:: ipython
 
     In [1]: from IPython.parallel import Client
     
     In [2]: %load_ext parallel_magic
     
     In [3]: c = Client(profile='mpi')
     
     In [4]: view = c[:]
     
     In [5]: view.activate()
     
     # run the contents of the file on each engine:
     In [6]: view.run('psum.py')
     
     In [6]: px a = np.random.rand(100)
     Parallel execution on engines: [0,1,2,3]
     
     In [8]: px s = psum(a)
     Parallel execution on engines: [0,1,2,3]
     
     In [9]: view['s']
     Out[9]: [187.451545803,187.451545803,187.451545803,187.451545803]
 
 Any Python code that makes calls to MPI can be used in this manner, including
 compiled C, C++ and Fortran libraries that have been exposed to Python.
 
 .. [MPI] Message Passing Interface.  http://www-unix.mcs.anl.gov/mpi/
 .. [mpi4py] MPI for Python. mpi4py: http://mpi4py.scipy.org/
 .. [OpenMPI] Open MPI. http://www.open-mpi.org/
 .. [PyTrilinos] PyTrilinos. http://trilinos.sandia.gov/packages/pytrilinos/
 
 

docs/source/parallel/parallel_multiengine.txt

@@ -1,843 +1,843
 .. _parallel_multiengine:
 
 ==========================
 IPython's Direct interface
 ==========================
 
 The direct, or multiengine, interface represents one possible way of working with a set of
 IPython engines. The basic idea behind the multiengine interface is that the
 capabilities of each engine are directly and explicitly exposed to the user.
 Thus, in the multiengine interface, each engine is given an id that is used to
 identify the engine and give it work to do. This interface is very intuitive
 and is designed with interactive usage in mind, and is the best place for
 new users of IPython to begin.
 
 Starting the IPython controller and engines
 ===========================================
 
 To follow along with this tutorial, you will need to start the IPython
 controller and four IPython engines. The simplest way of doing this is to use
 the :command:`ipcluster` command::
 
-    $ ipcluster start -n 4
+    $ ipcluster start n=4
     
 For more detailed information about starting the controller and engines, see
 our :ref:`introduction <ip1par>` to using IPython for parallel computing.
 
 Creating a ``Client`` instance
 ==============================
 
 The first step is to import the IPython :mod:`IPython.parallel`
 module and then create a :class:`.Client` instance:
 
 .. sourcecode:: ipython
 
     In [1]: from IPython.parallel import Client
     
     In [2]: rc = Client()
 
 This form assumes that the default connection information (stored in
 :file:`ipcontroller-client.json` found in :file:`IPYTHON_DIR/cluster_default/security`) is
 accurate. If the controller was started on a remote machine, you must copy that connection
 file to the client machine, or enter its contents as arguments to the Client constructor:
 
 .. sourcecode:: ipython
 
     # If you have copied the json connector file from the controller:
     In [2]: rc = Client('/path/to/ipcontroller-client.json')
     # or to connect with a specific profile you have set up:
     In [3]: rc = Client(profile='mpi')
     
 
 To make sure there are engines connected to the controller, users can get a list
 of engine ids:
 
 .. sourcecode:: ipython
 
     In [3]: rc.ids
     Out[3]: [0, 1, 2, 3]
 
 Here we see that there are four engines ready to do work for us.
 
 For direct execution, we will make use of a :class:`DirectView` object, which can be
 constructed via list-access to the client:
 
 .. sourcecode:: ipython
 
     In [4]: dview = rc[:] # use all engines
 
 .. seealso::
 
     For more information, see the in-depth explanation of :ref:`Views <parallel_details>`.
 
 
 Quick and easy parallelism
 ==========================
 
 In many cases, you simply want to apply a Python function to a sequence of
 objects, but *in parallel*. The client interface provides a simple way
 of accomplishing this: using the DirectView's :meth:`~DirectView.map` method.
 
 Parallel map
 ------------
 
 Python's builtin :func:`map` functions allows a function to be applied to a
 sequence element-by-element. This type of code is typically trivial to
 parallelize. In fact, since IPython's interface is all about functions anyway,
 you can just use the builtin :func:`map` with a :class:`RemoteFunction`, or a 
 DirectView's :meth:`map` method:
 
 .. sourcecode:: ipython
 
     In [62]: serial_result = map(lambda x:x**10, range(32))
     
     In [63]: parallel_result = dview.map_sync(lambda x: x**10, range(32))
 
     In [67]: serial_result==parallel_result
     Out[67]: True
 
 
 .. note::
 
     The :class:`DirectView`'s version of :meth:`map` does
     not do dynamic load balancing. For a load balanced version, use a
     :class:`LoadBalancedView`.
 
 .. seealso::
     
     :meth:`map` is implemented via :class:`ParallelFunction`.
 
 Remote function decorators
 --------------------------
 
 Remote functions are just like normal functions, but when they are called,
 they execute on one or more engines, rather than locally. IPython provides
 two decorators:
 
 .. sourcecode:: ipython
 
     In [10]: @dview.remote(block=True)
         ...: def getpid():
         ...:     import os
         ...:     return os.getpid()
         ...: 
 
     In [11]: getpid()
     Out[11]: [12345, 12346, 12347, 12348]
 
 The ``@parallel`` decorator creates parallel functions, that break up an element-wise
 operations and distribute them, reconstructing the result.
 
 .. sourcecode:: ipython
 
     In [12]: import numpy as np
     
     In [13]: A = np.random.random((64,48))
     
     In [14]: @dview.parallel(block=True)
         ...: def pmul(A,B):
         ...:     return A*B
     
     In [15]: C_local = A*A
     
     In [16]: C_remote = pmul(A,A)
     
     In [17]: (C_local == C_remote).all()
     Out[17]: True
 
 .. seealso::
 
     See the docstrings for the :func:`parallel` and :func:`remote` decorators for
     options.
 
 Calling Python functions
 ========================
 
 The most basic type of operation that can be performed on the engines is to
 execute Python code or call Python functions. Executing Python code can be
 done in blocking or non-blocking mode (non-blocking is default) using the
 :meth:`.View.execute` method, and calling functions can be done via the
 :meth:`.View.apply` method.
 
 apply
 -----
 
 The main method for doing remote execution (in fact, all methods that
 communicate with the engines are built on top of it), is :meth:`View.apply`.
 
 We strive to provide the cleanest interface we can, so `apply` has the following
 signature:
 
 .. sourcecode:: python
 
     view.apply(f, *args, **kwargs)
 
 There are various ways to call functions with IPython, and these flags are set as
 attributes of the View.  The ``DirectView`` has just two of these flags:
 
 dv.block : bool
     whether to wait for the result, or return an :class:`AsyncResult` object
     immediately
 dv.track : bool
     whether to instruct pyzmq to track when 
     This is primarily useful for non-copying sends of numpy arrays that you plan to
     edit in-place.  You need to know when it becomes safe to edit the buffer
     without corrupting the message.
 
 
 Creating a view is simple: index-access on a client creates a :class:`.DirectView`.
 
 .. sourcecode:: ipython
 
     In [4]: view = rc[1:3]
     Out[4]: <DirectView [1, 2]>
 
     In [5]: view.apply<tab>
     view.apply  view.apply_async  view.apply_sync
 
 For convenience, you can set block temporarily for a single call with the extra sync/async methods.
 
 Blocking execution
 ------------------
 
 In blocking mode, the :class:`.DirectView` object (called ``dview`` in
 these examples) submits the command to the controller, which places the
 command in the engines' queues for execution. The :meth:`apply` call then
 blocks until the engines are done executing the command:
 
 .. sourcecode:: ipython
 
     In [2]: dview = rc[:] # A DirectView of all engines
     In [3]: dview.block=True
     In [4]: dview['a'] = 5
 
     In [5]: dview['b'] = 10
 
     In [6]: dview.apply(lambda x: a+b+x, 27)
     Out[6]: [42, 42, 42, 42]
 
 You can also select blocking execution on a call-by-call basis with the :meth:`apply_sync`
 method:
 
     In [7]: dview.block=False
 
     In [8]: dview.apply_sync(lambda x: a+b+x, 27)
     Out[8]: [42, 42, 42, 42]
 
 Python commands can be executed as strings on specific engines by using a View's ``execute``
 method:
 
 .. sourcecode:: ipython
 
     In [6]: rc[::2].execute('c=a+b')
 
     In [7]: rc[1::2].execute('c=a-b')
 
     In [8]: dview['c'] # shorthand for dview.pull('c', block=True)
     Out[8]: [15, -5, 15, -5]
 
 
 Non-blocking execution
 ----------------------
 
 In non-blocking mode, :meth:`apply` submits the command to be executed and
 then returns a :class:`AsyncResult` object immediately. The
 :class:`AsyncResult` object gives you a way of getting a result at a later
 time through its :meth:`get` method.
 
 .. Note::
 
     The :class:`AsyncResult` object provides a superset of the interface in 
     :py:class:`multiprocessing.pool.AsyncResult`.  See the 
     `official Python documentation <http://docs.python.org/library/multiprocessing#multiprocessing.pool.AsyncResult>`_
     for more.
 
 
 This allows you to quickly submit long running commands without blocking your
 local Python/IPython session:
 
 .. sourcecode:: ipython
     
     # define our function
     In [6]: def wait(t):
        ...:     import time
        ...:     tic = time.time()
        ...:     time.sleep(t)
        ...:     return time.time()-tic
     
     # In non-blocking mode
     In [7]: ar = dview.apply_async(wait, 2)
 
     # Now block for the result
     In [8]: ar.get()
     Out[8]: [2.0006198883056641, 1.9997570514678955, 1.9996809959411621, 2.0003249645233154]
 
     # Again in non-blocking mode
     In [9]: ar = dview.apply_async(wait, 10)
 
     # Poll to see if the result is ready
     In [10]: ar.ready()
     Out[10]: False
     
     # ask for the result, but wait a maximum of 1 second:
     In [45]: ar.get(1)
     ---------------------------------------------------------------------------
     TimeoutError                              Traceback (most recent call last)
     /home/you/<ipython-input-45-7cd858bbb8e0> in <module>()
     ----> 1 ar.get(1)
 
     /path/to/site-packages/IPython/parallel/asyncresult.pyc in get(self, timeout)
          62                 raise self._exception
          63         else:
     ---> 64             raise error.TimeoutError("Result not ready.")
          65 
          66     def ready(self):
 
     TimeoutError: Result not ready.
 
 .. Note::
 
     Note the import inside the function. This is a common model, to ensure
     that the appropriate modules are imported where the task is run. You can
     also manually import modules into the engine(s) namespace(s) via 
     :meth:`view.execute('import numpy')`.
 
 Often, it is desirable to wait until a set of :class:`AsyncResult` objects
 are done. For this, there is a the method :meth:`wait`. This method takes a
 tuple of :class:`AsyncResult` objects (or `msg_ids` or indices to the client's History),
 and blocks until all of the associated results are ready:
 
 .. sourcecode:: ipython
 
     In [72]: dview.block=False
 
     # A trivial list of AsyncResults objects
     In [73]: pr_list = [dview.apply_async(wait, 3) for i in range(10)]
 
     # Wait until all of them are done
     In [74]: dview.wait(pr_list)
 
     # Then, their results are ready using get() or the `.r` attribute
     In [75]: pr_list[0].get()
     Out[75]: [2.9982571601867676, 2.9982588291168213, 2.9987530708312988, 2.9990990161895752]
 
 
 
 The ``block`` and ``targets`` keyword arguments and attributes
 --------------------------------------------------------------
 
 Most DirectView methods (excluding :meth:`apply` and :meth:`map`) accept ``block`` and
 ``targets`` as keyword arguments. As we have seen above, these keyword arguments control the
 blocking mode and which engines the command is applied to. The :class:`View` class also has
 :attr:`block` and :attr:`targets` attributes that control the default behavior when the keyword
 arguments are not provided. Thus the following logic is used for :attr:`block` and :attr:`targets`:
 
 * If no keyword argument is provided, the instance attributes are used.
 * Keyword argument, if provided override the instance attributes for
   the duration of a single call.
   
 The following examples demonstrate how to use the instance attributes:
 
 .. sourcecode:: ipython
 
     In [16]: dview.targets = [0,2]
     
     In [17]: dview.block = False
 
     In [18]: ar = dview.apply(lambda : 10)
 
     In [19]: ar.get()
     Out[19]: [10, 10]
 
     In [16]: dview.targets = v.client.ids # all engines (4)
     
     In [21]: dview.block = True
 
     In [22]: dview.apply(lambda : 42)
     Out[22]: [42, 42, 42, 42]
 
 The :attr:`block` and :attr:`targets` instance attributes of the
 :class:`.DirectView` also determine the behavior of the parallel magic commands.
 
 Parallel magic commands
 -----------------------
 
 .. warning::
 
     The magics have not been changed to work with the zeromq system. The
     magics do work, but *do not* print stdin/out like they used to in IPython.kernel.
 
 We provide a few IPython magic commands (``%px``, ``%autopx`` and ``%result``)
 that make it more pleasant to execute Python commands on the engines
 interactively. These are simply shortcuts to :meth:`execute` and
 :meth:`get_result` of the :class:`DirectView`. The ``%px`` magic executes a single 
 Python command on the engines specified by the :attr:`targets` attribute of the
 :class:`DirectView` instance:
 
 .. sourcecode:: ipython
 
     # load the parallel magic extension:
     In [21]: %load_ext parallelmagic
     
     # Create a DirectView for all targets
     In [22]: dv = rc[:]
     
     # Make this DirectView active for parallel magic commands
     In [23]: dv.activate()
 
     In [24]: dv.block=True
 
     In [25]: import numpy
 
     In [26]: %px import numpy
     Parallel execution on engines: [0, 1, 2, 3]
 
     In [27]: %px a = numpy.random.rand(2,2)
     Parallel execution on engines: [0, 1, 2, 3]
 
     In [28]: %px ev = numpy.linalg.eigvals(a)
     Parallel execution on engines: [0, 1, 2, 3]
 
     In [28]: dv['ev']
     Out[28]: [ array([ 1.09522024, -0.09645227]), 
                array([ 1.21435496, -0.35546712]), 
                array([ 0.72180653,  0.07133042]), 
                array([  1.46384341e+00,   1.04353244e-04])
              ]
 
 The ``%result`` magic gets the most recent result, or takes an argument
 specifying the index of the result to be requested. It is simply a shortcut to the
 :meth:`get_result` method:
 
 .. sourcecode:: ipython
     
     In [29]: dv.apply_async(lambda : ev)
     
     In [30]: %result
     Out[30]: [ [ 1.28167017  0.14197338],
                 [-0.14093616  1.27877273],
                 [-0.37023573  1.06779409],
                 [ 0.83664764 -0.25602658] ]
 
 The ``%autopx`` magic switches to a mode where everything you type is executed
 on the engines given by the :attr:`targets` attribute:
 
 .. sourcecode:: ipython
 
     In [30]: dv.block=False
 
     In [31]: %autopx
     Auto Parallel Enabled
     Type %autopx to disable
 
     In [32]: max_evals = []
     <IPython.parallel.AsyncResult object at 0x17b8a70>
 
     In [33]: for i in range(100):
        ....:     a = numpy.random.rand(10,10)
        ....:     a = a+a.transpose()
        ....:     evals = numpy.linalg.eigvals(a)
        ....:     max_evals.append(evals[0].real)
        ....:
        ....:
     <IPython.parallel.AsyncResult object at 0x17af8f0>
 
     In [34]: %autopx
     Auto Parallel Disabled
 
     In [35]: dv.block=True
 
     In [36]: px ans= "Average max eigenvalue is: %f"%(sum(max_evals)/len(max_evals))
     Parallel execution on engines: [0, 1, 2, 3]
     
     In [37]: dv['ans']
     Out[37]: [ 'Average max eigenvalue is:  10.1387247332',
                'Average max eigenvalue is:  10.2076902286',
                'Average max eigenvalue is:  10.1891484655',
                'Average max eigenvalue is:  10.1158837784',]
 
 
 Moving Python objects around
 ============================
 
 In addition to calling functions and executing code on engines, you can
 transfer Python objects to and from your IPython session and the engines. In
 IPython, these operations are called :meth:`push` (sending an object to the
 engines) and :meth:`pull` (getting an object from the engines).
 
 Basic push and pull
 -------------------
 
 Here are some examples of how you use :meth:`push` and :meth:`pull`:
 
 .. sourcecode:: ipython
 
     In [38]: dview.push(dict(a=1.03234,b=3453))
     Out[38]: [None,None,None,None]
 
     In [39]: dview.pull('a')
     Out[39]: [ 1.03234, 1.03234, 1.03234, 1.03234]
 
     In [40]: dview.pull('b', targets=0)
     Out[40]: 3453
 
     In [41]: dview.pull(('a','b'))
     Out[41]: [ [1.03234, 3453], [1.03234, 3453], [1.03234, 3453], [1.03234, 3453] ]
     
     In [43]: dview.push(dict(c='speed'))
     Out[43]: [None,None,None,None]
 
 In non-blocking mode :meth:`push` and :meth:`pull` also return
 :class:`AsyncResult` objects:
 
 .. sourcecode:: ipython
 
     In [48]: ar = dview.pull('a', block=False)
 
     In [49]: ar.get()
     Out[49]: [1.03234, 1.03234, 1.03234, 1.03234]
 
 
 Dictionary interface
 --------------------
 
 Since a Python namespace is just a :class:`dict`, :class:`DirectView` objects provide
 dictionary-style access by key and methods such as :meth:`get` and
 :meth:`update` for convenience. This make the remote namespaces of the engines
 appear as a local dictionary. Underneath, these methods call :meth:`apply`:
 
 .. sourcecode:: ipython
 
     In [51]: dview['a']=['foo','bar']
 
     In [52]: dview['a']
     Out[52]: [ ['foo', 'bar'], ['foo', 'bar'], ['foo', 'bar'], ['foo', 'bar'] ]
 
 Scatter and gather
 ------------------
 
 Sometimes it is useful to partition a sequence and push the partitions to
 different engines. In MPI language, this is know as scatter/gather and we
 follow that terminology. However, it is important to remember that in
 IPython's :class:`Client` class, :meth:`scatter` is from the
 interactive IPython session to the engines and :meth:`gather` is from the
 engines back to the interactive IPython session. For scatter/gather operations
 between engines, MPI should be used:
 
 .. sourcecode:: ipython
 
     In [58]: dview.scatter('a',range(16))
     Out[58]: [None,None,None,None]
 
     In [59]: dview['a']
     Out[59]: [ [0, 1, 2, 3], [4, 5, 6, 7], [8, 9, 10, 11], [12, 13, 14, 15] ]
 
     In [60]: dview.gather('a')
     Out[60]: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]
 
 Other things to look at
 =======================
 
 How to do parallel list comprehensions
 --------------------------------------
 
 In many cases list comprehensions are nicer than using the map function. While
 we don't have fully parallel list comprehensions, it is simple to get the
 basic effect using :meth:`scatter` and :meth:`gather`:
 
 .. sourcecode:: ipython
 
     In [66]: dview.scatter('x',range(64))
 
     In [67]: %px y = [i**10 for i in x]
     Parallel execution on engines: [0, 1, 2, 3]
     Out[67]:
 
     In [68]: y = dview.gather('y')
 
     In [69]: print y
     [0, 1, 1024, 59049, 1048576, 9765625, 60466176, 282475249, 1073741824,...]
 
 Remote imports
 --------------
 
 Sometimes you will want to import packages both in your interactive session
 and on your remote engines.  This can be done with the :class:`ContextManager`
 created by a DirectView's :meth:`sync_imports` method:
 
 .. sourcecode:: ipython
 
     In [69]: with dview.sync_imports():
         ...:     import numpy
     importing numpy on engine(s)
 
 Any imports made inside the block will also be performed on the view's engines.
 sync_imports also takes a `local` boolean flag that defaults to True, which specifies
 whether the local imports should also be performed.  However, support for `local=False`
 has not been implemented, so only packages that can be imported locally will work
 this way.
 
 You can also specify imports via the ``@require`` decorator.  This is a decorator
 designed for use in Dependencies, but can be used to handle remote imports as well.
 Modules or module names passed to ``@require`` will be imported before the decorated
 function is called.  If they cannot be imported, the decorated function will never
 execution, and will fail with an UnmetDependencyError.
 
 .. sourcecode:: ipython
 
     In [69]: from IPython.parallel import require
 
     In [70]: @requre('re'):
         ...: def findall(pat, x):
         ...:     # re is guaranteed to be available
         ...:     return re.findall(pat, x)
           
     # you can also pass modules themselves, that you already have locally:
     In [71]: @requre(time):
         ...: def wait(t):
         ...:     time.sleep(t)
         ...:     return t
 
 
 Parallel exceptions
 -------------------
 
 In the multiengine interface, parallel commands can raise Python exceptions,
 just like serial commands. But, it is a little subtle, because a single
 parallel command can actually raise multiple exceptions (one for each engine
 the command was run on). To express this idea, we have a
 :exc:`CompositeError` exception class that will be raised in most cases. The
 :exc:`CompositeError` class is a special type of exception that wraps one or
 more other types of exceptions. Here is how it works:
 
 .. sourcecode:: ipython
 
     In [76]: dview.block=True
 
     In [77]: dview.execute('1/0')
     ---------------------------------------------------------------------------
     CompositeError                            Traceback (most recent call last)
     /home/you/<ipython-input-10-15c2c22dec39> in <module>()
     ----> 1 dview.execute('1/0', block=True)
 
     /path/to/site-packages/IPython/parallel/view.py in execute(self, code, block)
         460                 default: self.block
         461         """
     --> 462         return self.apply_with_flags(util._execute, args=(code,), block=block)
         463 
         464     def run(self, filename, block=None):
 
     /home/you/<string> in apply_with_flags(self, f, args, kwargs, block, track)
 
     /path/to/site-packages/IPython/parallel/view.py in sync_results(f, self, *args, **kwargs)
          46 def sync_results(f, self, *args, **kwargs):
          47     """sync relevant results from self.client to our results attribute."""
     ---> 48     ret = f(self, *args, **kwargs)
          49     delta = self.outstanding.difference(self.client.outstanding)
          50     completed = self.outstanding.intersection(delta)
 
     /home/you/<string> in apply_with_flags(self, f, args, kwargs, block, track)
 
     /path/to/site-packages/IPython/parallel/view.py in save_ids(f, self, *args, **kwargs)
          35     n_previous = len(self.client.history)
          36     try:
     ---> 37         ret = f(self, *args, **kwargs)
          38     finally:
          39         nmsgs = len(self.client.history) - n_previous
 
     /path/to/site-packages/IPython/parallel/view.py in apply_with_flags(self, f, args, kwargs, block, track)
         398         if block:
         399             try:
     --> 400                 return ar.get()
         401             except KeyboardInterrupt:
         402                 pass
 
     /path/to/site-packages/IPython/parallel/asyncresult.pyc in get(self, timeout)
          87                 return self._result
          88             else:
     ---> 89                 raise self._exception
          90         else:
          91             raise error.TimeoutError("Result not ready.")
 
     CompositeError: one or more exceptions from call to method: _execute
     [0:apply]: ZeroDivisionError: integer division or modulo by zero
     [1:apply]: ZeroDivisionError: integer division or modulo by zero
     [2:apply]: ZeroDivisionError: integer division or modulo by zero
     [3:apply]: ZeroDivisionError: integer division or modulo by zero
 
 
 Notice how the error message printed when :exc:`CompositeError` is raised has
 information about the individual exceptions that were raised on each engine.
 If you want, you can even raise one of these original exceptions:
 
 .. sourcecode:: ipython
 
     In [80]: try:
        ....:     dview.execute('1/0')
        ....: except client.CompositeError, e:
        ....:     e.raise_exception()
        ....:     
        ....:     
     ---------------------------------------------------------------------------
     ZeroDivisionError                         Traceback (most recent call last)
 
     /ipython1-client-r3021/docs/examples/<ipython console> in <module>()
 
     /ipython1-client-r3021/ipython1/kernel/error.pyc in raise_exception(self, excid)
         156             raise IndexError("an exception with index %i does not exist"%excid)
         157         else:
     --> 158             raise et, ev, etb
         159 
         160 def collect_exceptions(rlist, method):
 
     ZeroDivisionError: integer division or modulo by zero
 
 If you are working in IPython, you can simple type ``%debug`` after one of
 these :exc:`CompositeError` exceptions is raised, and inspect the exception
 instance:
 
 .. sourcecode:: ipython
 
     In [81]: dview.execute('1/0')
     ---------------------------------------------------------------------------
     CompositeError                            Traceback (most recent call last)
     /home/you/<ipython-input-10-15c2c22dec39> in <module>()
     ----> 1 dview.execute('1/0', block=True)
 
     /path/to/site-packages/IPython/parallel/view.py in execute(self, code, block)
         460                 default: self.block
         461         """
     --> 462         return self.apply_with_flags(util._execute, args=(code,), block=block)
         463 
         464     def run(self, filename, block=None):
 
     /home/you/<string> in apply_with_flags(self, f, args, kwargs, block, track)
 
     /path/to/site-packages/IPython/parallel/view.py in sync_results(f, self, *args, **kwargs)
          46 def sync_results(f, self, *args, **kwargs):
          47     """sync relevant results from self.client to our results attribute."""
     ---> 48     ret = f(self, *args, **kwargs)
          49     delta = self.outstanding.difference(self.client.outstanding)
          50     completed = self.outstanding.intersection(delta)
 
     /home/you/<string> in apply_with_flags(self, f, args, kwargs, block, track)
 
     /path/to/site-packages/IPython/parallel/view.py in save_ids(f, self, *args, **kwargs)
          35     n_previous = len(self.client.history)
          36     try:
     ---> 37         ret = f(self, *args, **kwargs)
          38     finally:
          39         nmsgs = len(self.client.history) - n_previous
 
     /path/to/site-packages/IPython/parallel/view.py in apply_with_flags(self, f, args, kwargs, block, track)
         398         if block:
         399             try:
     --> 400                 return ar.get()
         401             except KeyboardInterrupt:
         402                 pass
 
     /path/to/site-packages/IPython/parallel/asyncresult.pyc in get(self, timeout)
          87                 return self._result
          88             else:
     ---> 89                 raise self._exception
          90         else:
          91             raise error.TimeoutError("Result not ready.")
 
     CompositeError: one or more exceptions from call to method: _execute
     [0:apply]: ZeroDivisionError: integer division or modulo by zero
     [1:apply]: ZeroDivisionError: integer division or modulo by zero
     [2:apply]: ZeroDivisionError: integer division or modulo by zero
     [3:apply]: ZeroDivisionError: integer division or modulo by zero
     
     In [82]: %debug
     > /path/to/site-packages/IPython/parallel/asyncresult.py(80)get()
          79             else:
     ---> 80                 raise self._exception
          81         else:
     
     
     # With the debugger running, e is the exceptions instance.  We can tab complete
     # on it and see the extra methods that are available.
     ipdb> e.
     e.__class__         e.__getitem__       e.__new__           e.__setstate__      e.args
     e.__delattr__       e.__getslice__      e.__reduce__        e.__str__           e.elist
     e.__dict__          e.__hash__          e.__reduce_ex__     e.__weakref__       e.message
     e.__doc__           e.__init__          e.__repr__          e._get_engine_str   e.print_tracebacks
     e.__getattribute__  e.__module__        e.__setattr__       e._get_traceback    e.raise_exception
     ipdb> e.print_tracebacks()
     [0:apply]: 
     Traceback (most recent call last):
       File "/path/to/site-packages/IPython/parallel/streamkernel.py", line 332, in apply_request
         exec code in working, working
       File "<string>", line 1, in <module>
       File "/path/to/site-packages/IPython/parallel/client.py", line 69, in _execute
         exec code in globals()
       File "<string>", line 1, in <module>
     ZeroDivisionError: integer division or modulo by zero
 
 
     [1:apply]: 
     Traceback (most recent call last):
       File "/path/to/site-packages/IPython/parallel/streamkernel.py", line 332, in apply_request
         exec code in working, working
       File "<string>", line 1, in <module>
       File "/path/to/site-packages/IPython/parallel/client.py", line 69, in _execute
         exec code in globals()
       File "<string>", line 1, in <module>
     ZeroDivisionError: integer division or modulo by zero
 
 
     [2:apply]: 
     Traceback (most recent call last):
       File "/path/to/site-packages/IPython/parallel/streamkernel.py", line 332, in apply_request
         exec code in working, working
       File "<string>", line 1, in <module>
       File "/path/to/site-packages/IPython/parallel/client.py", line 69, in _execute
         exec code in globals()
       File "<string>", line 1, in <module>
     ZeroDivisionError: integer division or modulo by zero
 
 
     [3:apply]: 
     Traceback (most recent call last):
       File "/path/to/site-packages/IPython/parallel/streamkernel.py", line 332, in apply_request
         exec code in working, working
       File "<string>", line 1, in <module>
       File "/path/to/site-packages/IPython/parallel/client.py", line 69, in _execute
         exec code in globals()
       File "<string>", line 1, in <module>
     ZeroDivisionError: integer division or modulo by zero
     
 
 .. note::
 
     TODO: The above tracebacks are not up to date
 
 
 All of this same error handling magic even works in non-blocking mode:
 
 .. sourcecode:: ipython
 
     In [83]: dview.block=False
 
     In [84]: ar = dview.execute('1/0')
 
     In [85]: ar.get()
     ---------------------------------------------------------------------------
     CompositeError                            Traceback (most recent call last)
     /Users/minrk/<ipython-input-3-8531eb3d26fb> in <module>()
     ----> 1 ar.get()
 
     /path/to/site-packages/IPython/parallel/asyncresult.pyc in get(self, timeout)
          78                 return self._result
          79             else:
     ---> 80                 raise self._exception
          81         else:
          82             raise error.TimeoutError("Result not ready.")
 
     CompositeError: one or more exceptions from call to method: _execute
     [0:apply]: ZeroDivisionError: integer division or modulo by zero
     [1:apply]: ZeroDivisionError: integer division or modulo by zero
     [2:apply]: ZeroDivisionError: integer division or modulo by zero
     [3:apply]: ZeroDivisionError: integer division or modulo by zero
 

docs/source/parallel/parallel_process.txt

@@ -1,506 +1,507
 .. _parallel_process:
 
 ===========================================
 Starting the IPython controller and engines
 ===========================================
 
 To use IPython for parallel computing, you need to start one instance of
 the controller and one or more instances of the engine. The controller
 and each engine can run on different machines or on the same machine.
 Because of this, there are many different possibilities.
 
 Broadly speaking, there are two ways of going about starting a controller and engines:
 
 * In an automated manner using the :command:`ipcluster` command.
 * In a more manual way using the :command:`ipcontroller` and 
   :command:`ipengine` commands.
 
 This document describes both of these methods. We recommend that new users
 start with the :command:`ipcluster` command as it simplifies many common usage
 cases.
 
 General considerations
 ======================
 
 Before delving into the details about how you can start a controller and
 engines using the various methods, we outline some of the general issues that
 come up when starting the controller and engines. These things come up no
 matter which method you use to start your IPython cluster.
 
 Let's say that you want to start the controller on ``host0`` and engines on
 hosts ``host1``-``hostn``. The following steps are then required:
 
 1. Start the controller on ``host0`` by running :command:`ipcontroller` on
    ``host0``.
 2. Move the JSON file (:file:`ipcontroller-engine.json`) created by the 
    controller from ``host0`` to hosts ``host1``-``hostn``.
 3. Start the engines on hosts ``host1``-``hostn`` by running 
    :command:`ipengine`.  This command has to be told where the JSON file 
    (:file:`ipcontroller-engine.json`) is located.
 
 At this point, the controller and engines will be connected. By default, the JSON files
 created by the controller are put into the :file:`~/.ipython/cluster_default/security`
 directory. If the engines share a filesystem with the controller, step 2 can be skipped as
 the engines will automatically look at that location.
 
 The final step required to actually use the running controller from a client is to move
 the JSON file :file:`ipcontroller-client.json` from ``host0`` to any host where clients
 will be run. If these file are put into the :file:`~/.ipython/cluster_default/security`
 directory of the client's host, they will be found automatically. Otherwise, the full path
 to them has to be passed to the client's constructor.
 
 Using :command:`ipcluster`
 ===========================
 
 The :command:`ipcluster` command provides a simple way of starting a
 controller and engines in the following situations:
 
 1. When the controller and engines are all run on localhost. This is useful
    for testing or running on a multicore computer.
-2. When engines are started using the :command:`mpirun` command that comes 
+2. When engines are started using the :command:`mpiexec` command that comes 
    with most MPI [MPI]_ implementations
 3. When engines are started using the PBS [PBS]_ batch system 
    (or other `qsub` systems, such as SGE).
 4. When the controller is started on localhost and the engines are started on 
    remote nodes using :command:`ssh`.
 5. When engines are started using the Windows HPC Server batch system.
 
 .. note::
 
     Currently :command:`ipcluster` requires that the
     :file:`~/.ipython/cluster_<profile>/security` directory live on a shared filesystem that is
     seen by both the controller and engines. If you don't have a shared file
     system you will need to use :command:`ipcontroller` and
     :command:`ipengine` directly.
 
 Under the hood, :command:`ipcluster` just uses :command:`ipcontroller`
 and :command:`ipengine` to perform the steps described above.
 
 The simplest way to use ipcluster requires no configuration, and will
 launch a controller and a number of engines on the local machine. For instance, 
 to start one controller and 4 engines on localhost, just do::
 
-    $ ipcluster start -n 4
+    $ ipcluster start n=4
 
-To see other command line options for the local mode, do::
+To see other command line options, do::
 
     $ ipcluster -h
 
 
 Configuring an IPython cluster
 ==============================
 
 Cluster configurations are stored as `profiles`.  You can create a new profile with::
 
-    $ ipcluster create -p myprofile
+    $ ipcluster create profile=myprofile
 
 This will create the directory :file:`IPYTHONDIR/cluster_myprofile`, and populate it
 with the default configuration files for the three IPython cluster commands. Once
 you edit those files, you can continue to call ipcluster/ipcontroller/ipengine
-with no arguments beyond ``-p myprofile``, and any configuration will be maintained.
+with no arguments beyond ``p=myprofile``, and any configuration will be maintained.
 
 There is no limit to the number of profiles you can have, so you can maintain a profile for each
 of your common use cases. The default profile will be used whenever the
 profile argument is not specified, so edit :file:`IPYTHONDIR/cluster_default/*_config.py` to
 represent your most common use case.
 
 The configuration files are loaded with commented-out settings and explanations,
 which should cover most of the available possibilities.
 
 Using various batch systems with :command:`ipcluster`
 ------------------------------------------------------
 
 :command:`ipcluster` has a notion of Launchers that can start controllers
 and engines with various remote execution schemes.  Currently supported
-models include `mpiexec`, PBS-style (Torque, SGE), and Windows HPC Server.
+models include :command:`ssh`, :command`mpiexec`, PBS-style (Torque, SGE),
+and Windows HPC Server.
 
 .. note::
 
     The Launchers and configuration are designed in such a way that advanced
     users can subclass and configure them to fit their own system that we
     have not yet supported (such as Condor)
 
 Using :command:`ipcluster` in mpiexec/mpirun mode
 --------------------------------------------------
 
 
 The mpiexec/mpirun mode is useful if you:
 
 1. Have MPI installed.
 2. Your systems are configured to use the :command:`mpiexec` or
    :command:`mpirun` commands to start MPI processes.
 
 If these are satisfied, you can create a new profile::
 
-    $ ipcluster create -p mpi
+    $ ipcluster create profile=mpi
 
 and edit the file :file:`IPYTHONDIR/cluster_mpi/ipcluster_config.py`.
 
 There, instruct ipcluster to use the MPIExec launchers by adding the lines:
 
 .. sourcecode:: python
 
-    c.Global.engine_launcher = 'IPython.parallel.apps.launcher.MPIExecEngineSetLauncher'
+    c.IPClusterEnginesApp.engine_launcher = 'IPython.parallel.apps.launcher.MPIExecEngineSetLauncher'
 
 If the default MPI configuration is correct, then you can now start your cluster, with::
 
-    $ ipcluster start -n 4 -p mpi
+    $ ipcluster start n=4 profile=mpi
 
 This does the following:
 
 1. Starts the IPython controller on current host.
 2. Uses :command:`mpiexec` to start 4 engines.
 
 If you have a reason to also start the Controller with mpi, you can specify:
 
 .. sourcecode:: python
 
-    c.Global.controller_launcher = 'IPython.parallel.apps.launcher.MPIExecControllerLauncher'
+    c.IPClusterStartApp.controller_launcher = 'IPython.parallel.apps.launcher.MPIExecControllerLauncher'
 
 .. note::
 
     The Controller *will not* be in the same MPI universe as the engines, so there is not
     much reason to do this unless sysadmins demand it.
 
 On newer MPI implementations (such as OpenMPI), this will work even if you
 don't make any calls to MPI or call :func:`MPI_Init`. However, older MPI
 implementations actually require each process to call :func:`MPI_Init` upon
 starting. The easiest way of having this done is to install the mpi4py
 [mpi4py]_ package and then specify the ``c.MPI.use`` option in :file:`ipengine_config.py`:
 
 .. sourcecode:: python
 
     c.MPI.use = 'mpi4py'
 
 Unfortunately, even this won't work for some MPI implementations. If you are
 having problems with this, you will likely have to use a custom Python
 executable that itself calls :func:`MPI_Init` at the appropriate time.
 Fortunately, mpi4py comes with such a custom Python executable that is easy to
 install and use. However, this custom Python executable approach will not work
 with :command:`ipcluster` currently.
 
 More details on using MPI with IPython can be found :ref:`here <parallelmpi>`.
 
 
 Using :command:`ipcluster` in PBS mode
 ---------------------------------------
 
 The PBS mode uses the Portable Batch System [PBS]_ to start the engines. 
 
 As usual, we will start by creating a fresh profile::
 
-    $ ipcluster create -p pbs
+    $ ipcluster create profile=pbs
 
 And in :file:`ipcluster_config.py`, we will select the PBS launchers for the controller
 and engines:
 
 .. sourcecode:: python
 
     c.Global.controller_launcher = 'IPython.parallel.apps.launcher.PBSControllerLauncher'
     c.Global.engine_launcher = 'IPython.parallel.apps.launcher.PBSEngineSetLauncher'
 
 IPython does provide simple default batch templates for PBS and SGE, but you may need
 to specify your own. Here is a sample PBS script template:
 
 .. sourcecode:: bash
 
     #PBS -N ipython
     #PBS -j oe
     #PBS -l walltime=00:10:00
     #PBS -l nodes=${n/4}:ppn=4
     #PBS -q $queue
 
     cd $$PBS_O_WORKDIR
     export PATH=$$HOME/usr/local/bin
     export PYTHONPATH=$$HOME/usr/local/lib/python2.7/site-packages
-    /usr/local/bin/mpiexec -n ${n} ipengine --cluster_dir=${cluster_dir}
+    /usr/local/bin/mpiexec -n ${n} ipengine cluster_dir=${cluster_dir}
 
 There are a few important points about this template:
 
 1. This template will be rendered at runtime using IPython's :mod:`Itpl` 
    template engine.
 
 2. Instead of putting in the actual number of engines, use the notation
    ``${n}`` to indicate the number of engines to be started. You can also uses
    expressions like ``${n/4}`` in the template to indicate the number of 
    nodes. There will always be a ${n} and ${cluster_dir} variable passed to the template.
    These allow the batch system to know how many engines, and where the configuration
    files reside. The same is true for the batch queue, with the template variable ``$queue``.
 
 3. Because ``$`` is a special character used by the template engine, you must  
    escape any ``$`` by using ``$$``.  This is important when referring to 
    environment variables in the template, or in SGE, where the config lines start
    with ``#$``, which will have to be ``#$$``.
 
 4. Any options to :command:`ipengine` can be given in the batch script 
    template, or in :file:`ipengine_config.py`.
 
 5. Depending on the configuration of you system, you may have to set 
    environment variables in the script template.
 
 The controller template should be similar, but simpler:
 
 .. sourcecode:: bash
 
     #PBS -N ipython
     #PBS -j oe
     #PBS -l walltime=00:10:00
     #PBS -l nodes=1:ppn=4
     #PBS -q $queue
 
     cd $$PBS_O_WORKDIR
     export PATH=$$HOME/usr/local/bin
     export PYTHONPATH=$$HOME/usr/local/lib/python2.7/site-packages
-    ipcontroller --cluster_dir=${cluster_dir}
+    ipcontroller cluster_dir=${cluster_dir}
 
 
 Once you have created these scripts, save them with names like
 :file:`pbs.engine.template`. Now you can load them into the :file:`ipcluster_config` with:
 
 .. sourcecode:: python
 
     c.PBSEngineSetLauncher.batch_template_file = "pbs.engine.template"
     
     c.PBSControllerLauncher.batch_template_file = "pbs.controller.template"
         
 
 Alternately, you can just define the templates as strings inside :file:`ipcluster_config`.
 
 Whether you are using your own templates or our defaults, the extra configurables available are
 the number of engines to launch (``$n``, and the batch system queue to which the jobs are to be
 submitted (``$queue``)). These are configurables, and can be specified in
 :file:`ipcluster_config`:
 
 .. sourcecode:: python
 
     c.PBSLauncher.queue = 'veryshort.q'
     c.PBSEngineSetLauncher.n = 64
 
 Note that assuming you are running PBS on a multi-node cluster, the Controller's default behavior
 of listening only on localhost is likely too restrictive.  In this case, also assuming the
 nodes are safely behind a firewall, you can simply instruct the Controller to listen for
 connections on all its interfaces, by adding in :file:`ipcontroller_config`:
 
 .. sourcecode:: python
 
     c.RegistrationFactory.ip = '*'
 
 You can now run the cluster with::
 
-    $ ipcluster start -p pbs -n 128
+    $ ipcluster start profile=pbs n=128
 
 Additional configuration options can be found in the PBS section of :file:`ipcluster_config`.
 
 .. note::
 
     Due to the flexibility of configuration, the PBS launchers work with simple changes
     to the template for other :command:`qsub`-using systems, such as Sun Grid Engine,
     and with further configuration in similar batch systems like Condor.
 
 
 Using :command:`ipcluster` in SSH mode
 ---------------------------------------
 
 
 The SSH mode uses :command:`ssh` to execute :command:`ipengine` on remote
 nodes and :command:`ipcontroller` can be run remotely as well, or on localhost.
 
 .. note::
 
     When using this mode it highly recommended that you have set up SSH keys
     and are using ssh-agent [SSH]_ for password-less logins.
 
 As usual, we start by creating a clean profile::
 
-    $ ipcluster create -p ssh
+    $ ipcluster create profile= ssh
 
 To use this mode, select the SSH launchers in :file:`ipcluster_config.py`:
 
 .. sourcecode:: python
 
     c.Global.engine_launcher = 'IPython.parallel.apps.launcher.SSHEngineSetLauncher'
     # and if the Controller is also to be remote:
     c.Global.controller_launcher = 'IPython.parallel.apps.launcher.SSHControllerLauncher'
     
 
 The controller's remote location and configuration can be specified:
 
 .. sourcecode:: python
 
     # Set the user and hostname for the controller
     # c.SSHControllerLauncher.hostname = 'controller.example.com'
     # c.SSHControllerLauncher.user = os.environ.get('USER','username')
 
     # Set the arguments to be passed to ipcontroller
     # note that remotely launched ipcontroller will not get the contents of 
     # the local ipcontroller_config.py unless it resides on the *remote host*
-    # in the location specified by the --cluster_dir argument.
+    # in the location specified by the `cluster_dir` argument.
     # c.SSHControllerLauncher.program_args = ['-r', '-ip', '0.0.0.0', '--cluster_dir', '/path/to/cd']
 
 .. note::
 
     SSH mode does not do any file movement, so you will need to distribute configuration
     files manually.  To aid in this, the `reuse_files` flag defaults to True for ssh-launched
     Controllers, so you will only need to do this once, unless you override this flag back
     to False.
 
 Engines are specified in a dictionary, by hostname and the number of engines to be run
 on that host.
 
 .. sourcecode:: python
 
     c.SSHEngineSetLauncher.engines = { 'host1.example.com' : 2,
                 'host2.example.com' : 5,
-                'host3.example.com' : (1, ['--cluster_dir', '/home/different/location']),
+                'host3.example.com' : (1, ['cluster_dir=/home/different/location']),
                 'host4.example.com' : 8 }
 
 * The `engines` dict, where the keys are the host we want to run engines on and
   the value is the number of engines to run on that host.
 * on host3, the value is a tuple, where the number of engines is first, and the arguments
   to be passed to :command:`ipengine` are the second element.
 
 For engines without explicitly specified arguments, the default arguments are set in
 a single location:
 
 .. sourcecode:: python
 
     c.SSHEngineSetLauncher.engine_args = ['--cluster_dir', '/path/to/cluster_ssh']
 
 Current limitations of the SSH mode of :command:`ipcluster` are:
 
 * Untested on Windows. Would require a working :command:`ssh` on Windows.
   Also, we are using shell scripts to setup and execute commands on remote
   hosts.
 * No file movement - 
 
 Using the :command:`ipcontroller` and :command:`ipengine` commands
 ====================================================================
 
 It is also possible to use the :command:`ipcontroller` and :command:`ipengine`
 commands to start your controller and engines. This approach gives you full
 control over all aspects of the startup process.
 
 Starting the controller and engine on your local machine
 --------------------------------------------------------
 
 To use :command:`ipcontroller` and :command:`ipengine` to start things on your
 local machine, do the following.
 
 First start the controller::
 
     $ ipcontroller
     
 Next, start however many instances of the engine you want using (repeatedly)
 the command::
 
     $ ipengine
 
 The engines should start and automatically connect to the controller using the
 JSON files in :file:`~/.ipython/cluster_default/security`. You are now ready to use the
 controller and engines from IPython.
 
 .. warning:: 
     
     The order of the above operations may be important.  You *must*
     start the controller before the engines, unless you are reusing connection
     information (via `-r`), in which case ordering is not important.
 
 .. note::
 
     On some platforms (OS X), to put the controller and engine into the
     background you may need to give these commands in the form ``(ipcontroller
     &)`` and ``(ipengine &)`` (with the parentheses) for them to work
     properly.
 
 Starting the controller and engines on different hosts
 ------------------------------------------------------
 
 When the controller and engines are running on different hosts, things are
 slightly more complicated, but the underlying ideas are the same:
 
 1. Start the controller on a host using :command:`ipcontroller`.
 2. Copy :file:`ipcontroller-engine.json` from :file:`~/.ipython/cluster_<profile>/security` on
    the controller's host to the host where the engines will run.
 3. Use :command:`ipengine` on the engine's hosts to start the engines.
 
 The only thing you have to be careful of is to tell :command:`ipengine` where
 the :file:`ipcontroller-engine.json` file is located. There are two ways you
 can do this:
 
 * Put :file:`ipcontroller-engine.json` in the :file:`~/.ipython/cluster_<profile>/security`
   directory on the engine's host, where it will be found automatically.
 * Call :command:`ipengine` with the ``--file=full_path_to_the_file``
   flag.
 
 The ``--file`` flag works like this::
 
     $ ipengine --file=/path/to/my/ipcontroller-engine.json
 
 .. note::
     
     If the controller's and engine's hosts all have a shared file system  
     (:file:`~/.ipython/cluster_<profile>/security` is the same on all of them), then things
     will just work!
 
 Make JSON files persistent
 --------------------------
 
 At fist glance it may seem that that managing the JSON files is a bit
 annoying. Going back to the house and key analogy, copying the JSON around
 each time you start the controller is like having to make a new key every time
 you want to unlock the door and enter your house. As with your house, you want
 to be able to create the key (or JSON file) once, and then simply use it at
 any point in the future.
 
-To do this, the only thing you have to do is specify the `-r` flag, so that
+To do this, the only thing you have to do is specify the `--reuse` flag, so that
 the connection information in the JSON files remains accurate::
 
-    $ ipcontroller -r
+    $ ipcontroller --reuse
 
 Then, just copy the JSON files over the first time and you are set. You can
 start and stop the controller and engines any many times as you want in the
 future, just make sure to tell the controller to reuse the file.
 
 .. note::
 
     You may ask the question: what ports does the controller listen on if you
     don't tell is to use specific ones? The default is to use high random port
     numbers. We do this for two reasons: i) to increase security through
     obscurity and ii) to multiple controllers on a given host to start and
     automatically use different ports.
 
 Log files
 ---------
 
 All of the components of IPython have log files associated with them.
 These log files can be extremely useful in debugging problems with
 IPython and can be found in the directory :file:`~/.ipython/cluster_<profile>/log`.
 Sending the log files to us will often help us to debug any problems.
 
 
 Configuring `ipcontroller`
 ---------------------------
 
 Ports and addresses
 *******************
 
 
 Database Backend
 ****************
 
 
 .. seealso::
 
     
 
 Configuring `ipengine`
 -----------------------
 
 .. note::
 
     TODO
 
 
 
 .. [PBS] Portable Batch System.  http://www.openpbs.org/
 .. [SSH] SSH-Agent http://en.wikipedia.org/wiki/ssh-agent

docs/source/parallel/parallel_task.txt

@@ -1,442 +1,442
 .. _parallel_task:
 
 ==========================
 The IPython task interface
 ==========================
 
 The task interface to the cluster presents the engines as a fault tolerant,
 dynamic load-balanced system of workers. Unlike the multiengine interface, in
 the task interface the user have no direct access to individual engines. By
 allowing the IPython scheduler to assign work, this interface is simultaneously
 simpler and more powerful.
 
 Best of all, the user can use both of these interfaces running at the same time
 to take advantage of their respective strengths. When the user can break up
 the user's work into segments that do not depend on previous execution, the
 task interface is ideal. But it also has more power and flexibility, allowing
 the user to guide the distribution of jobs, without having to assign tasks to
 engines explicitly.
 
 Starting the IPython controller and engines
 ===========================================
 
 To follow along with this tutorial, you will need to start the IPython
 controller and four IPython engines. The simplest way of doing this is to use
 the :command:`ipcluster` command::
 
-	$ ipcluster start -n 4
+	$ ipcluster start n=4
 	
 For more detailed information about starting the controller and engines, see
 our :ref:`introduction <ip1par>` to using IPython for parallel computing.
 
 Creating a ``Client`` instance
 ==============================
 
 The first step is to import the IPython :mod:`IPython.parallel`
 module and then create a :class:`.Client` instance, and we will also be using
 a :class:`LoadBalancedView`, here called `lview`:
 
 .. sourcecode:: ipython
 
     In [1]: from IPython.parallel import Client
 
     In [2]: rc = Client()
     
 
 This form assumes that the controller was started on localhost with default
 configuration. If not, the location of the controller must be given as an
 argument to the constructor:
 
 .. sourcecode:: ipython
 
     # for a visible LAN controller listening on an external port:
     In [2]: rc = Client('tcp://192.168.1.16:10101')
     # or to connect with a specific profile you have set up:
     In [3]: rc = Client(profile='mpi')
 
 For load-balanced execution, we will make use of a :class:`LoadBalancedView` object, which can
 be constructed via the client's :meth:`load_balanced_view` method:
 
 .. sourcecode:: ipython
 
     In [4]: lview = rc.load_balanced_view() # default load-balanced view
 
 .. seealso::
 
     For more information, see the in-depth explanation of :ref:`Views <parallel_details>`.
 
 
 Quick and easy parallelism
 ==========================
 
 In many cases, you simply want to apply a Python function to a sequence of
 objects, but *in parallel*. Like the multiengine interface, these can be
 implemented via the task interface. The exact same tools can perform these
 actions in load-balanced ways as well as multiplexed ways: a parallel version
 of :func:`map` and :func:`@parallel` function decorator. If one specifies the
 argument `balanced=True`, then they are dynamically load balanced. Thus, if the
 execution time per item varies significantly, you should use the versions in
 the task interface.
 
 Parallel map
 ------------
 
 To load-balance :meth:`map`,simply use a LoadBalancedView:
 
 .. sourcecode:: ipython
     
     In [62]: lview.block = True
     
 	In [63]: serial_result = map(lambda x:x**10, range(32))
 
 	In [64]: parallel_result = lview.map(lambda x:x**10, range(32))
 
 	In [65]: serial_result==parallel_result
 	Out[65]: True
 
 Parallel function decorator
 ---------------------------
 
 Parallel functions are just like normal function, but they can be called on
 sequences and *in parallel*. The multiengine interface provides a decorator
 that turns any Python function into a parallel function:
 
 .. sourcecode:: ipython
 
     In [10]: @lview.parallel()
        ....: def f(x):
        ....:     return 10.0*x**4
        ....:
 
     In [11]: f.map(range(32))    # this is done in parallel
     Out[11]: [0.0,10.0,160.0,...]
 
 .. _parallel_dependencies:
 
 Dependencies
 ============
 
 Often, pure atomic load-balancing is too primitive for your work. In these cases, you
 may want to associate some kind of `Dependency` that describes when, where, or whether
 a task can be run.  In IPython, we provide two types of dependencies: 
 `Functional Dependencies`_ and `Graph Dependencies`_
 
 .. note::
 
     It is important to note that the pure ZeroMQ scheduler does not support dependencies,
     and you will see errors or warnings if you try to use dependencies with the pure
     scheduler.
 
 Functional Dependencies
 -----------------------
 
 Functional dependencies are used to determine whether a given engine is capable of running
 a particular task.  This is implemented via a special :class:`Exception` class, 
 :class:`UnmetDependency`, found in `IPython.parallel.error`.  Its use is very simple: 
 if a task fails with an UnmetDependency exception, then the scheduler, instead of relaying
 the error up to the client like any other error, catches the error, and submits the task
 to a different engine.  This will repeat indefinitely, and a task will never be submitted
 to a given engine a second time.
 
 You can manually raise the :class:`UnmetDependency` yourself, but IPython has provided 
 some decorators for facilitating this behavior.
 
 There are two decorators and a class used for functional dependencies:
 
 .. sourcecode:: ipython
 
     In [9]: from IPython.parallel import depend, require, dependent
 
 @require
 ********
 
 The simplest sort of dependency is requiring that a Python module is available. The
 ``@require`` decorator lets you define a function that will only run on engines where names
 you specify are importable:
 
 .. sourcecode:: ipython
 
     In [10]: @require('numpy', 'zmq')
         ...: def myfunc():
         ...:     return dostuff()
 
 Now, any time you apply :func:`myfunc`, the task will only run on a machine that has
 numpy and pyzmq available, and when :func:`myfunc` is called, numpy and zmq will be imported.
 
 @depend
 *******
 
 The ``@depend`` decorator lets you decorate any function with any *other* function to
 evaluate the dependency. The dependency function will be called at the start of the task,
 and if it returns ``False``, then the dependency will be considered unmet, and the task
 will be assigned to another engine. If the dependency returns *anything other than
 ``False``*, the rest of the task will continue.
 
 .. sourcecode:: ipython
 
     In [10]: def platform_specific(plat):
         ...:    import sys
         ...:    return sys.platform == plat
     
     In [11]: @depend(platform_specific, 'darwin')
         ...: def mactask():
         ...:    do_mac_stuff()
 
     In [12]: @depend(platform_specific, 'nt')
         ...: def wintask():
         ...:    do_windows_stuff()
     
 In this case, any time you apply ``mytask``, it will only run on an OSX machine.
 ``@depend`` is just like ``apply``, in that it has a ``@depend(f,*args,**kwargs)``
 signature.
 
 dependents
 **********
 
 You don't have to use the decorators on your tasks, if for instance you may want
 to run tasks with a single function but varying dependencies, you can directly construct
 the :class:`dependent` object that the decorators use:
 
 .. sourcecode::ipython
 
     In [13]: def mytask(*args):
         ...:    dostuff()
     
     In [14]: mactask = dependent(mytask, platform_specific, 'darwin')
     # this is the same as decorating the declaration of mytask with @depend
     # but you can do it again:
     
     In [15]: wintask = dependent(mytask, platform_specific, 'nt')
     
     # in general:
     In [16]: t = dependent(f, g, *dargs, **dkwargs)
     
     # is equivalent to:
     In [17]: @depend(g, *dargs, **dkwargs)
         ...: def t(a,b,c):
         ...:     # contents of f
 
 Graph Dependencies
 ------------------
 
 Sometimes you want to restrict the time and/or location to run a given task as a function
 of the time and/or location of other tasks. This is implemented via a subclass of
 :class:`set`, called a :class:`Dependency`. A Dependency is just a set of `msg_ids`
 corresponding to tasks, and a few attributes to guide how to decide when the Dependency
 has been met.
 
 The switches we provide for interpreting whether a given dependency set has been met:
 
 any|all
     Whether the dependency is considered met if *any* of the dependencies are done, or
     only after *all* of them have finished.  This is set by a Dependency's :attr:`all`
     boolean attribute, which defaults to ``True``.
 
 success [default: True]
     Whether to consider tasks that succeeded as fulfilling dependencies.
 
 failure [default : False]
     Whether to consider tasks that failed as fulfilling dependencies.
     using `failure=True,success=False` is useful for setting up cleanup tasks, to be run
     only when tasks have failed.
 
 Sometimes you want to run a task after another, but only if that task succeeded. In this case,
 ``success`` should be ``True`` and ``failure`` should be ``False``. However sometimes you may
 not care whether the task succeeds, and always want the second task to run, in which case you
 should use `success=failure=True`. The default behavior is to only use successes.
 
 There are other switches for interpretation that are made at the *task* level.  These are
 specified via keyword arguments to the client's :meth:`apply` method.
 
 after,follow
     You may want to run a task *after* a given set of dependencies have been run and/or
     run it *where* another set of dependencies are met. To support this, every task has an
     `after` dependency to restrict time, and a `follow` dependency to restrict
     destination.
 
 timeout
     You may also want to set a time-limit for how long the scheduler should wait before a
     task's dependencies are met. This is done via a `timeout`, which defaults to 0, which
     indicates that the task should never timeout. If the timeout is reached, and the
     scheduler still hasn't been able to assign the task to an engine, the task will fail
     with a :class:`DependencyTimeout`.
 
 .. note::
 
     Dependencies only work within the task scheduler. You cannot instruct a load-balanced
     task to run after a job submitted via the MUX interface.
 
 The simplest form of Dependencies is with `all=True,success=True,failure=False`. In these cases,
 you can skip using Dependency objects, and just pass msg_ids or AsyncResult objects as the
 `follow` and `after` keywords to :meth:`client.apply`:
 
 .. sourcecode:: ipython
 
     In [14]: client.block=False
 
     In [15]: ar = lview.apply(f, args, kwargs)
 
     In [16]: ar2 = lview.apply(f2)
 
     In [17]: ar3 = lview.apply_with_flags(f3, after=[ar,ar2])
 
     In [17]: ar4 = lview.apply_with_flags(f3, follow=[ar], timeout=2.5)
 
 
 .. seealso::
 
     Some parallel workloads can be described as a `Directed Acyclic Graph
     <http://en.wikipedia.org/wiki/Directed_acyclic_graph>`_, or DAG. See :ref:`DAG
     Dependencies <dag_dependencies>` for an example demonstrating how to use map a NetworkX DAG
     onto task dependencies.
 
 
 
 
 Impossible Dependencies
 ***********************
 
 The schedulers do perform some analysis on graph dependencies to determine whether they
 are not possible to be met. If the scheduler does discover that a dependency cannot be
 met, then the task will fail with an :class:`ImpossibleDependency` error. This way, if the
 scheduler realized that a task can never be run, it won't sit indefinitely in the
 scheduler clogging the pipeline.
 
 The basic cases that are checked:
 
 * depending on nonexistent messages
 * `follow` dependencies were run on more than one machine and `all=True`
 * any dependencies failed and `all=True,success=True,failures=False`
 * all dependencies failed and `all=False,success=True,failure=False`
 
 .. warning::
 
     This analysis has not been proven to be rigorous, so it is likely possible for tasks
     to become impossible to run in obscure situations, so a timeout may be a good choice.
 
 
 Retries and Resubmit
 ====================
 
 Retries
 -------
 
 Another flag for tasks is `retries`.  This is an integer, specifying how many times
 a task should be resubmitted after failure.  This is useful for tasks that should still run
 if their engine was shutdown, or may have some statistical chance of failing.  The default
 is to not retry tasks.
 
 Resubmit
 --------
 
 Sometimes you may want to re-run a task. This could be because it failed for some reason, and
 you have fixed the error, or because you want to restore the cluster to an interrupted state.
 For this, the :class:`Client` has a :meth:`rc.resubmit` method.  This simply takes one or more
 msg_ids, and returns an :class:`AsyncHubResult` for the result(s).  You cannot resubmit
 a task that is pending - only those that have finished, either successful or unsuccessful.
 
 .. _parallel_schedulers:
 
 Schedulers
 ==========
 
 There are a variety of valid ways to determine where jobs should be assigned in a 
 load-balancing situation.  In IPython, we support several standard schemes, and
-even make it easy to define your own.  The scheme can be selected via the ``--scheme``
-argument to :command:`ipcontroller`, or in the :attr:`HubFactory.scheme` attribute
+even make it easy to define your own.  The scheme can be selected via the ``scheme``
+argument to :command:`ipcontroller`, or in the :attr:`TaskScheduler.schemename` attribute
 of a controller config object.
 
 The built-in routing schemes:
 
 To select one of these schemes, simply do::
 
-    $ ipcontroller --scheme <schemename>
+    $ ipcontroller scheme=<schemename>
     for instance:
-    $ ipcontroller --scheme lru
+    $ ipcontroller scheme=lru
 
 lru: Least Recently Used
 
     Always assign work to the least-recently-used engine.  A close relative of
     round-robin, it will be fair with respect to the number of tasks, agnostic 
     with respect to runtime of each task.
 
 plainrandom: Plain Random
 
     Randomly picks an engine on which to run.
 
 twobin: Two-Bin Random
 
     **Requires numpy**
     
     Pick two engines at random, and use the LRU of the two. This is known to be better
     than plain random in many cases, but requires a small amount of computation.
 
 leastload: Least Load
 
     **This is the default scheme**
 
     Always assign tasks to the engine with the fewest outstanding tasks (LRU breaks tie).
 
 weighted: Weighted Two-Bin Random
 
     **Requires numpy**
     
     Pick two engines at random using the number of outstanding tasks as inverse weights,
     and use the one with the lower load.
 
 
 Pure ZMQ Scheduler
 ------------------
 
 For maximum throughput, the 'pure' scheme is not Python at all, but a C-level
 :class:`MonitoredQueue` from PyZMQ, which uses a ZeroMQ ``XREQ`` socket to perform all
 load-balancing. This scheduler does not support any of the advanced features of the Python
 :class:`.Scheduler`.
 
 Disabled features when using the ZMQ Scheduler:
 
 * Engine unregistration
     Task farming will be disabled if an engine unregisters.  
     Further, if an engine is unregistered during computation, the scheduler may not recover.
 * Dependencies
     Since there is no Python logic inside the Scheduler, routing decisions cannot be made
     based on message content.
 * Early destination notification
     The Python schedulers know which engine gets which task, and notify the Hub.  This 
     allows graceful handling of Engines coming and going.  There is no way to know
     where ZeroMQ messages have gone, so there is no way to know what tasks are on which
     engine until they *finish*.  This makes recovery from engine shutdown very difficult.
 
 
 .. note::
 
     TODO: performance comparisons
 
 
 
 
 More details
 ============
 
 The :class:`LoadBalancedView` has many more powerful features that allow quite a bit
 of flexibility in how tasks are defined and run. The next places to look are
 in the following classes:
 
 * :class:`~IPython.parallel.client.view.LoadBalancedView`
 * :class:`~IPython.parallel.client.asyncresult.AsyncResult`
 * :meth:`~IPython.parallel.client.view.LoadBalancedView.apply`
 * :mod:`~IPython.parallel.controller.dependency`
 
 The following is an overview of how to use these classes together:
 
 1. Create a :class:`Client` and :class:`LoadBalancedView`
 2. Define some functions to be run as tasks
 3. Submit your tasks to using the :meth:`apply` method of your
    :class:`LoadBalancedView` instance.
 4. Use :meth:`Client.get_result` to get the results of the
    tasks, or use the :meth:`AsyncResult.get` method of the results to wait
    for and then receive the results.
 
 .. seealso::
 
     A demo of :ref:`DAG Dependencies <dag_dependencies>` with NetworkX and IPython.

docs/source/parallel/parallel_winhpc.txt

@@ -1,334 +1,334
 ============================================
 Getting started with Windows HPC Server 2008
 ============================================
 
 .. note::
 
     Not adapted to zmq yet
 
 Introduction
 ============
 
 The Python programming language is an increasingly popular language for
 numerical computing. This is due to a unique combination of factors. First,
 Python is a high-level and *interactive* language that is well matched to
 interactive numerical work. Second, it is easy (often times trivial) to
 integrate legacy C/C++/Fortran code into Python. Third, a large number of
 high-quality open source projects provide all the needed building blocks for
 numerical computing: numerical arrays (NumPy), algorithms (SciPy), 2D/3D
 Visualization (Matplotlib, Mayavi, Chaco), Symbolic Mathematics (Sage, Sympy)
 and others.
 
 The IPython project is a core part of this open-source toolchain and is
 focused on creating a comprehensive environment for interactive and
 exploratory computing in the Python programming language. It enables all of
 the above tools to be used interactively and consists of two main components:
 
 * An enhanced interactive Python shell with support for interactive plotting
   and visualization.
 * An architecture for interactive parallel computing.
 
 With these components, it is possible to perform all aspects of a parallel
 computation interactively. This type of workflow is particularly relevant in
 scientific and numerical computing where algorithms, code and data are
 continually evolving as the user/developer explores a problem. The broad
 treads in computing (commodity clusters, multicore, cloud computing, etc.)
 make these capabilities of IPython particularly relevant.
 
 While IPython is a cross platform tool, it has particularly strong support for
 Windows based compute clusters running Windows HPC Server 2008. This document
 describes how to get started with IPython on Windows HPC Server 2008. The
 content and emphasis here is practical: installing IPython, configuring
 IPython to use the Windows job scheduler and running example parallel programs
 interactively. A more complete description of IPython's parallel computing
 capabilities can be found in IPython's online documentation
 (http://ipython.scipy.org/moin/Documentation).
 
 Setting up your Windows cluster
 ===============================
 
 This document assumes that you already have a cluster running Windows
 HPC Server 2008. Here is a broad overview of what is involved with setting up
 such a cluster:
 
 1. Install Windows Server 2008 on the head and compute nodes in the cluster.
 2. Setup the network configuration on each host. Each host should have a 
    static IP address.
 3. On the head node, activate the "Active Directory Domain Services" role 
    and make the head node the domain controller.
 4. Join the compute nodes to the newly created Active Directory (AD) domain.
 5. Setup user accounts in the domain with shared home directories.
 6. Install the HPC Pack 2008 on the head node to create a cluster.
 7. Install the HPC Pack 2008 on the compute nodes.
 
 More details about installing and configuring Windows HPC Server 2008 can be
 found on the Windows HPC Home Page (http://www.microsoft.com/hpc). Regardless
 of what steps you follow to set up your cluster, the remainder of this
 document will assume that:
 
 * There are domain users that can log on to the AD domain and submit jobs
   to the cluster scheduler.
 * These domain users have shared home directories. While shared home
   directories are not required to use IPython, they make it much easier to
   use IPython.
 
 Installation of IPython and its dependencies
 ============================================
 
 IPython and all of its dependencies are freely available and open source.
 These packages provide a powerful and cost-effective approach to numerical and
 scientific computing on Windows. The following dependencies are needed to run
 IPython on Windows:
 
 * Python 2.6 or 2.7 (http://www.python.org)
 * pywin32 (http://sourceforge.net/projects/pywin32/)
 * PyReadline (https://launchpad.net/pyreadline)
 * pyzmq (http://github.com/zeromq/pyzmq/downloads)
 * IPython (http://ipython.scipy.org)
 
 In addition, the following dependencies are needed to run the demos described
 in this document.
 
 * NumPy and SciPy (http://www.scipy.org)
 * Matplotlib (http://matplotlib.sourceforge.net/)
 
 The easiest way of obtaining these dependencies is through the Enthought
 Python Distribution (EPD) (http://www.enthought.com/products/epd.php). EPD is
 produced by Enthought, Inc. and contains all of these packages and others in a
 single installer and is available free for academic users. While it is also
 possible to download and install each package individually, this is a tedious
 process. Thus, we highly recommend using EPD to install these packages on
 Windows.
 
 Regardless of how you install the dependencies, here are the steps you will
 need to follow:
 
 1. Install all of the packages listed above, either individually or using EPD
    on the head node, compute nodes and user workstations.
 
 2. Make sure that :file:`C:\\Python27` and :file:`C:\\Python27\\Scripts` are
    in the system :envvar:`%PATH%` variable on each node.
 
 3. Install the latest development version of IPython. This can be done by
    downloading the the development version from the IPython website
    (http://ipython.scipy.org) and following the installation instructions.
 
 Further details about installing IPython or its dependencies can be found in
 the online IPython documentation (http://ipython.scipy.org/moin/Documentation)
 Once you are finished with the installation, you can try IPython out by
 opening a Windows Command Prompt and typing ``ipython``. This will
 start IPython's interactive shell and you should see something like the
 following screenshot:
 
 .. image:: ipython_shell.*
 
 Starting an IPython cluster
 ===========================
 
 To use IPython's parallel computing capabilities, you will need to start an
 IPython cluster. An IPython cluster consists of one controller and multiple
 engines:
 
 IPython controller
     The IPython controller manages the engines and acts as a gateway between 
     the engines and the client, which runs in the user's interactive IPython
     session. The controller is started using the :command:`ipcontroller`
     command.
 
 IPython engine
     IPython engines run a user's Python code in parallel on the compute nodes. 
     Engines are starting using the :command:`ipengine` command.
 
 Once these processes are started, a user can run Python code interactively and
 in parallel on the engines from within the IPython shell using an appropriate
 client. This includes the ability to interact with, plot and visualize data
 from the engines.
 
 IPython has a command line program called :command:`ipcluster` that automates
 all aspects of starting the controller and engines on the compute nodes.
 :command:`ipcluster` has full support for the Windows HPC job scheduler,
 meaning that :command:`ipcluster` can use this job scheduler to start the
 controller and engines. In our experience, the Windows HPC job scheduler is
 particularly well suited for interactive applications, such as IPython. Once
 :command:`ipcluster` is configured properly, a user can start an IPython
 cluster from their local workstation almost instantly, without having to log
 on to the head node (as is typically required by Unix based job schedulers).
 This enables a user to move seamlessly between serial and parallel
 computations.
 
 In this section we show how to use :command:`ipcluster` to start an IPython
 cluster using the Windows HPC Server 2008 job scheduler. To make sure that
 :command:`ipcluster` is installed and working properly, you should first try
 to start an IPython cluster on your local host. To do this, open a Windows
 Command Prompt and type the following command::
 
-    ipcluster start -n 2
+    ipcluster start n=2
 
 You should see a number of messages printed to the screen, ending with 
 "IPython cluster: started". The result should look something like the following
 screenshot:
 
 .. image:: ipcluster_start.*
 
 At this point, the controller and two engines are running on your local host.
 This configuration is useful for testing and for situations where you want to
 take advantage of multiple cores on your local computer. 
 
 Now that we have confirmed that :command:`ipcluster` is working properly, we
 describe how to configure and run an IPython cluster on an actual compute
 cluster running Windows HPC Server 2008. Here is an outline of the needed
 steps:
 
-1. Create a cluster profile using: ``ipcluster create -p mycluster``
+1. Create a cluster profile using: ``ipcluster create profile=mycluster``
 
 2. Edit configuration files in the directory :file:`.ipython\\cluster_mycluster`
 
-3. Start the cluster using: ``ipcluser start -p mycluster -n 32``
+3. Start the cluster using: ``ipcluser start profile=mycluster n=32``
 
 Creating a cluster profile
 --------------------------
 
 In most cases, you will have to create a cluster profile to use IPython on a
 cluster. A cluster profile is a name (like "mycluster") that is associated
 with a particular cluster configuration. The profile name is used by
 :command:`ipcluster` when working with the cluster.
 
 Associated with each cluster profile is a cluster directory. This cluster
 directory is a specially named directory (typically located in the
 :file:`.ipython` subdirectory of your home directory) that contains the
 configuration files for a particular cluster profile, as well as log files and
 security keys. The naming convention for cluster directories is:
 :file:`cluster_<profile name>`. Thus, the cluster directory for a profile named
 "foo" would be :file:`.ipython\\cluster_foo`.
 
 To create a new cluster profile (named "mycluster") and the associated cluster
 directory, type the following command at the Windows Command Prompt::
 
-    ipcluster create -p mycluster
+    ipcluster create profile=mycluster
 
 The output of this command is shown in the screenshot below. Notice how
 :command:`ipcluster` prints out the location of the newly created cluster
 directory.
 
 .. image:: ipcluster_create.*
 
 Configuring a cluster profile
 -----------------------------
 
 Next, you will need to configure the newly created cluster profile by editing
 the following configuration files in the cluster directory:
 
 * :file:`ipcluster_config.py`
 * :file:`ipcontroller_config.py`
 * :file:`ipengine_config.py`
 
 When :command:`ipcluster` is run, these configuration files are used to 
 determine how the engines and controller will be started. In most cases,
 you will only have to set a few of the attributes in these files.
 
 To configure :command:`ipcluster` to use the Windows HPC job scheduler, you
 will need to edit the following attributes in the file
 :file:`ipcluster_config.py`::
 
     # Set these at the top of the file to tell ipcluster to use the
     # Windows HPC job scheduler.
     c.Global.controller_launcher = \
         'IPython.parallel.apps.launcher.WindowsHPCControllerLauncher'
     c.Global.engine_launcher = \
         'IPython.parallel.apps.launcher.WindowsHPCEngineSetLauncher'
 
     # Set these to the host name of the scheduler (head node) of your cluster.
     c.WindowsHPCControllerLauncher.scheduler = 'HEADNODE'
     c.WindowsHPCEngineSetLauncher.scheduler = 'HEADNODE'
 
 There are a number of other configuration attributes that can be set, but
 in most cases these will be sufficient to get you started.
 
 .. warning::
     If any of your configuration attributes involve specifying the location
     of shared directories or files, you must make sure that you use UNC paths
     like :file:`\\\\host\\share`. It is also important that you specify
     these paths using raw Python strings: ``r'\\host\share'`` to make sure
     that the backslashes are properly escaped.
 
 Starting the cluster profile
 ----------------------------
 
 Once a cluster profile has been configured, starting an IPython cluster using
 the profile is simple::
 
-    ipcluster start -p mycluster -n 32
+    ipcluster start profile=mycluster n=32
 
 The ``-n`` option tells :command:`ipcluster` how many engines to start (in
 this case 32). Stopping the cluster is as simple as typing Control-C.
 
 Using the HPC Job Manager
 -------------------------
 
 When ``ipcluster start`` is run the first time, :command:`ipcluster` creates
 two XML job description files in the cluster directory:
 
 * :file:`ipcontroller_job.xml`
 * :file:`ipengineset_job.xml`
 
 Once these files have been created, they can be imported into the HPC Job
 Manager application. Then, the controller and engines for that profile can be
 started using the HPC Job Manager directly, without using :command:`ipcluster`.
 However, anytime the cluster profile is re-configured, ``ipcluster start``
 must be run again to regenerate the XML job description files. The
 following screenshot shows what the HPC Job Manager interface looks like
 with a running IPython cluster.
 
 .. image:: hpc_job_manager.*
 
 Performing a simple interactive parallel computation
 ====================================================
 
 Once you have started your IPython cluster, you can start to use it. To do
 this, open up a new Windows Command Prompt and start up IPython's interactive
 shell by typing::
 
     ipython
 
 Then you can create a :class:`MultiEngineClient` instance for your profile and
 use the resulting instance to do a simple interactive parallel computation. In
 the code and screenshot that follows, we take a simple Python function and
 apply it to each element of an array of integers in parallel using the
 :meth:`MultiEngineClient.map` method:
 
 .. sourcecode:: ipython
 
     In [1]: from IPython.parallel import *
 
     In [2]: c = MultiEngineClient(profile='mycluster')
 
     In [3]: mec.get_ids()
     Out[3]: [0, 1, 2, 3, 4, 5, 67, 8, 9, 10, 11, 12, 13, 14]
 
     In [4]: def f(x):
        ...:     return x**10
 
     In [5]: mec.map(f, range(15))  # f is applied in parallel
     Out[5]:
     [0,
      1,
      1024,
      59049,
      1048576,
      9765625,
      60466176,
      282475249,
      1073741824,
      3486784401L,
      10000000000L,
      25937424601L,
      61917364224L,
      137858491849L,
      289254654976L]
 
 The :meth:`map` method has the same signature as Python's builtin :func:`map`
 function, but runs the calculation in parallel. More involved examples of using
 :class:`MultiEngineClient` are provided in the examples that follow.
 
 .. image:: mec_simple.*