|
|
#!/usr/bin/env python
|
|
|
# encoding: utf-8
|
|
|
"""
|
|
|
The ipcluster application.
|
|
|
"""
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
|
# Copyright (C) 2008-2009 The IPython Development Team
|
|
|
#
|
|
|
# Distributed under the terms of the BSD License. The full license is in
|
|
|
# the file COPYING, distributed as part of this software.
|
|
|
#-----------------------------------------------------------------------------
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
|
# Imports
|
|
|
#-----------------------------------------------------------------------------
|
|
|
|
|
|
import logging
|
|
|
import os
|
|
|
import signal
|
|
|
|
|
|
if os.name=='posix':
|
|
|
from twisted.scripts._twistd_unix import daemonize
|
|
|
|
|
|
from twisted.internet import reactor, defer
|
|
|
from twisted.python import log, failure
|
|
|
|
|
|
|
|
|
from IPython.external.argparse import ArgumentParser, SUPPRESS
|
|
|
from IPython.utils.importstring import import_item
|
|
|
from IPython.kernel.clusterdir import (
|
|
|
ApplicationWithClusterDir, ClusterDirConfigLoader,
|
|
|
ClusterDirError, PIDFileError
|
|
|
)
|
|
|
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
|
# Module level variables
|
|
|
#-----------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
default_config_file_name = u'ipcluster_config.py'
|
|
|
|
|
|
|
|
|
_description = """\
|
|
|
Start an IPython cluster for parallel computing.\n\n
|
|
|
|
|
|
An IPython cluster consists of 1 controller and 1 or more engines.
|
|
|
This command automates the startup of these processes using a wide
|
|
|
range of startup methods (SSH, local processes, PBS, mpiexec,
|
|
|
Windows HPC Server 2008). To start a cluster with 4 engines on your
|
|
|
local host simply do 'ipcluster start -n 4'. For more complex usage
|
|
|
you will typically do 'ipcluster create -p mycluster', then edit
|
|
|
configuration files, followed by 'ipcluster start -p mycluster -n 4'.
|
|
|
"""
|
|
|
|
|
|
|
|
|
# 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
|
|
|
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
|
# Command line options
|
|
|
#-----------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
class IPClusterAppConfigLoader(ClusterDirConfigLoader):
|
|
|
|
|
|
def _add_arguments(self):
|
|
|
# Don't call ClusterDirConfigLoader._add_arguments as we don't want
|
|
|
# its defaults on self.parser. Instead, we will put those on
|
|
|
# default options on our subparsers.
|
|
|
|
|
|
# This has all the common options that all subcommands use
|
|
|
parent_parser1 = ArgumentParser(
|
|
|
add_help=False,
|
|
|
argument_default=SUPPRESS
|
|
|
)
|
|
|
self._add_ipython_dir(parent_parser1)
|
|
|
self._add_log_level(parent_parser1)
|
|
|
|
|
|
# This has all the common options that other subcommands use
|
|
|
parent_parser2 = ArgumentParser(
|
|
|
add_help=False,
|
|
|
argument_default=SUPPRESS
|
|
|
)
|
|
|
self._add_cluster_profile(parent_parser2)
|
|
|
self._add_cluster_dir(parent_parser2)
|
|
|
self._add_work_dir(parent_parser2)
|
|
|
paa = parent_parser2.add_argument
|
|
|
paa('--log-to-file',
|
|
|
action='store_true', dest='Global.log_to_file',
|
|
|
help='Log to a file in the log directory (default is stdout)')
|
|
|
|
|
|
# Create the object used to create the subparsers.
|
|
|
subparsers = self.parser.add_subparsers(
|
|
|
dest='Global.subcommand',
|
|
|
title='ipcluster subcommands',
|
|
|
description=
|
|
|
"""ipcluster has a variety of subcommands. The general way of
|
|
|
running ipcluster is 'ipcluster <cmd> [options]'. To get help
|
|
|
on a particular subcommand do 'ipcluster <cmd> -h'."""
|
|
|
# help="For more help, type 'ipcluster <cmd> -h'",
|
|
|
)
|
|
|
|
|
|
# The "list" subcommand parser
|
|
|
parser_list = subparsers.add_parser(
|
|
|
'list',
|
|
|
parents=[parent_parser1],
|
|
|
argument_default=SUPPRESS,
|
|
|
help="List all clusters in cwd and ipython_dir.",
|
|
|
description=
|
|
|
"""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>'."""
|
|
|
)
|
|
|
|
|
|
# The "create" subcommand parser
|
|
|
parser_create = subparsers.add_parser(
|
|
|
'create',
|
|
|
parents=[parent_parser1, parent_parser2],
|
|
|
argument_default=SUPPRESS,
|
|
|
help="Create a new cluster directory.",
|
|
|
description=
|
|
|
"""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 -p mycluster', which will put the directory
|
|
|
in '<ipython_dir>/cluster_mycluster'.
|
|
|
"""
|
|
|
)
|
|
|
paa = parser_create.add_argument
|
|
|
paa('--reset-config',
|
|
|
dest='Global.reset_config', action='store_true',
|
|
|
help=
|
|
|
"""Recopy the default config files to the cluster directory.
|
|
|
You will loose any modifications you have made to these files.""")
|
|
|
|
|
|
# The "start" subcommand parser
|
|
|
parser_start = subparsers.add_parser(
|
|
|
'start',
|
|
|
parents=[parent_parser1, parent_parser2],
|
|
|
argument_default=SUPPRESS,
|
|
|
help="Start a cluster.",
|
|
|
description=
|
|
|
"""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 -p <profile>`,
|
|
|
otherwise use the '--cluster-dir' option.
|
|
|
"""
|
|
|
)
|
|
|
paa = parser_start.add_argument
|
|
|
paa('-n', '--number',
|
|
|
type=int, dest='Global.n',
|
|
|
help='The number of engines to start.',
|
|
|
metavar='Global.n')
|
|
|
paa('--clean-logs',
|
|
|
dest='Global.clean_logs', action='store_true',
|
|
|
help='Delete old log flies before starting.')
|
|
|
paa('--no-clean-logs',
|
|
|
dest='Global.clean_logs', action='store_false',
|
|
|
help="Don't delete old log flies before starting.")
|
|
|
paa('--daemon',
|
|
|
dest='Global.daemonize', action='store_true',
|
|
|
help='Daemonize the ipcluster program. This implies --log-to-file')
|
|
|
paa('--no-daemon',
|
|
|
dest='Global.daemonize', action='store_false',
|
|
|
help="Dont't daemonize the ipcluster program.")
|
|
|
|
|
|
# The "stop" subcommand parser
|
|
|
parser_stop = subparsers.add_parser(
|
|
|
'stop',
|
|
|
parents=[parent_parser1, parent_parser2],
|
|
|
argument_default=SUPPRESS,
|
|
|
help="Stop a running cluster.",
|
|
|
description=
|
|
|
"""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 -p <profile>`, otherwise
|
|
|
use the '--cluster-dir' option.
|
|
|
"""
|
|
|
)
|
|
|
paa = parser_stop.add_argument
|
|
|
paa('--signal',
|
|
|
dest='Global.signal', type=int,
|
|
|
help="The signal number to use in stopping the cluster (default=2).",
|
|
|
metavar="Global.signal")
|
|
|
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
|
# Main application
|
|
|
#-----------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
class IPClusterApp(ApplicationWithClusterDir):
|
|
|
|
|
|
name = u'ipcluster'
|
|
|
description = _description
|
|
|
usage = None
|
|
|
command_line_loader = IPClusterAppConfigLoader
|
|
|
default_config_file_name = default_config_file_name
|
|
|
default_log_level = logging.INFO
|
|
|
auto_create_cluster_dir = False
|
|
|
|
|
|
def create_default_config(self):
|
|
|
super(IPClusterApp, self).create_default_config()
|
|
|
self.default_config.Global.controller_launcher = \
|
|
|
'IPython.kernel.launcher.LocalControllerLauncher'
|
|
|
self.default_config.Global.engine_launcher = \
|
|
|
'IPython.kernel.launcher.LocalEngineSetLauncher'
|
|
|
self.default_config.Global.n = 2
|
|
|
self.default_config.Global.reset_config = False
|
|
|
self.default_config.Global.clean_logs = True
|
|
|
self.default_config.Global.signal = 2
|
|
|
self.default_config.Global.daemonize = False
|
|
|
|
|
|
def find_resources(self):
|
|
|
subcommand = self.command_line_config.Global.subcommand
|
|
|
if subcommand=='list':
|
|
|
self.list_cluster_dirs()
|
|
|
# Exit immediately because there is nothing left to do.
|
|
|
self.exit()
|
|
|
elif subcommand=='create':
|
|
|
self.auto_create_cluster_dir = True
|
|
|
super(IPClusterApp, self).find_resources()
|
|
|
elif subcommand=='start' or subcommand=='stop':
|
|
|
self.auto_create_cluster_dir = True
|
|
|
try:
|
|
|
super(IPClusterApp, self).find_resources()
|
|
|
except ClusterDirError:
|
|
|
raise ClusterDirError(
|
|
|
"Could not find a cluster directory. A cluster dir must "
|
|
|
"be created before running 'ipcluster start'. Do "
|
|
|
"'ipcluster create -h' or 'ipcluster list -h' for more "
|
|
|
"information about creating and listing cluster dirs."
|
|
|
)
|
|
|
|
|
|
def list_cluster_dirs(self):
|
|
|
# Find the search paths
|
|
|
cluster_dir_paths = os.environ.get('IPCLUSTER_DIR_PATH','')
|
|
|
if cluster_dir_paths:
|
|
|
cluster_dir_paths = cluster_dir_paths.split(':')
|
|
|
else:
|
|
|
cluster_dir_paths = []
|
|
|
try:
|
|
|
ipython_dir = self.command_line_config.Global.ipython_dir
|
|
|
except AttributeError:
|
|
|
ipython_dir = self.default_config.Global.ipython_dir
|
|
|
paths = [os.getcwd(), ipython_dir] + \
|
|
|
cluster_dir_paths
|
|
|
paths = list(set(paths))
|
|
|
|
|
|
self.log.info('Searching for cluster dirs in paths: %r' % paths)
|
|
|
for path in paths:
|
|
|
files = os.listdir(path)
|
|
|
for f in files:
|
|
|
full_path = os.path.join(path, f)
|
|
|
if os.path.isdir(full_path) and f.startswith('cluster_'):
|
|
|
profile = full_path.split('_')[-1]
|
|
|
start_cmd = 'ipcluster start -p %s -n 4' % profile
|
|
|
print start_cmd + " ==> " + full_path
|
|
|
|
|
|
def pre_construct(self):
|
|
|
# IPClusterApp.pre_construct() is where we cd to the working directory.
|
|
|
super(IPClusterApp, self).pre_construct()
|
|
|
config = self.master_config
|
|
|
try:
|
|
|
daemon = config.Global.daemonize
|
|
|
if daemon:
|
|
|
config.Global.log_to_file = True
|
|
|
except AttributeError:
|
|
|
pass
|
|
|
|
|
|
def construct(self):
|
|
|
config = self.master_config
|
|
|
subcmd = config.Global.subcommand
|
|
|
reset = config.Global.reset_config
|
|
|
if subcmd == 'list':
|
|
|
return
|
|
|
if subcmd == 'create':
|
|
|
self.log.info('Copying default config files to cluster directory '
|
|
|
'[overwrite=%r]' % (reset,))
|
|
|
self.cluster_dir_obj.copy_all_config_files(overwrite=reset)
|
|
|
if subcmd =='start':
|
|
|
self.cluster_dir_obj.copy_all_config_files(overwrite=False)
|
|
|
self.start_logging()
|
|
|
reactor.callWhenRunning(self.start_launchers)
|
|
|
|
|
|
def start_launchers(self):
|
|
|
config = self.master_config
|
|
|
|
|
|
# Create the launchers. In both bases, we set the work_dir of
|
|
|
# the launcher to the cluster_dir. This is where the launcher's
|
|
|
# subprocesses will be launched. It is not where the controller
|
|
|
# and engine will be launched.
|
|
|
el_class = import_item(config.Global.engine_launcher)
|
|
|
self.engine_launcher = el_class(
|
|
|
work_dir=self.cluster_dir, config=config
|
|
|
)
|
|
|
cl_class = import_item(config.Global.controller_launcher)
|
|
|
self.controller_launcher = cl_class(
|
|
|
work_dir=self.cluster_dir, config=config
|
|
|
)
|
|
|
|
|
|
# Setup signals
|
|
|
signal.signal(signal.SIGINT, self.sigint_handler)
|
|
|
|
|
|
# Setup the observing of stopping. If the controller dies, shut
|
|
|
# everything down as that will be completely fatal for the engines.
|
|
|
d1 = self.controller_launcher.observe_stop()
|
|
|
d1.addCallback(self.stop_launchers)
|
|
|
# But, we don't monitor the stopping of engines. An engine dying
|
|
|
# is just fine and in principle a user could start a new engine.
|
|
|
# Also, if we did monitor engine stopping, it is difficult to
|
|
|
# know what to do when only some engines die. Currently, the
|
|
|
# observing of engine stopping is inconsistent. Some launchers
|
|
|
# might trigger on a single engine stopping, other wait until
|
|
|
# all stop. TODO: think more about how to handle this.
|
|
|
|
|
|
# Start the controller and engines
|
|
|
self._stopping = False # Make sure stop_launchers is not called 2x.
|
|
|
d = self.start_controller()
|
|
|
d.addCallback(self.start_engines)
|
|
|
d.addCallback(self.startup_message)
|
|
|
# If the controller or engines fail to start, stop everything
|
|
|
d.addErrback(self.stop_launchers)
|
|
|
return d
|
|
|
|
|
|
def startup_message(self, r=None):
|
|
|
log.msg("IPython cluster: started")
|
|
|
return r
|
|
|
|
|
|
def start_controller(self, r=None):
|
|
|
# log.msg("In start_controller")
|
|
|
config = self.master_config
|
|
|
d = self.controller_launcher.start(
|
|
|
cluster_dir=config.Global.cluster_dir
|
|
|
)
|
|
|
return d
|
|
|
|
|
|
def start_engines(self, r=None):
|
|
|
# log.msg("In start_engines")
|
|
|
config = self.master_config
|
|
|
d = self.engine_launcher.start(
|
|
|
config.Global.n,
|
|
|
cluster_dir=config.Global.cluster_dir
|
|
|
)
|
|
|
return d
|
|
|
|
|
|
def stop_controller(self, r=None):
|
|
|
# log.msg("In stop_controller")
|
|
|
if self.controller_launcher.running:
|
|
|
d = self.controller_launcher.stop()
|
|
|
d.addErrback(self.log_err)
|
|
|
return d
|
|
|
else:
|
|
|
return defer.succeed(None)
|
|
|
|
|
|
def stop_engines(self, r=None):
|
|
|
# log.msg("In stop_engines")
|
|
|
if self.engine_launcher.running:
|
|
|
d = self.engine_launcher.stop()
|
|
|
d.addErrback(self.log_err)
|
|
|
return d
|
|
|
else:
|
|
|
return defer.succeed(None)
|
|
|
|
|
|
def log_err(self, f):
|
|
|
log.msg(f.getTraceback())
|
|
|
return None
|
|
|
|
|
|
def stop_launchers(self, r=None):
|
|
|
if not self._stopping:
|
|
|
self._stopping = True
|
|
|
if isinstance(r, failure.Failure):
|
|
|
log.msg('Unexpected error in ipcluster:')
|
|
|
log.msg(r.getTraceback())
|
|
|
log.msg("IPython cluster: stopping")
|
|
|
# These return deferreds. We are not doing anything with them
|
|
|
# but we are holding refs to them as a reminder that they
|
|
|
# do return deferreds.
|
|
|
d1 = self.stop_engines()
|
|
|
d2 = self.stop_controller()
|
|
|
# Wait a few seconds to let things shut down.
|
|
|
reactor.callLater(4.0, reactor.stop)
|
|
|
|
|
|
def sigint_handler(self, signum, frame):
|
|
|
self.stop_launchers()
|
|
|
|
|
|
def start_logging(self):
|
|
|
# Remove old log files of the controller and engine
|
|
|
if self.master_config.Global.clean_logs:
|
|
|
log_dir = self.master_config.Global.log_dir
|
|
|
for f in os.listdir(log_dir):
|
|
|
if f.startswith('ipengine' + '-'):
|
|
|
if f.endswith('.log') or f.endswith('.out') or f.endswith('.err'):
|
|
|
os.remove(os.path.join(log_dir, f))
|
|
|
if f.startswith('ipcontroller' + '-'):
|
|
|
if f.endswith('.log') or f.endswith('.out') or f.endswith('.err'):
|
|
|
os.remove(os.path.join(log_dir, f))
|
|
|
# This will remote old log files for ipcluster itself
|
|
|
super(IPClusterApp, self).start_logging()
|
|
|
|
|
|
def start_app(self):
|
|
|
"""Start the application, depending on what subcommand is used."""
|
|
|
subcmd = self.master_config.Global.subcommand
|
|
|
if subcmd=='create' or subcmd=='list':
|
|
|
return
|
|
|
elif subcmd=='start':
|
|
|
self.start_app_start()
|
|
|
elif subcmd=='stop':
|
|
|
self.start_app_stop()
|
|
|
|
|
|
def start_app_start(self):
|
|
|
"""Start the app for the start subcommand."""
|
|
|
config = self.master_config
|
|
|
# First see if the cluster is already running
|
|
|
try:
|
|
|
pid = self.get_pid_from_file()
|
|
|
except PIDFileError:
|
|
|
pass
|
|
|
else:
|
|
|
self.log.critical(
|
|
|
'Cluster is already running with [pid=%s]. '
|
|
|
'use "ipcluster stop" to stop the cluster.' % pid
|
|
|
)
|
|
|
# Here I exit with a unusual exit status that other processes
|
|
|
# can watch for to learn how I existed.
|
|
|
self.exit(ALREADY_STARTED)
|
|
|
|
|
|
# Now log and daemonize
|
|
|
self.log.info(
|
|
|
'Starting ipcluster with [daemon=%r]' % config.Global.daemonize
|
|
|
)
|
|
|
# TODO: Get daemonize working on Windows or as a Windows Server.
|
|
|
if config.Global.daemonize:
|
|
|
if os.name=='posix':
|
|
|
daemonize()
|
|
|
|
|
|
# Now write the new pid file AFTER our new forked pid is active.
|
|
|
self.write_pid_file()
|
|
|
reactor.addSystemEventTrigger('during','shutdown', self.remove_pid_file)
|
|
|
reactor.run()
|
|
|
|
|
|
def start_app_stop(self):
|
|
|
"""Start the app for the stop subcommand."""
|
|
|
config = self.master_config
|
|
|
try:
|
|
|
pid = self.get_pid_from_file()
|
|
|
except PIDFileError:
|
|
|
self.log.critical(
|
|
|
'Problem reading pid file, cluster is probably not running.'
|
|
|
)
|
|
|
# Here I exit with a unusual exit status that other processes
|
|
|
# can watch for to learn how I existed.
|
|
|
self.exit(ALREADY_STOPPED)
|
|
|
else:
|
|
|
if os.name=='posix':
|
|
|
sig = config.Global.signal
|
|
|
self.log.info(
|
|
|
"Stopping cluster [pid=%r] with [signal=%r]" % (pid, sig)
|
|
|
)
|
|
|
os.kill(pid, sig)
|
|
|
elif os.name=='nt':
|
|
|
# As of right now, we don't support daemonize on Windows, so
|
|
|
# stop will not do anything. Minimally, it should clean up the
|
|
|
# old .pid files.
|
|
|
self.remove_pid_file()
|
|
|
|
|
|
|
|
|
def launch_new_instance():
|
|
|
"""Create and run the IPython cluster."""
|
|
|
app = IPClusterApp()
|
|
|
app.start()
|
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
|
launch_new_instance()
|
|
|
|
|
|
|
