#!/usr/bin/env python
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
import logging
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')
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.append(expand_item(params, c["command"]))
return r
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")
elif "regex" in c:
pattern = re.compile(c["regex"])
if "filter" in c:
+ # 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:
+ # generate a list of lists, where items are grouped on common
+ # subexpression match
_, items = var_items(p, c, "group")
groups = {}
for i in items:
add_to_group(groups, match)
return [groups[k] for k in groups]
elif "extract" in c:
+ # generate a list of lists, where items are split by
+ # subexpression match
_, items = var_items(p, c, "extract")
r = []
for i in items:
r.append(list(match.groups()))
return r
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 = []
return expand_item(p, p[m.group(1)])
else:
return subst.do_substitution(p, c)
-
- raise EvaluationError("expand_item() unexpected parameter type %s" % (type(c))
+ else:
+ raise EvaluationError("expand_item() unexpected parameter type %s" % type(c))
# Evaluate in a list context
# "p" is the parameter scope, "value" will be evaluated
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 if no non-zero return codes
-success = not any([status != 0 for status in 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