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 EX_TEMPFAIL => 75;
101 $ENV{"TMPDIR"} ||= "/tmp";
102 unless (defined $ENV{"CRUNCH_TMP"}) {
103 $ENV{"CRUNCH_TMP"} = $ENV{"TMPDIR"} . "/crunch-job";
104 if ($ENV{"USER"} ne "crunch" && $< != 0) {
105 # use a tmp dir unique for my uid
106 $ENV{"CRUNCH_TMP"} .= "-$<";
110 # Create the tmp directory if it does not exist
111 if ( ! -d $ENV{"CRUNCH_TMP"} ) {
112 make_path $ENV{"CRUNCH_TMP"} or die "Failed to create temporary working directory: " . $ENV{"CRUNCH_TMP"};
115 $ENV{"JOB_WORK"} = $ENV{"CRUNCH_TMP"} . "/work";
116 $ENV{"CRUNCH_INSTALL"} = "$ENV{CRUNCH_TMP}/opt";
117 $ENV{"CRUNCH_WORK"} = $ENV{"JOB_WORK"}; # deprecated
118 mkdir ($ENV{"JOB_WORK"});
126 GetOptions('force-unlock' => \$force_unlock,
127 'git-dir=s' => \$git_dir,
128 'job=s' => \$jobspec,
129 'job-api-token=s' => \$job_api_token,
130 'no-clear-tmp' => \$no_clear_tmp,
131 'resume-stash=s' => \$resume_stash,
134 if (defined $job_api_token) {
135 $ENV{ARVADOS_API_TOKEN} = $job_api_token;
138 my $have_slurm = exists $ENV{SLURM_JOBID} && exists $ENV{SLURM_NODELIST};
144 $main::ENV{CRUNCH_DEBUG} = 1;
148 $main::ENV{CRUNCH_DEBUG} = 0;
153 my $arv = Arvados->new('apiVersion' => 'v1');
161 my $User = api_call("users/current");
163 if ($jobspec =~ /^[-a-z\d]+$/)
165 # $jobspec is an Arvados UUID, not a JSON job specification
166 $Job = api_call("jobs/get", uuid => $jobspec);
167 if (!$force_unlock) {
168 # Claim this job, and make sure nobody else does
169 eval { api_call("jobs/lock", uuid => $Job->{uuid}); };
171 Log(undef, "Error while locking job, exiting ".EX_TEMPFAIL);
178 $Job = JSON::decode_json($jobspec);
182 map { croak ("No $_ specified") unless $Job->{$_} }
183 qw(script script_version script_parameters);
186 $Job->{'is_locked_by_uuid'} = $User->{'uuid'};
187 $Job->{'started_at'} = gmtime;
188 $Job->{'state'} = 'Running';
190 $Job = api_call("jobs/create", job => $Job);
192 $job_id = $Job->{'uuid'};
194 my $keep_logfile = $job_id . '.log.txt';
195 log_writer_start($keep_logfile);
197 $Job->{'runtime_constraints'} ||= {};
198 $Job->{'runtime_constraints'}->{'max_tasks_per_node'} ||= 0;
199 my $max_ncpus = $Job->{'runtime_constraints'}->{'max_tasks_per_node'};
201 my $gem_versions = `gem list --quiet arvados-cli 2>/dev/null`;
203 $gem_versions =~ s/^arvados-cli \(/ with arvados-cli Gem version(s) /;
204 chomp($gem_versions);
205 chop($gem_versions); # Closing parentheses
210 "running from " . ((-e $0) ? realpath($0) : "stdin") . $gem_versions);
212 Log (undef, "check slurm allocation");
215 # Should use $ENV{SLURM_TASKS_PER_NODE} instead of sinfo? (eg. "4(x3),2,4(x2)")
219 my $localcpus = 0 + `grep -cw ^processor /proc/cpuinfo` || 1;
220 push @sinfo, "$localcpus localhost";
222 if (exists $ENV{SLURM_NODELIST})
224 push @sinfo, `sinfo -h --format='%c %N' --nodes=\Q$ENV{SLURM_NODELIST}\E`;
228 my ($ncpus, $slurm_nodelist) = split;
229 $ncpus = $max_ncpus if $max_ncpus && $ncpus > $max_ncpus;
232 while ($slurm_nodelist =~ s/^([^\[,]+?(\[.*?\])?)(,|$)//)
235 if ($nodelist =~ /\[((\d+)(-(\d+))?(,(\d+)(-(\d+))?)*)\]/)
238 foreach (split (",", $ranges))
251 push @nodelist, map {
253 $n =~ s/\[[-,\d]+\]/$_/;
260 push @nodelist, $nodelist;
263 foreach my $nodename (@nodelist)
265 Log (undef, "node $nodename - $ncpus slots");
266 my $node = { name => $nodename,
270 foreach my $cpu (1..$ncpus)
272 push @slot, { node => $node,
276 push @node, @nodelist;
281 # Ensure that we get one jobstep running on each allocated node before
282 # we start overloading nodes with concurrent steps
284 @slot = sort { $a->{cpu} <=> $b->{cpu} } @slot;
287 $Job->update_attributes(
288 'tasks_summary' => { 'failed' => 0,
293 Log (undef, "start");
294 $SIG{'INT'} = sub { $main::please_freeze = 1; };
295 $SIG{'QUIT'} = sub { $main::please_freeze = 1; };
296 $SIG{'TERM'} = \&croak;
297 $SIG{'TSTP'} = sub { $main::please_freeze = 1; };
298 $SIG{'ALRM'} = sub { $main::please_info = 1; };
299 $SIG{'CONT'} = sub { $main::please_continue = 1; };
300 $SIG{'HUP'} = sub { $main::please_refresh = 1; };
302 $main::please_freeze = 0;
303 $main::please_info = 0;
304 $main::please_continue = 0;
305 $main::please_refresh = 0;
306 my $jobsteps_must_output_keys = 0; # becomes 1 when any task outputs a key
308 grep { $ENV{$1} = $2 if /^(NOCACHE.*?)=(.*)/ } split ("\n", $$Job{knobs});
309 $ENV{"CRUNCH_JOB_UUID"} = $job_id;
310 $ENV{"JOB_UUID"} = $job_id;
313 my @jobstep_todo = ();
314 my @jobstep_done = ();
315 my @jobstep_tomerge = ();
316 my $jobstep_tomerge_level = 0;
318 my $squeue_kill_checked;
319 my $latest_refresh = scalar time;
323 if (defined $Job->{thawedfromkey})
325 thaw ($Job->{thawedfromkey});
329 my $first_task = api_call("job_tasks/create", job_task => {
330 'job_uuid' => $Job->{'uuid'},
335 push @jobstep, { 'level' => 0,
337 'arvados_task' => $first_task,
339 push @jobstep_todo, 0;
345 must_lock_now("$ENV{CRUNCH_TMP}/.lock", "a job is already running here.");
348 my $build_script = handle_readall(\*DATA);
349 my $nodelist = join(",", @node);
350 my $git_tar_count = 0;
352 if (!defined $no_clear_tmp) {
353 # Clean out crunch_tmp/work, crunch_tmp/opt, crunch_tmp/src*
354 Log (undef, "Clean work dirs");
356 my $cleanpid = fork();
359 # Find FUSE mounts that look like Keep mounts (the mount path has the
360 # word "keep") and unmount them. Then clean up work directories.
361 # TODO: When #5036 is done and widely deployed, we can get rid of the
362 # regular expression and just unmount everything with type fuse.keep.
363 srun (["srun", "--nodelist=$nodelist", "-D", $ENV{'TMPDIR'}],
364 ['bash', '-ec', 'mount -t fuse,fuse.keep | awk \'($3 ~ /\ykeep\y/){print $3}\' | xargs -r -n 1 fusermount -u -z; sleep 1; rm -rf $JOB_WORK $CRUNCH_INSTALL $CRUNCH_TMP/task $CRUNCH_TMP/src*']);
369 last if $cleanpid == waitpid (-1, WNOHANG);
370 freeze_if_want_freeze ($cleanpid);
371 select (undef, undef, undef, 0.1);
373 Log (undef, "Cleanup command exited ".exit_status_s($?));
376 # If this job requires a Docker image, install that.
377 my $docker_bin = "/usr/bin/docker.io";
378 my ($docker_locator, $docker_stream, $docker_hash);
379 if ($docker_locator = $Job->{docker_image_locator}) {
380 ($docker_stream, $docker_hash) = find_docker_image($docker_locator);
383 croak("No Docker image hash found from locator $docker_locator");
385 $docker_stream =~ s/^\.//;
386 my $docker_install_script = qq{
387 if ! $docker_bin images -q --no-trunc | grep -qxF \Q$docker_hash\E; then
388 arv-get \Q$docker_locator$docker_stream/$docker_hash.tar\E | $docker_bin load
391 my $docker_pid = fork();
392 if ($docker_pid == 0)
394 srun (["srun", "--nodelist=" . join(',', @node)],
395 ["/bin/sh", "-ec", $docker_install_script]);
400 last if $docker_pid == waitpid (-1, WNOHANG);
401 freeze_if_want_freeze ($docker_pid);
402 select (undef, undef, undef, 0.1);
406 croak("Installing Docker image from $docker_locator exited "
410 if ($Job->{arvados_sdk_version}) {
411 # The job also specifies an Arvados SDK version. Add the SDKs to the
412 # tar file for the build script to install.
413 Log(undef, sprintf("Packing Arvados SDK version %s for installation",
414 $Job->{arvados_sdk_version}));
415 add_git_archive("git", "--git-dir=$git_dir", "archive",
416 "--prefix=.arvados.sdk/",
417 $Job->{arvados_sdk_version}, "sdk");
421 if (!defined $git_dir && $Job->{'script_version'} =~ m{^/}) {
422 # If script_version looks like an absolute path, *and* the --git-dir
423 # argument was not given -- which implies we were not invoked by
424 # crunch-dispatch -- we will use the given path as a working
425 # directory instead of resolving script_version to a git commit (or
426 # doing anything else with git).
427 $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{'script_version'};
428 $ENV{"CRUNCH_SRC"} = $Job->{'script_version'};
431 # Resolve the given script_version to a git commit sha1. Also, if
432 # the repository is remote, clone it into our local filesystem: this
433 # ensures "git archive" will work, and is necessary to reliably
434 # resolve a symbolic script_version like "master^".
435 $ENV{"CRUNCH_SRC"} = "$ENV{CRUNCH_TMP}/src";
437 Log (undef, "Looking for version ".$Job->{script_version}." from repository ".$Job->{repository});
439 $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{script_version};
441 # If we're running under crunch-dispatch, it will have already
442 # pulled the appropriate source tree into its own repository, and
443 # given us that repo's path as $git_dir.
445 # If we're running a "local" job, we might have to fetch content
446 # from a remote repository.
448 # (Currently crunch-dispatch gives a local path with --git-dir, but
449 # we might as well accept URLs there too in case it changes its
451 my $repo = $git_dir || $Job->{'repository'};
453 # Repository can be remote or local. If remote, we'll need to fetch it
454 # to a local dir before doing `git log` et al.
457 if ($repo =~ m{://|^[^/]*:}) {
458 # $repo is a git url we can clone, like git:// or https:// or
459 # file:/// or [user@]host:repo.git. Note "user/name@host:foo" is
460 # not recognized here because distinguishing that from a local
461 # path is too fragile. If you really need something strange here,
462 # use the ssh:// form.
463 $repo_location = 'remote';
464 } elsif ($repo =~ m{^\.*/}) {
465 # $repo is a local path to a git index. We'll also resolve ../foo
466 # to ../foo/.git if the latter is a directory. To help
467 # disambiguate local paths from named hosted repositories, this
468 # form must be given as ./ or ../ if it's a relative path.
469 if (-d "$repo/.git") {
470 $repo = "$repo/.git";
472 $repo_location = 'local';
474 # $repo is none of the above. It must be the name of a hosted
476 my $arv_repo_list = api_call("repositories/list",
477 'filters' => [['name','=',$repo]]);
478 my @repos_found = @{$arv_repo_list->{'items'}};
479 my $n_found = $arv_repo_list->{'serverResponse'}->{'items_available'};
481 Log(undef, "Repository '$repo' -> "
482 . join(", ", map { $_->{'uuid'} } @repos_found));
485 croak("Error: Found $n_found repositories with name '$repo'.");
487 $repo = $repos_found[0]->{'fetch_url'};
488 $repo_location = 'remote';
490 Log(undef, "Using $repo_location repository '$repo'");
491 $ENV{"CRUNCH_SRC_URL"} = $repo;
493 # Resolve given script_version (we'll call that $treeish here) to a
494 # commit sha1 ($commit).
495 my $treeish = $Job->{'script_version'};
497 if ($repo_location eq 'remote') {
498 # We minimize excess object-fetching by re-using the same bare
499 # repository in CRUNCH_TMP/.git for multiple crunch-jobs -- we
500 # just keep adding remotes to it as needed.
501 my $local_repo = $ENV{'CRUNCH_TMP'}."/.git";
502 my $gitcmd = "git --git-dir=\Q$local_repo\E";
504 # Set up our local repo for caching remote objects, making
506 if (!-d $local_repo) {
507 make_path($local_repo) or croak("Error: could not create $local_repo");
509 # This works (exits 0 and doesn't delete fetched objects) even
510 # if $local_repo is already initialized:
511 `$gitcmd init --bare`;
513 croak("Error: $gitcmd init --bare exited ".exit_status_s($?));
516 # If $treeish looks like a hash (or abbrev hash) we look it up in
517 # our local cache first, since that's cheaper. (We don't want to
518 # do that with tags/branches though -- those change over time, so
519 # they should always be resolved by the remote repo.)
520 if ($treeish =~ /^[0-9a-f]{7,40}$/s) {
521 # Hide stderr because it's normal for this to fail:
522 my $sha1 = `$gitcmd rev-list -n1 ''\Q$treeish\E 2>/dev/null`;
524 # Careful not to resolve a branch named abcdeff to commit 1234567:
525 $sha1 =~ /^$treeish/ &&
526 $sha1 =~ /^([0-9a-f]{40})$/s) {
528 Log(undef, "Commit $commit already present in $local_repo");
532 if (!defined $commit) {
533 # If $treeish isn't just a hash or abbrev hash, or isn't here
534 # yet, we need to fetch the remote to resolve it correctly.
536 # First, remove all local heads. This prevents a name that does
537 # not exist on the remote from resolving to (or colliding with)
538 # a previously fetched branch or tag (possibly from a different
540 remove_tree("$local_repo/refs/heads", {keep_root => 1});
542 Log(undef, "Fetching objects from $repo to $local_repo");
543 `$gitcmd fetch --no-progress --tags ''\Q$repo\E \Q+refs/heads/*:refs/heads/*\E`;
545 croak("Error: `$gitcmd fetch` exited ".exit_status_s($?));
549 # Now that the data is all here, we will use our local repo for
550 # the rest of our git activities.
554 my $gitcmd = "git --git-dir=\Q$repo\E";
555 my $sha1 = `$gitcmd rev-list -n1 ''\Q$treeish\E`;
556 unless ($? == 0 && $sha1 =~ /^([0-9a-f]{40})$/) {
557 croak("`$gitcmd rev-list` exited "
559 .", '$treeish' not found. Giving up.");
562 Log(undef, "Version $treeish is commit $commit");
564 if ($commit ne $Job->{'script_version'}) {
565 # Record the real commit id in the database, frozentokey, logs,
566 # etc. -- instead of an abbreviation or a branch name which can
567 # become ambiguous or point to a different commit in the future.
568 if (!$Job->update_attributes('script_version' => $commit)) {
569 croak("Error: failed to update job's script_version attribute");
573 $ENV{"CRUNCH_SRC_COMMIT"} = $commit;
574 add_git_archive("$gitcmd archive ''\Q$commit\E");
577 my $git_archive = combined_git_archive();
578 if (!defined $git_archive) {
579 Log(undef, "Skip install phase (no git archive)");
581 Log(undef, "Warning: This probably means workers have no source tree!");
585 Log(undef, "Run install script on all workers");
587 my @srunargs = ("srun",
588 "--nodelist=$nodelist",
589 "-D", $ENV{'TMPDIR'}, "--job-name=$job_id");
590 my @execargs = ("sh", "-c",
591 "mkdir -p $ENV{CRUNCH_INSTALL} && cd $ENV{CRUNCH_TMP} && perl -");
593 my $installpid = fork();
594 if ($installpid == 0)
596 srun (\@srunargs, \@execargs, {}, $build_script . $git_archive);
601 last if $installpid == waitpid (-1, WNOHANG);
602 freeze_if_want_freeze ($installpid);
603 select (undef, undef, undef, 0.1);
605 my $install_exited = $?;
606 Log (undef, "Install script exited ".exit_status_s($install_exited));
607 foreach my $tar_filename (map { tar_filename_n($_); } (1..$git_tar_count)) {
608 unlink($tar_filename);
610 exit (1) if $install_exited != 0;
613 foreach (qw (script script_version script_parameters runtime_constraints))
617 (ref($Job->{$_}) ? JSON::encode_json($Job->{$_}) : $Job->{$_}));
619 foreach (split (/\n/, $Job->{knobs}))
621 Log (undef, "knob " . $_);
626 $main::success = undef;
632 my $thisround_succeeded = 0;
633 my $thisround_failed = 0;
634 my $thisround_failed_multiple = 0;
636 @jobstep_todo = sort { $jobstep[$a]->{level} <=> $jobstep[$b]->{level}
637 or $a <=> $b } @jobstep_todo;
638 my $level = $jobstep[$jobstep_todo[0]]->{level};
639 Log (undef, "start level $level");
644 my @freeslot = (0..$#slot);
647 my $progress_is_dirty = 1;
648 my $progress_stats_updated = 0;
650 update_progress_stats();
655 for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++)
657 my $id = $jobstep_todo[$todo_ptr];
658 my $Jobstep = $jobstep[$id];
659 if ($Jobstep->{level} != $level)
664 pipe $reader{$id}, "writer" or croak ($!);
665 my $flags = fcntl ($reader{$id}, F_GETFL, 0) or croak ($!);
666 fcntl ($reader{$id}, F_SETFL, $flags | O_NONBLOCK) or croak ($!);
668 my $childslot = $freeslot[0];
669 my $childnode = $slot[$childslot]->{node};
670 my $childslotname = join (".",
671 $slot[$childslot]->{node}->{name},
672 $slot[$childslot]->{cpu});
673 my $childpid = fork();
676 $SIG{'INT'} = 'DEFAULT';
677 $SIG{'QUIT'} = 'DEFAULT';
678 $SIG{'TERM'} = 'DEFAULT';
680 foreach (values (%reader))
684 fcntl ("writer", F_SETFL, 0) or croak ($!); # no close-on-exec
685 open(STDOUT,">&writer");
686 open(STDERR,">&writer");
691 delete $ENV{"GNUPGHOME"};
692 $ENV{"TASK_UUID"} = $Jobstep->{'arvados_task'}->{'uuid'};
693 $ENV{"TASK_QSEQUENCE"} = $id;
694 $ENV{"TASK_SEQUENCE"} = $level;
695 $ENV{"JOB_SCRIPT"} = $Job->{script};
696 while (my ($param, $value) = each %{$Job->{script_parameters}}) {
697 $param =~ tr/a-z/A-Z/;
698 $ENV{"JOB_PARAMETER_$param"} = $value;
700 $ENV{"TASK_SLOT_NODE"} = $slot[$childslot]->{node}->{name};
701 $ENV{"TASK_SLOT_NUMBER"} = $slot[$childslot]->{cpu};
702 $ENV{"TASK_WORK"} = $ENV{"CRUNCH_TMP"}."/task/$childslotname";
703 $ENV{"HOME"} = $ENV{"TASK_WORK"};
704 $ENV{"TASK_KEEPMOUNT"} = $ENV{"TASK_WORK"}.".keep";
705 $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
706 $ENV{"CRUNCH_NODE_SLOTS"} = $slot[$childslot]->{node}->{ncpus};
707 $ENV{"PATH"} = $ENV{"CRUNCH_INSTALL"} . "/bin:" . $ENV{"PATH"};
713 "--nodelist=".$childnode->{name},
714 qw(-n1 -c1 -N1 -D), $ENV{'TMPDIR'},
715 "--job-name=$job_id.$id.$$",
718 "if [ -e $ENV{TASK_WORK} ]; then rm -rf $ENV{TASK_WORK}; fi; "
719 ."mkdir -p $ENV{CRUNCH_TMP} $ENV{JOB_WORK} $ENV{TASK_WORK} $ENV{TASK_KEEPMOUNT} "
720 ."&& cd $ENV{CRUNCH_TMP} ";
721 $command .= "&& exec arv-mount --by-id --allow-other $ENV{TASK_KEEPMOUNT} --exec ";
724 my $cidfile = "$ENV{CRUNCH_TMP}/$ENV{TASK_UUID}.cid";
725 $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -cgroup-parent=docker -cgroup-cid=$cidfile -poll=10000 ";
726 $command .= "$docker_bin run --rm=true --attach=stdout --attach=stderr --attach=stdin -i --user=crunch --cidfile=$cidfile --sig-proxy ";
728 # Dynamically configure the container to use the host system as its
729 # DNS server. Get the host's global addresses from the ip command,
730 # and turn them into docker --dns options using gawk.
732 q{$(ip -o address show scope global |
733 gawk 'match($4, /^([0-9\.:]+)\//, x){print "--dns", x[1]}') };
735 # The source tree and $destdir directory (which we have
736 # installed on the worker host) are available in the container,
737 # under the same path.
738 $command .= "--volume=\Q$ENV{CRUNCH_SRC}:$ENV{CRUNCH_SRC}:ro\E ";
739 $command .= "--volume=\Q$ENV{CRUNCH_INSTALL}:$ENV{CRUNCH_INSTALL}:ro\E ";
741 # Currently, we make arv-mount's mount point appear at /keep
742 # inside the container (instead of using the same path as the
743 # host like we do with CRUNCH_SRC and CRUNCH_INSTALL). However,
744 # crunch scripts and utilities must not rely on this. They must
745 # use $TASK_KEEPMOUNT.
746 $command .= "--volume=\Q$ENV{TASK_KEEPMOUNT}:/keep:ro\E ";
747 $ENV{TASK_KEEPMOUNT} = "/keep";
749 # TASK_WORK is almost exactly like a docker data volume: it
750 # starts out empty, is writable, and persists until no
751 # containers use it any more. We don't use --volumes-from to
752 # share it with other containers: it is only accessible to this
753 # task, and it goes away when this task stops.
755 # However, a docker data volume is writable only by root unless
756 # the mount point already happens to exist in the container with
757 # different permissions. Therefore, we [1] assume /tmp already
758 # exists in the image and is writable by the crunch user; [2]
759 # avoid putting TASK_WORK inside CRUNCH_TMP (which won't be
760 # writable if they are created by docker while setting up the
761 # other --volumes); and [3] create $TASK_WORK inside the
762 # container using $build_script.
763 $command .= "--volume=/tmp ";
764 $ENV{"TASK_WORK"} = "/tmp/crunch-job-task-work/$childslotname";
765 $ENV{"HOME"} = $ENV{"TASK_WORK"};
766 $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
768 # TODO: Share a single JOB_WORK volume across all task
769 # containers on a given worker node, and delete it when the job
770 # ends (and, in case that doesn't work, when the next job
773 # For now, use the same approach as TASK_WORK above.
774 $ENV{"JOB_WORK"} = "/tmp/crunch-job-work";
776 while (my ($env_key, $env_val) = each %ENV)
778 if ($env_key =~ /^(ARVADOS|CRUNCH|JOB|TASK)_/) {
779 $command .= "--env=\Q$env_key=$env_val\E ";
782 $command .= "--env=\QHOME=$ENV{HOME}\E ";
783 $command .= "\Q$docker_hash\E ";
784 $command .= "stdbuf --output=0 --error=0 ";
785 $command .= "perl - $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
788 $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -poll=10000 ";
789 $command .= "stdbuf --output=0 --error=0 ";
790 $command .= "perl - $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
793 my @execargs = ('bash', '-c', $command);
794 srun (\@srunargs, \@execargs, undef, $build_script);
795 # exec() failed, we assume nothing happened.
796 die "srun() failed on build script\n";
799 if (!defined $childpid)
806 $proc{$childpid} = { jobstep => $id,
809 jobstepname => "$job_id.$id.$childpid",
811 croak ("assert failed: \$slot[$childslot]->{'pid'} exists") if exists $slot[$childslot]->{pid};
812 $slot[$childslot]->{pid} = $childpid;
814 Log ($id, "job_task ".$Jobstep->{'arvados_task'}->{'uuid'});
815 Log ($id, "child $childpid started on $childslotname");
816 $Jobstep->{starttime} = time;
817 $Jobstep->{node} = $childnode->{name};
818 $Jobstep->{slotindex} = $childslot;
819 delete $Jobstep->{stderr};
820 delete $Jobstep->{finishtime};
822 $Jobstep->{'arvados_task'}->{started_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{starttime});
823 $Jobstep->{'arvados_task'}->save;
825 splice @jobstep_todo, $todo_ptr, 1;
828 $progress_is_dirty = 1;
832 (@slot > @freeslot && $todo_ptr+1 > $#jobstep_todo))
834 last THISROUND if $main::please_freeze;
835 if ($main::please_info)
837 $main::please_info = 0;
839 create_output_collection();
841 update_progress_stats();
848 check_refresh_wanted();
850 update_progress_stats();
851 select (undef, undef, undef, 0.1);
853 elsif (time - $progress_stats_updated >= 30)
855 update_progress_stats();
857 if (($thisround_failed_multiple >= 8 && $thisround_succeeded == 0) ||
858 ($thisround_failed_multiple >= 16 && $thisround_failed_multiple > $thisround_succeeded))
860 my $message = "Repeated failure rate too high ($thisround_failed_multiple/"
861 .($thisround_failed+$thisround_succeeded)
862 .") -- giving up on this round";
863 Log (undef, $message);
867 # move slots from freeslot to holdslot (or back to freeslot) if necessary
868 for (my $i=$#freeslot; $i>=0; $i--) {
869 if ($slot[$freeslot[$i]]->{node}->{hold_until} > scalar time) {
870 push @holdslot, (splice @freeslot, $i, 1);
873 for (my $i=$#holdslot; $i>=0; $i--) {
874 if ($slot[$holdslot[$i]]->{node}->{hold_until} <= scalar time) {
875 push @freeslot, (splice @holdslot, $i, 1);
879 # give up if no nodes are succeeding
880 if (!grep { $_->{node}->{losing_streak} == 0 &&
881 $_->{node}->{hold_count} < 4 } @slot) {
882 my $message = "Every node has failed -- giving up on this round";
883 Log (undef, $message);
890 push @freeslot, splice @holdslot;
891 map { $slot[$freeslot[$_]]->{node}->{losing_streak} = 0 } (0..$#freeslot);
894 Log (undef, "wait for last ".(scalar keys %proc)." children to finish");
897 if ($main::please_continue) {
898 $main::please_continue = 0;
901 $main::please_info = 0, freeze(), create_output_collection(), save_meta(1) if $main::please_info;
905 check_refresh_wanted();
907 update_progress_stats();
908 select (undef, undef, undef, 0.1);
909 killem (keys %proc) if $main::please_freeze;
913 update_progress_stats();
914 freeze_if_want_freeze();
917 if (!defined $main::success)
920 $thisround_succeeded == 0 &&
921 ($thisround_failed == 0 || $thisround_failed > 4))
923 my $message = "stop because $thisround_failed tasks failed and none succeeded";
924 Log (undef, $message);
933 goto ONELEVEL if !defined $main::success;
936 release_allocation();
938 my $collated_output = &create_output_collection();
940 if (!$collated_output) {
941 Log (undef, "Failed to write output collection");
944 Log(undef, "output hash " . $collated_output);
945 $Job->update_attributes('output' => $collated_output);
948 Log (undef, "finish");
953 if ($collated_output && $main::success) {
954 $final_state = 'Complete';
956 $final_state = 'Failed';
958 $Job->update_attributes('state' => $final_state);
960 exit (($final_state eq 'Complete') ? 0 : 1);
964 sub update_progress_stats
966 $progress_stats_updated = time;
967 return if !$progress_is_dirty;
968 my ($todo, $done, $running) = (scalar @jobstep_todo,
969 scalar @jobstep_done,
970 scalar @slot - scalar @freeslot - scalar @holdslot);
971 $Job->{'tasks_summary'} ||= {};
972 $Job->{'tasks_summary'}->{'todo'} = $todo;
973 $Job->{'tasks_summary'}->{'done'} = $done;
974 $Job->{'tasks_summary'}->{'running'} = $running;
975 $Job->update_attributes('tasks_summary' => $Job->{'tasks_summary'});
976 Log (undef, "status: $done done, $running running, $todo todo");
977 $progress_is_dirty = 0;
984 my $pid = waitpid (-1, WNOHANG);
985 return 0 if $pid <= 0;
987 my $whatslot = ($slot[$proc{$pid}->{slot}]->{node}->{name}
989 . $slot[$proc{$pid}->{slot}]->{cpu});
990 my $jobstepid = $proc{$pid}->{jobstep};
991 my $elapsed = time - $proc{$pid}->{time};
992 my $Jobstep = $jobstep[$jobstepid];
994 my $childstatus = $?;
995 my $exitvalue = $childstatus >> 8;
996 my $exitinfo = "exit ".exit_status_s($childstatus);
997 $Jobstep->{'arvados_task'}->reload;
998 my $task_success = $Jobstep->{'arvados_task'}->{success};
1000 Log ($jobstepid, "child $pid on $whatslot $exitinfo success=$task_success");
1002 if (!defined $task_success) {
1003 # task did not indicate one way or the other --> fail
1004 $Jobstep->{'arvados_task'}->{success} = 0;
1005 $Jobstep->{'arvados_task'}->save;
1012 $temporary_fail ||= $Jobstep->{node_fail};
1013 $temporary_fail ||= ($exitvalue == 111);
1015 ++$thisround_failed;
1016 ++$thisround_failed_multiple if $Jobstep->{'failures'} >= 1;
1018 # Check for signs of a failed or misconfigured node
1019 if (++$slot[$proc{$pid}->{slot}]->{node}->{losing_streak} >=
1020 2+$slot[$proc{$pid}->{slot}]->{node}->{ncpus}) {
1021 # Don't count this against jobstep failure thresholds if this
1022 # node is already suspected faulty and srun exited quickly
1023 if ($slot[$proc{$pid}->{slot}]->{node}->{hold_until} &&
1025 Log ($jobstepid, "blaming failure on suspect node " .
1026 $slot[$proc{$pid}->{slot}]->{node}->{name});
1027 $temporary_fail ||= 1;
1029 ban_node_by_slot($proc{$pid}->{slot});
1032 Log ($jobstepid, sprintf('failure (#%d, %s) after %d seconds',
1033 ++$Jobstep->{'failures'},
1034 $temporary_fail ? 'temporary ' : 'permanent',
1037 if (!$temporary_fail || $Jobstep->{'failures'} >= 3) {
1038 # Give up on this task, and the whole job
1040 $main::please_freeze = 1;
1042 # Put this task back on the todo queue
1043 push @jobstep_todo, $jobstepid;
1044 $Job->{'tasks_summary'}->{'failed'}++;
1048 ++$thisround_succeeded;
1049 $slot[$proc{$pid}->{slot}]->{node}->{losing_streak} = 0;
1050 $slot[$proc{$pid}->{slot}]->{node}->{hold_until} = 0;
1051 push @jobstep_done, $jobstepid;
1052 Log ($jobstepid, "success in $elapsed seconds");
1054 $Jobstep->{exitcode} = $childstatus;
1055 $Jobstep->{finishtime} = time;
1056 $Jobstep->{'arvados_task'}->{finished_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{finishtime});
1057 $Jobstep->{'arvados_task'}->save;
1058 process_stderr ($jobstepid, $task_success);
1059 Log ($jobstepid, "output " . $Jobstep->{'arvados_task'}->{output});
1061 close $reader{$jobstepid};
1062 delete $reader{$jobstepid};
1063 delete $slot[$proc{$pid}->{slot}]->{pid};
1064 push @freeslot, $proc{$pid}->{slot};
1067 if ($task_success) {
1069 my $newtask_list = [];
1070 my $newtask_results;
1072 $newtask_results = api_call(
1075 'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid}
1077 'order' => 'qsequence',
1078 'offset' => scalar(@$newtask_list),
1080 push(@$newtask_list, @{$newtask_results->{items}});
1081 } while (@{$newtask_results->{items}});
1082 foreach my $arvados_task (@$newtask_list) {
1084 'level' => $arvados_task->{'sequence'},
1086 'arvados_task' => $arvados_task
1088 push @jobstep, $jobstep;
1089 push @jobstep_todo, $#jobstep;
1093 $progress_is_dirty = 1;
1097 sub check_refresh_wanted
1099 my @stat = stat $ENV{"CRUNCH_REFRESH_TRIGGER"};
1100 if (@stat && $stat[9] > $latest_refresh) {
1101 $latest_refresh = scalar time;
1102 my $Job2 = api_call("jobs/get", uuid => $jobspec);
1103 for my $attr ('cancelled_at',
1104 'cancelled_by_user_uuid',
1105 'cancelled_by_client_uuid',
1107 $Job->{$attr} = $Job2->{$attr};
1109 if ($Job->{'state'} ne "Running") {
1110 if ($Job->{'state'} eq "Cancelled") {
1111 Log (undef, "Job cancelled at " . $Job->{'cancelled_at'} . " by user " . $Job->{'cancelled_by_user_uuid'});
1113 Log (undef, "Job state unexpectedly changed to " . $Job->{'state'});
1116 $main::please_freeze = 1;
1123 # return if the kill list was checked <4 seconds ago
1124 if (defined $squeue_kill_checked && $squeue_kill_checked > time - 4)
1128 $squeue_kill_checked = time;
1130 # use killem() on procs whose killtime is reached
1133 if (exists $proc{$_}->{killtime}
1134 && $proc{$_}->{killtime} <= time)
1140 # return if the squeue was checked <60 seconds ago
1141 if (defined $squeue_checked && $squeue_checked > time - 60)
1145 $squeue_checked = time;
1149 # here is an opportunity to check for mysterious problems with local procs
1153 # get a list of steps still running
1154 my @squeue = `squeue -s -h -o '%i %j' && echo ok`;
1156 if ($squeue[-1] ne "ok")
1162 # which of my jobsteps are running, according to squeue?
1166 if (/^(\d+)\.(\d+) (\S+)/)
1168 if ($1 eq $ENV{SLURM_JOBID})
1175 # which of my active child procs (>60s old) were not mentioned by squeue?
1176 foreach (keys %proc)
1178 if ($proc{$_}->{time} < time - 60
1179 && !exists $ok{$proc{$_}->{jobstepname}}
1180 && !exists $proc{$_}->{killtime})
1182 # kill this proc if it hasn't exited in 30 seconds
1183 $proc{$_}->{killtime} = time + 30;
1189 sub release_allocation
1193 Log (undef, "release job allocation");
1194 system "scancel $ENV{SLURM_JOBID}";
1202 foreach my $job (keys %reader)
1205 while (0 < sysread ($reader{$job}, $buf, 8192))
1207 print STDERR $buf if $ENV{CRUNCH_DEBUG};
1208 $jobstep[$job]->{stderr} .= $buf;
1209 preprocess_stderr ($job);
1210 if (length ($jobstep[$job]->{stderr}) > 16384)
1212 substr ($jobstep[$job]->{stderr}, 0, 8192) = "";
1221 sub preprocess_stderr
1225 while ($jobstep[$job]->{stderr} =~ /^(.*?)\n/) {
1227 substr $jobstep[$job]->{stderr}, 0, 1+length($line), "";
1228 Log ($job, "stderr $line");
1229 if ($line =~ /srun: error: (SLURM job $ENV{SLURM_JOB_ID} has expired|Unable to confirm allocation for job $ENV{SLURM_JOB_ID})/) {
1231 $main::please_freeze = 1;
1233 elsif ($line =~ /srun: error: (Node failure on|Unable to create job step) /) {
1234 $jobstep[$job]->{node_fail} = 1;
1235 ban_node_by_slot($jobstep[$job]->{slotindex});
1244 my $task_success = shift;
1245 preprocess_stderr ($job);
1248 Log ($job, "stderr $_");
1249 } split ("\n", $jobstep[$job]->{stderr});
1256 if (!open($keep, "-|", "arv-get", "--retries", retry_count(), $hash)) {
1257 Log(undef, "fetch_block run error from arv-get $hash: $!");
1260 my $output_block = "";
1263 my $bytes = sysread($keep, $buf, 1024 * 1024);
1264 if (!defined $bytes) {
1265 Log(undef, "fetch_block read error from arv-get: $!");
1266 $output_block = undef;
1268 } elsif ($bytes == 0) {
1269 # sysread returns 0 at the end of the pipe.
1272 # some bytes were read into buf.
1273 $output_block .= $buf;
1278 Log(undef, "fetch_block arv-get exited " . exit_status_s($?));
1279 $output_block = undef;
1281 return $output_block;
1284 # Create a collection by concatenating the output of all tasks (each
1285 # task's output is either a manifest fragment, a locator for a
1286 # manifest fragment stored in Keep, or nothing at all). Return the
1287 # portable_data_hash of the new collection.
1288 sub create_output_collection
1290 Log (undef, "collate");
1292 my ($child_out, $child_in);
1293 my $pid = open2($child_out, $child_in, 'python', '-c', q{
1296 print (arvados.api("v1").collections().
1297 create(body={"manifest_text": sys.stdin.read()}).
1298 execute(num_retries=int(sys.argv[1]))["portable_data_hash"])
1305 my $output = $_->{'arvados_task'}->{output};
1306 next if (!defined($output));
1308 if ($output =~ /^[0-9a-f]{32}(\+\S+)*$/) {
1309 $next_write = fetch_block($output);
1311 $next_write = $output;
1313 if (defined($next_write)) {
1314 if (!defined(syswrite($child_in, $next_write))) {
1315 # There's been an error writing. Stop the loop.
1316 # We'll log details about the exit code later.
1320 my $uuid = $_->{'arvados_task'}->{'uuid'};
1321 Log (undef, "Error retrieving '$output' output by task $task_idx ($uuid)");
1328 my $s = IO::Select->new($child_out);
1329 if ($s->can_read(120)) {
1330 sysread($child_out, $joboutput, 64 * 1024 * 1024);
1333 Log(undef, "output collection creation exited " . exit_status_s($?));
1339 Log (undef, "timed out while creating output collection");
1340 foreach my $signal (2, 2, 2, 15, 15, 9) {
1341 kill($signal, $pid);
1342 last if waitpid($pid, WNOHANG) == -1;
1356 my $sig = 2; # SIGINT first
1357 if (exists $proc{$_}->{"sent_$sig"} &&
1358 time - $proc{$_}->{"sent_$sig"} > 4)
1360 $sig = 15; # SIGTERM if SIGINT doesn't work
1362 if (exists $proc{$_}->{"sent_$sig"} &&
1363 time - $proc{$_}->{"sent_$sig"} > 4)
1365 $sig = 9; # SIGKILL if SIGTERM doesn't work
1367 if (!exists $proc{$_}->{"sent_$sig"})
1369 Log ($proc{$_}->{jobstep}, "sending 2x signal $sig to pid $_");
1371 select (undef, undef, undef, 0.1);
1374 kill $sig, $_; # srun wants two SIGINT to really interrupt
1376 $proc{$_}->{"sent_$sig"} = time;
1377 $proc{$_}->{"killedafter"} = time - $proc{$_}->{"time"};
1387 vec($bits,fileno($_),1) = 1;
1393 # Send log output to Keep via arv-put.
1395 # $log_pipe_in and $log_pipe_out are the input and output filehandles to the arv-put pipe.
1396 # $log_pipe_pid is the pid of the arv-put subprocess.
1398 # The only functions that should access these variables directly are:
1400 # log_writer_start($logfilename)
1401 # Starts an arv-put pipe, reading data on stdin and writing it to
1402 # a $logfilename file in an output collection.
1404 # log_writer_send($txt)
1405 # Writes $txt to the output log collection.
1407 # log_writer_finish()
1408 # Closes the arv-put pipe and returns the output that it produces.
1410 # log_writer_is_active()
1411 # Returns a true value if there is currently a live arv-put
1412 # process, false otherwise.
1414 my ($log_pipe_in, $log_pipe_out, $log_pipe_pid);
1416 sub log_writer_start($)
1418 my $logfilename = shift;
1419 $log_pipe_pid = open2($log_pipe_out, $log_pipe_in,
1421 '--portable-data-hash',
1422 '--project-uuid', $Job->{owner_uuid},
1424 '--name', $logfilename,
1425 '--filename', $logfilename,
1429 sub log_writer_send($)
1432 print $log_pipe_in $txt;
1435 sub log_writer_finish()
1437 return unless $log_pipe_pid;
1439 close($log_pipe_in);
1442 my $s = IO::Select->new($log_pipe_out);
1443 if ($s->can_read(120)) {
1444 sysread($log_pipe_out, $arv_put_output, 1024);
1445 chomp($arv_put_output);
1447 Log (undef, "timed out reading from 'arv-put'");
1450 waitpid($log_pipe_pid, 0);
1451 $log_pipe_pid = $log_pipe_in = $log_pipe_out = undef;
1453 Log("log_writer_finish: arv-put exited ".exit_status_s($?))
1456 return $arv_put_output;
1459 sub log_writer_is_active() {
1460 return $log_pipe_pid;
1463 sub Log # ($jobstep_id, $logmessage)
1465 if ($_[1] =~ /\n/) {
1466 for my $line (split (/\n/, $_[1])) {
1471 my $fh = select STDERR; $|=1; select $fh;
1472 my $message = sprintf ("%s %d %s %s", $job_id, $$, @_);
1473 $message =~ s{([^ -\176])}{"\\" . sprintf ("%03o", ord($1))}ge;
1476 if (log_writer_is_active() || -t STDERR) {
1477 my @gmtime = gmtime;
1478 $datetime = sprintf ("%04d-%02d-%02d_%02d:%02d:%02d",
1479 $gmtime[5]+1900, $gmtime[4]+1, @gmtime[3,2,1,0]);
1481 print STDERR ((-t STDERR) ? ($datetime." ".$message) : $message);
1483 if (log_writer_is_active()) {
1484 log_writer_send($datetime . " " . $message);
1491 my ($package, $file, $line) = caller;
1492 my $message = "@_ at $file line $line\n";
1493 Log (undef, $message);
1494 freeze() if @jobstep_todo;
1495 create_output_collection() if @jobstep_todo;
1505 if ($Job->{'state'} eq 'Cancelled') {
1506 $Job->update_attributes('finished_at' => scalar gmtime);
1508 $Job->update_attributes('state' => 'Failed');
1515 my $justcheckpoint = shift; # false if this will be the last meta saved
1516 return if $justcheckpoint; # checkpointing is not relevant post-Warehouse.pm
1517 return unless log_writer_is_active();
1519 my $loglocator = log_writer_finish();
1520 Log (undef, "log manifest is $loglocator");
1521 $Job->{'log'} = $loglocator;
1522 $Job->update_attributes('log', $loglocator);
1526 sub freeze_if_want_freeze
1528 if ($main::please_freeze)
1530 release_allocation();
1533 # kill some srun procs before freeze+stop
1534 map { $proc{$_} = {} } @_;
1537 killem (keys %proc);
1538 select (undef, undef, undef, 0.1);
1540 while (($died = waitpid (-1, WNOHANG)) > 0)
1542 delete $proc{$died};
1547 create_output_collection();
1557 Log (undef, "Freeze not implemented");
1564 croak ("Thaw not implemented");
1580 $s =~ s{\\(.)}{$1 eq "n" ? "\n" : $1}ge;
1587 my $srunargs = shift;
1588 my $execargs = shift;
1589 my $opts = shift || {};
1591 my $args = $have_slurm ? [@$srunargs, @$execargs] : $execargs;
1593 $Data::Dumper::Terse = 1;
1594 $Data::Dumper::Indent = 0;
1595 my $show_cmd = Dumper($args);
1596 $show_cmd =~ s/(TOKEN\\*=)[^\s\']+/${1}[...]/g;
1597 $show_cmd =~ s/\n/ /g;
1598 warn "starting: $show_cmd\n";
1600 if (defined $stdin) {
1601 my $child = open STDIN, "-|";
1602 defined $child or die "no fork: $!";
1604 print $stdin or die $!;
1605 close STDOUT or die $!;
1610 return system (@$args) if $opts->{fork};
1613 warn "ENV size is ".length(join(" ",%ENV));
1614 die "exec failed: $!: @$args";
1618 sub ban_node_by_slot {
1619 # Don't start any new jobsteps on this node for 60 seconds
1621 $slot[$slotid]->{node}->{hold_until} = 60 + scalar time;
1622 $slot[$slotid]->{node}->{hold_count}++;
1623 Log (undef, "backing off node " . $slot[$slotid]->{node}->{name} . " for 60 seconds");
1628 my ($lockfile, $error_message) = @_;
1629 open L, ">", $lockfile or croak("$lockfile: $!");
1630 if (!flock L, LOCK_EX|LOCK_NB) {
1631 croak("Can't lock $lockfile: $error_message\n");
1635 sub find_docker_image {
1636 # Given a Keep locator, check to see if it contains a Docker image.
1637 # If so, return its stream name and Docker hash.
1638 # If not, return undef for both values.
1639 my $locator = shift;
1640 my ($streamname, $filename);
1641 my $image = api_call("collections/get", uuid => $locator);
1643 foreach my $line (split(/\n/, $image->{manifest_text})) {
1644 my @tokens = split(/\s+/, $line);
1646 $streamname = shift(@tokens);
1647 foreach my $filedata (grep(/^\d+:\d+:/, @tokens)) {
1648 if (defined($filename)) {
1649 return (undef, undef); # More than one file in the Collection.
1651 $filename = (split(/:/, $filedata, 3))[2];
1656 if (defined($filename) and ($filename =~ /^([0-9A-Fa-f]{64})\.tar$/)) {
1657 return ($streamname, $1);
1659 return (undef, undef);
1664 # Calculate the number of times an operation should be retried,
1665 # assuming exponential backoff, and that we're willing to retry as
1666 # long as tasks have been running. Enforce a minimum of 3 retries.
1667 my ($starttime, $endtime, $timediff, $retries);
1669 $starttime = $jobstep[0]->{starttime};
1670 $endtime = $jobstep[-1]->{finishtime};
1672 if (!defined($starttime)) {
1674 } elsif (!defined($endtime)) {
1675 $timediff = time - $starttime;
1677 $timediff = ($endtime - $starttime) - (time - $endtime);
1679 if ($timediff > 0) {
1680 $retries = int(log($timediff) / log(2));
1682 $retries = 1; # Use the minimum.
1684 return ($retries > 3) ? $retries : 3;
1688 # Pass in two function references.
1689 # This method will be called with the remaining arguments.
1690 # If it dies, retry it with exponential backoff until it succeeds,
1691 # or until the current retry_count is exhausted. After each failure
1692 # that can be retried, the second function will be called with
1693 # the current try count (0-based), next try time, and error message.
1694 my $operation = shift;
1695 my $retry_callback = shift;
1696 my $retries = retry_count();
1697 foreach my $try_count (0..$retries) {
1698 my $next_try = time + (2 ** $try_count);
1699 my $result = eval { $operation->(@_); };
1702 } elsif ($try_count < $retries) {
1703 $retry_callback->($try_count, $next_try, $@);
1704 my $sleep_time = $next_try - time;
1705 sleep($sleep_time) if ($sleep_time > 0);
1708 # Ensure the error message ends in a newline, so Perl doesn't add
1709 # retry_op's line number to it.
1715 # Pass in a /-separated API method name, and arguments for it.
1716 # This function will call that method, retrying as needed until
1717 # the current retry_count is exhausted, with a log on the first failure.
1718 my $method_name = shift;
1719 my $log_api_retry = sub {
1720 my ($try_count, $next_try_at, $errmsg) = @_;
1721 $errmsg =~ s/\s*\bat \Q$0\E line \d+\.?\s*//;
1722 $errmsg =~ s/\s/ /g;
1723 $errmsg =~ s/\s+$//;
1725 if ($next_try_at < time) {
1726 $retry_msg = "Retrying.";
1728 my $next_try_fmt = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($next_try_at);
1729 $retry_msg = "Retrying at $next_try_fmt.";
1731 Log(undef, "API method $method_name failed: $errmsg. $retry_msg");
1734 foreach my $key (split(/\//, $method_name)) {
1735 $method = $method->{$key};
1737 return retry_op(sub { $method->execute(@_); }, $log_api_retry, @_);
1741 # Given a $?, return a human-readable exit code string like "0" or
1742 # "1" or "0 with signal 1" or "1 with signal 11".
1743 my $exitcode = shift;
1744 my $s = $exitcode >> 8;
1745 if ($exitcode & 0x7f) {
1746 $s .= " with signal " . ($exitcode & 0x7f);
1748 if ($exitcode & 0x80) {
1749 $s .= " with core dump";
1754 sub handle_readall {
1755 # Pass in a glob reference to a file handle.
1756 # Read all its contents and return them as a string.
1757 my $fh_glob_ref = shift;
1759 return <$fh_glob_ref>;
1762 sub tar_filename_n {
1764 return sprintf("%s/git.%s.%d.tar", $ENV{CRUNCH_TMP}, $job_id, $n);
1767 sub add_git_archive {
1768 # Pass in a git archive command as a string or list, a la system().
1769 # This method will save its output to be included in the archive sent to the
1773 if (!open(GIT_ARCHIVE, ">", tar_filename_n($git_tar_count))) {
1774 croak("Failed to save git archive: $!");
1776 my $git_pid = open2(">&GIT_ARCHIVE", $git_input, @_);
1778 waitpid($git_pid, 0);
1781 croak("Failed to save git archive: git exited " . exit_status_s($?));
1785 sub combined_git_archive {
1786 # Combine all saved tar archives into a single archive, then return its
1787 # contents in a string. Return undef if no archives have been saved.
1788 if ($git_tar_count < 1) {
1791 my $base_tar_name = tar_filename_n(1);
1792 foreach my $tar_to_append (map { tar_filename_n($_); } (2..$git_tar_count)) {
1793 my $tar_exit = system("tar", "-Af", $base_tar_name, $tar_to_append);
1794 if ($tar_exit != 0) {
1795 croak("Error preparing build archive: tar -A exited " .
1796 exit_status_s($tar_exit));
1799 if (!open(GIT_TAR, "<", $base_tar_name)) {
1800 croak("Could not open build archive: $!");
1802 my $tar_contents = handle_readall(\*GIT_TAR);
1804 return $tar_contents;
1810 # This is crunch-job's internal dispatch script. crunch-job running on the API
1811 # server invokes this script on individual compute nodes, or localhost if we're
1812 # running a job locally. It gets called in two modes:
1814 # * No arguments: Installation mode. Read a tar archive from the DATA
1815 # file handle; it includes the Crunch script's source code, and
1816 # maybe SDKs as well. Those should be installed in the proper
1817 # locations. This runs outside of any Docker container, so don't try to
1818 # introspect Crunch's runtime environment.
1820 # * With arguments: Crunch script run mode. This script should set up the
1821 # environment, then run the command specified in the arguments. This runs
1822 # inside any Docker container.
1825 use File::Path qw( make_path remove_tree );
1826 use POSIX qw(getcwd);
1828 # Map SDK subdirectories to the path environments they belong to.
1829 my %SDK_ENVVARS = ("perl/lib" => "PERLLIB", "ruby/lib" => "RUBYLIB");
1831 my $destdir = $ENV{"CRUNCH_SRC"};
1832 my $commit = $ENV{"CRUNCH_SRC_COMMIT"};
1833 my $repo = $ENV{"CRUNCH_SRC_URL"};
1834 my $install_dir = $ENV{"CRUNCH_INSTALL"} || (getcwd() . "/opt");
1835 my $job_work = $ENV{"JOB_WORK"};
1836 my $task_work = $ENV{"TASK_WORK"};
1838 for my $dir ($destdir, $job_work, $task_work) {
1841 -e $dir or die "Failed to create temporary directory ($dir): $!";
1846 remove_tree($task_work, {keep_root => 1});
1849 open(STDOUT_ORIG, ">&", STDOUT);
1850 open(STDERR_ORIG, ">&", STDERR);
1851 open(STDOUT, ">>", "$destdir.log");
1852 open(STDERR, ">&", STDOUT);
1854 ### Crunch script run mode
1856 # We want to do routine logging during task 0 only. This gives the user
1857 # the information they need, but avoids repeating the information for every
1860 if ($ENV{TASK_SEQUENCE} eq "0") {
1863 printf STDERR_ORIG "[Crunch] $msg\n", @_;
1869 my $python_src = "$install_dir/python";
1870 my $venv_dir = "$job_work/.arvados.venv";
1871 my $venv_built = -e "$venv_dir/bin/activate";
1872 if ((!$venv_built) and (-d $python_src) and can_run("virtualenv")) {
1873 shell_or_die("virtualenv", "--quiet", "--system-site-packages",
1874 "--python=python2.7", $venv_dir);
1875 shell_or_die("$venv_dir/bin/pip", "--quiet", "install", "-I", $python_src);
1877 $Log->("Built Python SDK virtualenv");
1880 my $pip_bin = "pip";
1882 $Log->("Running in Python SDK virtualenv");
1883 $pip_bin = "$venv_dir/bin/pip";
1884 my $orig_argv = join(" ", map { quotemeta($_); } @ARGV);
1885 @ARGV = ("/bin/sh", "-ec",
1886 ". \Q$venv_dir/bin/activate\E; exec $orig_argv");
1887 } elsif (-d $python_src) {
1888 $Log->("Warning: virtualenv not found inside Docker container default " .
1889 "\$PATH. Can't install Python SDK.");
1892 my $pkgs = `(\Q$pip_bin\E freeze 2>/dev/null | grep arvados) || dpkg-query --show '*arvados*'`;
1894 $Log->("Using Arvados SDK:");
1895 foreach my $line (split /\n/, $pkgs) {
1899 $Log->("Arvados SDK packages not found");
1902 while (my ($sdk_dir, $sdk_envkey) = each(%SDK_ENVVARS)) {
1903 my $sdk_path = "$install_dir/$sdk_dir";
1905 if ($ENV{$sdk_envkey}) {
1906 $ENV{$sdk_envkey} = "$sdk_path:" . $ENV{$sdk_envkey};
1908 $ENV{$sdk_envkey} = $sdk_path;
1910 $Log->("Arvados SDK added to %s", $sdk_envkey);
1916 open(STDOUT, ">&", STDOUT_ORIG);
1917 open(STDERR, ">&", STDERR_ORIG);
1919 die "Cannot exec `@ARGV`: $!";
1922 ### Installation mode
1923 open L, ">", "$destdir.lock" or die "$destdir.lock: $!";
1925 if (readlink ("$destdir.commit") eq $commit && -d $destdir) {
1926 # This version already installed -> nothing to do.
1930 unlink "$destdir.commit";
1933 if (!open(TARX, "|-", "tar", "-xC", $destdir)) {
1934 die "Error launching 'tar -xC $destdir': $!";
1936 # If we send too much data to tar in one write (> 4-5 MiB), it stops, and we
1937 # get SIGPIPE. We must feed it data incrementally.
1939 while (read(DATA, $tar_input, 65536)) {
1940 print TARX $tar_input;
1943 die "'tar -xC $destdir' exited $?: $!";
1948 my $sdk_root = "$destdir/.arvados.sdk/sdk";
1950 foreach my $sdk_lang (("python",
1951 map { (split /\//, $_, 2)[0]; } keys(%SDK_ENVVARS))) {
1952 if (-d "$sdk_root/$sdk_lang") {
1953 if (!rename("$sdk_root/$sdk_lang", "$install_dir/$sdk_lang")) {
1954 die "Failed to install $sdk_lang SDK: $!";
1960 my $python_dir = "$install_dir/python";
1961 if ((-d $python_dir) and can_run("python2.7") and
1962 (system("python2.7", "$python_dir/setup.py", "--quiet", "egg_info") != 0)) {
1963 # egg_info failed, probably when it asked git for a build tag.
1964 # Specify no build tag.
1965 open(my $pysdk_cfg, ">>", "$python_dir/setup.cfg");
1966 print $pysdk_cfg "\n[egg_info]\ntag_build =\n";
1970 if (-e "$destdir/crunch_scripts/install") {
1971 shell_or_die ("$destdir/crunch_scripts/install", $install_dir);
1972 } elsif (!-e "./install.sh" && -e "./tests/autotests.sh") {
1974 shell_or_die ("./tests/autotests.sh", $install_dir);
1975 } elsif (-e "./install.sh") {
1976 shell_or_die ("./install.sh", $install_dir);
1980 unlink "$destdir.commit.new";
1981 symlink ($commit, "$destdir.commit.new") or die "$destdir.commit.new: $!";
1982 rename ("$destdir.commit.new", "$destdir.commit") or die "$destdir.commit: $!";
1988 my $command_name = shift;
1989 open(my $which, "-|", "which", $command_name);
1990 while (<$which>) { }
1997 if ($ENV{"DEBUG"}) {
1998 print STDERR "@_\n";
2000 if (system (@_) != 0) {
2002 my $exitstatus = sprintf("exit %d signal %d", $? >> 8, $? & 0x7f);
2003 open STDERR, ">&STDERR_ORIG";
2004 system ("cat $destdir.log >&2");
2005 die "@_ failed ($err): $exitstatus";