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 $cgroup_root = "/sys/fs/cgroup";
130 my $docker_bin = "docker.io";
131 my $docker_run_args = "";
132 GetOptions('force-unlock' => \$force_unlock,
133 'git-dir=s' => \$git_dir,
134 'job=s' => \$jobspec,
135 'job-api-token=s' => \$job_api_token,
136 'no-clear-tmp' => \$no_clear_tmp,
137 'resume-stash=s' => \$resume_stash,
138 'cgroup-root=s' => \$cgroup_root,
139 'docker-bin=s' => \$docker_bin,
140 'docker-run-args=s' => \$docker_run_args,
143 if (defined $job_api_token) {
144 $ENV{ARVADOS_API_TOKEN} = $job_api_token;
147 my $have_slurm = exists $ENV{SLURM_JOB_ID} && exists $ENV{SLURM_NODELIST};
152 $main::ENV{CRUNCH_DEBUG} = 1;
156 $main::ENV{CRUNCH_DEBUG} = 0;
159 my $arv = Arvados->new('apiVersion' => 'v1');
168 if ($jobspec =~ /^[-a-z\d]+$/)
170 # $jobspec is an Arvados UUID, not a JSON job specification
171 $Job = api_call("jobs/get", uuid => $jobspec);
176 $local_job = JSON::decode_json($jobspec);
180 # Make sure our workers (our slurm nodes, localhost, or whatever) are
181 # at least able to run basic commands: they aren't down or severely
184 if (($Job || $local_job)->{docker_image_locator}) {
185 $cmd = [$docker_bin, 'ps', '-q'];
187 Log(undef, "Sanity check is `@$cmd`");
188 my ($exited, $stdout, $stderr) = srun_sync(
189 ["srun", "--nodes=\Q$ENV{SLURM_NNODES}\E", "--ntasks-per-node=1"],
191 {label => "sanity check"});
193 Log(undef, "Sanity check failed: ".exit_status_s($exited));
196 Log(undef, "Sanity check OK");
199 my $User = api_call("users/current");
202 if (!$force_unlock) {
203 # Claim this job, and make sure nobody else does
204 eval { api_call("jobs/lock", uuid => $Job->{uuid}); };
206 Log(undef, "Error while locking job, exiting ".EX_TEMPFAIL);
215 map { croak ("No $_ specified") unless $local_job->{$_} }
216 qw(script script_version script_parameters);
219 $local_job->{'is_locked_by_uuid'} = $User->{'uuid'};
220 $local_job->{'started_at'} = gmtime;
221 $local_job->{'state'} = 'Running';
223 $Job = api_call("jobs/create", job => $local_job);
225 $job_id = $Job->{'uuid'};
227 my $keep_logfile = $job_id . '.log.txt';
228 log_writer_start($keep_logfile);
230 $Job->{'runtime_constraints'} ||= {};
231 $Job->{'runtime_constraints'}->{'max_tasks_per_node'} ||= 0;
232 my $max_ncpus = $Job->{'runtime_constraints'}->{'max_tasks_per_node'};
234 my $gem_versions = `gem list --quiet arvados-cli 2>/dev/null`;
236 $gem_versions =~ s/^arvados-cli \(/ with arvados-cli Gem version(s) /;
237 chomp($gem_versions);
238 chop($gem_versions); # Closing parentheses
243 "running from " . ((-e $0) ? realpath($0) : "stdin") . $gem_versions);
245 Log (undef, "check slurm allocation");
248 # Should use $ENV{SLURM_TASKS_PER_NODE} instead of sinfo? (eg. "4(x3),2,4(x2)")
252 my $localcpus = 0 + `grep -cw ^processor /proc/cpuinfo` || 1;
253 push @sinfo, "$localcpus localhost";
255 if (exists $ENV{SLURM_NODELIST})
257 push @sinfo, `sinfo -h --format='%c %N' --nodes=\Q$ENV{SLURM_NODELIST}\E`;
261 my ($ncpus, $slurm_nodelist) = split;
262 $ncpus = $max_ncpus if $max_ncpus && $ncpus > $max_ncpus;
265 while ($slurm_nodelist =~ s/^([^\[,]+?(\[.*?\])?)(,|$)//)
268 if ($nodelist =~ /\[((\d+)(-(\d+))?(,(\d+)(-(\d+))?)*)\]/)
271 foreach (split (",", $ranges))
284 push @nodelist, map {
286 $n =~ s/\[[-,\d]+\]/$_/;
293 push @nodelist, $nodelist;
296 foreach my $nodename (@nodelist)
298 Log (undef, "node $nodename - $ncpus slots");
299 my $node = { name => $nodename,
301 # The number of consecutive times a task has been dispatched
302 # to this node and failed.
304 # The number of consecutive times that SLURM has reported
305 # a node failure since the last successful task.
307 # Don't dispatch work to this node until this time
308 # (in seconds since the epoch) has passed.
310 foreach my $cpu (1..$ncpus)
312 push @slot, { node => $node,
316 push @node, @nodelist;
321 # Ensure that we get one jobstep running on each allocated node before
322 # we start overloading nodes with concurrent steps
324 @slot = sort { $a->{cpu} <=> $b->{cpu} } @slot;
327 $Job->update_attributes(
328 'tasks_summary' => { 'failed' => 0,
333 Log (undef, "start");
334 $SIG{'INT'} = sub { $main::please_freeze = 1; };
335 $SIG{'QUIT'} = sub { $main::please_freeze = 1; };
336 $SIG{'TERM'} = \&croak;
337 $SIG{'TSTP'} = sub { $main::please_freeze = 1; };
338 $SIG{'ALRM'} = sub { $main::please_info = 1; };
339 $SIG{'CONT'} = sub { $main::please_continue = 1; };
340 $SIG{'HUP'} = sub { $main::please_refresh = 1; };
342 $main::please_freeze = 0;
343 $main::please_info = 0;
344 $main::please_continue = 0;
345 $main::please_refresh = 0;
346 my $jobsteps_must_output_keys = 0; # becomes 1 when any task outputs a key
348 grep { $ENV{$1} = $2 if /^(NOCACHE.*?)=(.*)/ } split ("\n", $$Job{knobs});
349 $ENV{"CRUNCH_JOB_UUID"} = $job_id;
350 $ENV{"JOB_UUID"} = $job_id;
353 my @jobstep_todo = ();
354 my @jobstep_done = ();
355 my @jobstep_tomerge = ();
356 my $jobstep_tomerge_level = 0;
357 my $squeue_checked = 0;
358 my $sinfo_checked = 0;
359 my $latest_refresh = scalar time;
363 if (defined $Job->{thawedfromkey})
365 thaw ($Job->{thawedfromkey});
369 my $first_task = api_call("job_tasks/create", job_task => {
370 'job_uuid' => $Job->{'uuid'},
375 push @jobstep, { 'level' => 0,
377 'arvados_task' => $first_task,
379 push @jobstep_todo, 0;
385 must_lock_now("$ENV{CRUNCH_TMP}/.lock", "a job is already running here.");
388 my $build_script = handle_readall(\*DATA);
389 my $nodelist = join(",", @node);
390 my $git_tar_count = 0;
392 if (!defined $no_clear_tmp) {
393 # Find FUSE mounts under $CRUNCH_TMP and unmount them. Then clean
394 # up work directories crunch_tmp/work, crunch_tmp/opt,
397 # TODO: When #5036 is done and widely deployed, we can limit mount's
398 # -t option to simply fuse.keep.
399 my ($exited, $stdout, $stderr) = srun_sync(
400 ["srun", "--nodelist=$nodelist", "-D", $ENV{'TMPDIR'}],
401 ['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'],
402 {label => "clean work dirs"});
404 exit(EX_RETRY_UNLOCKED);
408 # If this job requires a Docker image, install that.
409 my ($docker_locator, $docker_stream, $docker_hash, $docker_limitmem, $dockeruserarg);
410 if ($docker_locator = $Job->{docker_image_locator}) {
411 Log (undef, "Install docker image $docker_locator");
412 ($docker_stream, $docker_hash) = find_docker_image($docker_locator);
415 croak("No Docker image hash found from locator $docker_locator");
417 Log (undef, "docker image hash is $docker_hash");
418 $docker_stream =~ s/^\.//;
419 my $docker_install_script = qq{
420 if $docker_bin images -q --no-trunc --all | grep -xF \Q$docker_hash\E >/dev/null; then
423 declare -a exit_codes=("\${PIPESTATUS[@]}")
424 if [ 0 != "\${exit_codes[0]}" ]; then
425 exit "\${exit_codes[0]}" # `docker images` failed
426 elif [ 1 != "\${exit_codes[1]}" ]; then
427 exit "\${exit_codes[1]}" # `grep` encountered an error
429 # Everything worked fine, but grep didn't find the image on this host.
430 arv-get \Q$docker_locator$docker_stream/$docker_hash.tar\E | $docker_bin load
434 my ($exited, $stdout, $stderr) = srun_sync(
435 ["srun", "--nodelist=" . join(',', @node)],
436 ["/bin/bash", "-o", "pipefail", "-ec", $docker_install_script],
437 {label => "load docker image"});
440 exit(EX_RETRY_UNLOCKED);
443 # Determine whether this version of Docker supports memory+swap limits.
444 ($exited, $stdout, $stderr) = srun_sync(
445 ["srun", "--nodes=1"],
446 [$docker_bin, 'run', '--help'],
447 {label => "check --memory-swap feature"});
448 $docker_limitmem = ($stdout =~ /--memory-swap/);
450 # Find a non-root Docker user to use.
451 # Tries the default user for the container, then 'crunch', then 'nobody',
452 # testing for whether the actual user id is non-zero. This defends against
453 # mistakes but not malice, but we intend to harden the security in the future
454 # so we don't want anyone getting used to their jobs running as root in their
456 my @tryusers = ("", "crunch", "nobody");
457 foreach my $try_user (@tryusers) {
460 if ($try_user eq "") {
461 $label = "check whether default user is UID 0";
464 $label = "check whether user '$try_user' is UID 0";
465 $try_user_arg = "--user=$try_user";
467 my ($exited, $stdout, $stderr) = srun_sync(
468 ["srun", "--nodes=1"],
470 "$docker_bin run $docker_run_args $try_user_arg $docker_hash id --user"],
473 if ($exited == 0 && $stdout =~ /^\d+$/ && $stdout > 0) {
474 $dockeruserarg = $try_user_arg;
475 if ($try_user eq "") {
476 Log(undef, "Container will run with default user");
478 Log(undef, "Container will run with $dockeruserarg");
484 if (!defined $dockeruserarg) {
485 croak("Could not find a user in container that is not UID 0 (tried default user, @tryusers) or there was a problem running 'id' in the container.");
488 if ($Job->{arvados_sdk_version}) {
489 # The job also specifies an Arvados SDK version. Add the SDKs to the
490 # tar file for the build script to install.
491 Log(undef, sprintf("Packing Arvados SDK version %s for installation",
492 $Job->{arvados_sdk_version}));
493 add_git_archive("git", "--git-dir=$git_dir", "archive",
494 "--prefix=.arvados.sdk/",
495 $Job->{arvados_sdk_version}, "sdk");
499 if (!defined $git_dir && $Job->{'script_version'} =~ m{^/}) {
500 # If script_version looks like an absolute path, *and* the --git-dir
501 # argument was not given -- which implies we were not invoked by
502 # crunch-dispatch -- we will use the given path as a working
503 # directory instead of resolving script_version to a git commit (or
504 # doing anything else with git).
505 $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{'script_version'};
506 $ENV{"CRUNCH_SRC"} = $Job->{'script_version'};
509 # Resolve the given script_version to a git commit sha1. Also, if
510 # the repository is remote, clone it into our local filesystem: this
511 # ensures "git archive" will work, and is necessary to reliably
512 # resolve a symbolic script_version like "master^".
513 $ENV{"CRUNCH_SRC"} = "$ENV{CRUNCH_TMP}/src";
515 Log (undef, "Looking for version ".$Job->{script_version}." from repository ".$Job->{repository});
517 $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{script_version};
519 # If we're running under crunch-dispatch, it will have already
520 # pulled the appropriate source tree into its own repository, and
521 # given us that repo's path as $git_dir.
523 # If we're running a "local" job, we might have to fetch content
524 # from a remote repository.
526 # (Currently crunch-dispatch gives a local path with --git-dir, but
527 # we might as well accept URLs there too in case it changes its
529 my $repo = $git_dir || $Job->{'repository'};
531 # Repository can be remote or local. If remote, we'll need to fetch it
532 # to a local dir before doing `git log` et al.
535 if ($repo =~ m{://|^[^/]*:}) {
536 # $repo is a git url we can clone, like git:// or https:// or
537 # file:/// or [user@]host:repo.git. Note "user/name@host:foo" is
538 # not recognized here because distinguishing that from a local
539 # path is too fragile. If you really need something strange here,
540 # use the ssh:// form.
541 $repo_location = 'remote';
542 } elsif ($repo =~ m{^\.*/}) {
543 # $repo is a local path to a git index. We'll also resolve ../foo
544 # to ../foo/.git if the latter is a directory. To help
545 # disambiguate local paths from named hosted repositories, this
546 # form must be given as ./ or ../ if it's a relative path.
547 if (-d "$repo/.git") {
548 $repo = "$repo/.git";
550 $repo_location = 'local';
552 # $repo is none of the above. It must be the name of a hosted
554 my $arv_repo_list = api_call("repositories/list",
555 'filters' => [['name','=',$repo]]);
556 my @repos_found = @{$arv_repo_list->{'items'}};
557 my $n_found = $arv_repo_list->{'serverResponse'}->{'items_available'};
559 Log(undef, "Repository '$repo' -> "
560 . join(", ", map { $_->{'uuid'} } @repos_found));
563 croak("Error: Found $n_found repositories with name '$repo'.");
565 $repo = $repos_found[0]->{'fetch_url'};
566 $repo_location = 'remote';
568 Log(undef, "Using $repo_location repository '$repo'");
569 $ENV{"CRUNCH_SRC_URL"} = $repo;
571 # Resolve given script_version (we'll call that $treeish here) to a
572 # commit sha1 ($commit).
573 my $treeish = $Job->{'script_version'};
575 if ($repo_location eq 'remote') {
576 # We minimize excess object-fetching by re-using the same bare
577 # repository in CRUNCH_TMP/.git for multiple crunch-jobs -- we
578 # just keep adding remotes to it as needed.
579 my $local_repo = $ENV{'CRUNCH_TMP'}."/.git";
580 my $gitcmd = "git --git-dir=\Q$local_repo\E";
582 # Set up our local repo for caching remote objects, making
584 if (!-d $local_repo) {
585 make_path($local_repo) or croak("Error: could not create $local_repo");
587 # This works (exits 0 and doesn't delete fetched objects) even
588 # if $local_repo is already initialized:
589 `$gitcmd init --bare`;
591 croak("Error: $gitcmd init --bare exited ".exit_status_s($?));
594 # If $treeish looks like a hash (or abbrev hash) we look it up in
595 # our local cache first, since that's cheaper. (We don't want to
596 # do that with tags/branches though -- those change over time, so
597 # they should always be resolved by the remote repo.)
598 if ($treeish =~ /^[0-9a-f]{7,40}$/s) {
599 # Hide stderr because it's normal for this to fail:
600 my $sha1 = `$gitcmd rev-list -n1 ''\Q$treeish\E 2>/dev/null`;
602 # Careful not to resolve a branch named abcdeff to commit 1234567:
603 $sha1 =~ /^$treeish/ &&
604 $sha1 =~ /^([0-9a-f]{40})$/s) {
606 Log(undef, "Commit $commit already present in $local_repo");
610 if (!defined $commit) {
611 # If $treeish isn't just a hash or abbrev hash, or isn't here
612 # yet, we need to fetch the remote to resolve it correctly.
614 # First, remove all local heads. This prevents a name that does
615 # not exist on the remote from resolving to (or colliding with)
616 # a previously fetched branch or tag (possibly from a different
618 remove_tree("$local_repo/refs/heads", {keep_root => 1});
620 Log(undef, "Fetching objects from $repo to $local_repo");
621 `$gitcmd fetch --no-progress --tags ''\Q$repo\E \Q+refs/heads/*:refs/heads/*\E`;
623 croak("Error: `$gitcmd fetch` exited ".exit_status_s($?));
627 # Now that the data is all here, we will use our local repo for
628 # the rest of our git activities.
632 my $gitcmd = "git --git-dir=\Q$repo\E";
633 my $sha1 = `$gitcmd rev-list -n1 ''\Q$treeish\E`;
634 unless ($? == 0 && $sha1 =~ /^([0-9a-f]{40})$/) {
635 croak("`$gitcmd rev-list` exited "
637 .", '$treeish' not found, giving up");
640 Log(undef, "Version $treeish is commit $commit");
642 if ($commit ne $Job->{'script_version'}) {
643 # Record the real commit id in the database, frozentokey, logs,
644 # etc. -- instead of an abbreviation or a branch name which can
645 # become ambiguous or point to a different commit in the future.
646 if (!$Job->update_attributes('script_version' => $commit)) {
647 croak("Error: failed to update job's script_version attribute");
651 $ENV{"CRUNCH_SRC_COMMIT"} = $commit;
652 add_git_archive("$gitcmd archive ''\Q$commit\E");
655 my $git_archive = combined_git_archive();
656 if (!defined $git_archive) {
657 Log(undef, "Skip install phase (no git archive)");
659 Log(undef, "Warning: This probably means workers have no source tree!");
664 my $install_script_tries_left = 3;
665 for (my $attempts = 0; $attempts < 3; $attempts++) {
666 my @srunargs = ("srun",
667 "--nodelist=$nodelist",
668 "-D", $ENV{'TMPDIR'}, "--job-name=$job_id");
669 my @execargs = ("sh", "-c",
670 "mkdir -p $ENV{CRUNCH_INSTALL} && cd $ENV{CRUNCH_TMP} && perl -");
672 $ENV{"CRUNCH_GIT_ARCHIVE_HASH"} = md5_hex($git_archive);
673 my ($stdout, $stderr);
674 ($exited, $stdout, $stderr) = srun_sync(
675 \@srunargs, \@execargs,
676 {label => "run install script on all workers"},
677 $build_script . $git_archive);
679 my $stderr_anything_from_script = 0;
680 for my $line (split(/\n/, $stderr)) {
681 if ($line !~ /^(srun: error: |starting: \[)/) {
682 $stderr_anything_from_script = 1;
686 last if $exited == 0 || $main::please_freeze;
688 # If the install script fails but doesn't print an error message,
689 # the next thing anyone is likely to do is just run it again in
690 # case it was a transient problem like "slurm communication fails
691 # because the network isn't reliable enough". So we'll just do
692 # that ourselves (up to 3 attempts in total). OTOH, if there is an
693 # error message, the problem is more likely to have a real fix and
694 # we should fail the job so the fixing process can start, instead
695 # of doing 2 more attempts.
696 last if $stderr_anything_from_script;
699 foreach my $tar_filename (map { tar_filename_n($_); } (1..$git_tar_count)) {
700 unlink($tar_filename);
708 foreach (qw (script script_version script_parameters runtime_constraints))
712 (ref($Job->{$_}) ? JSON::encode_json($Job->{$_}) : $Job->{$_}));
714 foreach (split (/\n/, $Job->{knobs}))
716 Log (undef, "knob " . $_);
721 $main::success = undef;
727 my $thisround_succeeded = 0;
728 my $thisround_failed = 0;
729 my $thisround_failed_multiple = 0;
730 my $working_slot_count = scalar(@slot);
732 @jobstep_todo = sort { $jobstep[$a]->{level} <=> $jobstep[$b]->{level}
733 or $a <=> $b } @jobstep_todo;
734 my $level = $jobstep[$jobstep_todo[0]]->{level};
736 my $initial_tasks_this_level = 0;
737 foreach my $id (@jobstep_todo) {
738 $initial_tasks_this_level++ if ($jobstep[$id]->{level} == $level);
741 # If the number of tasks scheduled at this level #T is smaller than the number
742 # of slots available #S, only use the first #T slots, or the first slot on
743 # each node, whichever number is greater.
745 # When we dispatch tasks later, we'll allocate whole-node resources like RAM
746 # based on these numbers. Using fewer slots makes more resources available
747 # to each individual task, which should normally be a better strategy when
748 # there are fewer of them running with less parallelism.
750 # Note that this calculation is not redone if the initial tasks at
751 # this level queue more tasks at the same level. This may harm
752 # overall task throughput for that level.
754 if ($initial_tasks_this_level < @node) {
755 @freeslot = (0..$#node);
756 } elsif ($initial_tasks_this_level < @slot) {
757 @freeslot = (0..$initial_tasks_this_level - 1);
759 @freeslot = (0..$#slot);
761 my $round_num_freeslots = scalar(@freeslot);
762 print STDERR "crunch-job have ${round_num_freeslots} free slots for ${initial_tasks_this_level} initial tasks at this level, ".scalar(@node)." nodes, and ".scalar(@slot)." slots\n";
764 my %round_max_slots = ();
765 for (my $ii = $#freeslot; $ii >= 0; $ii--) {
766 my $this_slot = $slot[$freeslot[$ii]];
767 my $node_name = $this_slot->{node}->{name};
768 $round_max_slots{$node_name} ||= $this_slot->{cpu};
769 last if (scalar(keys(%round_max_slots)) >= @node);
772 Log(undef, "start level $level with $round_num_freeslots slots");
775 my $progress_is_dirty = 1;
776 my $progress_stats_updated = 0;
778 update_progress_stats();
782 for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++)
784 # Don't create new tasks if we already know the job's final result.
785 last if defined($main::success);
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_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
835 $ENV{"CRUNCH_NODE_SLOTS"} = $round_max_slots{$ENV{TASK_SLOT_NODE}};
836 $ENV{"PATH"} = $ENV{"CRUNCH_INSTALL"} . "/bin:" . $ENV{"PATH"};
838 my $keep_mnt = $ENV{"TASK_WORK"}.".keep";
844 "--nodelist=".$childnode->{name},
845 qw(-n1 -c1 -N1 -D), $ENV{'TMPDIR'},
846 "--job-name=$job_id.$id.$$",
849 my $stdbuf = " stdbuf --output=0 --error=0 ";
851 my $arv_file_cache = "";
852 if (defined($Job->{'runtime_constraints'}->{'keep_cache_mb_per_task'})) {
853 $arv_file_cache = "--file-cache=" . ($Job->{'runtime_constraints'}->{'keep_cache_mb_per_task'} * 1024 * 1024);
857 "if [ -e \Q$ENV{TASK_WORK}\E ]; then rm -rf \Q$ENV{TASK_WORK}\E; fi; "
858 ."mkdir -p \Q$ENV{CRUNCH_TMP}\E \Q$ENV{JOB_WORK}\E \Q$ENV{TASK_WORK}\E \Q$keep_mnt\E "
859 ."&& cd \Q$ENV{CRUNCH_TMP}\E "
860 # These environment variables get used explicitly later in
861 # $command. No tool is expected to read these values directly.
862 .q{&& MEM=$(awk '($1 == "MemTotal:"){print $2}' </proc/meminfo) }
863 .q{&& SWAP=$(awk '($1 == "SwapTotal:"){print $2}' </proc/meminfo) }
864 ."&& MEMLIMIT=\$(( (\$MEM * 95) / ($ENV{CRUNCH_NODE_SLOTS} * 100) )) "
865 ."&& let SWAPLIMIT=\$MEMLIMIT+\$SWAP "
866 .q{&& declare -a VOLUMES=() }
867 .q{&& if which crunchrunner >/dev/null ; then VOLUMES+=("--volume=$(which crunchrunner):/usr/local/bin/crunchrunner") ; fi }
868 .q{&& if test -f /etc/ssl/certs/ca-certificates.crt ; then VOLUMES+=("--volume=/etc/ssl/certs/ca-certificates.crt:/etc/arvados/ca-certificates.crt") ; }
869 .q{elif test -f /etc/pki/tls/certs/ca-bundle.crt ; then VOLUMES+=("--volume=/etc/pki/tls/certs/ca-bundle.crt:/etc/arvados/ca-certificates.crt") ; fi };
871 $command .= "&& exec arv-mount --read-write --mount-by-pdh=by_pdh --mount-tmp=tmp --crunchstat-interval=10 --allow-other $arv_file_cache \Q$keep_mnt\E --exec ";
872 $ENV{TASK_KEEPMOUNT} = "$keep_mnt/by_pdh";
873 $ENV{TASK_KEEPMOUNT_TMP} = "$keep_mnt/tmp";
877 my $containername = "$Jobstep->{arvados_task}->{uuid}-$Jobstep->{failures}";
878 my $cidfile = "$ENV{CRUNCH_TMP}/$containername.cid";
879 $command .= "crunchstat -cgroup-root=\Q$cgroup_root\E -cgroup-parent=docker -cgroup-cid=$cidfile -poll=10000 ";
880 $command .= "$docker_bin run $docker_run_args --name=$containername --attach=stdout --attach=stderr --attach=stdin -i \Q$dockeruserarg\E --cidfile=$cidfile --sig-proxy ";
881 # We only set memory limits if Docker lets us limit both memory and swap.
882 # Memory limits alone have been supported longer, but subprocesses tend
883 # to get SIGKILL if they exceed that without any swap limit set.
884 # See #5642 for additional background.
885 if ($docker_limitmem) {
886 $command .= "--memory=\${MEMLIMIT}k --memory-swap=\${SWAPLIMIT}k ";
889 # The source tree and $destdir directory (which we have
890 # installed on the worker host) are available in the container,
891 # under the same path.
892 $command .= "--volume=\Q$ENV{CRUNCH_SRC}:$ENV{CRUNCH_SRC}:ro\E ";
893 $command .= "--volume=\Q$ENV{CRUNCH_INSTALL}:$ENV{CRUNCH_INSTALL}:ro\E ";
895 # Currently, we make the "by_pdh" directory in arv-mount's mount
896 # point appear at /keep inside the container (instead of using
897 # the same path as the host like we do with CRUNCH_SRC and
898 # CRUNCH_INSTALL). However, crunch scripts and utilities must
899 # not rely on this. They must use $TASK_KEEPMOUNT.
900 $command .= "--volume=\Q$ENV{TASK_KEEPMOUNT}:/keep:ro\E ";
901 $ENV{TASK_KEEPMOUNT} = "/keep";
903 # Ditto TASK_KEEPMOUNT_TMP, as /keep_tmp.
904 $command .= "--volume=\Q$ENV{TASK_KEEPMOUNT_TMP}:/keep_tmp\E ";
905 $ENV{TASK_KEEPMOUNT_TMP} = "/keep_tmp";
907 # TASK_WORK is almost exactly like a docker data volume: it
908 # starts out empty, is writable, and persists until no
909 # containers use it any more. We don't use --volumes-from to
910 # share it with other containers: it is only accessible to this
911 # task, and it goes away when this task stops.
913 # However, a docker data volume is writable only by root unless
914 # the mount point already happens to exist in the container with
915 # different permissions. Therefore, we [1] assume /tmp already
916 # exists in the image and is writable by the crunch user; [2]
917 # avoid putting TASK_WORK inside CRUNCH_TMP (which won't be
918 # writable if they are created by docker while setting up the
919 # other --volumes); and [3] create $TASK_WORK inside the
920 # container using $build_script.
921 $command .= "--volume=/tmp ";
922 $ENV{"TASK_WORK"} = "/tmp/crunch-job-task-work/$childslotname";
923 $ENV{"HOME"} = $ENV{"TASK_WORK"};
924 $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
926 # TODO: Share a single JOB_WORK volume across all task
927 # containers on a given worker node, and delete it when the job
928 # ends (and, in case that doesn't work, when the next job
931 # For now, use the same approach as TASK_WORK above.
932 $ENV{"JOB_WORK"} = "/tmp/crunch-job-work";
934 # Bind mount the crunchrunner binary and host TLS certificates file into
936 $command .= '"${VOLUMES[@]}" ';
938 while (my ($env_key, $env_val) = each %ENV)
940 if ($env_key =~ /^(ARVADOS|CRUNCH|JOB|TASK)_/) {
941 $command .= "--env=\Q$env_key=$env_val\E ";
944 $command .= "--env=\QHOME=$ENV{HOME}\E ";
945 $command .= "\Q$docker_hash\E ";
947 if ($Job->{arvados_sdk_version}) {
949 $command .= "perl - \Q$ENV{CRUNCH_SRC}/crunch_scripts/$Job->{script}\E";
951 $command .= "/bin/sh -c \'python -c " .
952 '"from pkg_resources import get_distribution as get; print \"Using Arvados SDK version\", get(\"arvados-python-client\").version"' .
953 ">&2 2>/dev/null; " .
954 "mkdir -p \"$ENV{JOB_WORK}\" \"$ENV{TASK_WORK}\" && " .
955 "if which stdbuf >/dev/null ; then " .
956 " exec $stdbuf \Q$ENV{CRUNCH_SRC}/crunch_scripts/$Job->{script}\E ;" .
958 " exec \Q$ENV{CRUNCH_SRC}/crunch_scripts/$Job->{script}\E ;" .
963 $command .= "crunchstat -cgroup-root=\Q$cgroup_root\E -poll=10000 ";
965 $command .= "perl - $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
968 my @execargs = ('bash', '-c', $command);
969 srun (\@srunargs, \@execargs, undef, $build_script);
970 # exec() failed, we assume nothing happened.
971 die "srun() failed on build script\n";
974 if (!defined $childpid)
985 jobstepname => "$job_id.$id.$childpid",
987 croak ("assert failed: \$slot[$childslot]->{'pid'} exists") if exists $slot[$childslot]->{pid};
988 $slot[$childslot]->{pid} = $childpid;
990 Log ($id, "job_task ".$Jobstep->{'arvados_task'}->{'uuid'});
991 Log ($id, "child $childpid started on $childslotname");
992 $Jobstep->{starttime} = time;
993 $Jobstep->{node} = $childnode->{name};
994 $Jobstep->{slotindex} = $childslot;
995 delete $Jobstep->{stderr};
996 delete $Jobstep->{finishtime};
997 delete $Jobstep->{tempfail};
999 $Jobstep->{'arvados_task'}->{started_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{starttime});
1000 $Jobstep->{'arvados_task'}->save;
1002 splice @jobstep_todo, $todo_ptr, 1;
1005 $progress_is_dirty = 1;
1009 ($round_num_freeslots > @freeslot && $todo_ptr+1 > $#jobstep_todo))
1011 last THISROUND if $main::please_freeze;
1012 if ($main::please_info)
1014 $main::please_info = 0;
1016 create_output_collection();
1018 update_progress_stats();
1023 if (!$gotsome || ($latest_refresh + 2 < scalar time))
1025 check_refresh_wanted();
1027 update_progress_stats();
1029 elsif (time - $progress_stats_updated >= 30 || $progress_is_dirty)
1031 update_progress_stats();
1034 select (undef, undef, undef, 0.1);
1036 $working_slot_count = scalar(grep { $_->{node}->{fail_count} == 0 &&
1037 $_->{node}->{hold_count} < 4 } @slot);
1038 if (($thisround_failed_multiple >= 8 && $thisround_succeeded == 0) ||
1039 ($thisround_failed_multiple >= 16 && $thisround_failed_multiple > $thisround_succeeded))
1041 my $message = "Repeated failure rate too high ($thisround_failed_multiple/"
1042 .($thisround_failed+$thisround_succeeded)
1043 .") -- giving up on this round";
1044 Log (undef, $message);
1048 # move slots from freeslot to holdslot (or back to freeslot) if necessary
1049 for (my $i=$#freeslot; $i>=0; $i--) {
1050 if ($slot[$freeslot[$i]]->{node}->{hold_until} > scalar time) {
1051 push @holdslot, (splice @freeslot, $i, 1);
1054 for (my $i=$#holdslot; $i>=0; $i--) {
1055 if ($slot[$holdslot[$i]]->{node}->{hold_until} <= scalar time) {
1056 push @freeslot, (splice @holdslot, $i, 1);
1060 # give up if no nodes are succeeding
1061 if ($working_slot_count < 1) {
1062 Log(undef, "Every node has failed -- giving up");
1069 push @freeslot, splice @holdslot;
1070 map { $slot[$freeslot[$_]]->{node}->{losing_streak} = 0 } (0..$#freeslot);
1073 Log (undef, "wait for last ".(scalar keys %proc)." children to finish");
1076 if ($main::please_continue) {
1077 $main::please_continue = 0;
1080 $main::please_info = 0, freeze(), create_output_collection(), save_meta(1) if $main::please_info;
1082 if (!reapchildren())
1084 check_refresh_wanted();
1086 update_progress_stats();
1087 select (undef, undef, undef, 0.1);
1088 killem (keys %proc) if $main::please_freeze;
1092 update_progress_stats();
1093 freeze_if_want_freeze();
1096 if (!defined $main::success)
1098 if (!@jobstep_todo) {
1100 } elsif ($working_slot_count < 1) {
1101 save_output_collection();
1103 exit(EX_RETRY_UNLOCKED);
1104 } elsif ($thisround_succeeded == 0 &&
1105 ($thisround_failed == 0 || $thisround_failed > 4)) {
1106 my $message = "stop because $thisround_failed tasks failed and none succeeded";
1107 Log (undef, $message);
1112 goto ONELEVEL if !defined $main::success;
1115 release_allocation();
1117 my $collated_output = save_output_collection();
1118 Log (undef, "finish");
1123 if ($collated_output && $main::success) {
1124 $final_state = 'Complete';
1126 $final_state = 'Failed';
1128 $Job->update_attributes('state' => $final_state);
1130 exit (($final_state eq 'Complete') ? 0 : 1);
1134 sub update_progress_stats
1136 $progress_stats_updated = time;
1137 return if !$progress_is_dirty;
1138 my ($todo, $done, $running) = (scalar @jobstep_todo,
1139 scalar @jobstep_done,
1140 scalar keys(%proc));
1141 $Job->{'tasks_summary'} ||= {};
1142 $Job->{'tasks_summary'}->{'todo'} = $todo;
1143 $Job->{'tasks_summary'}->{'done'} = $done;
1144 $Job->{'tasks_summary'}->{'running'} = $running;
1145 $Job->update_attributes('tasks_summary' => $Job->{'tasks_summary'});
1146 Log (undef, "status: $done done, $running running, $todo todo");
1147 $progress_is_dirty = 0;
1154 my $children_reaped = 0;
1155 my @successful_task_uuids = ();
1157 while((my $pid = waitpid (-1, WNOHANG)) > 0)
1159 my $childstatus = $?;
1161 my $whatslot = ($slot[$proc{$pid}->{slot}]->{node}->{name}
1163 . $slot[$proc{$pid}->{slot}]->{cpu});
1164 my $jobstepidx = $proc{$pid}->{jobstepidx};
1167 my $elapsed = time - $proc{$pid}->{time};
1168 my $Jobstep = $jobstep[$jobstepidx];
1170 my $exitvalue = $childstatus >> 8;
1171 my $exitinfo = "exit ".exit_status_s($childstatus);
1172 $Jobstep->{'arvados_task'}->reload;
1173 my $task_success = $Jobstep->{'arvados_task'}->{success};
1175 Log ($jobstepidx, "child $pid on $whatslot $exitinfo success=$task_success");
1177 if (!defined $task_success) {
1178 # task did not indicate one way or the other --> fail
1179 Log($jobstepidx, sprintf(
1180 "ERROR: Task process exited %s, but never updated its task record to indicate success and record its output.",
1181 exit_status_s($childstatus)));
1182 $Jobstep->{'arvados_task'}->{success} = 0;
1183 $Jobstep->{'arvados_task'}->save;
1190 $temporary_fail ||= $Jobstep->{tempfail};
1191 $temporary_fail ||= ($exitvalue == TASK_TEMPFAIL);
1193 ++$thisround_failed;
1194 ++$thisround_failed_multiple if $Jobstep->{'failures'} >= 1;
1196 # Check for signs of a failed or misconfigured node
1197 if (++$slot[$proc{$pid}->{slot}]->{node}->{losing_streak} >=
1198 2+$slot[$proc{$pid}->{slot}]->{node}->{ncpus}) {
1199 # Don't count this against jobstep failure thresholds if this
1200 # node is already suspected faulty and srun exited quickly
1201 if ($slot[$proc{$pid}->{slot}]->{node}->{hold_until} &&
1203 Log ($jobstepidx, "blaming failure on suspect node " .
1204 $slot[$proc{$pid}->{slot}]->{node}->{name});
1205 $temporary_fail ||= 1;
1207 ban_node_by_slot($proc{$pid}->{slot});
1210 Log ($jobstepidx, sprintf('failure (#%d, %s) after %d seconds',
1211 ++$Jobstep->{'failures'},
1212 $temporary_fail ? 'temporary' : 'permanent',
1215 if (!$temporary_fail || $Jobstep->{'failures'} >= 3) {
1216 # Give up on this task, and the whole job
1219 # Put this task back on the todo queue
1220 push @jobstep_todo, $jobstepidx;
1221 $Job->{'tasks_summary'}->{'failed'}++;
1225 push @successful_task_uuids, $Jobstep->{'arvados_task'}->{uuid};
1226 ++$thisround_succeeded;
1227 $slot[$proc{$pid}->{slot}]->{node}->{losing_streak} = 0;
1228 $slot[$proc{$pid}->{slot}]->{node}->{hold_until} = 0;
1229 $slot[$proc{$pid}->{slot}]->{node}->{fail_count} = 0;
1230 push @jobstep_done, $jobstepidx;
1231 Log ($jobstepidx, "success in $elapsed seconds");
1233 $Jobstep->{exitcode} = $childstatus;
1234 $Jobstep->{finishtime} = time;
1235 $Jobstep->{'arvados_task'}->{finished_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{finishtime});
1236 $Jobstep->{'arvados_task'}->save;
1237 process_stderr_final ($jobstepidx);
1238 Log ($jobstepidx, sprintf("task output (%d bytes): %s",
1239 length($Jobstep->{'arvados_task'}->{output}),
1240 $Jobstep->{'arvados_task'}->{output}));
1242 close $reader{$jobstepidx};
1243 delete $reader{$jobstepidx};
1244 delete $slot[$proc{$pid}->{slot}]->{pid};
1245 push @freeslot, $proc{$pid}->{slot};
1248 $progress_is_dirty = 1;
1251 if (scalar(@successful_task_uuids) > 0)
1253 Log (undef, sprintf("%d tasks exited (%d succeeded), checking for new tasks from API server.", $children_reaped, scalar(@successful_task_uuids)));
1255 my $newtask_list = [];
1256 my $newtask_results;
1258 $newtask_results = api_call(
1260 'filters' => [["created_by_job_task_uuid","in",\@successful_task_uuids]],
1261 'order' => 'qsequence',
1262 'offset' => scalar(@$newtask_list),
1264 push(@$newtask_list, @{$newtask_results->{items}});
1265 } while (@{$newtask_results->{items}});
1266 Log (undef, sprintf("Got %d new tasks from API server.", scalar(@$newtask_list)));
1267 foreach my $arvados_task (@$newtask_list) {
1269 'level' => $arvados_task->{'sequence'},
1271 'arvados_task' => $arvados_task
1273 push @jobstep, $jobstep;
1274 push @jobstep_todo, $#jobstep;
1278 return $children_reaped;
1281 sub check_refresh_wanted
1283 my @stat = stat $ENV{"CRUNCH_REFRESH_TRIGGER"};
1285 $stat[9] > $latest_refresh &&
1286 # ...and we have actually locked the job record...
1287 $job_id eq $Job->{'uuid'}) {
1288 $latest_refresh = scalar time;
1289 my $Job2 = api_call("jobs/get", uuid => $jobspec);
1290 for my $attr ('cancelled_at',
1291 'cancelled_by_user_uuid',
1292 'cancelled_by_client_uuid',
1294 $Job->{$attr} = $Job2->{$attr};
1296 if ($Job->{'state'} ne "Running") {
1297 if ($Job->{'state'} eq "Cancelled") {
1298 Log (undef, "Job cancelled at " . $Job->{'cancelled_at'} . " by user " . $Job->{'cancelled_by_user_uuid'});
1300 Log (undef, "Job state unexpectedly changed to " . $Job->{'state'});
1303 $main::please_freeze = 1;
1310 my $last_squeue_check = $squeue_checked;
1312 # Do not call `squeue` or check the kill list more than once every
1314 return if $last_squeue_check > time - 15;
1315 $squeue_checked = time;
1317 # Look for children from which we haven't received stderr data since
1318 # the last squeue check. If no such children exist, all procs are
1319 # alive and there's no need to even look at squeue.
1321 # As long as the crunchstat poll interval (10s) is shorter than the
1322 # squeue check interval (15s) this should make the squeue check an
1324 my $silent_procs = 0;
1325 for my $js (map {$jobstep[$_->{jobstepidx}]} values %proc)
1327 if (!exists($js->{stderr_at}))
1329 $js->{stderr_at} = 0;
1331 if ($js->{stderr_at} < $last_squeue_check)
1336 return if $silent_procs == 0;
1338 # use killem() on procs whose killtime is reached
1339 while (my ($pid, $procinfo) = each %proc)
1341 my $js = $jobstep[$procinfo->{jobstepidx}];
1342 if (exists $procinfo->{killtime}
1343 && $procinfo->{killtime} <= time
1344 && $js->{stderr_at} < $last_squeue_check)
1347 if ($js->{stderr_at}) {
1348 $sincewhen = " in last " . (time - $js->{stderr_at}) . "s";
1350 Log($procinfo->{jobstepidx}, "killing orphaned srun process $pid (task not in slurm queue, no stderr received$sincewhen)");
1357 # here is an opportunity to check for mysterious problems with local procs
1361 # Get a list of steps still running. Note: squeue(1) says --steps
1362 # selects a format (which we override anyway) and allows us to
1363 # specify which steps we're interested in (which we don't).
1364 # Importantly, it also changes the meaning of %j from "job name" to
1365 # "step name" and (although this isn't mentioned explicitly in the
1366 # docs) switches from "one line per job" mode to "one line per step"
1367 # mode. Without it, we'd just get a list of one job, instead of a
1369 my @squeue = `squeue --jobs=\Q$ENV{SLURM_JOB_ID}\E --steps --format='%j' --noheader`;
1372 Log(undef, "warning: squeue exit status $? ($!)");
1377 # which of my jobsteps are running, according to squeue?
1379 for my $jobstepname (@squeue)
1381 $ok{$jobstepname} = 1;
1384 # Check for child procs >60s old and not mentioned by squeue.
1385 while (my ($pid, $procinfo) = each %proc)
1387 if ($procinfo->{time} < time - 60
1388 && $procinfo->{jobstepname}
1389 && !exists $ok{$procinfo->{jobstepname}}
1390 && !exists $procinfo->{killtime})
1392 # According to slurm, this task has ended (successfully or not)
1393 # -- but our srun child hasn't exited. First we must wait (30
1394 # seconds) in case this is just a race between communication
1395 # channels. Then, if our srun child process still hasn't
1396 # terminated, we'll conclude some slurm communication
1397 # error/delay has caused the task to die without notifying srun,
1398 # and we'll kill srun ourselves.
1399 $procinfo->{killtime} = time + 30;
1400 Log($procinfo->{jobstepidx}, "notice: task is not in slurm queue but srun process $pid has not exited");
1407 my $last_sinfo_check = $sinfo_checked;
1409 # Do not call `sinfo` more than once every 15 seconds.
1410 return if $last_sinfo_check > time - 15;
1411 $sinfo_checked = time;
1413 my @sinfo = `sinfo --nodes=\Q$ENV{SLURM_NODELIST}\E --noheader -o "%t"`;
1416 Log(undef, "warning: sinfo exit status $? ($!)");
1423 if ($_ != "alloc" && $_ != "alloc*") {
1424 $main::please_freeze = 1;
1429 sub release_allocation
1433 Log (undef, "release job allocation");
1434 system "scancel $ENV{SLURM_JOB_ID}";
1443 my $sel = IO::Select->new();
1444 foreach my $jobstepidx (keys %reader)
1446 my $fd = $reader{$jobstepidx};
1448 $fd_job{$fd} = $jobstepidx;
1450 if (my $stdout_fd = $jobstep[$jobstepidx]->{stdout_r}) {
1451 $sel->add($stdout_fd);
1452 $fd_job{$stdout_fd} = $jobstepidx;
1455 # select on all reader fds with 0.1s timeout
1456 my @ready_fds = $sel->can_read(0.1);
1457 foreach my $fd (@ready_fds)
1460 if (0 < sysread ($fd, $buf, 65536))
1463 print STDERR $buf if $ENV{CRUNCH_DEBUG};
1465 my $jobstepidx = $fd_job{$fd};
1466 if ($jobstep[$jobstepidx]->{stdout_r} == $fd) {
1467 $jobstep[$jobstepidx]->{stdout_captured} .= $buf;
1471 $jobstep[$jobstepidx]->{stderr_at} = time;
1472 $jobstep[$jobstepidx]->{stderr} .= $buf;
1474 # Consume everything up to the last \n
1475 preprocess_stderr ($jobstepidx);
1477 if (length ($jobstep[$jobstepidx]->{stderr}) > 16384)
1479 # If we get a lot of stderr without a newline, chop off the
1480 # front to avoid letting our buffer grow indefinitely.
1481 substr ($jobstep[$jobstepidx]->{stderr},
1482 0, length($jobstep[$jobstepidx]->{stderr}) - 8192) = "";
1490 # Consume all full lines of stderr for a jobstep. Everything after the
1491 # last newline will remain in $jobstep[$jobstepidx]->{stderr} after
1493 sub preprocess_stderr
1495 my $jobstepidx = shift;
1496 # slotindex is only defined for children running Arvados job tasks.
1497 # Be prepared to handle the undef case (for setup srun calls, etc.).
1498 my $job_slot_index = $jobstep[$jobstepidx]->{slotindex};
1500 while ($jobstep[$jobstepidx]->{stderr} =~ /^(.*?)\n/) {
1502 substr $jobstep[$jobstepidx]->{stderr}, 0, 1+length($line), "";
1503 Log ($jobstepidx, "stderr $line");
1504 if ($line =~ /srun: error: (SLURM job $ENV{SLURM_JOB_ID} has expired|Unable to confirm allocation for job $ENV{SLURM_JOB_ID})/) {
1506 $main::please_freeze = 1;
1508 elsif ($line =~ /srun: error: (Node failure on|Aborting, .*\bio error\b)/) {
1509 $jobstep[$jobstepidx]->{tempfail} = 1;
1510 if (defined($job_slot_index)) {
1511 $slot[$job_slot_index]->{node}->{fail_count}++;
1512 ban_node_by_slot($job_slot_index);
1515 elsif ($line =~ /srun: error: (Unable to create job step|.*: Communication connection failure)/) {
1516 $jobstep[$jobstepidx]->{tempfail} = 1;
1517 ban_node_by_slot($job_slot_index) if (defined($job_slot_index));
1519 elsif ($line =~ /\bKeep(Read|Write|Request)Error:/) {
1520 $jobstep[$jobstepidx]->{tempfail} = 1;
1526 sub process_stderr_final
1528 my $jobstepidx = shift;
1529 preprocess_stderr ($jobstepidx);
1532 Log ($jobstepidx, "stderr $_");
1533 } split ("\n", $jobstep[$jobstepidx]->{stderr});
1534 $jobstep[$jobstepidx]->{stderr} = '';
1541 if (!open($keep, "-|", "arv-get", "--retries", retry_count(), $hash)) {
1542 Log(undef, "fetch_block run error from arv-get $hash: $!");
1545 my $output_block = "";
1548 my $bytes = sysread($keep, $buf, 1024 * 1024);
1549 if (!defined $bytes) {
1550 Log(undef, "fetch_block read error from arv-get: $!");
1551 $output_block = undef;
1553 } elsif ($bytes == 0) {
1554 # sysread returns 0 at the end of the pipe.
1557 # some bytes were read into buf.
1558 $output_block .= $buf;
1563 Log(undef, "fetch_block arv-get exited " . exit_status_s($?));
1564 $output_block = undef;
1566 return $output_block;
1569 # Create a collection by concatenating the output of all tasks (each
1570 # task's output is either a manifest fragment, a locator for a
1571 # manifest fragment stored in Keep, or nothing at all). Return the
1572 # portable_data_hash of the new collection.
1573 sub create_output_collection
1575 Log (undef, "collate");
1577 my ($child_out, $child_in);
1578 my $pid = open2($child_out, $child_in, 'python', '-c', q{
1581 print (arvados.api("v1").collections().
1582 create(body={"manifest_text": sys.stdin.read()}).
1583 execute(num_retries=int(sys.argv[1]))["portable_data_hash"])
1587 my $manifest_size = 0;
1591 my $output = $_->{'arvados_task'}->{output};
1592 next if (!defined($output));
1594 if ($output =~ /^[0-9a-f]{32}(\+\S+)*$/) {
1595 $next_write = fetch_block($output);
1597 $next_write = $output;
1599 if (defined($next_write)) {
1600 if (!defined(syswrite($child_in, $next_write))) {
1601 # There's been an error writing. Stop the loop.
1602 # We'll log details about the exit code later.
1605 $manifest_size += length($next_write);
1608 my $uuid = $_->{'arvados_task'}->{'uuid'};
1609 Log (undef, "Error retrieving '$output' output by task $task_idx ($uuid)");
1614 Log(undef, "collated output manifest text to send to API server is $manifest_size bytes with access tokens");
1617 my $s = IO::Select->new($child_out);
1618 if ($s->can_read(120)) {
1619 sysread($child_out, $joboutput, 1024 * 1024);
1622 Log(undef, "output collection creation exited " . exit_status_s($?));
1628 Log (undef, "timed out while creating output collection");
1629 foreach my $signal (2, 2, 2, 15, 15, 9) {
1630 kill($signal, $pid);
1631 last if waitpid($pid, WNOHANG) == -1;
1640 # Calls create_output_collection, logs the result, and returns it.
1641 # If that was successful, save that as the output in the job record.
1642 sub save_output_collection {
1643 my $collated_output = create_output_collection();
1645 if (!$collated_output) {
1646 Log(undef, "Failed to write output collection");
1649 Log(undef, "job output $collated_output");
1650 $Job->update_attributes('output' => $collated_output);
1652 return $collated_output;
1659 my $sig = 2; # SIGINT first
1660 if (exists $proc{$_}->{"sent_$sig"} &&
1661 time - $proc{$_}->{"sent_$sig"} > 4)
1663 $sig = 15; # SIGTERM if SIGINT doesn't work
1665 if (exists $proc{$_}->{"sent_$sig"} &&
1666 time - $proc{$_}->{"sent_$sig"} > 4)
1668 $sig = 9; # SIGKILL if SIGTERM doesn't work
1670 if (!exists $proc{$_}->{"sent_$sig"})
1672 Log ($proc{$_}->{jobstepidx}, "sending 2x signal $sig to pid $_");
1674 select (undef, undef, undef, 0.1);
1677 kill $sig, $_; # srun wants two SIGINT to really interrupt
1679 $proc{$_}->{"sent_$sig"} = time;
1680 $proc{$_}->{"killedafter"} = time - $proc{$_}->{"time"};
1690 vec($bits,fileno($_),1) = 1;
1696 # Send log output to Keep via arv-put.
1698 # $log_pipe_in and $log_pipe_out are the input and output filehandles to the arv-put pipe.
1699 # $log_pipe_out_buf is a string containing all output read from arv-put so far.
1700 # $log_pipe_out_select is an IO::Select object around $log_pipe_out.
1701 # $log_pipe_pid is the pid of the arv-put subprocess.
1703 # The only functions that should access these variables directly are:
1705 # log_writer_start($logfilename)
1706 # Starts an arv-put pipe, reading data on stdin and writing it to
1707 # a $logfilename file in an output collection.
1709 # log_writer_read_output([$timeout])
1710 # Read output from $log_pipe_out and append it to $log_pipe_out_buf.
1711 # Passes $timeout to the select() call, with a default of 0.01.
1712 # Returns the result of the last read() call on $log_pipe_out, or
1713 # -1 if read() wasn't called because select() timed out.
1714 # Only other log_writer_* functions should need to call this.
1716 # log_writer_send($txt)
1717 # Writes $txt to the output log collection.
1719 # log_writer_finish()
1720 # Closes the arv-put pipe and returns the output that it produces.
1722 # log_writer_is_active()
1723 # Returns a true value if there is currently a live arv-put
1724 # process, false otherwise.
1726 my ($log_pipe_in, $log_pipe_out, $log_pipe_out_buf, $log_pipe_out_select,
1729 sub log_writer_start($)
1731 my $logfilename = shift;
1732 $log_pipe_pid = open2($log_pipe_out, $log_pipe_in,
1736 '--filename', $logfilename,
1738 $log_pipe_out_buf = "";
1739 $log_pipe_out_select = IO::Select->new($log_pipe_out);
1742 sub log_writer_read_output {
1743 my $timeout = shift || 0.01;
1745 while ($read && $log_pipe_out_select->can_read($timeout)) {
1746 $read = read($log_pipe_out, $log_pipe_out_buf, 65536,
1747 length($log_pipe_out_buf));
1749 if (!defined($read)) {
1750 Log(undef, "error reading log manifest from arv-put: $!");
1755 sub log_writer_send($)
1758 print $log_pipe_in $txt;
1759 log_writer_read_output();
1762 sub log_writer_finish()
1764 return unless $log_pipe_pid;
1766 close($log_pipe_in);
1768 my $logger_failed = 0;
1769 my $read_result = log_writer_read_output(120);
1770 if ($read_result == -1) {
1771 $logger_failed = -1;
1772 Log (undef, "timed out reading from 'arv-put'");
1773 } elsif ($read_result != 0) {
1774 $logger_failed = -2;
1775 Log(undef, "failed to read arv-put log manifest to EOF");
1778 waitpid($log_pipe_pid, 0);
1780 $logger_failed ||= $?;
1781 Log(undef, "log_writer_finish: arv-put exited " . exit_status_s($?))
1784 close($log_pipe_out);
1785 my $arv_put_output = $logger_failed ? undef : $log_pipe_out_buf;
1786 $log_pipe_pid = $log_pipe_in = $log_pipe_out = $log_pipe_out_buf =
1787 $log_pipe_out_select = undef;
1789 return $arv_put_output;
1792 sub log_writer_is_active() {
1793 return $log_pipe_pid;
1796 sub Log # ($jobstepidx, $logmessage)
1798 my ($jobstepidx, $logmessage) = @_;
1799 if ($logmessage =~ /\n/) {
1800 for my $line (split (/\n/, $_[1])) {
1801 Log ($jobstepidx, $line);
1805 my $fh = select STDERR; $|=1; select $fh;
1807 if (defined($jobstepidx) && exists($jobstep[$jobstepidx]->{arvados_task})) {
1808 $task_qseq = $jobstepidx;
1810 my $message = sprintf ("%s %d %s %s", $job_id, $$, $task_qseq, $logmessage);
1811 $message =~ s{([^ -\176])}{"\\" . sprintf ("%03o", ord($1))}ge;
1814 if (log_writer_is_active() || -t STDERR) {
1815 my @gmtime = gmtime;
1816 $datetime = sprintf ("%04d-%02d-%02d_%02d:%02d:%02d",
1817 $gmtime[5]+1900, $gmtime[4]+1, @gmtime[3,2,1,0]);
1819 print STDERR ((-t STDERR) ? ($datetime." ".$message) : $message);
1821 if (log_writer_is_active()) {
1822 log_writer_send($datetime . " " . $message);
1829 my ($package, $file, $line) = caller;
1830 my $message = "@_ at $file line $line\n";
1831 Log (undef, $message);
1832 freeze() if @jobstep_todo;
1833 create_output_collection() if @jobstep_todo;
1843 if ($Job->{'state'} eq 'Cancelled') {
1844 $Job->update_attributes('finished_at' => scalar gmtime);
1846 $Job->update_attributes('state' => 'Failed');
1853 my $justcheckpoint = shift; # false if this will be the last meta saved
1854 return if $justcheckpoint; # checkpointing is not relevant post-Warehouse.pm
1855 return unless log_writer_is_active();
1856 my $log_manifest = log_writer_finish();
1857 return unless defined($log_manifest);
1860 my $prev_log_coll = api_call("collections/get", uuid => $Job->{log});
1861 $log_manifest = $prev_log_coll->{manifest_text} . $log_manifest;
1864 my $log_coll = api_call(
1865 "collections/create", ensure_unique_name => 1, collection => {
1866 manifest_text => $log_manifest,
1867 owner_uuid => $Job->{owner_uuid},
1868 name => sprintf("Log from %s job %s", $Job->{script}, $Job->{uuid}),
1870 Log(undef, "log collection is " . $log_coll->{portable_data_hash});
1871 $Job->update_attributes('log' => $log_coll->{portable_data_hash});
1875 sub freeze_if_want_freeze
1877 if ($main::please_freeze)
1879 release_allocation();
1882 # kill some srun procs before freeze+stop
1883 map { $proc{$_} = {} } @_;
1886 killem (keys %proc);
1887 select (undef, undef, undef, 0.1);
1889 while (($died = waitpid (-1, WNOHANG)) > 0)
1891 delete $proc{$died};
1896 create_output_collection();
1906 Log (undef, "Freeze not implemented");
1913 croak ("Thaw not implemented");
1929 $s =~ s{\\(.)}{$1 eq "n" ? "\n" : $1}ge;
1935 my $srunargs = shift;
1936 my $execargs = shift;
1937 my $opts = shift || {};
1940 my $label = exists $opts->{label} ? $opts->{label} : "@$execargs";
1941 Log (undef, "$label: start");
1943 my ($stderr_r, $stderr_w);
1944 pipe $stderr_r, $stderr_w or croak("pipe() failed: $!");
1946 my ($stdout_r, $stdout_w);
1947 pipe $stdout_r, $stdout_w or croak("pipe() failed: $!");
1949 my $srunpid = fork();
1954 fcntl($stderr_w, F_SETFL, 0) or croak($!); # no close-on-exec
1955 fcntl($stdout_w, F_SETFL, 0) or croak($!);
1956 open(STDERR, ">&", $stderr_w);
1957 open(STDOUT, ">&", $stdout_w);
1958 srun ($srunargs, $execargs, $opts, $stdin);
1964 set_nonblocking($stderr_r);
1965 set_nonblocking($stdout_r);
1967 # Add entries to @jobstep and %proc so check_squeue() and
1968 # freeze_if_want_freeze() can treat it like a job task process.
1972 stderr_captured => '',
1973 stdout_r => $stdout_r,
1974 stdout_captured => '',
1976 my $jobstepidx = $#jobstep;
1978 jobstepidx => $jobstepidx,
1980 $reader{$jobstepidx} = $stderr_r;
1982 while ($srunpid != waitpid ($srunpid, WNOHANG)) {
1983 my $busy = readfrompipes();
1984 if (!$busy || ($latest_refresh + 2 < scalar time)) {
1985 check_refresh_wanted();
1990 select(undef, undef, undef, 0.1);
1992 killem(keys %proc) if $main::please_freeze;
1996 1 while readfrompipes();
1997 process_stderr_final ($jobstepidx);
1999 Log (undef, "$label: exit ".exit_status_s($exited));
2003 delete $proc{$srunpid};
2004 delete $reader{$jobstepidx};
2006 my $j = pop @jobstep;
2007 # If the srun showed signs of tempfail, ensure the caller treats that as a
2009 if ($main::please_freeze || $j->{tempfail}) {
2012 return ($exited, $j->{stdout_captured}, $j->{stderr_captured});
2018 my $srunargs = shift;
2019 my $execargs = shift;
2020 my $opts = shift || {};
2022 my $args = $have_slurm ? [@$srunargs, @$execargs] : $execargs;
2024 $Data::Dumper::Terse = 1;
2025 $Data::Dumper::Indent = 0;
2026 my $show_cmd = Dumper($args);
2027 $show_cmd =~ s/(TOKEN\\*=)[^\s\']+/${1}[...]/g;
2028 $show_cmd =~ s/\n/ /g;
2029 if ($opts->{fork}) {
2030 Log(undef, "starting: $show_cmd");
2032 # This is a child process: parent is in charge of reading our
2033 # stderr and copying it to Log() if needed.
2034 warn "starting: $show_cmd\n";
2037 if (defined $stdin) {
2038 my $child = open STDIN, "-|";
2039 defined $child or die "no fork: $!";
2041 print $stdin or die $!;
2042 close STDOUT or die $!;
2047 return system (@$args) if $opts->{fork};
2050 warn "ENV size is ".length(join(" ",%ENV));
2051 die "exec failed: $!: @$args";
2055 sub ban_node_by_slot {
2056 # Don't start any new jobsteps on this node for 60 seconds
2058 $slot[$slotid]->{node}->{hold_until} = 60 + scalar time;
2059 $slot[$slotid]->{node}->{hold_count}++;
2060 Log (undef, "backing off node " . $slot[$slotid]->{node}->{name} . " for 60 seconds");
2065 my ($lockfile, $error_message) = @_;
2066 open L, ">", $lockfile or croak("$lockfile: $!");
2067 if (!flock L, LOCK_EX|LOCK_NB) {
2068 croak("Can't lock $lockfile: $error_message\n");
2072 sub find_docker_image {
2073 # Given a Keep locator, check to see if it contains a Docker image.
2074 # If so, return its stream name and Docker hash.
2075 # If not, return undef for both values.
2076 my $locator = shift;
2077 my ($streamname, $filename);
2078 my $image = api_call("collections/get", uuid => $locator);
2080 foreach my $line (split(/\n/, $image->{manifest_text})) {
2081 my @tokens = split(/\s+/, $line);
2083 $streamname = shift(@tokens);
2084 foreach my $filedata (grep(/^\d+:\d+:/, @tokens)) {
2085 if (defined($filename)) {
2086 return (undef, undef); # More than one file in the Collection.
2088 $filename = (split(/:/, $filedata, 3))[2];
2093 if (defined($filename) and ($filename =~ /^([0-9A-Fa-f]{64})\.tar$/)) {
2094 return ($streamname, $1);
2096 return (undef, undef);
2101 # Calculate the number of times an operation should be retried,
2102 # assuming exponential backoff, and that we're willing to retry as
2103 # long as tasks have been running. Enforce a minimum of 3 retries.
2104 my ($starttime, $endtime, $timediff, $retries);
2106 $starttime = $jobstep[0]->{starttime};
2107 $endtime = $jobstep[-1]->{finishtime};
2109 if (!defined($starttime)) {
2111 } elsif (!defined($endtime)) {
2112 $timediff = time - $starttime;
2114 $timediff = ($endtime - $starttime) - (time - $endtime);
2116 if ($timediff > 0) {
2117 $retries = int(log($timediff) / log(2));
2119 $retries = 1; # Use the minimum.
2121 return ($retries > 3) ? $retries : 3;
2125 # Pass in two function references.
2126 # This method will be called with the remaining arguments.
2127 # If it dies, retry it with exponential backoff until it succeeds,
2128 # or until the current retry_count is exhausted. After each failure
2129 # that can be retried, the second function will be called with
2130 # the current try count (0-based), next try time, and error message.
2131 my $operation = shift;
2132 my $retry_callback = shift;
2133 my $retries = retry_count();
2134 foreach my $try_count (0..$retries) {
2135 my $next_try = time + (2 ** $try_count);
2136 my $result = eval { $operation->(@_); };
2139 } elsif ($try_count < $retries) {
2140 $retry_callback->($try_count, $next_try, $@);
2141 my $sleep_time = $next_try - time;
2142 sleep($sleep_time) if ($sleep_time > 0);
2145 # Ensure the error message ends in a newline, so Perl doesn't add
2146 # retry_op's line number to it.
2152 # Pass in a /-separated API method name, and arguments for it.
2153 # This function will call that method, retrying as needed until
2154 # the current retry_count is exhausted, with a log on the first failure.
2155 my $method_name = shift;
2156 my $log_api_retry = sub {
2157 my ($try_count, $next_try_at, $errmsg) = @_;
2158 $errmsg =~ s/\s*\bat \Q$0\E line \d+\.?\s*//;
2159 $errmsg =~ s/\s/ /g;
2160 $errmsg =~ s/\s+$//;
2162 if ($next_try_at < time) {
2163 $retry_msg = "Retrying.";
2165 my $next_try_fmt = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($next_try_at);
2166 $retry_msg = "Retrying at $next_try_fmt.";
2168 Log(undef, "API method $method_name failed: $errmsg. $retry_msg");
2171 foreach my $key (split(/\//, $method_name)) {
2172 $method = $method->{$key};
2174 return retry_op(sub { $method->execute(@_); }, $log_api_retry, @_);
2178 # Given a $?, return a human-readable exit code string like "0" or
2179 # "1" or "0 with signal 1" or "1 with signal 11".
2180 my $exitcode = shift;
2181 my $s = $exitcode >> 8;
2182 if ($exitcode & 0x7f) {
2183 $s .= " with signal " . ($exitcode & 0x7f);
2185 if ($exitcode & 0x80) {
2186 $s .= " with core dump";
2191 sub handle_readall {
2192 # Pass in a glob reference to a file handle.
2193 # Read all its contents and return them as a string.
2194 my $fh_glob_ref = shift;
2196 return <$fh_glob_ref>;
2199 sub tar_filename_n {
2201 return sprintf("%s/git.%s.%d.tar", $ENV{CRUNCH_TMP}, $job_id, $n);
2204 sub add_git_archive {
2205 # Pass in a git archive command as a string or list, a la system().
2206 # This method will save its output to be included in the archive sent to the
2210 if (!open(GIT_ARCHIVE, ">", tar_filename_n($git_tar_count))) {
2211 croak("Failed to save git archive: $!");
2213 my $git_pid = open2(">&GIT_ARCHIVE", $git_input, @_);
2215 waitpid($git_pid, 0);
2218 croak("Failed to save git archive: git exited " . exit_status_s($?));
2222 sub combined_git_archive {
2223 # Combine all saved tar archives into a single archive, then return its
2224 # contents in a string. Return undef if no archives have been saved.
2225 if ($git_tar_count < 1) {
2228 my $base_tar_name = tar_filename_n(1);
2229 foreach my $tar_to_append (map { tar_filename_n($_); } (2..$git_tar_count)) {
2230 my $tar_exit = system("tar", "-Af", $base_tar_name, $tar_to_append);
2231 if ($tar_exit != 0) {
2232 croak("Error preparing build archive: tar -A exited " .
2233 exit_status_s($tar_exit));
2236 if (!open(GIT_TAR, "<", $base_tar_name)) {
2237 croak("Could not open build archive: $!");
2239 my $tar_contents = handle_readall(\*GIT_TAR);
2241 return $tar_contents;
2244 sub set_nonblocking {
2246 my $flags = fcntl ($fh, F_GETFL, 0) or croak ($!);
2247 fcntl ($fh, F_SETFL, $flags | O_NONBLOCK) or croak ($!);
2253 # This is crunch-job's internal dispatch script. crunch-job running on the API
2254 # server invokes this script on individual compute nodes, or localhost if we're
2255 # running a job locally. It gets called in two modes:
2257 # * No arguments: Installation mode. Read a tar archive from the DATA
2258 # file handle; it includes the Crunch script's source code, and
2259 # maybe SDKs as well. Those should be installed in the proper
2260 # locations. This runs outside of any Docker container, so don't try to
2261 # introspect Crunch's runtime environment.
2263 # * With arguments: Crunch script run mode. This script should set up the
2264 # environment, then run the command specified in the arguments. This runs
2265 # inside any Docker container.
2268 use File::Path qw( make_path remove_tree );
2269 use POSIX qw(getcwd);
2271 use constant TASK_TEMPFAIL => 111;
2273 # Map SDK subdirectories to the path environments they belong to.
2274 my %SDK_ENVVARS = ("perl/lib" => "PERLLIB", "ruby/lib" => "RUBYLIB");
2276 my $destdir = $ENV{"CRUNCH_SRC"};
2277 my $archive_hash = $ENV{"CRUNCH_GIT_ARCHIVE_HASH"};
2278 my $repo = $ENV{"CRUNCH_SRC_URL"};
2279 my $install_dir = $ENV{"CRUNCH_INSTALL"} || (getcwd() . "/opt");
2280 my $job_work = $ENV{"JOB_WORK"};
2281 my $task_work = $ENV{"TASK_WORK"};
2283 open(STDOUT_ORIG, ">&", STDOUT);
2284 open(STDERR_ORIG, ">&", STDERR);
2286 for my $dir ($destdir, $job_work, $task_work) {
2289 -e $dir or die "Failed to create temporary directory ($dir): $!";
2294 remove_tree($task_work, {keep_root => 1});
2297 ### Crunch script run mode
2299 # We want to do routine logging during task 0 only. This gives the user
2300 # the information they need, but avoids repeating the information for every
2303 if ($ENV{TASK_SEQUENCE} eq "0") {
2306 printf STDERR_ORIG "[Crunch] $msg\n", @_;
2312 my $python_src = "$install_dir/python";
2313 my $venv_dir = "$job_work/.arvados.venv";
2314 my $venv_built = -e "$venv_dir/bin/activate";
2315 if ((!$venv_built) and (-d $python_src) and can_run("virtualenv")) {
2316 shell_or_die(undef, "virtualenv", "--quiet", "--system-site-packages",
2317 "--python=python2.7", $venv_dir);
2318 shell_or_die(TASK_TEMPFAIL, "$venv_dir/bin/pip", "--quiet", "install", "-I", $python_src);
2320 $Log->("Built Python SDK virtualenv");
2323 my @pysdk_version_cmd = ("python", "-c",
2324 "from pkg_resources import get_distribution as get; print get('arvados-python-client').version");
2326 $Log->("Running in Python SDK virtualenv");
2327 @pysdk_version_cmd = ();
2328 my $orig_argv = join(" ", map { quotemeta($_); } @ARGV);
2329 @ARGV = ("/bin/sh", "-ec",
2330 ". \Q$venv_dir/bin/activate\E; exec $orig_argv");
2331 } elsif (-d $python_src) {
2332 $Log->("Warning: virtualenv not found inside Docker container default " .
2333 "\$PATH. Can't install Python SDK.");
2336 if (@pysdk_version_cmd) {
2337 open(my $pysdk_version_pipe, "-|", @pysdk_version_cmd);
2338 my $pysdk_version = <$pysdk_version_pipe>;
2339 close($pysdk_version_pipe);
2341 chomp($pysdk_version);
2342 $Log->("Using Arvados SDK version $pysdk_version");
2344 # A lot could've gone wrong here, but pretty much all of it means that
2345 # Python won't be able to load the Arvados SDK.
2346 $Log->("Warning: Arvados SDK not found");
2350 while (my ($sdk_dir, $sdk_envkey) = each(%SDK_ENVVARS)) {
2351 my $sdk_path = "$install_dir/$sdk_dir";
2353 if ($ENV{$sdk_envkey}) {
2354 $ENV{$sdk_envkey} = "$sdk_path:" . $ENV{$sdk_envkey};
2356 $ENV{$sdk_envkey} = $sdk_path;
2358 $Log->("Arvados SDK added to %s", $sdk_envkey);
2363 die "Cannot exec `@ARGV`: $!";
2366 ### Installation mode
2367 open L, ">", "$destdir.lock" or die "$destdir.lock: $!";
2369 if (readlink ("$destdir.archive_hash") eq $archive_hash && -d $destdir) {
2370 # This exact git archive (source + arvados sdk) is already installed
2371 # here, so there's no need to reinstall it.
2373 # We must consume our DATA section, though: otherwise the process
2374 # feeding it to us will get SIGPIPE.
2376 while (read(DATA, $buf, 65536)) { }
2381 unlink "$destdir.archive_hash";
2385 # Ignore SIGPIPE: we check retval of close() instead. See perlipc(1).
2386 local $SIG{PIPE} = "IGNORE";
2387 warn "Extracting archive: $archive_hash\n";
2388 # --ignore-zeros is necessary sometimes: depending on how much NUL
2389 # padding tar -A put on our combined archive (which in turn depends
2390 # on the length of the component archives) tar without
2391 # --ignore-zeros will exit before consuming stdin and cause close()
2392 # to fail on the resulting SIGPIPE.
2393 if (!open(TARX, "|-", "tar", "--ignore-zeros", "-xC", $destdir)) {
2394 die "Error launching 'tar -xC $destdir': $!";
2396 # If we send too much data to tar in one write (> 4-5 MiB), it stops, and we
2397 # get SIGPIPE. We must feed it data incrementally.
2399 while (read(DATA, $tar_input, 65536)) {
2400 print TARX $tar_input;
2403 die "'tar -xC $destdir' exited $?: $!";
2409 my $sdk_root = "$destdir/.arvados.sdk/sdk";
2411 foreach my $sdk_lang (("python",
2412 map { (split /\//, $_, 2)[0]; } keys(%SDK_ENVVARS))) {
2413 if (-d "$sdk_root/$sdk_lang") {
2414 if (!rename("$sdk_root/$sdk_lang", "$install_dir/$sdk_lang")) {
2415 die "Failed to install $sdk_lang SDK: $!";
2421 my $python_dir = "$install_dir/python";
2422 if ((-d $python_dir) and can_run("python2.7")) {
2423 open(my $egg_info_pipe, "-|",
2424 "python2.7 \Q$python_dir/setup.py\E egg_info 2>&1 >/dev/null");
2425 my @egg_info_errors = <$egg_info_pipe>;
2426 close($egg_info_pipe);
2429 if (@egg_info_errors and (($egg_info_errors[-1] =~ /\bgit\b/) or ($egg_info_errors[-1] =~ /\[Errno 2\]/))) {
2430 # egg_info apparently failed because it couldn't ask git for a build tag.
2431 # Specify no build tag.
2432 open(my $pysdk_cfg, ">>", "$python_dir/setup.cfg");
2433 print $pysdk_cfg "\n[egg_info]\ntag_build =\n";
2436 my $egg_info_exit = $? >> 8;
2437 foreach my $errline (@egg_info_errors) {
2440 warn "python setup.py egg_info failed: exit $egg_info_exit";
2441 exit ($egg_info_exit || 1);
2446 # Hide messages from the install script (unless it fails: shell_or_die
2447 # will show $destdir.log in that case).
2448 open(STDOUT, ">>", "$destdir.log");
2449 open(STDERR, ">&", STDOUT);
2451 if (-e "$destdir/crunch_scripts/install") {
2452 shell_or_die (undef, "$destdir/crunch_scripts/install", $install_dir);
2453 } elsif (!-e "./install.sh" && -e "./tests/autotests.sh") {
2455 shell_or_die (undef, "./tests/autotests.sh", $install_dir);
2456 } elsif (-e "./install.sh") {
2457 shell_or_die (undef, "./install.sh", $install_dir);
2460 if ($archive_hash) {
2461 unlink "$destdir.archive_hash.new";
2462 symlink ($archive_hash, "$destdir.archive_hash.new") or die "$destdir.archive_hash.new: $!";
2463 rename ("$destdir.archive_hash.new", "$destdir.archive_hash") or die "$destdir.archive_hash: $!";
2469 my $command_name = shift;
2470 open(my $which, "-|", "which", $command_name);
2471 while (<$which>) { }
2478 my $exitcode = shift;
2480 if ($ENV{"DEBUG"}) {
2481 print STDERR "@_\n";
2483 if (system (@_) != 0) {
2486 my $exitstatus = sprintf("exit %d signal %d", $code >> 8, $code & 0x7f);
2487 open STDERR, ">&STDERR_ORIG";
2488 system ("cat $destdir.log >&2");
2489 warn "@_ failed ($err): $exitstatus";
2490 if (defined($exitcode)) {
2494 exit (($code >> 8) || 1);