madcow: A simple tool for parallelizing bash commands across many Longhorn servers.

madcow.py

#!/usr/bin/env python

import os, errno, os.path
import argparse

JOB_TEMPLATE = """
#!/bin/sh
#-----------------------------------------------------------------------------
#$ -V
#$ -cwd
#$ -N %(name)s%(i)02d
#$ -j y
#$ -o %(base_dir)s/%(name)s/logs/command.%(i)02d.log
#$ -pe 1way 8
#$ -q %(queue)s
#$ -P data
#$ -l h_rt=%(hours)d:00:00
#$ -A %(project)s
#-----------------------------------------------------------------------------
parallel < %(base_dir)s/%(name)s/commands/command.%(i)02d
"""

queue = "long"
project = "isogram"
num_hosts = 20
base_dir = "/home/01813/roller/parallel"
hours = 24
name = "redo"

def argmin(w):
    return min(enumerate(w), key=lambda x: x[1])[0]

def mkdir_p(path):
    try:
        os.makedirs(path)
    except OSError as exc: # Python >2.5
        if exc.errno == errno.EEXIST and os.path.isdir(path):
            pass
        else: raise


parser = argparse.ArgumentParser(
            description='MadCow: a system for distributing bash commands across Longhorn')
parser.add_argument("--input", "-i", default="-", metavar="FILE",
                    help="The input of commands to run, one per line.")
parser.add_argument("--hours", "-t", default=24, type=int, metavar="HOURS",
                    help="The maximum number of hours to allow one computer to use.")
parser.add_argument("--name", "-n", metavar="NAME",
                    help="The name of the job.")
parser.add_argument("--num-hosts", "-k", metavar="#", type=int, default=num_hosts,
                    help="The number of hosts/jobs to create.")
parser.add_argument("--base-dir", "-d", metavar="DIR", default=base_dir,
                    help="The path to output job files, logs, etc.")
parser.add_argument("--project", "-p", metavar="PROJECT", default=project,
                    help="The TACC project to dock compute hours from.")
parser.add_argument("--queue", "-q", metavar="QUEUE", default=queue,
                    help="The queue to place the jobs in.")

args = parser.parse_args()
queue = args.queue
project = args.project
num_hosts = args.num_hosts
base_dir = args.base_dir
hours = args.hours
name = args.name


if args.input == "-":
    commandfile = sys.stdin
else:
    commandfile = open(args.input)


mkdir_p(base_dir)
mkdir_p(os.path.join(base_dir, name))
for d in ("commands", "jobs", "logs"):
    mkdir_p(os.path.join(base_dir, name, d))



# split up the computations
weights = [0] * num_hosts
bins = [list() for k in range(num_hosts)]

lst = commandfile.read().split("\n")

for inpt in reversed(lst):
    inpt = inpt.strip()
    if not inpt: continue
    try:
        size, cmd = inpt.split("\t")
        size = int(size)
    except ValueError:
        size, cmd = 1, inpt

    bin = argmin(weights)
    weights[bin] += size
    bins[bin].append(cmd)


for i, bin in enumerate(bins):
    f = open("%s/%s/commands/command.%02d" % (base_dir, name, i), "w")
    for command in bin:
        f.write(command + "\n")
    f.close()

    f = open("%s/%s/jobs/job.%02d" % (base_dir, name, i), "w")
    template = JOB_TEMPLATE % {
        'base_dir': base_dir,
        'project': project,
        'hours': hours,
        'name': name,
        'i': i,
        'queue': queue,
    }
    f.write(template)
    f.close()