Requirements

• Programs qsub, qstat, and qdel are required to submit, monitor, and delete queue-jobs.

• Your func(task) must be part of an importable python module.

• Your tasks and their results must be able to serialize using pickle.

• Both worker-nodes and process-node can read and write from and to a common path in the file-system.

Features

• Respects fair-share, i.e. slots are only occupied when the compute is done.

• No spawning of additional threads. Neither on the process-node, nor on the worker-nodes.

• No need for databases or web-servers.

• Queue-jobs with error-state 'E' can be deleted, and resubmitted until your predefined upper limit is reached.

• Can bundle tasks on worker-nodes to avoid start-up-overhead with many small tasks.

Alternatives

When you do not share resources with other users, and when you have some administrative power you might want to use one of these:

• Dask has a job_queue which also supports other flavors such as PBS, SLURM.

• pyABC.sge has a sge.map() very much like our one.

• ipyparallel

Inner workings

• map() makes a work_dir because the mapping and reducing takes place in the file-system. You can set work_dir manually to make sure both worker-nodes and process-node can reach it.

• map() serializes your tasks using pickle into separate files in work_dir/{ichunk:09d}.pkl.

• map() reads all environment-variables in its process.

• map() creates the worker-node-script in work_dir/worker_node_script.py. It contains and exports the process’ environment-variables into the batch-job’s context. It reads the chunk of tasks in work_dir/{ichunk:09d}.pkl, imports and runs your func(task), and finally writes the result back to work_dir/{ichunk:09d}.pkl.out.

• map() submits queue-jobs. The stdout and stderr of the tasks are written to work_dir/{ichunk:09d}.pkl.o and work_dir/{ichunk:09d}.pkl.e respectively. By default, shutil.which("python") is used to process the worker-node-script.

• When all queue-jobs are submitted, map() monitors their progress. In case a queue-job runs into an error-state ('E' by default) the job wFill be deleted and resubmitted until a maximum number of resubmissions is reached.

• When no more queue-jobs are running or pending, map() will reduce the results from work_dir/{ichunk:09d}.pkl.out.

• In case of non zero stderr in any task, a missing result, or on the user’s request, the work_dir will be kept for inspection. Otherwise its removed.

Testing

py.test -s .

from queue_map_reduce import dummy_queue

dummy_queue.init_queue_state(
    path="tests/resources/dummy_queue_state.json"
)