cluster-func 0.0.6
Run a function many times on many processes / machines
License: MIT
Keywords: threading multiprocessing schedule scheduling batch process processing queue queueing cluster qsub pbs

# Cluster-func

Cluster-func is a command line tool that lets you scale embarassingly parallel 
solution written in Python with zero additional code.

## Install

With pip
```bash
pip install cluster-func
```

From source
```bash
git clone git://github.com/enewel101/cluster-func
cd cluster-func
python setup.py develop
```

## Why?

Too often I find myself writing multiprocessing and job-division / scheduling
boilerplate code, which, despite being conceptually straightforward, is tedious
and error-prone.

Sure, `xargs` is nice, assuming that you can conveniently get the shell to
spit out your argument values.  But that's often not the case, and 
you may want your arguments to be arbitrary python types instead of strings.

And, sure, Hadoop is nice too, assuming you've got a lot of time to burn
configuring it, and that your mappers and reducers don't use too much 
memory, and that you've loaded all your data in the hdfs, 
and that the results from your maps don't 
yield too much data and overwhelm the network.  Oh, and assuming
you enjoy writing boilerplate mapper and reducer code... wait, maybe hadoop
isn't so nice... (OK, OK, it does have its place!)

## Basic usage
Cluster-func is designed for situations where you need to run a single function
on many different arguments.  This kind embarassingly parallelizable problem
comes up pretty often.  At it's minimum, a solution involves defining
**A)** the function to be repeatedly called, and **B)** all of the different 
arguments on which it should be called.  

Cluster-func assumes that you have defined
these in the form of a *callable* and an *iterable* within a python script, and
then it handles the business of spreading the work across cores and machines.

The nice thing about this approach is that you unavoidably define these two
things when you write your code for a single process anyway.  So you'll get
multiprocessing and cluster processing basically for free!

The tool has two modes.  In **direct mode**, it runs your function on the cpus of
a single machine.  In **dispatch mode**, it breaks the work into subjobs that can
be run on separate machines, and optionally submits them to a job scheduler
using `qsub`.

### Direct mode
This command:
```bash
$ cluf my_script.py 
``` 
will look in `my_script.py` for a function called `target` and an iterable called
`args`, and it will run `target` on every single value yielded from `args`,
spawning as many workers as there are cpu cores available to your user on the
machine.  

To use other names for you target function or arguments iterable,
or to use a different number of worker processes, do this:
```bash
$ cluf my_script.py --target=my_func --args=my_iterable --processes=12	# short options -t, -a, -p
```

If `args` yields a tuple, its contents will be unpacked and interpreted as the
positional arguments for one invocation of the target function.  If you need
greater control, for example, to provide keyword arguments, then see 
**<a href="#arguments-iterable">Arguments iterable</a>**

`args` can also be a callable that *returns* an iterable (including a generator),
which is often more convenient.

So, using `cluf` in direct mode lets you multiprocess any script without pools
or queues or even importing multiprocessing.  But, if you really need to scale
up, then you'll want to use the dispatch mode.

### Dispatch mode
The main use of dispatch mode is to spread the work in a cluster that
uses the `qsub` command for job submission.  But you can still use `cluf` to 
spread work between machines don't use `qsub`.

Dispatch mode is implicitly invoked when you specify a number of compute nodes
to use (you can force the running mode using `--mode`, see **<a
href="#reference">Reference</a>**).
For example, this command:
```bash
$ cluf my_script.py --nodes=10 --queue	# short options -n, -q
```
will break the work into 10 subjobs, and submit them using qsub.  It does
this by writing small shell scripts, each of which is responsible for calling 
the target function on a subset of the arguments yielded by the iterable.

(To learn about setting PBS directives for your subjob scripts, see **<a
href="#pbs-options">PBS
options</a>** below.)

Because each subjob script is a valid shell script, you can manually run them
on separate machines in case they aren't part of a cluster that uses `qsub`.
Just leave off the `--queue` option, and the scripts will be created but not
enqueued.  This is also a good way to test run one of the subjobs
before submitting it.

To divide the work properly, it's important that your argument iterable yields
the same arguments in the same order on each machine.  If you can't or don't
want write your iterable that way, see **<a href="#how-work-is-divided">How work is divided</a>** for other options.

And that's the end of the basic usage guide.  This will cover you for the most 
basic usecases.  To learn about more cool features, read on!

## <a name="arguments-iterable">Arguments iterable</a>
The main usecase imagined is one where the arguments iterable yields either
single, bare arguments, or tuples of positional arguments.  Of course, the 
Python language provides a very
flexible way of calling functions, allowing you to mix positional arguments
and keyword arguments.  If you need that flexibility, then set up your iterator
to yield `cluster_func.Arguments` objects.  This class acts as a proxy, and 
will be used to call your target function in exactly the way you called the 
`Arguments` constructor.

Here is the `Arguments` class in action:
```python
>>> from cluster_func import Arguments
>>> my_args = Arguments(0, *[1,2,3], four=4, **{'five':5, 'six':6})
>>> my_args
Arguments(0, 1, 2, 3, four=4, five=5, six=6)
>>>
>>> # Your target function would be called the same way you called Arguments, e.g.:
>>> my_target(0, *[1,2,3], four=4, **{'five':5, 'six':6})
```

## <a name="how-work-is-divided">How work is divided</a>
By default, work is divided by assuming that the arguments iterator will yield
the same arguments in the same order during each subjob.  Each subjob can then
execute the target function only on those arguments assigned to it.  

For example, if there were 10 subjobs, subjob 7 would run arguments 7, 17, 27,
... etc.  For ease of explanation, we'll call that subset "bin 7".

If you open the subjob scripts in an editor, you'll find that they actually
call `cluf` itself in *direct mode*.  In other words, when you run
`cluf` in dispatch mode, it creates scripts that call `cluf` in direct mode.

These direct-mode invocations use the `--bin` option, which
is what instructs `cluf` to only run arguments that fall into that subjob's 
bin.  For example, this command:
```bash
$ cluf my_script --bin=0/3		# short option: -b
```
would run `cluf` in direct mode, but only execute iterations falling into bin 0 
out of 3, i.e., iterations 0, 3, 6, 9, etc.  (Bins are zero-indexed.)

You can use this to start subjobs manually if you like.
You can assign multiple bins to one subjob, For example, the option
`--bins=0-2,5/10` will assign bins 0, 1, 2, and 5 (out of a total of 10 bins).

### If your iterable is not stable
The default approach to binning assumes that the arguments iterable will 
yield the same arguments in the same order during execution of each subjob.
If you can't ensure that, then binning can be based on the arguments themselves,
instead of their order.

There are two alernative ways to handle binning: using *argument hashing* and
*direct assignment*.

### Argument hashing
By specifying the `--hash` option, you can instruct `cluf` to hash one or more
of the arguments to determine its bin.

For example, doing:
```bash
$ cluf example --nodes=12 --hash=0		# short options: -n and -x
```
will instruct `cluf` to hash the first argument of each iteration to decide
which bin the iteration belongs to.  

Before hashing, `cluf` calls `str` on the argument.
It's important that the argument selected for hashing has a
stable string representation that reflects its value. Using objects that don't
implement `__str__` won't work, both because their string representation
doesn't reflect their value, and because their memory address appears within
it, which will be different in each subjob.  However, for this purpose, a list
would be considered "hashable" (provided it's individual elements are).  On the
other hand, dict's and sets are not suitable, because they are unordered, so
their string representation is not stable.  One safe approach is to simply 
provide one argument that is a unique ID, and select it for hashing.

Ideally the argument selected for hashing should be unique throughout
iteration, since repeated values would be assigned to the same subjob.  But
occaisional repetitions won't imbalance load much.  To help achieve uniqueness
you can provide combinations of arguments to be hashed.  If you provided 
arguments as keyword arguments (using an Arguments object) you can select them 
too.

For example, this:
```bash
$ cluf example --nodes=12 --hash=0-2,5,my_kwarg	# short options: -n and -x
```
will hash arguents in positions 0,1,2, and 5, along with the keyword argument 
`my_kwarg`.  If any hashed arguments are missing in an 
iteration (becase, recall, invocations may use different numbers of arguments), 
they are simply ommitted when calculating the hash.

### Direct assignment
The final method for dividing work is to include an argument that explicitly 
specifies the
bin for each iteration.  To activate direct assignment, and to specify which
argument should be interpreted as the bin, use `--key` option:

```bash
$ cluf example --nodes=12 --key=2 		# short options: -n and -k
```
In the command above, the argument in position 2 (the third argument)
 will be interpreted as
the bin for each iteration.  You can also specify a keyword argument by name.

You should only use direct assignment if you really have to, because it's more
error prone, and it makes it more difficult to change the number of bins.
It also
introduces job division logic into your script which `cluf` was designed to
prevent.

## `cluf_options` and `.clufrc`
For more extensive configuration, you can include a dictionary named 
`cluf_options` in your target script to
control the behavior of `cluf`.  This can be more convenient than the command
line if you have to set a lot of options, and helps to document the options
you used.  You can also set options globally in a file at `~/.clufrc`. 

All options that can be set on the command line can be set within `cluf_options`
or `.clufrc`, plus a few extras.  There is one exception, however, which is
the option to force direct / dispatch mode (`--mode` or `-m`), which can only 
be set on the command line.

`cluf_options` should be a dictionary whose keys are the long option names,
and whose values are strings representing the option values as you would enter
them on the command line.  `.clufrc` should be a valid JSON object, with the
same key-value format.  See the following examples:

**Using `cluf_options`**
(in `my_script.py`)
```python
cluf_options = {
	'hash': (0,1,2,5),	# Or use '0-2,5'.  Applies to .clufrc as well
	'nodes': 12,
	'queue': True		# Flag options based on truthiness of value
}
```

**Using `.clufrc`**
(in `~/.clufrc`)
```json
{
	"hash": [0,1,2,5],
	"nodes": 12,
	"queue": true
}
```

In some cases you may want to use `cluf_options` to simply modify (e.g. add to) 
the options in `.clufrc`, rather than overriding them.  You can access the 
`.clufrc` options by importing them, using `from clusterf_func import RC_PARAMS`.

For the most part, any option that can be set on the command line can be set in
`cluf_options` and `.clufrc`, and vice versa, but there are a few options that
can *only* be set in `cluf_options` and `.clufrc`.  We cover those now along
with some options that are just less convenient to set on the command line.  
See **<a href="#reference">Reference</a>** for all available options.

### Environment variables
(Note: This option can be set on the command line but is somewhat less 
convenient.)
Let's suppose execution of your target script requires certain environment
variables to be set.  If you run `cluf` in *direct mode*, there's nothing to
think about—your script will inheret the current environment.

For example, if you did:
```bash
MYENV=1 cluf my_script.py
```
The value of `MYENV` would be seen by your script.

However, if you run `cluf` in dispatch mode, then the job scripts will
run in a different environment.  Use the `env` option to specify any 
environment variables that should be set when running the subjobs.  `env` should
be a dictionary within `cluf_options` or `.clufrc` that has variable names
as keys and values as, well, values:

(in `my_script.py`)
```python
cluf_options = {
	'env': {
		'MYENV': 'foo',
		'OTHER_ENV': 'bar',
	}
}
```

(in `~/.clufrc`)
```json
{
	"env": {
		"MYENV": "foo",
		"OTHER_ENV": "bar"
	}
}
```

If provided on the commandline, the contents of `env` will be pasted as-is
in front of the line that runs the subjob within the subjob script.  So the
equivalent `env` specification would be:
```bash
cluf my_script.py --nodes=4 --env='MYENV=foo OTHER_ENV=bar'
```

Each of these methods will cause the invocation of subjobs within subjob
scripts to look like this:
```bash
MYENV=foo OTHER_ENV=bar cluf my_script [...options]
```

Note that, regardless of where it is specified, if you want to provide a 
value for an environment variable that contains a space or other character
needing escaping, keep in mind that a round of interpretation by the 
command line, Python, or JSON will have occurred (depending on where it was set)

So, e.g., this won't work:
```bash
cluf my_script.py --nodes=4 --env='MYENV=foo bar'
```
Nor will, this:
```bash
cluf my_script.py --nodes=4 --env=MYENV=foo\ bar
```
But This will work:
```bash
cluf my_script.py --nodes=4 --env='MYENV=foo\ bar'
```
And so will this:
```bash
cluf my_script.py --nodes=4 --env=MYENV=foo\\\ bar
```

### <a name="pbs-options">PBS options</a>
(This option cannot be set on the command line.)
Portable Batch System options control how the cluster scheduler schedules your
job, and allows you to request specific compute resources and specify the
amount of time that your job should run.
In general PBS options should be set using a dictionary of key-value pairs
using the option name as the key.  For example, to request that your
subjobs run on compute nodes with at least 4 cpu cores and 2 gpus, you can do
something like this:

(in `my_script.py`)
```python
cluf_options = {
	'pbs_options': {'ppn': 4, 'gpus': 2}
}
```

(in `~/.clufrc`)
```json
{
	"pbs_options": {"ppn": 4, "gpus": 2}
}
```

The `ppn` (processes per node) option should usually match the number of worker
processes set by the  `processes` option (whether set on the command line, 
`cluf_options`, or in `.clufrc`).  So if `ppn` isn't explicitly set in your
PBS options, but `processes` is set, then it will default to the value of 
`processes`.  You can still set a different value for ppn, if e.g.  your target 
function itself spawns proceses.

There are three special options whose names differ from the 
PBS option names slightly, and these options are set to defaults unless 
specifically overridden.

- `'name'`: the name of your subjobs as they appear in the job scheduler.
	This is also used to name your subjob scripts (by appending `'.pbs'`).
	The default is the format string `'{target}-{subjob}-{num_subjobs}'`,
	with the fields being interpolated by the target module's name, the 
	subjob number, and the total number of subjobs respectively.
	If you override
	this you can also use those format fields, and you must at least use
	the `{subjob}` field to ensure that each of your subjobs gets a 
	unique name (otherwise the subjob scripts will overwrite one another).
- `'stdout'`: the path at which to place stdout captured from your subjobs,
	relative to the `jobs_dir` if set (if not set then it is relative to the 
	current working directory).
	This defaults to the subjob name plus `'.stdout'`
	As for the `'name'` option, 
	if you override this, make sure that the paths for subjobs
	are unique by using the `{node_num}` field somewhere.
- `'stderr'`: similar to `'stdout'`.  Defaults to the subjob name plus 
	`'.stderr'`

The combinations of PBS options that are available and/or required depends on 
the setup of your cluster.  Usually a system is configured with smart defaults
so that you can queue simple jobs without setting any PBS options.


### Additional statements
(This option can be set on the command line but may be more convenient to set
within your target script.)
You can also specify any additional statements that you want to appear in your
subjob scripts.  This gives you more flexibility than simply setting environment 
variables.   You can include an external script, whose statements get merged
into the job script before the line that runs subjob using
 `prepend_script`.  Include an external script *after* the line that runs
the subjob using the `append_script` option.  

If you just want to add a statement or two, it maybe more convenient to 
put them in your target script (or `.clufrc`).
The options `prepend_statements` and `append_statements` can be used provide a 
list of shell statements that get inserted right before or after the line
that actually runs the subjob within subjob scripts.  Each element of the list
should be a valid shell statement which will appear on its own line when merged
into the jobscripts.  The options aren't available on the command line.

# <a name="reference">Reference</a>

The `cluf` command has lots of options, which can be specified in three
different places:

 1. as command line arguments, or
 2. within your target module, inside a dictionary called `cluf_options`, or
 3. in the `~/.clufrc` file, in the form of a JSON object

These locations are in decreasing order of precedence, i.e. the command line
overrides all other options, and the `.clufrc` file doesn't override the others.

Options given in the `cluf_options` dictionary in the target module or in the
`.clufrc` JSON object should be identified by the long option name, without
the leading '--'.


### All `cluf` options

<pre>
usage: cluf [-h] [-j JOBS_DIR] [-t TARGET] [-a ARGS] [-q] [-p PROCESSES]
            [-b BINS] [-e ENV] [-P PREPEND_SCRIPT] [-A APPEND_SCRIPT]
            [-m {dispatch,direct}] [-x HASH | -k KEY]
            [-n NODES | -i ITERATIONS]
            target_module

Run a function many times on multiple processors and machines.

positional arguments:
  target_module         path to the python module that contains the target
                        function to be run many times.

optional arguments:
  -h, --help            show this help message and exit
  -j JOBS_DIR, --jobs-dir JOBS_DIR
                        Specify a directory in which to store job scripts and
                        the files generated from the stdout and stdin during
                        job execution. This directory will be made if it
                        doesn't exist. This option only takes effect in
                        dispatch mode.
  -t TARGET, --target TARGET
                        Alternate name for the target callable to be invoked
                        repeatedly. Default is "target".
  -a ARGS, --args ARGS  Alternate name for the arguments iterable. Default is
                        "args".
  -q, --queue           Enqueue the generated scripts using qsub. This option
                        only takes effect in dispatch mode.
  -p PROCESSES, --processes PROCESSES
                        Number of processors to use.
  -b BINS, --bins BINS  Optionally specify a portion of the work to be done.
                        Should take the form "x/y" meaning "do the x-th
                        section out of y total sections. For example, "0/2"
                        means divide the work into two halves, and do the
                        first (0th) half. Note that x and y should be
                        integers, and x should be from 0 to y-1 inclusive.
                        This option only takes effect in direct mode.
  -e ENV, --env ENV     Provide environment variables that should be set when
                        running sub-jobs. This is for use in dispatch mode,
                        since job scripts will run in a different environment.
                        In direct mode, the environment is inherited. The
                        value of this option should be an enquoted string of
                        space-separated key=value pairs. For example: $ cluf
                        my_script -n 12 -e 'FOO=bar BAZ="fizz bang"'will set
                        FOO equal to "bar" and BAZ equal to "fizz bang". This
                        option only takes effect in dispatch mode.
  -P PREPEND_SCRIPT, --prepend-script PREPEND_SCRIPT
                        Path to a script whose contents should be included at
                        the beginning of subjob scripts, being executed before
                        running the subjob. You can include multiple comma-
                        separated paths. This option only takes effect in
                        dispatch mode.
  -A APPEND_SCRIPT, --append-script APPEND_SCRIPT
                        Path to a script whose contents should be included at
                        the end of subjob scripts, being executed after the
                        subjob completes. You can include multiple comma-
                        separated paths. This option only takes effect in
                        dispatch mode.
  -m {dispatch,direct}, --mode {dispatch,direct}
                        Explicitly set the mode of operation. Can be set to
                        "direct" or "dispatch". In direct mode the job is run,
                        whereas in dispatch mode a script for the job(s) is
                        created and optionally enqueued. Setting either -n or
                        -i implicitly sets the mode of operation to
                        "dispatch", unless specified otherwise.
  -x HASH, --hash HASH  Specify an argument or set of arguments to be used to
                        determine which bin an iteration belons in. These
                        arguments should have a stable string representation
                        (i.e. no unordered containers or memory addresses) and
                        should be unique over the argumetns iterable. This
                        should only be set if automatic binning won't work,
                        i.e. if your argument iterable is not stable.
  -k KEY, --key KEY     Integer specifying the positional argument to use as
                        the bin for each iteration. That key argument should
                        always take on a value that is an integer between 0
                        and num_bins-1. This should only be used if you really
                        need to control binning. Prefer to rely on automatic
                        binning (if your iterable is stable), or use the
                        -xoption, which is more flexible and less error-prone.
  -n NODES, --nodes NODES
                        Number of compute nodes. This option causes the
                        command to operate in dispatch mode, unless the mode
                        is explicitly set
  -i ITERATIONS, --iterations ITERATIONS
                        Approximate number of iterations per compute node.
                        This option causes the command to operate in dispatch
                        mode, unless the mode is explicitly set. Note that
                        using this instead of --nodes (-n) can lead to delay
                        because the total number of iterations has to be
                        counted to determine the number of compute nodes
                        needed.
</pre>