2 # -*- mode: perl; perl-indent-level: 2; indent-tabs-mode: nil; -*-
6 crunch-job: Execute job steps, save snapshots as requested, collate output.
10 Obtain job details from Arvados, run tasks on compute nodes (typically
11 invoked by scheduler on controller):
13 crunch-job --job x-y-z --git-dir /path/to/repo/.git
15 Obtain job details from command line, run tasks on local machine
16 (typically invoked by application or developer on VM):
18 crunch-job --job '{"script_version":"/path/to/working/tree","script":"scriptname",...}'
20 crunch-job --job '{"repository":"https://github.com/curoverse/arvados.git","script_version":"master","script":"scriptname",...}'
28 If the job is already locked, steal the lock and run it anyway.
32 Path to a .git directory (or a git URL) where the commit given in the
33 job's C<script_version> attribute is to be found. If this is I<not>
34 given, the job's C<repository> attribute will be used.
38 Arvados API authorization token to use during the course of the job.
42 Do not clear per-job/task temporary directories during initial job
43 setup. This can speed up development and debugging when running jobs
48 UUID of the job to run, or a JSON-encoded job resource without a
49 UUID. If the latter is given, a new job object will be created.
53 =head1 RUNNING JOBS LOCALLY
55 crunch-job's log messages appear on stderr along with the job tasks'
56 stderr streams. The log is saved in Keep at each checkpoint and when
59 If the job succeeds, the job's output locator is printed on stdout.
61 While the job is running, the following signals are accepted:
65 =item control-C, SIGINT, SIGQUIT
67 Save a checkpoint, terminate any job tasks that are running, and stop.
71 Save a checkpoint and continue.
75 Refresh node allocation (i.e., check whether any nodes have been added
76 or unallocated) and attributes of the Job record that should affect
77 behavior (e.g., cancel job if cancelled_at becomes non-nil).
85 use POSIX ':sys_wait_h';
86 use POSIX qw(strftime);
87 use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
91 use Digest::MD5 qw(md5_hex);
97 use File::Path qw( make_path remove_tree );
99 use constant TASK_TEMPFAIL => 111;
100 use constant EX_TEMPFAIL => 75;
101 use constant EX_RETRY_UNLOCKED => 93;
103 $ENV{"TMPDIR"} ||= "/tmp";
104 unless (defined $ENV{"CRUNCH_TMP"}) {
105 $ENV{"CRUNCH_TMP"} = $ENV{"TMPDIR"} . "/crunch-job";
106 if ($ENV{"USER"} ne "crunch" && $< != 0) {
107 # use a tmp dir unique for my uid
108 $ENV{"CRUNCH_TMP"} .= "-$<";
112 # Create the tmp directory if it does not exist
113 if ( ! -d $ENV{"CRUNCH_TMP"} ) {
114 make_path $ENV{"CRUNCH_TMP"} or die "Failed to create temporary working directory: " . $ENV{"CRUNCH_TMP"};
117 $ENV{"JOB_WORK"} = $ENV{"CRUNCH_TMP"} . "/work";
118 $ENV{"CRUNCH_INSTALL"} = "$ENV{CRUNCH_TMP}/opt";
119 $ENV{"CRUNCH_WORK"} = $ENV{"JOB_WORK"}; # deprecated
120 mkdir ($ENV{"JOB_WORK"});
129 my $docker_bin = "docker.io";
130 GetOptions('force-unlock' => \$force_unlock,
131 'git-dir=s' => \$git_dir,
132 'job=s' => \$jobspec,
133 'job-api-token=s' => \$job_api_token,
134 'no-clear-tmp' => \$no_clear_tmp,
135 'resume-stash=s' => \$resume_stash,
136 'docker-bin=s' => \$docker_bin,
139 if (defined $job_api_token) {
140 $ENV{ARVADOS_API_TOKEN} = $job_api_token;
143 my $have_slurm = exists $ENV{SLURM_JOB_ID} && exists $ENV{SLURM_NODELIST};
148 $main::ENV{CRUNCH_DEBUG} = 1;
152 $main::ENV{CRUNCH_DEBUG} = 0;
155 my $arv = Arvados->new('apiVersion' => 'v1');
164 if ($jobspec =~ /^[-a-z\d]+$/)
166 # $jobspec is an Arvados UUID, not a JSON job specification
167 $Job = api_call("jobs/get", uuid => $jobspec);
172 $local_job = JSON::decode_json($jobspec);
176 # Make sure our workers (our slurm nodes, localhost, or whatever) are
177 # at least able to run basic commands: they aren't down or severely
180 if (($Job || $local_job)->{docker_image_locator}) {
181 $cmd = [$docker_bin, 'ps', '-q'];
183 Log(undef, "Sanity check is `@$cmd`");
184 srun(["srun", "--nodes=\Q$ENV{SLURM_NNODES}\E", "--ntasks-per-node=1"],
188 Log(undef, "Sanity check failed: ".exit_status_s($?));
191 Log(undef, "Sanity check OK");
194 my $User = api_call("users/current");
197 if (!$force_unlock) {
198 # Claim this job, and make sure nobody else does
199 eval { api_call("jobs/lock", uuid => $Job->{uuid}); };
201 Log(undef, "Error while locking job, exiting ".EX_TEMPFAIL);
210 map { croak ("No $_ specified") unless $local_job->{$_} }
211 qw(script script_version script_parameters);
214 $local_job->{'is_locked_by_uuid'} = $User->{'uuid'};
215 $local_job->{'started_at'} = gmtime;
216 $local_job->{'state'} = 'Running';
218 $Job = api_call("jobs/create", job => $local_job);
220 $job_id = $Job->{'uuid'};
222 my $keep_logfile = $job_id . '.log.txt';
223 log_writer_start($keep_logfile);
225 $Job->{'runtime_constraints'} ||= {};
226 $Job->{'runtime_constraints'}->{'max_tasks_per_node'} ||= 0;
227 my $max_ncpus = $Job->{'runtime_constraints'}->{'max_tasks_per_node'};
229 my $gem_versions = `gem list --quiet arvados-cli 2>/dev/null`;
231 $gem_versions =~ s/^arvados-cli \(/ with arvados-cli Gem version(s) /;
232 chomp($gem_versions);
233 chop($gem_versions); # Closing parentheses
238 "running from " . ((-e $0) ? realpath($0) : "stdin") . $gem_versions);
240 Log (undef, "check slurm allocation");
243 # Should use $ENV{SLURM_TASKS_PER_NODE} instead of sinfo? (eg. "4(x3),2,4(x2)")
247 my $localcpus = 0 + `grep -cw ^processor /proc/cpuinfo` || 1;
248 push @sinfo, "$localcpus localhost";
250 if (exists $ENV{SLURM_NODELIST})
252 push @sinfo, `sinfo -h --format='%c %N' --nodes=\Q$ENV{SLURM_NODELIST}\E`;
256 my ($ncpus, $slurm_nodelist) = split;
257 $ncpus = $max_ncpus if $max_ncpus && $ncpus > $max_ncpus;
260 while ($slurm_nodelist =~ s/^([^\[,]+?(\[.*?\])?)(,|$)//)
263 if ($nodelist =~ /\[((\d+)(-(\d+))?(,(\d+)(-(\d+))?)*)\]/)
266 foreach (split (",", $ranges))
279 push @nodelist, map {
281 $n =~ s/\[[-,\d]+\]/$_/;
288 push @nodelist, $nodelist;
291 foreach my $nodename (@nodelist)
293 Log (undef, "node $nodename - $ncpus slots");
294 my $node = { name => $nodename,
296 # The number of consecutive times a task has been dispatched
297 # to this node and failed.
299 # The number of consecutive times that SLURM has reported
300 # a node failure since the last successful task.
302 # Don't dispatch work to this node until this time
303 # (in seconds since the epoch) has passed.
305 foreach my $cpu (1..$ncpus)
307 push @slot, { node => $node,
311 push @node, @nodelist;
316 # Ensure that we get one jobstep running on each allocated node before
317 # we start overloading nodes with concurrent steps
319 @slot = sort { $a->{cpu} <=> $b->{cpu} } @slot;
322 $Job->update_attributes(
323 'tasks_summary' => { 'failed' => 0,
328 Log (undef, "start");
329 $SIG{'INT'} = sub { $main::please_freeze = 1; };
330 $SIG{'QUIT'} = sub { $main::please_freeze = 1; };
331 $SIG{'TERM'} = \&croak;
332 $SIG{'TSTP'} = sub { $main::please_freeze = 1; };
333 $SIG{'ALRM'} = sub { $main::please_info = 1; };
334 $SIG{'CONT'} = sub { $main::please_continue = 1; };
335 $SIG{'HUP'} = sub { $main::please_refresh = 1; };
337 $main::please_freeze = 0;
338 $main::please_info = 0;
339 $main::please_continue = 0;
340 $main::please_refresh = 0;
341 my $jobsteps_must_output_keys = 0; # becomes 1 when any task outputs a key
343 grep { $ENV{$1} = $2 if /^(NOCACHE.*?)=(.*)/ } split ("\n", $$Job{knobs});
344 $ENV{"CRUNCH_JOB_UUID"} = $job_id;
345 $ENV{"JOB_UUID"} = $job_id;
348 my @jobstep_todo = ();
349 my @jobstep_done = ();
350 my @jobstep_tomerge = ();
351 my $jobstep_tomerge_level = 0;
352 my $squeue_checked = 0;
353 my $latest_refresh = scalar time;
357 if (defined $Job->{thawedfromkey})
359 thaw ($Job->{thawedfromkey});
363 my $first_task = api_call("job_tasks/create", job_task => {
364 'job_uuid' => $Job->{'uuid'},
369 push @jobstep, { 'level' => 0,
371 'arvados_task' => $first_task,
373 push @jobstep_todo, 0;
379 must_lock_now("$ENV{CRUNCH_TMP}/.lock", "a job is already running here.");
382 my $build_script = handle_readall(\*DATA);
383 my $nodelist = join(",", @node);
384 my $git_tar_count = 0;
386 if (!defined $no_clear_tmp) {
387 # Clean out crunch_tmp/work, crunch_tmp/opt, crunch_tmp/src*
388 Log (undef, "Clean work dirs");
390 my $cleanpid = fork();
393 # Find FUSE mounts under $CRUNCH_TMP and unmount them.
394 # Then clean up work directories.
395 # TODO: When #5036 is done and widely deployed, we can limit mount's
396 # -t option to simply fuse.keep.
397 srun (["srun", "--nodelist=$nodelist", "-D", $ENV{'TMPDIR'}],
398 ['bash', '-ec', '-o', 'pipefail', 'mount -t fuse,fuse.keep | awk "(index(\$3, \"$CRUNCH_TMP\") == 1){print \$3}" | xargs -r -n 1 fusermount -u -z; sleep 1; rm -rf $JOB_WORK $CRUNCH_INSTALL $CRUNCH_TMP/task $CRUNCH_TMP/src* $CRUNCH_TMP/*.cid']);
403 last if $cleanpid == waitpid (-1, WNOHANG);
404 freeze_if_want_freeze ($cleanpid);
405 select (undef, undef, undef, 0.1);
408 Log(undef, "Clean work dirs: exit ".exit_status_s($?));
409 exit(EX_RETRY_UNLOCKED);
413 # If this job requires a Docker image, install that.
414 my ($docker_locator, $docker_stream, $docker_hash, $docker_limitmem);
415 if ($docker_locator = $Job->{docker_image_locator}) {
416 ($docker_stream, $docker_hash) = find_docker_image($docker_locator);
419 croak("No Docker image hash found from locator $docker_locator");
421 $docker_stream =~ s/^\.//;
422 my $docker_install_script = qq{
423 if ! $docker_bin images -q --no-trunc --all | grep -qxF \Q$docker_hash\E; then
424 arv-get \Q$docker_locator$docker_stream/$docker_hash.tar\E | $docker_bin load
427 my $docker_pid = fork();
428 if ($docker_pid == 0)
430 srun (["srun", "--nodelist=" . join(',', @node)],
431 ["/bin/sh", "-ec", $docker_install_script]);
436 last if $docker_pid == waitpid (-1, WNOHANG);
437 freeze_if_want_freeze ($docker_pid);
438 select (undef, undef, undef, 0.1);
442 croak("Installing Docker image from $docker_locator exited "
446 # Determine whether this version of Docker supports memory+swap limits.
447 srun(["srun", "--nodelist=" . $node[0]],
448 ["/bin/sh", "-ec", "$docker_bin run --help | grep -qe --memory-swap="],
450 $docker_limitmem = ($? == 0);
452 if ($Job->{arvados_sdk_version}) {
453 # The job also specifies an Arvados SDK version. Add the SDKs to the
454 # tar file for the build script to install.
455 Log(undef, sprintf("Packing Arvados SDK version %s for installation",
456 $Job->{arvados_sdk_version}));
457 add_git_archive("git", "--git-dir=$git_dir", "archive",
458 "--prefix=.arvados.sdk/",
459 $Job->{arvados_sdk_version}, "sdk");
463 if (!defined $git_dir && $Job->{'script_version'} =~ m{^/}) {
464 # If script_version looks like an absolute path, *and* the --git-dir
465 # argument was not given -- which implies we were not invoked by
466 # crunch-dispatch -- we will use the given path as a working
467 # directory instead of resolving script_version to a git commit (or
468 # doing anything else with git).
469 $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{'script_version'};
470 $ENV{"CRUNCH_SRC"} = $Job->{'script_version'};
473 # Resolve the given script_version to a git commit sha1. Also, if
474 # the repository is remote, clone it into our local filesystem: this
475 # ensures "git archive" will work, and is necessary to reliably
476 # resolve a symbolic script_version like "master^".
477 $ENV{"CRUNCH_SRC"} = "$ENV{CRUNCH_TMP}/src";
479 Log (undef, "Looking for version ".$Job->{script_version}." from repository ".$Job->{repository});
481 $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{script_version};
483 # If we're running under crunch-dispatch, it will have already
484 # pulled the appropriate source tree into its own repository, and
485 # given us that repo's path as $git_dir.
487 # If we're running a "local" job, we might have to fetch content
488 # from a remote repository.
490 # (Currently crunch-dispatch gives a local path with --git-dir, but
491 # we might as well accept URLs there too in case it changes its
493 my $repo = $git_dir || $Job->{'repository'};
495 # Repository can be remote or local. If remote, we'll need to fetch it
496 # to a local dir before doing `git log` et al.
499 if ($repo =~ m{://|^[^/]*:}) {
500 # $repo is a git url we can clone, like git:// or https:// or
501 # file:/// or [user@]host:repo.git. Note "user/name@host:foo" is
502 # not recognized here because distinguishing that from a local
503 # path is too fragile. If you really need something strange here,
504 # use the ssh:// form.
505 $repo_location = 'remote';
506 } elsif ($repo =~ m{^\.*/}) {
507 # $repo is a local path to a git index. We'll also resolve ../foo
508 # to ../foo/.git if the latter is a directory. To help
509 # disambiguate local paths from named hosted repositories, this
510 # form must be given as ./ or ../ if it's a relative path.
511 if (-d "$repo/.git") {
512 $repo = "$repo/.git";
514 $repo_location = 'local';
516 # $repo is none of the above. It must be the name of a hosted
518 my $arv_repo_list = api_call("repositories/list",
519 'filters' => [['name','=',$repo]]);
520 my @repos_found = @{$arv_repo_list->{'items'}};
521 my $n_found = $arv_repo_list->{'serverResponse'}->{'items_available'};
523 Log(undef, "Repository '$repo' -> "
524 . join(", ", map { $_->{'uuid'} } @repos_found));
527 croak("Error: Found $n_found repositories with name '$repo'.");
529 $repo = $repos_found[0]->{'fetch_url'};
530 $repo_location = 'remote';
532 Log(undef, "Using $repo_location repository '$repo'");
533 $ENV{"CRUNCH_SRC_URL"} = $repo;
535 # Resolve given script_version (we'll call that $treeish here) to a
536 # commit sha1 ($commit).
537 my $treeish = $Job->{'script_version'};
539 if ($repo_location eq 'remote') {
540 # We minimize excess object-fetching by re-using the same bare
541 # repository in CRUNCH_TMP/.git for multiple crunch-jobs -- we
542 # just keep adding remotes to it as needed.
543 my $local_repo = $ENV{'CRUNCH_TMP'}."/.git";
544 my $gitcmd = "git --git-dir=\Q$local_repo\E";
546 # Set up our local repo for caching remote objects, making
548 if (!-d $local_repo) {
549 make_path($local_repo) or croak("Error: could not create $local_repo");
551 # This works (exits 0 and doesn't delete fetched objects) even
552 # if $local_repo is already initialized:
553 `$gitcmd init --bare`;
555 croak("Error: $gitcmd init --bare exited ".exit_status_s($?));
558 # If $treeish looks like a hash (or abbrev hash) we look it up in
559 # our local cache first, since that's cheaper. (We don't want to
560 # do that with tags/branches though -- those change over time, so
561 # they should always be resolved by the remote repo.)
562 if ($treeish =~ /^[0-9a-f]{7,40}$/s) {
563 # Hide stderr because it's normal for this to fail:
564 my $sha1 = `$gitcmd rev-list -n1 ''\Q$treeish\E 2>/dev/null`;
566 # Careful not to resolve a branch named abcdeff to commit 1234567:
567 $sha1 =~ /^$treeish/ &&
568 $sha1 =~ /^([0-9a-f]{40})$/s) {
570 Log(undef, "Commit $commit already present in $local_repo");
574 if (!defined $commit) {
575 # If $treeish isn't just a hash or abbrev hash, or isn't here
576 # yet, we need to fetch the remote to resolve it correctly.
578 # First, remove all local heads. This prevents a name that does
579 # not exist on the remote from resolving to (or colliding with)
580 # a previously fetched branch or tag (possibly from a different
582 remove_tree("$local_repo/refs/heads", {keep_root => 1});
584 Log(undef, "Fetching objects from $repo to $local_repo");
585 `$gitcmd fetch --no-progress --tags ''\Q$repo\E \Q+refs/heads/*:refs/heads/*\E`;
587 croak("Error: `$gitcmd fetch` exited ".exit_status_s($?));
591 # Now that the data is all here, we will use our local repo for
592 # the rest of our git activities.
596 my $gitcmd = "git --git-dir=\Q$repo\E";
597 my $sha1 = `$gitcmd rev-list -n1 ''\Q$treeish\E`;
598 unless ($? == 0 && $sha1 =~ /^([0-9a-f]{40})$/) {
599 croak("`$gitcmd rev-list` exited "
601 .", '$treeish' not found, giving up");
604 Log(undef, "Version $treeish is commit $commit");
606 if ($commit ne $Job->{'script_version'}) {
607 # Record the real commit id in the database, frozentokey, logs,
608 # etc. -- instead of an abbreviation or a branch name which can
609 # become ambiguous or point to a different commit in the future.
610 if (!$Job->update_attributes('script_version' => $commit)) {
611 croak("Error: failed to update job's script_version attribute");
615 $ENV{"CRUNCH_SRC_COMMIT"} = $commit;
616 add_git_archive("$gitcmd archive ''\Q$commit\E");
619 my $git_archive = combined_git_archive();
620 if (!defined $git_archive) {
621 Log(undef, "Skip install phase (no git archive)");
623 Log(undef, "Warning: This probably means workers have no source tree!");
628 my $install_script_tries_left = 3;
629 for (my $attempts = 0; $attempts < 3; $attempts++) {
630 Log(undef, "Run install script on all workers");
632 my @srunargs = ("srun",
633 "--nodelist=$nodelist",
634 "-D", $ENV{'TMPDIR'}, "--job-name=$job_id");
635 my @execargs = ("sh", "-c",
636 "mkdir -p $ENV{CRUNCH_INSTALL} && cd $ENV{CRUNCH_TMP} && perl -");
638 $ENV{"CRUNCH_GIT_ARCHIVE_HASH"} = md5_hex($git_archive);
639 my ($install_stderr_r, $install_stderr_w);
640 pipe $install_stderr_r, $install_stderr_w or croak("pipe() failed: $!");
641 set_nonblocking($install_stderr_r);
642 my $installpid = fork();
643 if ($installpid == 0)
645 close($install_stderr_r);
646 fcntl($install_stderr_w, F_SETFL, 0) or croak($!); # no close-on-exec
647 open(STDOUT, ">&", $install_stderr_w);
648 open(STDERR, ">&", $install_stderr_w);
649 srun (\@srunargs, \@execargs, {}, $build_script . $git_archive);
652 close($install_stderr_w);
653 # Tell freeze_if_want_freeze how to kill the child, otherwise the
654 # "waitpid(installpid)" loop won't get interrupted by a freeze:
655 $proc{$installpid} = {};
657 # Track whether anything appears on stderr other than slurm errors
658 # ("srun: ...") and the "starting: ..." message printed by the
659 # srun subroutine itself:
660 my $stderr_anything_from_script = 0;
661 my $match_our_own_errors = '^(srun: error: |starting: \[)';
662 while ($installpid != waitpid(-1, WNOHANG)) {
663 freeze_if_want_freeze ($installpid);
664 # Wait up to 0.1 seconds for something to appear on stderr, then
665 # do a non-blocking read.
666 my $bits = fhbits($install_stderr_r);
667 select ($bits, undef, $bits, 0.1);
668 if (0 < sysread ($install_stderr_r, $stderr_buf, 8192, length($stderr_buf)))
670 while ($stderr_buf =~ /^(.*?)\n/) {
672 substr $stderr_buf, 0, 1+length($line), "";
673 Log(undef, "stderr $line");
674 if ($line !~ /$match_our_own_errors/) {
675 $stderr_anything_from_script = 1;
680 delete $proc{$installpid};
681 $install_exited = $?;
682 close($install_stderr_r);
683 if (length($stderr_buf) > 0) {
684 if ($stderr_buf !~ /$match_our_own_errors/) {
685 $stderr_anything_from_script = 1;
687 Log(undef, "stderr $stderr_buf")
690 Log (undef, "Install script exited ".exit_status_s($install_exited));
691 last if $install_exited == 0 || $main::please_freeze;
692 # If the install script fails but doesn't print an error message,
693 # the next thing anyone is likely to do is just run it again in
694 # case it was a transient problem like "slurm communication fails
695 # because the network isn't reliable enough". So we'll just do
696 # that ourselves (up to 3 attempts in total). OTOH, if there is an
697 # error message, the problem is more likely to have a real fix and
698 # we should fail the job so the fixing process can start, instead
699 # of doing 2 more attempts.
700 last if $stderr_anything_from_script;
703 foreach my $tar_filename (map { tar_filename_n($_); } (1..$git_tar_count)) {
704 unlink($tar_filename);
707 if ($install_exited != 0) {
712 foreach (qw (script script_version script_parameters runtime_constraints))
716 (ref($Job->{$_}) ? JSON::encode_json($Job->{$_}) : $Job->{$_}));
718 foreach (split (/\n/, $Job->{knobs}))
720 Log (undef, "knob " . $_);
725 $main::success = undef;
731 my $thisround_succeeded = 0;
732 my $thisround_failed = 0;
733 my $thisround_failed_multiple = 0;
734 my $working_slot_count = scalar(@slot);
736 @jobstep_todo = sort { $jobstep[$a]->{level} <=> $jobstep[$b]->{level}
737 or $a <=> $b } @jobstep_todo;
738 my $level = $jobstep[$jobstep_todo[0]]->{level};
740 my $initial_tasks_this_level = 0;
741 foreach my $id (@jobstep_todo) {
742 $initial_tasks_this_level++ if ($jobstep[$id]->{level} == $level);
745 # If the number of tasks scheduled at this level #T is smaller than the number
746 # of slots available #S, only use the first #T slots, or the first slot on
747 # each node, whichever number is greater.
749 # When we dispatch tasks later, we'll allocate whole-node resources like RAM
750 # based on these numbers. Using fewer slots makes more resources available
751 # to each individual task, which should normally be a better strategy when
752 # there are fewer of them running with less parallelism.
754 # Note that this calculation is not redone if the initial tasks at
755 # this level queue more tasks at the same level. This may harm
756 # overall task throughput for that level.
758 if ($initial_tasks_this_level < @node) {
759 @freeslot = (0..$#node);
760 } elsif ($initial_tasks_this_level < @slot) {
761 @freeslot = (0..$initial_tasks_this_level - 1);
763 @freeslot = (0..$#slot);
765 my $round_num_freeslots = scalar(@freeslot);
767 my %round_max_slots = ();
768 for (my $ii = $#freeslot; $ii >= 0; $ii--) {
769 my $this_slot = $slot[$freeslot[$ii]];
770 my $node_name = $this_slot->{node}->{name};
771 $round_max_slots{$node_name} ||= $this_slot->{cpu};
772 last if (scalar(keys(%round_max_slots)) >= @node);
775 Log(undef, "start level $level with $round_num_freeslots slots");
778 my $progress_is_dirty = 1;
779 my $progress_stats_updated = 0;
781 update_progress_stats();
785 for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++)
787 my $id = $jobstep_todo[$todo_ptr];
788 my $Jobstep = $jobstep[$id];
789 if ($Jobstep->{level} != $level)
794 pipe $reader{$id}, "writer" or croak("pipe() failed: $!");
795 set_nonblocking($reader{$id});
797 my $childslot = $freeslot[0];
798 my $childnode = $slot[$childslot]->{node};
799 my $childslotname = join (".",
800 $slot[$childslot]->{node}->{name},
801 $slot[$childslot]->{cpu});
803 my $childpid = fork();
806 $SIG{'INT'} = 'DEFAULT';
807 $SIG{'QUIT'} = 'DEFAULT';
808 $SIG{'TERM'} = 'DEFAULT';
810 foreach (values (%reader))
814 fcntl ("writer", F_SETFL, 0) or croak ($!); # no close-on-exec
815 open(STDOUT,">&writer");
816 open(STDERR,">&writer");
821 delete $ENV{"GNUPGHOME"};
822 $ENV{"TASK_UUID"} = $Jobstep->{'arvados_task'}->{'uuid'};
823 $ENV{"TASK_QSEQUENCE"} = $id;
824 $ENV{"TASK_SEQUENCE"} = $level;
825 $ENV{"JOB_SCRIPT"} = $Job->{script};
826 while (my ($param, $value) = each %{$Job->{script_parameters}}) {
827 $param =~ tr/a-z/A-Z/;
828 $ENV{"JOB_PARAMETER_$param"} = $value;
830 $ENV{"TASK_SLOT_NODE"} = $slot[$childslot]->{node}->{name};
831 $ENV{"TASK_SLOT_NUMBER"} = $slot[$childslot]->{cpu};
832 $ENV{"TASK_WORK"} = $ENV{"CRUNCH_TMP"}."/task/$childslotname";
833 $ENV{"HOME"} = $ENV{"TASK_WORK"};
834 $ENV{"TASK_KEEPMOUNT"} = $ENV{"TASK_WORK"}.".keep";
835 $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
836 $ENV{"CRUNCH_NODE_SLOTS"} = $round_max_slots{$ENV{TASK_SLOT_NODE}};
837 $ENV{"PATH"} = $ENV{"CRUNCH_INSTALL"} . "/bin:" . $ENV{"PATH"};
843 "--nodelist=".$childnode->{name},
844 qw(-n1 -c1 -N1 -D), $ENV{'TMPDIR'},
845 "--job-name=$job_id.$id.$$",
848 "if [ -e $ENV{TASK_WORK} ]; then rm -rf $ENV{TASK_WORK}; fi; "
849 ."mkdir -p $ENV{CRUNCH_TMP} $ENV{JOB_WORK} $ENV{TASK_WORK} $ENV{TASK_KEEPMOUNT} "
850 ."&& cd $ENV{CRUNCH_TMP} "
851 # These environment variables get used explicitly later in
852 # $command. No tool is expected to read these values directly.
853 .q{&& MEM=$(awk '($1 == "MemTotal:"){print $2}' </proc/meminfo) }
854 .q{&& SWAP=$(awk '($1 == "SwapTotal:"){print $2}' </proc/meminfo) }
855 ."&& MEMLIMIT=\$(( (\$MEM * 95) / ($ENV{CRUNCH_NODE_SLOTS} * 100) )) "
856 ."&& let SWAPLIMIT=\$MEMLIMIT+\$SWAP ";
857 $command .= "&& exec arv-mount --by-id --allow-other $ENV{TASK_KEEPMOUNT} --exec ";
860 my $cidfile = "$ENV{CRUNCH_TMP}/$Jobstep->{arvados_task}->{uuid}-$Jobstep->{failures}.cid";
861 $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -cgroup-parent=docker -cgroup-cid=$cidfile -poll=10000 ";
862 $command .= "$docker_bin run --rm=true --attach=stdout --attach=stderr --attach=stdin -i --user=crunch --cidfile=$cidfile --sig-proxy ";
863 # We only set memory limits if Docker lets us limit both memory and swap.
864 # Memory limits alone have been supported longer, but subprocesses tend
865 # to get SIGKILL if they exceed that without any swap limit set.
866 # See #5642 for additional background.
867 if ($docker_limitmem) {
868 $command .= "--memory=\${MEMLIMIT}k --memory-swap=\${SWAPLIMIT}k ";
871 # The source tree and $destdir directory (which we have
872 # installed on the worker host) are available in the container,
873 # under the same path.
874 $command .= "--volume=\Q$ENV{CRUNCH_SRC}:$ENV{CRUNCH_SRC}:ro\E ";
875 $command .= "--volume=\Q$ENV{CRUNCH_INSTALL}:$ENV{CRUNCH_INSTALL}:ro\E ";
877 # Currently, we make arv-mount's mount point appear at /keep
878 # inside the container (instead of using the same path as the
879 # host like we do with CRUNCH_SRC and CRUNCH_INSTALL). However,
880 # crunch scripts and utilities must not rely on this. They must
881 # use $TASK_KEEPMOUNT.
882 $command .= "--volume=\Q$ENV{TASK_KEEPMOUNT}:/keep:ro\E ";
883 $ENV{TASK_KEEPMOUNT} = "/keep";
885 # TASK_WORK is almost exactly like a docker data volume: it
886 # starts out empty, is writable, and persists until no
887 # containers use it any more. We don't use --volumes-from to
888 # share it with other containers: it is only accessible to this
889 # task, and it goes away when this task stops.
891 # However, a docker data volume is writable only by root unless
892 # the mount point already happens to exist in the container with
893 # different permissions. Therefore, we [1] assume /tmp already
894 # exists in the image and is writable by the crunch user; [2]
895 # avoid putting TASK_WORK inside CRUNCH_TMP (which won't be
896 # writable if they are created by docker while setting up the
897 # other --volumes); and [3] create $TASK_WORK inside the
898 # container using $build_script.
899 $command .= "--volume=/tmp ";
900 $ENV{"TASK_WORK"} = "/tmp/crunch-job-task-work/$childslotname";
901 $ENV{"HOME"} = $ENV{"TASK_WORK"};
902 $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
904 # TODO: Share a single JOB_WORK volume across all task
905 # containers on a given worker node, and delete it when the job
906 # ends (and, in case that doesn't work, when the next job
909 # For now, use the same approach as TASK_WORK above.
910 $ENV{"JOB_WORK"} = "/tmp/crunch-job-work";
912 while (my ($env_key, $env_val) = each %ENV)
914 if ($env_key =~ /^(ARVADOS|CRUNCH|JOB|TASK)_/) {
915 $command .= "--env=\Q$env_key=$env_val\E ";
918 $command .= "--env=\QHOME=$ENV{HOME}\E ";
919 $command .= "\Q$docker_hash\E ";
920 $command .= "stdbuf --output=0 --error=0 ";
921 $command .= "perl - $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
924 $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -poll=10000 ";
925 $command .= "stdbuf --output=0 --error=0 ";
926 $command .= "perl - $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
929 my @execargs = ('bash', '-c', $command);
930 srun (\@srunargs, \@execargs, undef, $build_script);
931 # exec() failed, we assume nothing happened.
932 die "srun() failed on build script\n";
935 if (!defined $childpid)
942 $proc{$childpid} = { jobstep => $id,
945 jobstepname => "$job_id.$id.$childpid",
947 croak ("assert failed: \$slot[$childslot]->{'pid'} exists") if exists $slot[$childslot]->{pid};
948 $slot[$childslot]->{pid} = $childpid;
950 Log ($id, "job_task ".$Jobstep->{'arvados_task'}->{'uuid'});
951 Log ($id, "child $childpid started on $childslotname");
952 $Jobstep->{starttime} = time;
953 $Jobstep->{node} = $childnode->{name};
954 $Jobstep->{slotindex} = $childslot;
955 delete $Jobstep->{stderr};
956 delete $Jobstep->{finishtime};
957 delete $Jobstep->{tempfail};
959 $Jobstep->{'arvados_task'}->{started_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{starttime});
960 $Jobstep->{'arvados_task'}->save;
962 splice @jobstep_todo, $todo_ptr, 1;
965 $progress_is_dirty = 1;
969 ($round_num_freeslots > @freeslot && $todo_ptr+1 > $#jobstep_todo))
971 last THISROUND if $main::please_freeze || defined($main::success);
972 if ($main::please_info)
974 $main::please_info = 0;
976 create_output_collection();
978 update_progress_stats();
983 if (!$gotsome || ($latest_refresh + 2 < scalar time))
985 check_refresh_wanted();
987 update_progress_stats();
988 select (undef, undef, undef, 0.1);
990 elsif (time - $progress_stats_updated >= 30 || $progress_is_dirty)
992 update_progress_stats();
994 $working_slot_count = scalar(grep { $_->{node}->{fail_count} == 0 &&
995 $_->{node}->{hold_count} < 4 } @slot);
996 if (($thisround_failed_multiple >= 8 && $thisround_succeeded == 0) ||
997 ($thisround_failed_multiple >= 16 && $thisround_failed_multiple > $thisround_succeeded))
999 my $message = "Repeated failure rate too high ($thisround_failed_multiple/"
1000 .($thisround_failed+$thisround_succeeded)
1001 .") -- giving up on this round";
1002 Log (undef, $message);
1006 # move slots from freeslot to holdslot (or back to freeslot) if necessary
1007 for (my $i=$#freeslot; $i>=0; $i--) {
1008 if ($slot[$freeslot[$i]]->{node}->{hold_until} > scalar time) {
1009 push @holdslot, (splice @freeslot, $i, 1);
1012 for (my $i=$#holdslot; $i>=0; $i--) {
1013 if ($slot[$holdslot[$i]]->{node}->{hold_until} <= scalar time) {
1014 push @freeslot, (splice @holdslot, $i, 1);
1018 # give up if no nodes are succeeding
1019 if ($working_slot_count < 1) {
1020 Log(undef, "Every node has failed -- giving up");
1027 push @freeslot, splice @holdslot;
1028 map { $slot[$freeslot[$_]]->{node}->{losing_streak} = 0 } (0..$#freeslot);
1031 Log (undef, "wait for last ".(scalar keys %proc)." children to finish");
1034 if ($main::please_continue) {
1035 $main::please_continue = 0;
1038 $main::please_info = 0, freeze(), create_output_collection(), save_meta(1) if $main::please_info;
1040 if (!reapchildren())
1042 check_refresh_wanted();
1044 update_progress_stats();
1045 select (undef, undef, undef, 0.1);
1046 killem (keys %proc) if $main::please_freeze;
1050 update_progress_stats();
1051 freeze_if_want_freeze();
1054 if (!defined $main::success)
1056 if (!@jobstep_todo) {
1058 } elsif ($working_slot_count < 1) {
1059 save_output_collection();
1061 exit(EX_RETRY_UNLOCKED);
1062 } elsif ($thisround_succeeded == 0 &&
1063 ($thisround_failed == 0 || $thisround_failed > 4)) {
1064 my $message = "stop because $thisround_failed tasks failed and none succeeded";
1065 Log (undef, $message);
1070 goto ONELEVEL if !defined $main::success;
1073 release_allocation();
1075 my $collated_output = save_output_collection();
1076 Log (undef, "finish");
1081 if ($collated_output && $main::success) {
1082 $final_state = 'Complete';
1084 $final_state = 'Failed';
1086 $Job->update_attributes('state' => $final_state);
1088 exit (($final_state eq 'Complete') ? 0 : 1);
1092 sub update_progress_stats
1094 $progress_stats_updated = time;
1095 return if !$progress_is_dirty;
1096 my ($todo, $done, $running) = (scalar @jobstep_todo,
1097 scalar @jobstep_done,
1098 scalar keys(%proc));
1099 $Job->{'tasks_summary'} ||= {};
1100 $Job->{'tasks_summary'}->{'todo'} = $todo;
1101 $Job->{'tasks_summary'}->{'done'} = $done;
1102 $Job->{'tasks_summary'}->{'running'} = $running;
1103 $Job->update_attributes('tasks_summary' => $Job->{'tasks_summary'});
1104 Log (undef, "status: $done done, $running running, $todo todo");
1105 $progress_is_dirty = 0;
1112 my $pid = waitpid (-1, WNOHANG);
1113 return 0 if $pid <= 0;
1115 my $whatslot = ($slot[$proc{$pid}->{slot}]->{node}->{name}
1117 . $slot[$proc{$pid}->{slot}]->{cpu});
1118 my $jobstepid = $proc{$pid}->{jobstep};
1119 my $elapsed = time - $proc{$pid}->{time};
1120 my $Jobstep = $jobstep[$jobstepid];
1122 my $childstatus = $?;
1123 my $exitvalue = $childstatus >> 8;
1124 my $exitinfo = "exit ".exit_status_s($childstatus);
1125 $Jobstep->{'arvados_task'}->reload;
1126 my $task_success = $Jobstep->{'arvados_task'}->{success};
1128 Log ($jobstepid, "child $pid on $whatslot $exitinfo success=$task_success");
1130 if (!defined $task_success) {
1131 # task did not indicate one way or the other --> fail
1132 $Jobstep->{'arvados_task'}->{success} = 0;
1133 $Jobstep->{'arvados_task'}->save;
1140 $temporary_fail ||= $Jobstep->{tempfail};
1141 $temporary_fail ||= ($exitvalue == TASK_TEMPFAIL);
1143 ++$thisround_failed;
1144 ++$thisround_failed_multiple if $Jobstep->{'failures'} >= 1;
1146 # Check for signs of a failed or misconfigured node
1147 if (++$slot[$proc{$pid}->{slot}]->{node}->{losing_streak} >=
1148 2+$slot[$proc{$pid}->{slot}]->{node}->{ncpus}) {
1149 # Don't count this against jobstep failure thresholds if this
1150 # node is already suspected faulty and srun exited quickly
1151 if ($slot[$proc{$pid}->{slot}]->{node}->{hold_until} &&
1153 Log ($jobstepid, "blaming failure on suspect node " .
1154 $slot[$proc{$pid}->{slot}]->{node}->{name});
1155 $temporary_fail ||= 1;
1157 ban_node_by_slot($proc{$pid}->{slot});
1160 Log ($jobstepid, sprintf('failure (#%d, %s) after %d seconds',
1161 ++$Jobstep->{'failures'},
1162 $temporary_fail ? 'temporary' : 'permanent',
1165 if (!$temporary_fail || $Jobstep->{'failures'} >= 3) {
1166 # Give up on this task, and the whole job
1169 # Put this task back on the todo queue
1170 push @jobstep_todo, $jobstepid;
1171 $Job->{'tasks_summary'}->{'failed'}++;
1175 ++$thisround_succeeded;
1176 $slot[$proc{$pid}->{slot}]->{node}->{losing_streak} = 0;
1177 $slot[$proc{$pid}->{slot}]->{node}->{hold_until} = 0;
1178 $slot[$proc{$pid}->{slot}]->{node}->{fail_count} = 0;
1179 push @jobstep_done, $jobstepid;
1180 Log ($jobstepid, "success in $elapsed seconds");
1182 $Jobstep->{exitcode} = $childstatus;
1183 $Jobstep->{finishtime} = time;
1184 $Jobstep->{'arvados_task'}->{finished_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{finishtime});
1185 $Jobstep->{'arvados_task'}->save;
1186 process_stderr ($jobstepid, $task_success);
1187 Log ($jobstepid, sprintf("task output (%d bytes): %s",
1188 length($Jobstep->{'arvados_task'}->{output}),
1189 $Jobstep->{'arvados_task'}->{output}));
1191 close $reader{$jobstepid};
1192 delete $reader{$jobstepid};
1193 delete $slot[$proc{$pid}->{slot}]->{pid};
1194 push @freeslot, $proc{$pid}->{slot};
1197 if ($task_success) {
1199 my $newtask_list = [];
1200 my $newtask_results;
1202 $newtask_results = api_call(
1205 'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid}
1207 'order' => 'qsequence',
1208 'offset' => scalar(@$newtask_list),
1210 push(@$newtask_list, @{$newtask_results->{items}});
1211 } while (@{$newtask_results->{items}});
1212 foreach my $arvados_task (@$newtask_list) {
1214 'level' => $arvados_task->{'sequence'},
1216 'arvados_task' => $arvados_task
1218 push @jobstep, $jobstep;
1219 push @jobstep_todo, $#jobstep;
1223 $progress_is_dirty = 1;
1227 sub check_refresh_wanted
1229 my @stat = stat $ENV{"CRUNCH_REFRESH_TRIGGER"};
1230 if (@stat && $stat[9] > $latest_refresh) {
1231 $latest_refresh = scalar time;
1232 my $Job2 = api_call("jobs/get", uuid => $jobspec);
1233 for my $attr ('cancelled_at',
1234 'cancelled_by_user_uuid',
1235 'cancelled_by_client_uuid',
1237 $Job->{$attr} = $Job2->{$attr};
1239 if ($Job->{'state'} ne "Running") {
1240 if ($Job->{'state'} eq "Cancelled") {
1241 Log (undef, "Job cancelled at " . $Job->{'cancelled_at'} . " by user " . $Job->{'cancelled_by_user_uuid'});
1243 Log (undef, "Job state unexpectedly changed to " . $Job->{'state'});
1246 $main::please_freeze = 1;
1253 my $last_squeue_check = $squeue_checked;
1255 # Do not call `squeue` or check the kill list more than once every
1257 return if $last_squeue_check > time - 15;
1258 $squeue_checked = time;
1260 # Look for children from which we haven't received stderr data since
1261 # the last squeue check. If no such children exist, all procs are
1262 # alive and there's no need to even look at squeue.
1264 # As long as the crunchstat poll interval (10s) is shorter than the
1265 # squeue check interval (15s) this should make the squeue check an
1267 my $silent_procs = 0;
1268 for my $jobstep (values %proc)
1270 if ($jobstep->{stderr_at} < $last_squeue_check)
1275 return if $silent_procs == 0;
1277 # use killem() on procs whose killtime is reached
1278 while (my ($pid, $jobstep) = each %proc)
1280 if (exists $jobstep->{killtime}
1281 && $jobstep->{killtime} <= time
1282 && $jobstep->{stderr_at} < $last_squeue_check)
1285 if ($jobstep->{stderr_at}) {
1286 $sincewhen = " in last " . (time - $jobstep->{stderr_at}) . "s";
1288 Log($jobstep->{jobstep}, "killing orphaned srun process $pid (task not in slurm queue, no stderr received$sincewhen)");
1295 # here is an opportunity to check for mysterious problems with local procs
1299 # Get a list of steps still running. Note: squeue(1) says --steps
1300 # selects a format (which we override anyway) and allows us to
1301 # specify which steps we're interested in (which we don't).
1302 # Importantly, it also changes the meaning of %j from "job name" to
1303 # "step name" and (although this isn't mentioned explicitly in the
1304 # docs) switches from "one line per job" mode to "one line per step"
1305 # mode. Without it, we'd just get a list of one job, instead of a
1307 my @squeue = `squeue --jobs=\Q$ENV{SLURM_JOB_ID}\E --steps --format='%j' --noheader`;
1310 Log(undef, "warning: squeue exit status $? ($!)");
1315 # which of my jobsteps are running, according to squeue?
1317 for my $jobstepname (@squeue)
1319 $ok{$jobstepname} = 1;
1322 # Check for child procs >60s old and not mentioned by squeue.
1323 while (my ($pid, $jobstep) = each %proc)
1325 if ($jobstep->{time} < time - 60
1326 && $jobstep->{jobstepname}
1327 && !exists $ok{$jobstep->{jobstepname}}
1328 && !exists $jobstep->{killtime})
1330 # According to slurm, this task has ended (successfully or not)
1331 # -- but our srun child hasn't exited. First we must wait (30
1332 # seconds) in case this is just a race between communication
1333 # channels. Then, if our srun child process still hasn't
1334 # terminated, we'll conclude some slurm communication
1335 # error/delay has caused the task to die without notifying srun,
1336 # and we'll kill srun ourselves.
1337 $jobstep->{killtime} = time + 30;
1338 Log($jobstep->{jobstep}, "notice: task is not in slurm queue but srun process $pid has not exited");
1344 sub release_allocation
1348 Log (undef, "release job allocation");
1349 system "scancel $ENV{SLURM_JOB_ID}";
1357 foreach my $job (keys %reader)
1360 while (0 < sysread ($reader{$job}, $buf, 8192))
1362 print STDERR $buf if $ENV{CRUNCH_DEBUG};
1363 $jobstep[$job]->{stderr_at} = time;
1364 $jobstep[$job]->{stderr} .= $buf;
1365 preprocess_stderr ($job);
1366 if (length ($jobstep[$job]->{stderr}) > 16384)
1368 substr ($jobstep[$job]->{stderr}, 0, 8192) = "";
1377 sub preprocess_stderr
1381 while ($jobstep[$job]->{stderr} =~ /^(.*?)\n/) {
1383 substr $jobstep[$job]->{stderr}, 0, 1+length($line), "";
1384 Log ($job, "stderr $line");
1385 if ($line =~ /srun: error: (SLURM job $ENV{SLURM_JOB_ID} has expired|Unable to confirm allocation for job $ENV{SLURM_JOB_ID})/) {
1387 $main::please_freeze = 1;
1389 elsif ($line =~ /srun: error: Node failure on/) {
1390 my $job_slot_index = $jobstep[$job]->{slotindex};
1391 $slot[$job_slot_index]->{node}->{fail_count}++;
1392 $jobstep[$job]->{tempfail} = 1;
1393 ban_node_by_slot($job_slot_index);
1395 elsif ($line =~ /srun: error: (Unable to create job step|.*: Communication connection failure)/) {
1396 $jobstep[$job]->{tempfail} = 1;
1397 ban_node_by_slot($jobstep[$job]->{slotindex});
1399 elsif ($line =~ /arvados\.errors\.Keep/) {
1400 $jobstep[$job]->{tempfail} = 1;
1409 my $task_success = shift;
1410 preprocess_stderr ($job);
1413 Log ($job, "stderr $_");
1414 } split ("\n", $jobstep[$job]->{stderr});
1421 if (!open($keep, "-|", "arv-get", "--retries", retry_count(), $hash)) {
1422 Log(undef, "fetch_block run error from arv-get $hash: $!");
1425 my $output_block = "";
1428 my $bytes = sysread($keep, $buf, 1024 * 1024);
1429 if (!defined $bytes) {
1430 Log(undef, "fetch_block read error from arv-get: $!");
1431 $output_block = undef;
1433 } elsif ($bytes == 0) {
1434 # sysread returns 0 at the end of the pipe.
1437 # some bytes were read into buf.
1438 $output_block .= $buf;
1443 Log(undef, "fetch_block arv-get exited " . exit_status_s($?));
1444 $output_block = undef;
1446 return $output_block;
1449 # Create a collection by concatenating the output of all tasks (each
1450 # task's output is either a manifest fragment, a locator for a
1451 # manifest fragment stored in Keep, or nothing at all). Return the
1452 # portable_data_hash of the new collection.
1453 sub create_output_collection
1455 Log (undef, "collate");
1457 my ($child_out, $child_in);
1458 my $pid = open2($child_out, $child_in, 'python', '-c', q{
1461 print (arvados.api("v1").collections().
1462 create(body={"manifest_text": sys.stdin.read()}).
1463 execute(num_retries=int(sys.argv[1]))["portable_data_hash"])
1467 my $manifest_size = 0;
1471 my $output = $_->{'arvados_task'}->{output};
1472 next if (!defined($output));
1474 if ($output =~ /^[0-9a-f]{32}(\+\S+)*$/) {
1475 $next_write = fetch_block($output);
1477 $next_write = $output;
1479 if (defined($next_write)) {
1480 if (!defined(syswrite($child_in, $next_write))) {
1481 # There's been an error writing. Stop the loop.
1482 # We'll log details about the exit code later.
1485 $manifest_size += length($next_write);
1488 my $uuid = $_->{'arvados_task'}->{'uuid'};
1489 Log (undef, "Error retrieving '$output' output by task $task_idx ($uuid)");
1494 Log(undef, "collated output manifest text to send to API server is $manifest_size bytes with access tokens");
1497 my $s = IO::Select->new($child_out);
1498 if ($s->can_read(120)) {
1499 sysread($child_out, $joboutput, 1024 * 1024);
1502 Log(undef, "output collection creation exited " . exit_status_s($?));
1508 Log (undef, "timed out while creating output collection");
1509 foreach my $signal (2, 2, 2, 15, 15, 9) {
1510 kill($signal, $pid);
1511 last if waitpid($pid, WNOHANG) == -1;
1520 # Calls create_output_collection, logs the result, and returns it.
1521 # If that was successful, save that as the output in the job record.
1522 sub save_output_collection {
1523 my $collated_output = create_output_collection();
1525 if (!$collated_output) {
1526 Log(undef, "Failed to write output collection");
1529 Log(undef, "job output $collated_output");
1530 $Job->update_attributes('output' => $collated_output);
1532 return $collated_output;
1539 my $sig = 2; # SIGINT first
1540 if (exists $proc{$_}->{"sent_$sig"} &&
1541 time - $proc{$_}->{"sent_$sig"} > 4)
1543 $sig = 15; # SIGTERM if SIGINT doesn't work
1545 if (exists $proc{$_}->{"sent_$sig"} &&
1546 time - $proc{$_}->{"sent_$sig"} > 4)
1548 $sig = 9; # SIGKILL if SIGTERM doesn't work
1550 if (!exists $proc{$_}->{"sent_$sig"})
1552 Log ($proc{$_}->{jobstep}, "sending 2x signal $sig to pid $_");
1554 select (undef, undef, undef, 0.1);
1557 kill $sig, $_; # srun wants two SIGINT to really interrupt
1559 $proc{$_}->{"sent_$sig"} = time;
1560 $proc{$_}->{"killedafter"} = time - $proc{$_}->{"time"};
1570 vec($bits,fileno($_),1) = 1;
1576 # Send log output to Keep via arv-put.
1578 # $log_pipe_in and $log_pipe_out are the input and output filehandles to the arv-put pipe.
1579 # $log_pipe_out_buf is a string containing all output read from arv-put so far.
1580 # $log_pipe_out_select is an IO::Select object around $log_pipe_out.
1581 # $log_pipe_pid is the pid of the arv-put subprocess.
1583 # The only functions that should access these variables directly are:
1585 # log_writer_start($logfilename)
1586 # Starts an arv-put pipe, reading data on stdin and writing it to
1587 # a $logfilename file in an output collection.
1589 # log_writer_read_output([$timeout])
1590 # Read output from $log_pipe_out and append it to $log_pipe_out_buf.
1591 # Passes $timeout to the select() call, with a default of 0.01.
1592 # Returns the result of the last read() call on $log_pipe_out, or
1593 # -1 if read() wasn't called because select() timed out.
1594 # Only other log_writer_* functions should need to call this.
1596 # log_writer_send($txt)
1597 # Writes $txt to the output log collection.
1599 # log_writer_finish()
1600 # Closes the arv-put pipe and returns the output that it produces.
1602 # log_writer_is_active()
1603 # Returns a true value if there is currently a live arv-put
1604 # process, false otherwise.
1606 my ($log_pipe_in, $log_pipe_out, $log_pipe_out_buf, $log_pipe_out_select,
1609 sub log_writer_start($)
1611 my $logfilename = shift;
1612 $log_pipe_pid = open2($log_pipe_out, $log_pipe_in,
1616 '--filename', $logfilename,
1618 $log_pipe_out_buf = "";
1619 $log_pipe_out_select = IO::Select->new($log_pipe_out);
1622 sub log_writer_read_output {
1623 my $timeout = shift || 0.01;
1625 while ($read && $log_pipe_out_select->can_read($timeout)) {
1626 $read = read($log_pipe_out, $log_pipe_out_buf, 65536,
1627 length($log_pipe_out_buf));
1629 if (!defined($read)) {
1630 Log(undef, "error reading log manifest from arv-put: $!");
1635 sub log_writer_send($)
1638 print $log_pipe_in $txt;
1639 log_writer_read_output();
1642 sub log_writer_finish()
1644 return unless $log_pipe_pid;
1646 close($log_pipe_in);
1648 my $read_result = log_writer_read_output(120);
1649 if ($read_result == -1) {
1650 Log (undef, "timed out reading from 'arv-put'");
1651 } elsif ($read_result != 0) {
1652 Log(undef, "failed to read arv-put log manifest to EOF");
1655 waitpid($log_pipe_pid, 0);
1657 Log(undef, "log_writer_finish: arv-put exited " . exit_status_s($?))
1660 close($log_pipe_out);
1661 my $arv_put_output = $log_pipe_out_buf;
1662 $log_pipe_pid = $log_pipe_in = $log_pipe_out = $log_pipe_out_buf =
1663 $log_pipe_out_select = undef;
1665 return $arv_put_output;
1668 sub log_writer_is_active() {
1669 return $log_pipe_pid;
1672 sub Log # ($jobstep_id, $logmessage)
1674 if ($_[1] =~ /\n/) {
1675 for my $line (split (/\n/, $_[1])) {
1680 my $fh = select STDERR; $|=1; select $fh;
1681 my $message = sprintf ("%s %d %s %s", $job_id, $$, @_);
1682 $message =~ s{([^ -\176])}{"\\" . sprintf ("%03o", ord($1))}ge;
1685 if (log_writer_is_active() || -t STDERR) {
1686 my @gmtime = gmtime;
1687 $datetime = sprintf ("%04d-%02d-%02d_%02d:%02d:%02d",
1688 $gmtime[5]+1900, $gmtime[4]+1, @gmtime[3,2,1,0]);
1690 print STDERR ((-t STDERR) ? ($datetime." ".$message) : $message);
1692 if (log_writer_is_active()) {
1693 log_writer_send($datetime . " " . $message);
1700 my ($package, $file, $line) = caller;
1701 my $message = "@_ at $file line $line\n";
1702 Log (undef, $message);
1703 freeze() if @jobstep_todo;
1704 create_output_collection() if @jobstep_todo;
1714 if ($Job->{'state'} eq 'Cancelled') {
1715 $Job->update_attributes('finished_at' => scalar gmtime);
1717 $Job->update_attributes('state' => 'Failed');
1724 my $justcheckpoint = shift; # false if this will be the last meta saved
1725 return if $justcheckpoint; # checkpointing is not relevant post-Warehouse.pm
1726 return unless log_writer_is_active();
1728 my $log_manifest = "";
1730 my $prev_log_coll = api_call("collections/get", uuid => $Job->{log});
1731 $log_manifest .= $prev_log_coll->{manifest_text};
1733 $log_manifest .= log_writer_finish();
1735 my $log_coll = api_call(
1736 "collections/create", ensure_unique_name => 1, collection => {
1737 manifest_text => $log_manifest,
1738 owner_uuid => $Job->{owner_uuid},
1739 name => sprintf("Log from %s job %s", $Job->{script}, $Job->{uuid}),
1741 Log(undef, "log collection is " . $log_coll->{portable_data_hash});
1742 $Job->update_attributes('log' => $log_coll->{portable_data_hash});
1746 sub freeze_if_want_freeze
1748 if ($main::please_freeze)
1750 release_allocation();
1753 # kill some srun procs before freeze+stop
1754 map { $proc{$_} = {} } @_;
1757 killem (keys %proc);
1758 select (undef, undef, undef, 0.1);
1760 while (($died = waitpid (-1, WNOHANG)) > 0)
1762 delete $proc{$died};
1767 create_output_collection();
1777 Log (undef, "Freeze not implemented");
1784 croak ("Thaw not implemented");
1800 $s =~ s{\\(.)}{$1 eq "n" ? "\n" : $1}ge;
1807 my $srunargs = shift;
1808 my $execargs = shift;
1809 my $opts = shift || {};
1811 my $args = $have_slurm ? [@$srunargs, @$execargs] : $execargs;
1813 $Data::Dumper::Terse = 1;
1814 $Data::Dumper::Indent = 0;
1815 my $show_cmd = Dumper($args);
1816 $show_cmd =~ s/(TOKEN\\*=)[^\s\']+/${1}[...]/g;
1817 $show_cmd =~ s/\n/ /g;
1818 if ($opts->{fork}) {
1819 Log(undef, "starting: $show_cmd");
1821 # This is a child process: parent is in charge of reading our
1822 # stderr and copying it to Log() if needed.
1823 warn "starting: $show_cmd\n";
1826 if (defined $stdin) {
1827 my $child = open STDIN, "-|";
1828 defined $child or die "no fork: $!";
1830 print $stdin or die $!;
1831 close STDOUT or die $!;
1836 return system (@$args) if $opts->{fork};
1839 warn "ENV size is ".length(join(" ",%ENV));
1840 die "exec failed: $!: @$args";
1844 sub ban_node_by_slot {
1845 # Don't start any new jobsteps on this node for 60 seconds
1847 $slot[$slotid]->{node}->{hold_until} = 60 + scalar time;
1848 $slot[$slotid]->{node}->{hold_count}++;
1849 Log (undef, "backing off node " . $slot[$slotid]->{node}->{name} . " for 60 seconds");
1854 my ($lockfile, $error_message) = @_;
1855 open L, ">", $lockfile or croak("$lockfile: $!");
1856 if (!flock L, LOCK_EX|LOCK_NB) {
1857 croak("Can't lock $lockfile: $error_message\n");
1861 sub find_docker_image {
1862 # Given a Keep locator, check to see if it contains a Docker image.
1863 # If so, return its stream name and Docker hash.
1864 # If not, return undef for both values.
1865 my $locator = shift;
1866 my ($streamname, $filename);
1867 my $image = api_call("collections/get", uuid => $locator);
1869 foreach my $line (split(/\n/, $image->{manifest_text})) {
1870 my @tokens = split(/\s+/, $line);
1872 $streamname = shift(@tokens);
1873 foreach my $filedata (grep(/^\d+:\d+:/, @tokens)) {
1874 if (defined($filename)) {
1875 return (undef, undef); # More than one file in the Collection.
1877 $filename = (split(/:/, $filedata, 3))[2];
1882 if (defined($filename) and ($filename =~ /^([0-9A-Fa-f]{64})\.tar$/)) {
1883 return ($streamname, $1);
1885 return (undef, undef);
1890 # Calculate the number of times an operation should be retried,
1891 # assuming exponential backoff, and that we're willing to retry as
1892 # long as tasks have been running. Enforce a minimum of 3 retries.
1893 my ($starttime, $endtime, $timediff, $retries);
1895 $starttime = $jobstep[0]->{starttime};
1896 $endtime = $jobstep[-1]->{finishtime};
1898 if (!defined($starttime)) {
1900 } elsif (!defined($endtime)) {
1901 $timediff = time - $starttime;
1903 $timediff = ($endtime - $starttime) - (time - $endtime);
1905 if ($timediff > 0) {
1906 $retries = int(log($timediff) / log(2));
1908 $retries = 1; # Use the minimum.
1910 return ($retries > 3) ? $retries : 3;
1914 # Pass in two function references.
1915 # This method will be called with the remaining arguments.
1916 # If it dies, retry it with exponential backoff until it succeeds,
1917 # or until the current retry_count is exhausted. After each failure
1918 # that can be retried, the second function will be called with
1919 # the current try count (0-based), next try time, and error message.
1920 my $operation = shift;
1921 my $retry_callback = shift;
1922 my $retries = retry_count();
1923 foreach my $try_count (0..$retries) {
1924 my $next_try = time + (2 ** $try_count);
1925 my $result = eval { $operation->(@_); };
1928 } elsif ($try_count < $retries) {
1929 $retry_callback->($try_count, $next_try, $@);
1930 my $sleep_time = $next_try - time;
1931 sleep($sleep_time) if ($sleep_time > 0);
1934 # Ensure the error message ends in a newline, so Perl doesn't add
1935 # retry_op's line number to it.
1941 # Pass in a /-separated API method name, and arguments for it.
1942 # This function will call that method, retrying as needed until
1943 # the current retry_count is exhausted, with a log on the first failure.
1944 my $method_name = shift;
1945 my $log_api_retry = sub {
1946 my ($try_count, $next_try_at, $errmsg) = @_;
1947 $errmsg =~ s/\s*\bat \Q$0\E line \d+\.?\s*//;
1948 $errmsg =~ s/\s/ /g;
1949 $errmsg =~ s/\s+$//;
1951 if ($next_try_at < time) {
1952 $retry_msg = "Retrying.";
1954 my $next_try_fmt = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($next_try_at);
1955 $retry_msg = "Retrying at $next_try_fmt.";
1957 Log(undef, "API method $method_name failed: $errmsg. $retry_msg");
1960 foreach my $key (split(/\//, $method_name)) {
1961 $method = $method->{$key};
1963 return retry_op(sub { $method->execute(@_); }, $log_api_retry, @_);
1967 # Given a $?, return a human-readable exit code string like "0" or
1968 # "1" or "0 with signal 1" or "1 with signal 11".
1969 my $exitcode = shift;
1970 my $s = $exitcode >> 8;
1971 if ($exitcode & 0x7f) {
1972 $s .= " with signal " . ($exitcode & 0x7f);
1974 if ($exitcode & 0x80) {
1975 $s .= " with core dump";
1980 sub handle_readall {
1981 # Pass in a glob reference to a file handle.
1982 # Read all its contents and return them as a string.
1983 my $fh_glob_ref = shift;
1985 return <$fh_glob_ref>;
1988 sub tar_filename_n {
1990 return sprintf("%s/git.%s.%d.tar", $ENV{CRUNCH_TMP}, $job_id, $n);
1993 sub add_git_archive {
1994 # Pass in a git archive command as a string or list, a la system().
1995 # This method will save its output to be included in the archive sent to the
1999 if (!open(GIT_ARCHIVE, ">", tar_filename_n($git_tar_count))) {
2000 croak("Failed to save git archive: $!");
2002 my $git_pid = open2(">&GIT_ARCHIVE", $git_input, @_);
2004 waitpid($git_pid, 0);
2007 croak("Failed to save git archive: git exited " . exit_status_s($?));
2011 sub combined_git_archive {
2012 # Combine all saved tar archives into a single archive, then return its
2013 # contents in a string. Return undef if no archives have been saved.
2014 if ($git_tar_count < 1) {
2017 my $base_tar_name = tar_filename_n(1);
2018 foreach my $tar_to_append (map { tar_filename_n($_); } (2..$git_tar_count)) {
2019 my $tar_exit = system("tar", "-Af", $base_tar_name, $tar_to_append);
2020 if ($tar_exit != 0) {
2021 croak("Error preparing build archive: tar -A exited " .
2022 exit_status_s($tar_exit));
2025 if (!open(GIT_TAR, "<", $base_tar_name)) {
2026 croak("Could not open build archive: $!");
2028 my $tar_contents = handle_readall(\*GIT_TAR);
2030 return $tar_contents;
2033 sub set_nonblocking {
2035 my $flags = fcntl ($fh, F_GETFL, 0) or croak ($!);
2036 fcntl ($fh, F_SETFL, $flags | O_NONBLOCK) or croak ($!);
2042 # This is crunch-job's internal dispatch script. crunch-job running on the API
2043 # server invokes this script on individual compute nodes, or localhost if we're
2044 # running a job locally. It gets called in two modes:
2046 # * No arguments: Installation mode. Read a tar archive from the DATA
2047 # file handle; it includes the Crunch script's source code, and
2048 # maybe SDKs as well. Those should be installed in the proper
2049 # locations. This runs outside of any Docker container, so don't try to
2050 # introspect Crunch's runtime environment.
2052 # * With arguments: Crunch script run mode. This script should set up the
2053 # environment, then run the command specified in the arguments. This runs
2054 # inside any Docker container.
2057 use File::Path qw( make_path remove_tree );
2058 use POSIX qw(getcwd);
2060 use constant TASK_TEMPFAIL => 111;
2062 # Map SDK subdirectories to the path environments they belong to.
2063 my %SDK_ENVVARS = ("perl/lib" => "PERLLIB", "ruby/lib" => "RUBYLIB");
2065 my $destdir = $ENV{"CRUNCH_SRC"};
2066 my $archive_hash = $ENV{"CRUNCH_GIT_ARCHIVE_HASH"};
2067 my $repo = $ENV{"CRUNCH_SRC_URL"};
2068 my $install_dir = $ENV{"CRUNCH_INSTALL"} || (getcwd() . "/opt");
2069 my $job_work = $ENV{"JOB_WORK"};
2070 my $task_work = $ENV{"TASK_WORK"};
2072 open(STDOUT_ORIG, ">&", STDOUT);
2073 open(STDERR_ORIG, ">&", STDERR);
2075 for my $dir ($destdir, $job_work, $task_work) {
2078 -e $dir or die "Failed to create temporary directory ($dir): $!";
2083 remove_tree($task_work, {keep_root => 1});
2086 ### Crunch script run mode
2088 # We want to do routine logging during task 0 only. This gives the user
2089 # the information they need, but avoids repeating the information for every
2092 if ($ENV{TASK_SEQUENCE} eq "0") {
2095 printf STDERR_ORIG "[Crunch] $msg\n", @_;
2101 my $python_src = "$install_dir/python";
2102 my $venv_dir = "$job_work/.arvados.venv";
2103 my $venv_built = -e "$venv_dir/bin/activate";
2104 if ((!$venv_built) and (-d $python_src) and can_run("virtualenv")) {
2105 shell_or_die(undef, "virtualenv", "--quiet", "--system-site-packages",
2106 "--python=python2.7", $venv_dir);
2107 shell_or_die(TASK_TEMPFAIL, "$venv_dir/bin/pip", "--quiet", "install", "-I", $python_src);
2109 $Log->("Built Python SDK virtualenv");
2112 my $pip_bin = "pip";
2114 $Log->("Running in Python SDK virtualenv");
2115 $pip_bin = "$venv_dir/bin/pip";
2116 my $orig_argv = join(" ", map { quotemeta($_); } @ARGV);
2117 @ARGV = ("/bin/sh", "-ec",
2118 ". \Q$venv_dir/bin/activate\E; exec $orig_argv");
2119 } elsif (-d $python_src) {
2120 $Log->("Warning: virtualenv not found inside Docker container default " .
2121 "\$PATH. Can't install Python SDK.");
2124 my $pkgs = `(\Q$pip_bin\E freeze 2>/dev/null | grep arvados) || dpkg-query --show '*arvados*'`;
2126 $Log->("Using Arvados SDK:");
2127 foreach my $line (split /\n/, $pkgs) {
2131 $Log->("Arvados SDK packages not found");
2134 while (my ($sdk_dir, $sdk_envkey) = each(%SDK_ENVVARS)) {
2135 my $sdk_path = "$install_dir/$sdk_dir";
2137 if ($ENV{$sdk_envkey}) {
2138 $ENV{$sdk_envkey} = "$sdk_path:" . $ENV{$sdk_envkey};
2140 $ENV{$sdk_envkey} = $sdk_path;
2142 $Log->("Arvados SDK added to %s", $sdk_envkey);
2147 die "Cannot exec `@ARGV`: $!";
2150 ### Installation mode
2151 open L, ">", "$destdir.lock" or die "$destdir.lock: $!";
2153 if (readlink ("$destdir.archive_hash") eq $archive_hash && -d $destdir) {
2154 # This exact git archive (source + arvados sdk) is already installed
2155 # here, so there's no need to reinstall it.
2157 # We must consume our DATA section, though: otherwise the process
2158 # feeding it to us will get SIGPIPE.
2160 while (read(DATA, $buf, 65536)) { }
2165 unlink "$destdir.archive_hash";
2169 # Ignore SIGPIPE: we check retval of close() instead. See perlipc(1).
2170 local $SIG{PIPE} = "IGNORE";
2171 warn "Extracting archive: $archive_hash\n";
2172 # --ignore-zeros is necessary sometimes: depending on how much NUL
2173 # padding tar -A put on our combined archive (which in turn depends
2174 # on the length of the component archives) tar without
2175 # --ignore-zeros will exit before consuming stdin and cause close()
2176 # to fail on the resulting SIGPIPE.
2177 if (!open(TARX, "|-", "tar", "--ignore-zeros", "-xC", $destdir)) {
2178 die "Error launching 'tar -xC $destdir': $!";
2180 # If we send too much data to tar in one write (> 4-5 MiB), it stops, and we
2181 # get SIGPIPE. We must feed it data incrementally.
2183 while (read(DATA, $tar_input, 65536)) {
2184 print TARX $tar_input;
2187 die "'tar -xC $destdir' exited $?: $!";
2193 my $sdk_root = "$destdir/.arvados.sdk/sdk";
2195 foreach my $sdk_lang (("python",
2196 map { (split /\//, $_, 2)[0]; } keys(%SDK_ENVVARS))) {
2197 if (-d "$sdk_root/$sdk_lang") {
2198 if (!rename("$sdk_root/$sdk_lang", "$install_dir/$sdk_lang")) {
2199 die "Failed to install $sdk_lang SDK: $!";
2205 my $python_dir = "$install_dir/python";
2206 if ((-d $python_dir) and can_run("python2.7")) {
2207 open(my $egg_info_pipe, "-|",
2208 "python2.7 \Q$python_dir/setup.py\E egg_info 2>&1 >/dev/null");
2209 my @egg_info_errors = <$egg_info_pipe>;
2210 close($egg_info_pipe);
2213 if (@egg_info_errors and (($egg_info_errors[-1] =~ /\bgit\b/) or ($egg_info_errors[-1] =~ /\[Errno 2\]/))) {
2214 # egg_info apparently failed because it couldn't ask git for a build tag.
2215 # Specify no build tag.
2216 open(my $pysdk_cfg, ">>", "$python_dir/setup.cfg");
2217 print $pysdk_cfg "\n[egg_info]\ntag_build =\n";
2220 my $egg_info_exit = $? >> 8;
2221 foreach my $errline (@egg_info_errors) {
2224 warn "python setup.py egg_info failed: exit $egg_info_exit";
2225 exit ($egg_info_exit || 1);
2230 # Hide messages from the install script (unless it fails: shell_or_die
2231 # will show $destdir.log in that case).
2232 open(STDOUT, ">>", "$destdir.log");
2233 open(STDERR, ">&", STDOUT);
2235 if (-e "$destdir/crunch_scripts/install") {
2236 shell_or_die (undef, "$destdir/crunch_scripts/install", $install_dir);
2237 } elsif (!-e "./install.sh" && -e "./tests/autotests.sh") {
2239 shell_or_die (undef, "./tests/autotests.sh", $install_dir);
2240 } elsif (-e "./install.sh") {
2241 shell_or_die (undef, "./install.sh", $install_dir);
2244 if ($archive_hash) {
2245 unlink "$destdir.archive_hash.new";
2246 symlink ($archive_hash, "$destdir.archive_hash.new") or die "$destdir.archive_hash.new: $!";
2247 rename ("$destdir.archive_hash.new", "$destdir.archive_hash") or die "$destdir.archive_hash: $!";
2253 my $command_name = shift;
2254 open(my $which, "-|", "which", $command_name);
2255 while (<$which>) { }
2262 my $exitcode = shift;
2264 if ($ENV{"DEBUG"}) {
2265 print STDERR "@_\n";
2267 if (system (@_) != 0) {
2270 my $exitstatus = sprintf("exit %d signal %d", $code >> 8, $code & 0x7f);
2271 open STDERR, ">&STDERR_ORIG";
2272 system ("cat $destdir.log >&2");
2273 warn "@_ failed ($err): $exitstatus";
2274 if (defined($exitcode)) {
2278 exit (($code >> 8) || 1);