jobp = json.loads(args.script_parameters)
os.environ['JOB_UUID'] = 'zzzzz-8i9sb-1234567890abcde'
os.environ['TASK_UUID'] = 'zzzzz-ot0gb-1234567890abcde'
- os.environ['CRUNCH_SRC'] = '/tmp/crunche-src'
+ os.environ['CRUNCH_SRC'] = '/tmp/crunch-src'
if 'TASK_KEEPMOUNT' not in os.environ:
os.environ['TASK_KEEPMOUNT'] = '/keep'
-links = []
-
def sub_tmpdir(v):
return os.path.join(arvados.current_task().tmpdir, 'tmpdir')
gr[m] = []
gr[m].append(match.group(0))
+class EvaluationError(Exception):
+ pass
+
+# Return the name of variable ('var') that will take on each value in 'items'
+# when performing an inner substitution
def var_items(p, c, key):
+ if key not in c:
+ raise EvaluationError("'%s' was expected in 'p' but is missing" % key)
+
if "var" in c:
- # Var specifies
+ if not isinstance(c["var"], basestring):
+ raise EvaluationError("Value of 'var' must be a string")
+ # Var specifies the variable name for inner parameter substitution
return (c["var"], get_items(p, c[key]))
else:
+ # The component function ('key') value is a list, so return the list
+ # directly with no parameter selected.
if isinstance(c[key], list):
return (None, get_items(p, c[key]))
- m = re.match("^\$\((.*)\)$", c[key])
- if m and m.group(1) in p:
- return (m.group(1), get_items(p, c[key]))
+ elif isinstance(c[key], basestring):
+ # check if c[key] is a string that looks like a parameter
+ m = re.match("^\$\((.*)\)$", c[key])
+ if m and m.group(1) in p:
+ return (m.group(1), get_items(p, c[key]))
+ else:
+ # backwards compatible, foreach specifies bare parameter name to use
+ return (c[key], get_items(p, p[c[key]]))
else:
- # backwards compatible, foreach specifies bare parameter name to use
- return (c[key], get_items(p, p[c[key]]))
+ raise EvaluationError("Value of '%s' must be a string or list" % key)
+# "p" is the parameter scope, "c" is the item to be expanded.
+# If "c" is a dict, apply function expansion.
+# If "c" is a list, recursively expand each item and return a new list.
+# If "c" is a string, apply parameter substitution
def expand_item(p, c):
if isinstance(c, dict):
if "foreach" in c and "command" in c:
+ # Expand a command template for each item in the specified user
+ # parameter
var, items = var_items(p, c, "foreach")
+ if var is None:
+ raise EvaluationError("Must specify 'var' in foreach")
r = []
for i in items:
params = copy.copy(p)
params[var] = i
r.append(expand_item(params, c["command"]))
return r
- if "list" in c and "index" in c and "command" in c:
+ elif "list" in c and "index" in c and "command" in c:
+ # extract a single item from a list
var, items = var_items(p, c, "list")
+ if var is None:
+ raise EvaluationError("Must specify 'var' in list")
params = copy.copy(p)
params[var] = items[int(c["index"])]
return expand_item(params, c["command"])
- if "regex" in c:
+ elif "regex" in c:
pattern = re.compile(c["regex"])
if "filter" in c:
- var, items = var_items(p, c, "filter")
+ # filter list so that it only includes items that match a
+ # regular expression
+ _, items = var_items(p, c, "filter")
return [i for i in items if pattern.match(i)]
elif "group" in c:
- var, items = var_items(p, c, "group")
+ # generate a list of lists, where items are grouped on common
+ # subexpression match
+ _, items = var_items(p, c, "group")
groups = {}
for i in items:
match = pattern.match(i)
add_to_group(groups, match)
return [groups[k] for k in groups]
elif "extract" in c:
- var, items = var_items(p, c, "extract")
+ # generate a list of lists, where items are split by
+ # subexpression match
+ _, items = var_items(p, c, "extract")
r = []
for i in items:
match = pattern.match(i)
if match:
r.append(list(match.groups()))
return r
- if "batch" in c and "size" in c:
- var, items = var_items(p, c, "batch")
+ elif "batch" in c and "size" in c:
+ # generate a list of lists, where items are split into a batch size
+ _, items = var_items(p, c, "batch")
sz = int(c["size"])
r = []
for j in xrange(0, len(items), sz):
r.append(items[j:j+sz])
return r
+ raise EvaluationError("Missing valid list context function")
elif isinstance(c, list):
return [expand_item(p, arg) for arg in c]
elif isinstance(c, basestring):
return expand_item(p, p[m.group(1)])
else:
return subst.do_substitution(p, c)
+ else:
+ raise EvaluationError("expand_item() unexpected parameter type %s" % type(c))
- return []
-
+# Evaluate in a list context
+# "p" is the parameter scope, "value" will be evaluated
+# if "value" is a list after expansion, return that
+# if "value" is a path to a directory, return a list consisting of each entry in the directory
+# if "value" is a path to a file, return a list consisting of each line of the file
def get_items(p, value):
value = expand_item(p, value)
if isinstance(value, list):
with open(value) as f:
items = [line.rstrip("\r\n") for line in f]
return items
- raise Exception("get_items did not yield a list")
+ raise EvaluationError("get_items did not yield a list")
stdoutname = None
stdoutfile = None
stdinname = None
stdinfile = None
+# Construct the cross product of all values of each variable listed in fvars
def recursive_foreach(params, fvars):
var = fvars[0]
fvars = fvars[1:]
if not args.dry_run:
stdoutfile = open(stdoutname, "wb")
+ if "task.env" in taskp:
+ env = copy.copy(os.environ)
+ for k,v in taskp["task.env"].items():
+ env[k] = subst.do_substitution(taskp, v)
+ else:
+ env = None
+
logger.info("{}{}{}".format(' | '.join([' '.join(c) for c in cmd]), (" < " + stdinname) if stdinname is not None else "", (" > " + stdoutname) if stdoutname is not None else ""))
if args.dry_run:
logger.error(pprint.pformat(taskp))
sys.exit(1)
+# rcode holds the return codes produced by each subprocess
+rcode = {}
try:
subprocesses = []
close_streams = []
# this is an intermediate command in the pipeline, so its stdout should go to a pipe
next_stdout = subprocess.PIPE
- sp = subprocess.Popen(cmd[i], shell=False, stdin=next_stdin, stdout=next_stdout)
+ sp = subprocess.Popen(cmd[i], shell=False, stdin=next_stdin, stdout=next_stdout, env=env)
# Need to close the FDs on our side so that subcommands will get SIGPIPE if the
# consuming process ends prematurely.
active = 1
pids = set([s.pid for s in subprocesses])
- rcode = {}
while len(pids) > 0:
- (pid, status) = os.wait()
- pids.discard(pid)
- if not taskp.get("task.ignore_rcode"):
- rcode[pid] = (status >> 8)
+ try:
+ (pid, status) = os.wait()
+ except OSError as e:
+ if e.errno == errno.EINTR:
+ pass
+ else:
+ raise
else:
- rcode[pid] = 0
+ pids.discard(pid)
+ if not taskp.get("task.ignore_rcode"):
+ rcode[pid] = (status >> 8)
+ else:
+ rcode[pid] = 0
if sig.sig is not None:
logger.critical("terminating on signal %s" % sig.sig)
signal.signal(signal.SIGTERM, signal.SIG_DFL)
signal.signal(signal.SIGQUIT, signal.SIG_DFL)
-for l in links:
- os.unlink(l)
-
logger.info("the following output files will be saved to keep:")
-subprocess.call(["find", ".", "-type", "f", "-printf", "run-command: %12.12s %h/%f\\n"], stdout=sys.stderr)
+subprocess.call(["find", "-L", ".", "-type", "f", "-printf", "run-command: %12.12s %h/%f\\n"], stdout=sys.stderr, cwd=outdir)
logger.info("start writing output to keep")
-if "task.vwd" in taskp:
- if "task.foreach" in jobp:
- # This is a subtask, so don't merge with the original collection, that will happen at the end
- outcollection = vwd.checkin(subst.do_substitution(taskp, taskp["task.vwd"]), outdir, merge=False).manifest_text()
- else:
- # Just a single task, so do merge with the original collection
- outcollection = vwd.checkin(subst.do_substitution(taskp, taskp["task.vwd"]), outdir, merge=True).manifest_text()
-else:
- outcollection = robust_put.upload(outdir, logger)
+if "task.vwd" in taskp and "task.foreach" in jobp:
+ for root, dirs, files in os.walk(outdir):
+ for f in files:
+ s = os.lstat(os.path.join(root, f))
+ if stat.S_ISLNK(s.st_mode):
+ os.unlink(os.path.join(root, f))
+
+(outcollection, checkin_error) = vwd.checkin(outdir)
-success = reduce(lambda x, y: x & (y == 0), [True]+rcode.values())
+# Success if we ran any subprocess, and they all exited 0.
+success = rcode and all(status == 0 for status in rcode.itervalues()) and not checkin_error
api.job_tasks().update(uuid=arvados.current_task()['uuid'],
body={
- 'output': outcollection,
+ 'output': outcollection.manifest_text(),
'success': success,
'progress':1.0
}).execute()
-sys.exit(rcode)
+sys.exit(0 if success else 1)
projects / arvados.git / blobdiff