paralleljobs.py

   1 #!/usr/bin/env python
   2 """ A simple tool to run jobs from a database in parallel."""
   3
   4 __author__ = "Stefan Huber"
   5 __copyright__ = "Copyright 2013"
   6
   7 __version__ = "1.0"
   8 __license__ = "LGPL3"
   9
  10
  11 import sys, getopt, os
  12 import sqlite3
  13 import subprocess
  14
  15 verbose = False
  16
  17 def printStatusInfo(conn):
  18     c = conn.cursor()
  19
  20     c.execute("SELECT count(id) FROM jobs;")
  21     nototal, = c.fetchone()
  22
  23     c.execute("SELECT count(id) FROM jobs WHERE done=1;")
  24     nodone, = c.fetchone()
  25
  26     c.execute("SELECT sum(workloadestm) FROM jobs WHERE done=1;")
  27     wldone, = c.fetchone()
  28     if wldone == None:
  29         wldone = 0.0
  30
  31     c.execute("SELECT sum(workloadestm) FROM jobs;")
  32     wltotal, = c.fetchone()
  33
  34     c.close()
  35
  36     print(nototal, nodone, wldone, wltotal)
  37     perdone = 100.0*float(nodone)/float(nototal)
  38     perwl = 100.0*float(wldone)/float(wltotal)
  39
  40     print("%d (%.1f%%) of %d jobs done. %.1f%% of the workload finished." % \
  41             (nodone, perdone, nototal, perwl))
  42
  43 def createPropertiesTable(conn, propdef):
  44     conn.execute("BEGIN EXCLUSIVE")
  45
  46     c = conn.cursor()
  47     c.execute("SELECT count(name) FROM sqlite_master WHERE name='properties';")
  48     if c.fetchone() == (0,):
  49         print("Creating properties table.")
  50         sqlstmt = "CREATE TABLE properties (\
  51                 jobid INTEGER PRIMARY KEY,\
  52                 %s, \
  53                 FOREIGN KEY (jobid) REFERENCES jobs (id));" % (propdef,)
  54         c.execute(sqlstmt)
  55     c.close()
  56     conn.commit()
  57
  58 def runCmd(cmd):
  59     proc = subprocess.Popen(cmd, \
  60             stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
  61     out, err = proc.communicate()
  62     exitcode = proc.wait()
  63
  64     if verbose:
  65         print(out, err)
  66
  67     return exitcode, out, err
  68
  69 def processJob(conn, jobid):
  70     print("Process job %d" % (jobid))
  71
  72     c = conn.cursor()
  73     c.execute("SELECT cmd FROM jobs WHERE id=?", (jobid,))
  74     cmd, = c.fetchone()
  75
  76     ec, out, err = runCmd(cmd)
  77     c.execute("UPDATE jobs SET exitcode=?, done=1 WHERE id=?;", (ec, jobid))
  78
  79     propstr = []
  80     for l in out.splitlines():
  81         if l.startswith("DB-PROPERTIES:"):
  82             propstr += [l[14:]]
  83     for l in err.splitlines():
  84         if l.startswith("DB-PROPERTIES:"):
  85             propstr += [l[14:]]
  86
  87     prop = {}
  88     for ps in propstr:
  89         p = eval(ps)
  90         for k, v in p.iteritems():
  91             prop[k] = v
  92
  93     if len(prop) > 0:
  94         collist = ", ".join([str(k) for k in prop.keys()])
  95         collist = "jobid, " + collist
  96
  97         vallist = ", ".join(["?" for k in prop.keys()])
  98         vallist = "?, " + vallist
  99
 100         c = conn.cursor()
 101         sqlstmt = "INSERT INTO properties (%s) VALUES (%s);" % (collist,vallist)
 102         c.execute(sqlstmt, [jobid] + list(prop.values()))
 103
 104     c.close()
 105     conn.commit()
 106
 107 def insertJobs(conn, cmds):
 108     conn.execute("BEGIN EXCLUSIVE")
 109     conn.executemany("INSERT INTO jobs (cmd) VALUES (?);", cmds)
 110     conn.commit()
 111
 112 def createSchema(conn):
 113
 114     c = conn.cursor()
 115     c.execute("BEGIN EXCLUSIVE")
 116
 117     # Create table, if necessary
 118     c.execute("SELECT count(name) FROM sqlite_master WHERE name='jobs';")
 119     if c.fetchone() == (0,):
 120         print("Creating jobs table.")
 121         conn.execute("CREATE TABLE jobs ( \
 122                         id INTEGER PRIMARY KEY AUTOINCREMENT, \
 123                         cmd STRING NOT NULL, \
 124                         started BOOL DEFAULT (0) NOT NULL, \
 125                         done BOOL DEFAULT (0) NOT NULL, \
 126                         exitcode INTEGER, \
 127                         workloadestm REAL DEFAULT (1) NOT NULL)")
 128     c.close()
 129     conn.commit()
 130
 131 def getNextJobId(conn):
 132
 133     c = conn.cursor()
 134     c.execute("BEGIN EXCLUSIVE")
 135     c.execute("SELECT id FROM jobs WHERE NOT started=1 LIMIT 1;")
 136
 137     r = c.fetchone()
 138     if r == None:
 139         return None
 140
 141     jobid, = r
 142     conn.execute("UPDATE jobs SET started=1 WHERE id=?;", (jobid,))
 143
 144     c.close()
 145     conn.commit()
 146
 147     return jobid
 148
 149
 150
 151 def usage():
 152     """Print usage text of this program"""
 153
 154     print("""
 155 Take the jobs defined in jobs table of the given database and process one job
 156 after the other. Multiple instances may be launched against the same database.
 157
 158 Usage:
 159   {0} [OPTIONS] [COMMANDS] -d database
 160   {0} -h
 161
 162 COMMANDS:
 163   -c cmdfn        add jobs from the file with list of commands
 164   -h              print this text
 165   -s              print status information
 166   -w              work on the database
 167
 168 OPTIONS:
 169   -d database     the database to process
 170   -p cols-def     create properties table with SQL column spec
 171   -v              print output of the job's command
 172
 173 Commands may be combined in one call of {0}.
 174
 175 A list of jobs may be importet line-by-line from a file using the -c option.
 176 Every job may output to stdout or stderr a string of the form
 177     DB-PROPERTIES: {{ "key": "value", "key2": 1.23, "key3": True }}
 178 It is assumed that a table 'properties' exists with the columns jobid, key,
 179 key2, and key3. The corresponding values are inserted into this table. Using
 180 the option -p such a properties table can be created by giving a list of
 181 column definitions in SQL style.
 182
 183 The jobs table also contains a 'workloadestm' column that is used when
 184 estimating the finished workload so far. The entries default to 1 and may be
 185 set externally.
 186
 187 Examples:
 188   # create cmds.sh with jobs
 189   echo "ulimit -v 2000000 -t 1200; ./isprime 65535" > cmds.sh
 190   echo "ulimit -v 2000000 -t 1200; ./isprime 65537" >> cmds.sh
 191   # create an initial database, but do not work
 192   {0} -d jobs.db -c cmds.sh \\
 193       -p 'number INTEGER, time REAL, mem INTEGER'
 194   # launch two workers
 195   {0} -d jobs.db -w &
 196   {0} -d jobs.db -w &
 197   # print status info
 198   {0} -d jobs.db -s
 199 """.format(sys.argv[0]))
 200
 201
 202 if __name__ == "__main__":
 203
 204     nojobs = 1
 205     dbfn = None
 206     cmdfn = None
 207     propdef = None
 208     work = False
 209     status = False
 210
 211     try:
 212         opts, args = getopt.getopt(sys.argv[1:], "hd:c:p:wsv")
 213
 214         for opt, arg in opts:
 215             if opt == "-h":
 216                 usage()
 217                 sys.exit(os.EX_OK)
 218             elif opt == "-d":
 219                 dbfn = arg
 220             elif opt == "-c":
 221                 cmdfn = arg
 222             elif opt == "-p":
 223                 propdef = arg
 224             elif opt == "-w":
 225                 work = True
 226             elif opt == "-s":
 227                 status = True
 228             elif opt == "-v":
 229                 verbose = True
 230             else:
 231                 print("Unknown option '", opt, "'.")
 232
 233     except getopt.GetoptError as e:
 234         print("Error parsing arguments:", e)
 235         usage()
 236         sys.exit(os.EX_USAGE)
 237
 238     if dbfn == None:
 239         print("No database given.")
 240         sys.exit(os.EX_USAGE)
 241
 242     conn = sqlite3.connect(dbfn)
 243     createSchema(conn)
 244
 245     if status:
 246         printStatusInfo(conn)
 247
 248     if propdef != None:
 249         createPropertiesTable(conn, propdef)
 250
 251     if cmdfn != None:
 252         print("Adding jobs...")
 253         cmds = open(cmdfn).readlines()
 254         cmds = [(c.strip(),) for c in cmds]
 255         insertJobs(conn, cmds)
 256
 257     if work:
 258         while True:
 259             jobid = getNextJobId(conn)
 260             if jobid == None:
 261                 print("All jobs have been started.")
 262                 break
 263             processJob(conn, jobid)
 264
 265
 266     conn.close()
 267