sdk/cli/bin/crunch-job

   1 #!/usr/bin/perl
   2 # -*- mode: perl; perl-indent-level: 2; indent-tabs-mode: nil; -*-
   3
   4 =head1 NAME
   5
   6 crunch-job: Execute job steps, save snapshots as requested, collate output.
   7
   8 =head1 SYNOPSIS
   9
  10 Obtain job details from Arvados, run tasks on compute nodes (typically
  11 invoked by scheduler on controller):
  12
  13  crunch-job --job x-y-z --git-dir /path/to/repo/.git
  14
  15 Obtain job details from command line, run tasks on local machine
  16 (typically invoked by application or developer on VM):
  17
  18  crunch-job --job '{"script_version":"/path/to/working/tree","script":"scriptname",...}'
  19
  20  crunch-job --job '{"repository":"https://github.com/curoverse/arvados.git","script_version":"master","script":"scriptname",...}'
  21
  22 =head1 OPTIONS
  23
  24 =over
  25
  26 =item --force-unlock
  27
  28 If the job is already locked, steal the lock and run it anyway.
  29
  30 =item --git-dir
  31
  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.
  35
  36 =item --job-api-token
  37
  38 Arvados API authorization token to use during the course of the job.
  39
  40 =item --no-clear-tmp
  41
  42 Do not clear per-job/task temporary directories during initial job
  43 setup. This can speed up development and debugging when running jobs
  44 locally.
  45
  46 =item --job
  47
  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.
  50
  51 =back
  52
  53 =head1 RUNNING JOBS LOCALLY
  54
  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
  57 the job finishes.
  58
  59 If the job succeeds, the job's output locator is printed on stdout.
  60
  61 While the job is running, the following signals are accepted:
  62
  63 =over
  64
  65 =item control-C, SIGINT, SIGQUIT
  66
  67 Save a checkpoint, terminate any job tasks that are running, and stop.
  68
  69 =item SIGALRM
  70
  71 Save a checkpoint and continue.
  72
  73 =item SIGHUP
  74
  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).
  78
  79 =back
  80
  81 =cut
  82
  83
  84 use strict;
  85 use POSIX ':sys_wait_h';
  86 use POSIX qw(strftime);
  87 use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
  88 use Arvados;
  89 use Cwd qw(realpath);
  90 use Data::Dumper;
  91 use Digest::MD5 qw(md5_hex);
  92 use Getopt::Long;
  93 use IPC::Open2;
  94 use IO::Select;
  95 use File::Temp;
  96 use Fcntl ':flock';
  97 use File::Path qw( make_path remove_tree );
  98
  99 use constant TASK_TEMPFAIL => 111;
 100 use constant EX_TEMPFAIL => 75;
 101
 102 $ENV{"TMPDIR"} ||= "/tmp";
 103 unless (defined $ENV{"CRUNCH_TMP"}) {
 104   $ENV{"CRUNCH_TMP"} = $ENV{"TMPDIR"} . "/crunch-job";
 105   if ($ENV{"USER"} ne "crunch" && $< != 0) {
 106     # use a tmp dir unique for my uid
 107     $ENV{"CRUNCH_TMP"} .= "-$<";
 108   }
 109 }
 110
 111 # Create the tmp directory if it does not exist
 112 if ( ! -d $ENV{"CRUNCH_TMP"} ) {
 113   make_path $ENV{"CRUNCH_TMP"} or die "Failed to create temporary working directory: " . $ENV{"CRUNCH_TMP"};
 114 }
 115
 116 $ENV{"JOB_WORK"} = $ENV{"CRUNCH_TMP"} . "/work";
 117 $ENV{"CRUNCH_INSTALL"} = "$ENV{CRUNCH_TMP}/opt";
 118 $ENV{"CRUNCH_WORK"} = $ENV{"JOB_WORK"}; # deprecated
 119 mkdir ($ENV{"JOB_WORK"});
 120
 121 my $force_unlock;
 122 my $git_dir;
 123 my $jobspec;
 124 my $job_api_token;
 125 my $no_clear_tmp;
 126 my $resume_stash;
 127 GetOptions('force-unlock' => \$force_unlock,
 128            'git-dir=s' => \$git_dir,
 129            'job=s' => \$jobspec,
 130            'job-api-token=s' => \$job_api_token,
 131            'no-clear-tmp' => \$no_clear_tmp,
 132            'resume-stash=s' => \$resume_stash,
 133     );
 134
 135 if (defined $job_api_token) {
 136   $ENV{ARVADOS_API_TOKEN} = $job_api_token;
 137 }
 138
 139 my $have_slurm = exists $ENV{SLURM_JOBID} && exists $ENV{SLURM_NODELIST};
 140 my $local_job = 0;
 141
 142
 143 $SIG{'USR1'} = sub
 144 {
 145   $main::ENV{CRUNCH_DEBUG} = 1;
 146 };
 147 $SIG{'USR2'} = sub
 148 {
 149   $main::ENV{CRUNCH_DEBUG} = 0;
 150 };
 151
 152
 153
 154 my $arv = Arvados->new('apiVersion' => 'v1');
 155
 156 my $Job;
 157 my $job_id;
 158 my $dbh;
 159 my $sth;
 160 my @jobstep;
 161
 162 my $User = api_call("users/current");
 163
 164 if ($jobspec =~ /^[-a-z\d]+$/)
 165 {
 166   # $jobspec is an Arvados UUID, not a JSON job specification
 167   $Job = api_call("jobs/get", uuid => $jobspec);
 168   if (!$force_unlock) {
 169     # Claim this job, and make sure nobody else does
 170     eval { api_call("jobs/lock", uuid => $Job->{uuid}); };
 171     if ($@) {
 172       Log(undef, "Error while locking job, exiting ".EX_TEMPFAIL);
 173       exit EX_TEMPFAIL;
 174     };
 175   }
 176 }
 177 else
 178 {
 179   $Job = JSON::decode_json($jobspec);
 180
 181   if (!$resume_stash)
 182   {
 183     map { croak ("No $_ specified") unless $Job->{$_} }
 184     qw(script script_version script_parameters);
 185   }
 186
 187   $Job->{'is_locked_by_uuid'} = $User->{'uuid'};
 188   $Job->{'started_at'} = gmtime;
 189   $Job->{'state'} = 'Running';
 190
 191   $Job = api_call("jobs/create", job => $Job);
 192 }
 193 $job_id = $Job->{'uuid'};
 194
 195 my $keep_logfile = $job_id . '.log.txt';
 196 log_writer_start($keep_logfile);
 197
 198 $Job->{'runtime_constraints'} ||= {};
 199 $Job->{'runtime_constraints'}->{'max_tasks_per_node'} ||= 0;
 200 my $max_ncpus = $Job->{'runtime_constraints'}->{'max_tasks_per_node'};
 201
 202 my $gem_versions = `gem list --quiet arvados-cli 2>/dev/null`;
 203 if ($? == 0) {
 204   $gem_versions =~ s/^arvados-cli \(/ with arvados-cli Gem version(s) /;
 205   chomp($gem_versions);
 206   chop($gem_versions);  # Closing parentheses
 207 } else {
 208   $gem_versions = "";
 209 }
 210 Log(undef,
 211     "running from " . ((-e $0) ? realpath($0) : "stdin") . $gem_versions);
 212
 213 Log (undef, "check slurm allocation");
 214 my @slot;
 215 my @node;
 216 # Should use $ENV{SLURM_TASKS_PER_NODE} instead of sinfo? (eg. "4(x3),2,4(x2)")
 217 my @sinfo;
 218 if (!$have_slurm)
 219 {
 220   my $localcpus = 0 + `grep -cw ^processor /proc/cpuinfo` || 1;
 221   push @sinfo, "$localcpus localhost";
 222 }
 223 if (exists $ENV{SLURM_NODELIST})
 224 {
 225   push @sinfo, `sinfo -h --format='%c %N' --nodes=\Q$ENV{SLURM_NODELIST}\E`;
 226 }
 227 foreach (@sinfo)
 228 {
 229   my ($ncpus, $slurm_nodelist) = split;
 230   $ncpus = $max_ncpus if $max_ncpus && $ncpus > $max_ncpus;
 231
 232   my @nodelist;
 233   while ($slurm_nodelist =~ s/^([^\[,]+?(\[.*?\])?)(,|$)//)
 234   {
 235     my $nodelist = $1;
 236     if ($nodelist =~ /\[((\d+)(-(\d+))?(,(\d+)(-(\d+))?)*)\]/)
 237     {
 238       my $ranges = $1;
 239       foreach (split (",", $ranges))
 240       {
 241         my ($a, $b);
 242         if (/(\d+)-(\d+)/)
 243         {
 244           $a = $1;
 245           $b = $2;
 246         }
 247         else
 248         {
 249           $a = $_;
 250           $b = $_;
 251         }
 252         push @nodelist, map {
 253           my $n = $nodelist;
 254           $n =~ s/\[[-,\d]+\]/$_/;
 255           $n;
 256         } ($a..$b);
 257       }
 258     }
 259     else
 260     {
 261       push @nodelist, $nodelist;
 262     }
 263   }
 264   foreach my $nodename (@nodelist)
 265   {
 266     Log (undef, "node $nodename - $ncpus slots");
 267     my $node = { name => $nodename,
 268                  ncpus => $ncpus,
 269                  losing_streak => 0,
 270                  hold_until => 0 };
 271     foreach my $cpu (1..$ncpus)
 272     {
 273       push @slot, { node => $node,
 274                     cpu => $cpu };
 275     }
 276   }
 277   push @node, @nodelist;
 278 }
 279
 280
 281
 282 # Ensure that we get one jobstep running on each allocated node before
 283 # we start overloading nodes with concurrent steps
 284
 285 @slot = sort { $a->{cpu} <=> $b->{cpu} } @slot;
 286
 287
 288 $Job->update_attributes(
 289   'tasks_summary' => { 'failed' => 0,
 290                        'todo' => 1,
 291                        'running' => 0,
 292                        'done' => 0 });
 293
 294 Log (undef, "start");
 295 $SIG{'INT'} = sub { $main::please_freeze = 1; };
 296 $SIG{'QUIT'} = sub { $main::please_freeze = 1; };
 297 $SIG{'TERM'} = \&croak;
 298 $SIG{'TSTP'} = sub { $main::please_freeze = 1; };
 299 $SIG{'ALRM'} = sub { $main::please_info = 1; };
 300 $SIG{'CONT'} = sub { $main::please_continue = 1; };
 301 $SIG{'HUP'} = sub { $main::please_refresh = 1; };
 302
 303 $main::please_freeze = 0;
 304 $main::please_info = 0;
 305 $main::please_continue = 0;
 306 $main::please_refresh = 0;
 307 my $jobsteps_must_output_keys = 0;      # becomes 1 when any task outputs a key
 308
 309 grep { $ENV{$1} = $2 if /^(NOCACHE.*?)=(.*)/ } split ("\n", $$Job{knobs});
 310 $ENV{"CRUNCH_JOB_UUID"} = $job_id;
 311 $ENV{"JOB_UUID"} = $job_id;
 312
 313
 314 my @jobstep_todo = ();
 315 my @jobstep_done = ();
 316 my @jobstep_tomerge = ();
 317 my $jobstep_tomerge_level = 0;
 318 my $squeue_checked;
 319 my $squeue_kill_checked;
 320 my $latest_refresh = scalar time;
 321
 322
 323
 324 if (defined $Job->{thawedfromkey})
 325 {
 326   thaw ($Job->{thawedfromkey});
 327 }
 328 else
 329 {
 330   my $first_task = api_call("job_tasks/create", job_task => {
 331     'job_uuid' => $Job->{'uuid'},
 332     'sequence' => 0,
 333     'qsequence' => 0,
 334     'parameters' => {},
 335   });
 336   push @jobstep, { 'level' => 0,
 337                    'failures' => 0,
 338                    'arvados_task' => $first_task,
 339                  };
 340   push @jobstep_todo, 0;
 341 }
 342
 343
 344 if (!$have_slurm)
 345 {
 346   must_lock_now("$ENV{CRUNCH_TMP}/.lock", "a job is already running here.");
 347 }
 348
 349 my $build_script = handle_readall(\*DATA);
 350 my $nodelist = join(",", @node);
 351 my $git_tar_count = 0;
 352
 353 if (!defined $no_clear_tmp) {
 354   # Clean out crunch_tmp/work, crunch_tmp/opt, crunch_tmp/src*
 355   Log (undef, "Clean work dirs");
 356
 357   my $cleanpid = fork();
 358   if ($cleanpid == 0)
 359   {
 360     # Find FUSE mounts that look like Keep mounts (the mount path has the
 361     # word "keep") and unmount them.  Then clean up work directories.
 362     # TODO: When #5036 is done and widely deployed, we can get rid of the
 363     # regular expression and just unmount everything with type fuse.keep.
 364     srun (["srun", "--nodelist=$nodelist", "-D", $ENV{'TMPDIR'}],
 365           ['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* $CRUNCH_TMP/*.cid']);
 366     exit (1);
 367   }
 368   while (1)
 369   {
 370     last if $cleanpid == waitpid (-1, WNOHANG);
 371     freeze_if_want_freeze ($cleanpid);
 372     select (undef, undef, undef, 0.1);
 373   }
 374   Log (undef, "Cleanup command exited ".exit_status_s($?));
 375 }
 376
 377 # If this job requires a Docker image, install that.
 378 my $docker_bin = "/usr/bin/docker.io";
 379 my ($docker_locator, $docker_stream, $docker_hash, $docker_limitmem);
 380 if ($docker_locator = $Job->{docker_image_locator}) {
 381   ($docker_stream, $docker_hash) = find_docker_image($docker_locator);
 382   if (!$docker_hash)
 383   {
 384     croak("No Docker image hash found from locator $docker_locator");
 385   }
 386   $docker_stream =~ s/^\.//;
 387   my $docker_install_script = qq{
 388 if ! $docker_bin images -q --no-trunc --all | grep -qxF \Q$docker_hash\E; then
 389     arv-get \Q$docker_locator$docker_stream/$docker_hash.tar\E | $docker_bin load
 390 fi
 391 };
 392   my $docker_pid = fork();
 393   if ($docker_pid == 0)
 394   {
 395     srun (["srun", "--nodelist=" . join(',', @node)],
 396           ["/bin/sh", "-ec", $docker_install_script]);
 397     exit ($?);
 398   }
 399   while (1)
 400   {
 401     last if $docker_pid == waitpid (-1, WNOHANG);
 402     freeze_if_want_freeze ($docker_pid);
 403     select (undef, undef, undef, 0.1);
 404   }
 405   if ($? != 0)
 406   {
 407     croak("Installing Docker image from $docker_locator exited "
 408           .exit_status_s($?));
 409   }
 410
 411   # Determine whether this version of Docker supports memory+swap limits.
 412   srun(["srun", "--nodelist=" . $node[0]],
 413        ["/bin/sh", "-ec", "$docker_bin run --help | grep -qe --memory-swap="],
 414       {fork => 1});
 415   $docker_limitmem = ($? == 0);
 416
 417   if ($Job->{arvados_sdk_version}) {
 418     # The job also specifies an Arvados SDK version.  Add the SDKs to the
 419     # tar file for the build script to install.
 420     Log(undef, sprintf("Packing Arvados SDK version %s for installation",
 421                        $Job->{arvados_sdk_version}));
 422     add_git_archive("git", "--git-dir=$git_dir", "archive",
 423                     "--prefix=.arvados.sdk/",
 424                     $Job->{arvados_sdk_version}, "sdk");
 425   }
 426 }
 427
 428 if (!defined $git_dir && $Job->{'script_version'} =~ m{^/}) {
 429   # If script_version looks like an absolute path, *and* the --git-dir
 430   # argument was not given -- which implies we were not invoked by
 431   # crunch-dispatch -- we will use the given path as a working
 432   # directory instead of resolving script_version to a git commit (or
 433   # doing anything else with git).
 434   $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{'script_version'};
 435   $ENV{"CRUNCH_SRC"} = $Job->{'script_version'};
 436 }
 437 else {
 438   # Resolve the given script_version to a git commit sha1. Also, if
 439   # the repository is remote, clone it into our local filesystem: this
 440   # ensures "git archive" will work, and is necessary to reliably
 441   # resolve a symbolic script_version like "master^".
 442   $ENV{"CRUNCH_SRC"} = "$ENV{CRUNCH_TMP}/src";
 443
 444   Log (undef, "Looking for version ".$Job->{script_version}." from repository ".$Job->{repository});
 445
 446   $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{script_version};
 447
 448   # If we're running under crunch-dispatch, it will have already
 449   # pulled the appropriate source tree into its own repository, and
 450   # given us that repo's path as $git_dir.
 451   #
 452   # If we're running a "local" job, we might have to fetch content
 453   # from a remote repository.
 454   #
 455   # (Currently crunch-dispatch gives a local path with --git-dir, but
 456   # we might as well accept URLs there too in case it changes its
 457   # mind.)
 458   my $repo = $git_dir || $Job->{'repository'};
 459
 460   # Repository can be remote or local. If remote, we'll need to fetch it
 461   # to a local dir before doing `git log` et al.
 462   my $repo_location;
 463
 464   if ($repo =~ m{://|^[^/]*:}) {
 465     # $repo is a git url we can clone, like git:// or https:// or
 466     # file:/// or [user@]host:repo.git. Note "user/name@host:foo" is
 467     # not recognized here because distinguishing that from a local
 468     # path is too fragile. If you really need something strange here,
 469     # use the ssh:// form.
 470     $repo_location = 'remote';
 471   } elsif ($repo =~ m{^\.*/}) {
 472     # $repo is a local path to a git index. We'll also resolve ../foo
 473     # to ../foo/.git if the latter is a directory. To help
 474     # disambiguate local paths from named hosted repositories, this
 475     # form must be given as ./ or ../ if it's a relative path.
 476     if (-d "$repo/.git") {
 477       $repo = "$repo/.git";
 478     }
 479     $repo_location = 'local';
 480   } else {
 481     # $repo is none of the above. It must be the name of a hosted
 482     # repository.
 483     my $arv_repo_list = api_call("repositories/list",
 484                                  'filters' => [['name','=',$repo]]);
 485     my @repos_found = @{$arv_repo_list->{'items'}};
 486     my $n_found = $arv_repo_list->{'serverResponse'}->{'items_available'};
 487     if ($n_found > 0) {
 488       Log(undef, "Repository '$repo' -> "
 489           . join(", ", map { $_->{'uuid'} } @repos_found));
 490     }
 491     if ($n_found != 1) {
 492       croak("Error: Found $n_found repositories with name '$repo'.");
 493     }
 494     $repo = $repos_found[0]->{'fetch_url'};
 495     $repo_location = 'remote';
 496   }
 497   Log(undef, "Using $repo_location repository '$repo'");
 498   $ENV{"CRUNCH_SRC_URL"} = $repo;
 499
 500   # Resolve given script_version (we'll call that $treeish here) to a
 501   # commit sha1 ($commit).
 502   my $treeish = $Job->{'script_version'};
 503   my $commit;
 504   if ($repo_location eq 'remote') {
 505     # We minimize excess object-fetching by re-using the same bare
 506     # repository in CRUNCH_TMP/.git for multiple crunch-jobs -- we
 507     # just keep adding remotes to it as needed.
 508     my $local_repo = $ENV{'CRUNCH_TMP'}."/.git";
 509     my $gitcmd = "git --git-dir=\Q$local_repo\E";
 510
 511     # Set up our local repo for caching remote objects, making
 512     # archives, etc.
 513     if (!-d $local_repo) {
 514       make_path($local_repo) or croak("Error: could not create $local_repo");
 515     }
 516     # This works (exits 0 and doesn't delete fetched objects) even
 517     # if $local_repo is already initialized:
 518     `$gitcmd init --bare`;
 519     if ($?) {
 520       croak("Error: $gitcmd init --bare exited ".exit_status_s($?));
 521     }
 522
 523     # If $treeish looks like a hash (or abbrev hash) we look it up in
 524     # our local cache first, since that's cheaper. (We don't want to
 525     # do that with tags/branches though -- those change over time, so
 526     # they should always be resolved by the remote repo.)
 527     if ($treeish =~ /^[0-9a-f]{7,40}$/s) {
 528       # Hide stderr because it's normal for this to fail:
 529       my $sha1 = `$gitcmd rev-list -n1 ''\Q$treeish\E 2>/dev/null`;
 530       if ($? == 0 &&
 531           # Careful not to resolve a branch named abcdeff to commit 1234567:
 532           $sha1 =~ /^$treeish/ &&
 533           $sha1 =~ /^([0-9a-f]{40})$/s) {
 534         $commit = $1;
 535         Log(undef, "Commit $commit already present in $local_repo");
 536       }
 537     }
 538
 539     if (!defined $commit) {
 540       # If $treeish isn't just a hash or abbrev hash, or isn't here
 541       # yet, we need to fetch the remote to resolve it correctly.
 542
 543       # First, remove all local heads. This prevents a name that does
 544       # not exist on the remote from resolving to (or colliding with)
 545       # a previously fetched branch or tag (possibly from a different
 546       # remote).
 547       remove_tree("$local_repo/refs/heads", {keep_root => 1});
 548
 549       Log(undef, "Fetching objects from $repo to $local_repo");
 550       `$gitcmd fetch --no-progress --tags ''\Q$repo\E \Q+refs/heads/*:refs/heads/*\E`;
 551       if ($?) {
 552         croak("Error: `$gitcmd fetch` exited ".exit_status_s($?));
 553       }
 554     }
 555
 556     # Now that the data is all here, we will use our local repo for
 557     # the rest of our git activities.
 558     $repo = $local_repo;
 559   }
 560
 561   my $gitcmd = "git --git-dir=\Q$repo\E";
 562   my $sha1 = `$gitcmd rev-list -n1 ''\Q$treeish\E`;
 563   unless ($? == 0 && $sha1 =~ /^([0-9a-f]{40})$/) {
 564     croak("`$gitcmd rev-list` exited "
 565           .exit_status_s($?)
 566           .", '$treeish' not found. Giving up.");
 567   }
 568   $commit = $1;
 569   Log(undef, "Version $treeish is commit $commit");
 570
 571   if ($commit ne $Job->{'script_version'}) {
 572     # Record the real commit id in the database, frozentokey, logs,
 573     # etc. -- instead of an abbreviation or a branch name which can
 574     # become ambiguous or point to a different commit in the future.
 575     if (!$Job->update_attributes('script_version' => $commit)) {
 576       croak("Error: failed to update job's script_version attribute");
 577     }
 578   }
 579
 580   $ENV{"CRUNCH_SRC_COMMIT"} = $commit;
 581   add_git_archive("$gitcmd archive ''\Q$commit\E");
 582 }
 583
 584 my $git_archive = combined_git_archive();
 585 if (!defined $git_archive) {
 586   Log(undef, "Skip install phase (no git archive)");
 587   if ($have_slurm) {
 588     Log(undef, "Warning: This probably means workers have no source tree!");
 589   }
 590 }
 591 else {
 592   Log(undef, "Run install script on all workers");
 593
 594   my @srunargs = ("srun",
 595                   "--nodelist=$nodelist",
 596                   "-D", $ENV{'TMPDIR'}, "--job-name=$job_id");
 597   my @execargs = ("sh", "-c",
 598                   "mkdir -p $ENV{CRUNCH_INSTALL} && cd $ENV{CRUNCH_TMP} && perl -");
 599
 600   $ENV{"CRUNCH_GIT_ARCHIVE_HASH"} = md5_hex($git_archive);
 601   my $installpid = fork();
 602   if ($installpid == 0)
 603   {
 604     srun (\@srunargs, \@execargs, {}, $build_script . $git_archive);
 605     exit (1);
 606   }
 607   while (1)
 608   {
 609     last if $installpid == waitpid (-1, WNOHANG);
 610     freeze_if_want_freeze ($installpid);
 611     select (undef, undef, undef, 0.1);
 612   }
 613   my $install_exited = $?;
 614   Log (undef, "Install script exited ".exit_status_s($install_exited));
 615   foreach my $tar_filename (map { tar_filename_n($_); } (1..$git_tar_count)) {
 616     unlink($tar_filename);
 617   }
 618   exit (1) if $install_exited != 0;
 619 }
 620
 621 foreach (qw (script script_version script_parameters runtime_constraints))
 622 {
 623   Log (undef,
 624        "$_ " .
 625        (ref($Job->{$_}) ? JSON::encode_json($Job->{$_}) : $Job->{$_}));
 626 }
 627 foreach (split (/\n/, $Job->{knobs}))
 628 {
 629   Log (undef, "knob " . $_);
 630 }
 631
 632
 633
 634 $main::success = undef;
 635
 636
 637
 638 ONELEVEL:
 639
 640 my $thisround_succeeded = 0;
 641 my $thisround_failed = 0;
 642 my $thisround_failed_multiple = 0;
 643
 644 @jobstep_todo = sort { $jobstep[$a]->{level} <=> $jobstep[$b]->{level}
 645                        or $a <=> $b } @jobstep_todo;
 646 my $level = $jobstep[$jobstep_todo[0]]->{level};
 647
 648 my $initial_tasks_this_level = 0;
 649 foreach my $id (@jobstep_todo) {
 650   $initial_tasks_this_level++ if ($jobstep[$id]->{level} == $level);
 651 }
 652
 653 # If the number of tasks scheduled at this level #T is smaller than the number
 654 # of slots available #S, only use the first #T slots, or the first slot on
 655 # each node, whichever number is greater.
 656 #
 657 # When we dispatch tasks later, we'll allocate whole-node resources like RAM
 658 # based on these numbers.  Using fewer slots makes more resources available
 659 # to each individual task, which should normally be a better strategy when
 660 # there are fewer of them running with less parallelism.
 661 #
 662 # Note that this calculation is not redone if the initial tasks at
 663 # this level queue more tasks at the same level.  This may harm
 664 # overall task throughput for that level.
 665 my @freeslot;
 666 if ($initial_tasks_this_level < @node) {
 667   @freeslot = (0..$#node);
 668 } elsif ($initial_tasks_this_level < @slot) {
 669   @freeslot = (0..$initial_tasks_this_level - 1);
 670 } else {
 671   @freeslot = (0..$#slot);
 672 }
 673 my $round_num_freeslots = scalar(@freeslot);
 674
 675 my %round_max_slots = ();
 676 for (my $ii = $#freeslot; $ii >= 0; $ii--) {
 677   my $this_slot = $slot[$freeslot[$ii]];
 678   my $node_name = $this_slot->{node}->{name};
 679   $round_max_slots{$node_name} ||= $this_slot->{cpu};
 680   last if (scalar(keys(%round_max_slots)) >= @node);
 681 }
 682
 683 Log(undef, "start level $level with $round_num_freeslots slots");
 684 my %proc;
 685 my @holdslot;
 686 my %reader;
 687 my $progress_is_dirty = 1;
 688 my $progress_stats_updated = 0;
 689
 690 update_progress_stats();
 691
 692
 693 THISROUND:
 694 for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++)
 695 {
 696   my $id = $jobstep_todo[$todo_ptr];
 697   my $Jobstep = $jobstep[$id];
 698   if ($Jobstep->{level} != $level)
 699   {
 700     next;
 701   }
 702
 703   pipe $reader{$id}, "writer" or croak ($!);
 704   my $flags = fcntl ($reader{$id}, F_GETFL, 0) or croak ($!);
 705   fcntl ($reader{$id}, F_SETFL, $flags | O_NONBLOCK) or croak ($!);
 706
 707   my $childslot = $freeslot[0];
 708   my $childnode = $slot[$childslot]->{node};
 709   my $childslotname = join (".",
 710                             $slot[$childslot]->{node}->{name},
 711                             $slot[$childslot]->{cpu});
 712
 713   my $childpid = fork();
 714   if ($childpid == 0)
 715   {
 716     $SIG{'INT'} = 'DEFAULT';
 717     $SIG{'QUIT'} = 'DEFAULT';
 718     $SIG{'TERM'} = 'DEFAULT';
 719
 720     foreach (values (%reader))
 721     {
 722       close($_);
 723     }
 724     fcntl ("writer", F_SETFL, 0) or croak ($!); # no close-on-exec
 725     open(STDOUT,">&writer");
 726     open(STDERR,">&writer");
 727
 728     undef $dbh;
 729     undef $sth;
 730
 731     delete $ENV{"GNUPGHOME"};
 732     $ENV{"TASK_UUID"} = $Jobstep->{'arvados_task'}->{'uuid'};
 733     $ENV{"TASK_QSEQUENCE"} = $id;
 734     $ENV{"TASK_SEQUENCE"} = $level;
 735     $ENV{"JOB_SCRIPT"} = $Job->{script};
 736     while (my ($param, $value) = each %{$Job->{script_parameters}}) {
 737       $param =~ tr/a-z/A-Z/;
 738       $ENV{"JOB_PARAMETER_$param"} = $value;
 739     }
 740     $ENV{"TASK_SLOT_NODE"} = $slot[$childslot]->{node}->{name};
 741     $ENV{"TASK_SLOT_NUMBER"} = $slot[$childslot]->{cpu};
 742     $ENV{"TASK_WORK"} = $ENV{"CRUNCH_TMP"}."/task/$childslotname";
 743     $ENV{"HOME"} = $ENV{"TASK_WORK"};
 744     $ENV{"TASK_KEEPMOUNT"} = $ENV{"TASK_WORK"}.".keep";
 745     $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
 746     $ENV{"CRUNCH_NODE_SLOTS"} = $round_max_slots{$ENV{TASK_SLOT_NODE}};
 747     $ENV{"PATH"} = $ENV{"CRUNCH_INSTALL"} . "/bin:" . $ENV{"PATH"};
 748
 749     $ENV{"GZIP"} = "-n";
 750
 751     my @srunargs = (
 752       "srun",
 753       "--nodelist=".$childnode->{name},
 754       qw(-n1 -c1 -N1 -D), $ENV{'TMPDIR'},
 755       "--job-name=$job_id.$id.$$",
 756         );
 757     my $command =
 758         "if [ -e $ENV{TASK_WORK} ]; then rm -rf $ENV{TASK_WORK}; fi; "
 759         ."mkdir -p $ENV{CRUNCH_TMP} $ENV{JOB_WORK} $ENV{TASK_WORK} $ENV{TASK_KEEPMOUNT} "
 760         ."&& cd $ENV{CRUNCH_TMP} "
 761         # These environment variables get used explicitly later in
 762         # $command.  No tool is expected to read these values directly.
 763         .q{&& MEM=$(awk '($1 == "MemTotal:"){print $2}' </proc/meminfo) }
 764         .q{&& SWAP=$(awk '($1 == "SwapTotal:"){print $2}' </proc/meminfo) }
 765         ."&& MEMLIMIT=\$(( (\$MEM * 95) / ($ENV{CRUNCH_NODE_SLOTS} * 100) )) "
 766         ."&& let SWAPLIMIT=\$MEMLIMIT+\$SWAP ";
 767     $command .= "&& exec arv-mount --by-id --allow-other $ENV{TASK_KEEPMOUNT} --exec ";
 768     if ($docker_hash)
 769     {
 770       my $cidfile = "$ENV{CRUNCH_TMP}/$Jobstep->{arvados_task}->{uuid}-$Jobstep->{failures}.cid";
 771       $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -cgroup-parent=docker -cgroup-cid=$cidfile -poll=10000 ";
 772       $command .= "$docker_bin run --rm=true --attach=stdout --attach=stderr --attach=stdin -i --user=crunch --cidfile=$cidfile --sig-proxy ";
 773       # We only set memory limits if Docker lets us limit both memory and swap.
 774       # Memory limits alone have been supported longer, but subprocesses tend
 775       # to get SIGKILL if they exceed that without any swap limit set.
 776       # See #5642 for additional background.
 777       if ($docker_limitmem) {
 778         $command .= "--memory=\${MEMLIMIT}k --memory-swap=\${SWAPLIMIT}k ";
 779       }
 780
 781       # Dynamically configure the container to use the host system as its
 782       # DNS server.  Get the host's global addresses from the ip command,
 783       # and turn them into docker --dns options using gawk.
 784       $command .=
 785           q{$(ip -o address show scope global |
 786               gawk 'match($4, /^([0-9\.:]+)\//, x){print "--dns", x[1]}') };
 787
 788       # The source tree and $destdir directory (which we have
 789       # installed on the worker host) are available in the container,
 790       # under the same path.
 791       $command .= "--volume=\Q$ENV{CRUNCH_SRC}:$ENV{CRUNCH_SRC}:ro\E ";
 792       $command .= "--volume=\Q$ENV{CRUNCH_INSTALL}:$ENV{CRUNCH_INSTALL}:ro\E ";
 793
 794       # Currently, we make arv-mount's mount point appear at /keep
 795       # inside the container (instead of using the same path as the
 796       # host like we do with CRUNCH_SRC and CRUNCH_INSTALL). However,
 797       # crunch scripts and utilities must not rely on this. They must
 798       # use $TASK_KEEPMOUNT.
 799       $command .= "--volume=\Q$ENV{TASK_KEEPMOUNT}:/keep:ro\E ";
 800       $ENV{TASK_KEEPMOUNT} = "/keep";
 801
 802       # TASK_WORK is almost exactly like a docker data volume: it
 803       # starts out empty, is writable, and persists until no
 804       # containers use it any more. We don't use --volumes-from to
 805       # share it with other containers: it is only accessible to this
 806       # task, and it goes away when this task stops.
 807       #
 808       # However, a docker data volume is writable only by root unless
 809       # the mount point already happens to exist in the container with
 810       # different permissions. Therefore, we [1] assume /tmp already
 811       # exists in the image and is writable by the crunch user; [2]
 812       # avoid putting TASK_WORK inside CRUNCH_TMP (which won't be
 813       # writable if they are created by docker while setting up the
 814       # other --volumes); and [3] create $TASK_WORK inside the
 815       # container using $build_script.
 816       $command .= "--volume=/tmp ";
 817       $ENV{"TASK_WORK"} = "/tmp/crunch-job-task-work/$childslotname";
 818       $ENV{"HOME"} = $ENV{"TASK_WORK"};
 819       $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
 820
 821       # TODO: Share a single JOB_WORK volume across all task
 822       # containers on a given worker node, and delete it when the job
 823       # ends (and, in case that doesn't work, when the next job
 824       # starts).
 825       #
 826       # For now, use the same approach as TASK_WORK above.
 827       $ENV{"JOB_WORK"} = "/tmp/crunch-job-work";
 828
 829       while (my ($env_key, $env_val) = each %ENV)
 830       {
 831         if ($env_key =~ /^(ARVADOS|CRUNCH|JOB|TASK)_/) {
 832           $command .= "--env=\Q$env_key=$env_val\E ";
 833         }
 834       }
 835       $command .= "--env=\QHOME=$ENV{HOME}\E ";
 836       $command .= "\Q$docker_hash\E ";
 837       $command .= "stdbuf --output=0 --error=0 ";
 838       $command .= "perl - $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
 839     } else {
 840       # Non-docker run
 841       $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -poll=10000 ";
 842       $command .= "stdbuf --output=0 --error=0 ";
 843       $command .= "perl - $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
 844     }
 845
 846     my @execargs = ('bash', '-c', $command);
 847     srun (\@srunargs, \@execargs, undef, $build_script);
 848     # exec() failed, we assume nothing happened.
 849     die "srun() failed on build script\n";
 850   }
 851   close("writer");
 852   if (!defined $childpid)
 853   {
 854     close $reader{$id};
 855     delete $reader{$id};
 856     next;
 857   }
 858   shift @freeslot;
 859   $proc{$childpid} = { jobstep => $id,
 860                        time => time,
 861                        slot => $childslot,
 862                        jobstepname => "$job_id.$id.$childpid",
 863                      };
 864   croak ("assert failed: \$slot[$childslot]->{'pid'} exists") if exists $slot[$childslot]->{pid};
 865   $slot[$childslot]->{pid} = $childpid;
 866
 867   Log ($id, "job_task ".$Jobstep->{'arvados_task'}->{'uuid'});
 868   Log ($id, "child $childpid started on $childslotname");
 869   $Jobstep->{starttime} = time;
 870   $Jobstep->{node} = $childnode->{name};
 871   $Jobstep->{slotindex} = $childslot;
 872   delete $Jobstep->{stderr};
 873   delete $Jobstep->{finishtime};
 874
 875   $Jobstep->{'arvados_task'}->{started_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{starttime});
 876   $Jobstep->{'arvados_task'}->save;
 877
 878   splice @jobstep_todo, $todo_ptr, 1;
 879   --$todo_ptr;
 880
 881   $progress_is_dirty = 1;
 882
 883   while (!@freeslot
 884          ||
 885          ($round_num_freeslots > @freeslot && $todo_ptr+1 > $#jobstep_todo))
 886   {
 887     last THISROUND if $main::please_freeze || defined($main::success);
 888     if ($main::please_info)
 889     {
 890       $main::please_info = 0;
 891       freeze();
 892       create_output_collection();
 893       save_meta(1);
 894       update_progress_stats();
 895     }
 896     my $gotsome
 897         = readfrompipes ()
 898         + reapchildren ();
 899     if (!$gotsome)
 900     {
 901       check_refresh_wanted();
 902       check_squeue();
 903       update_progress_stats();
 904       select (undef, undef, undef, 0.1);
 905     }
 906     elsif (time - $progress_stats_updated >= 30)
 907     {
 908       update_progress_stats();
 909     }
 910     if (($thisround_failed_multiple >= 8 && $thisround_succeeded == 0) ||
 911         ($thisround_failed_multiple >= 16 && $thisround_failed_multiple > $thisround_succeeded))
 912     {
 913       my $message = "Repeated failure rate too high ($thisround_failed_multiple/"
 914           .($thisround_failed+$thisround_succeeded)
 915           .") -- giving up on this round";
 916       Log (undef, $message);
 917       last THISROUND;
 918     }
 919
 920     # move slots from freeslot to holdslot (or back to freeslot) if necessary
 921     for (my $i=$#freeslot; $i>=0; $i--) {
 922       if ($slot[$freeslot[$i]]->{node}->{hold_until} > scalar time) {
 923         push @holdslot, (splice @freeslot, $i, 1);
 924       }
 925     }
 926     for (my $i=$#holdslot; $i>=0; $i--) {
 927       if ($slot[$holdslot[$i]]->{node}->{hold_until} <= scalar time) {
 928         push @freeslot, (splice @holdslot, $i, 1);
 929       }
 930     }
 931
 932     # give up if no nodes are succeeding
 933     if (!grep { $_->{node}->{losing_streak} == 0 &&
 934                     $_->{node}->{hold_count} < 4 } @slot) {
 935       my $message = "Every node has failed -- giving up on this round";
 936       Log (undef, $message);
 937       last THISROUND;
 938     }
 939   }
 940 }
 941
 942
 943 push @freeslot, splice @holdslot;
 944 map { $slot[$freeslot[$_]]->{node}->{losing_streak} = 0 } (0..$#freeslot);
 945
 946
 947 Log (undef, "wait for last ".(scalar keys %proc)." children to finish");
 948 while (%proc)
 949 {
 950   if ($main::please_continue) {
 951     $main::please_continue = 0;
 952     goto THISROUND;
 953   }
 954   $main::please_info = 0, freeze(), create_output_collection(), save_meta(1) if $main::please_info;
 955   readfrompipes ();
 956   if (!reapchildren())
 957   {
 958     check_refresh_wanted();
 959     check_squeue();
 960     update_progress_stats();
 961     select (undef, undef, undef, 0.1);
 962     killem (keys %proc) if $main::please_freeze;
 963   }
 964 }
 965
 966 update_progress_stats();
 967 freeze_if_want_freeze();
 968
 969
 970 if (!defined $main::success)
 971 {
 972   if (@jobstep_todo &&
 973       $thisround_succeeded == 0 &&
 974       ($thisround_failed == 0 || $thisround_failed > 4))
 975   {
 976     my $message = "stop because $thisround_failed tasks failed and none succeeded";
 977     Log (undef, $message);
 978     $main::success = 0;
 979   }
 980   if (!@jobstep_todo)
 981   {
 982     $main::success = 1;
 983   }
 984 }
 985
 986 goto ONELEVEL if !defined $main::success;
 987
 988
 989 release_allocation();
 990 freeze();
 991 my $collated_output = &create_output_collection();
 992
 993 if (!$collated_output) {
 994   Log (undef, "Failed to write output collection");
 995 }
 996 else {
 997   Log(undef, "job output $collated_output");
 998   $Job->update_attributes('output' => $collated_output);
 999 }
1000
1001 Log (undef, "finish");
1002
1003 save_meta();
1004
1005 my $final_state;
1006 if ($collated_output && $main::success) {
1007   $final_state = 'Complete';
1008 } else {
1009   $final_state = 'Failed';
1010 }
1011 $Job->update_attributes('state' => $final_state);
1012
1013 exit (($final_state eq 'Complete') ? 0 : 1);
1014
1015
1016
1017 sub update_progress_stats
1018 {
1019   $progress_stats_updated = time;
1020   return if !$progress_is_dirty;
1021   my ($todo, $done, $running) = (scalar @jobstep_todo,
1022                                  scalar @jobstep_done,
1023                                  scalar @slot - scalar @freeslot - scalar @holdslot);
1024   $Job->{'tasks_summary'} ||= {};
1025   $Job->{'tasks_summary'}->{'todo'} = $todo;
1026   $Job->{'tasks_summary'}->{'done'} = $done;
1027   $Job->{'tasks_summary'}->{'running'} = $running;
1028   $Job->update_attributes('tasks_summary' => $Job->{'tasks_summary'});
1029   Log (undef, "status: $done done, $running running, $todo todo");
1030   $progress_is_dirty = 0;
1031 }
1032
1033
1034
1035 sub reapchildren
1036 {
1037   my $pid = waitpid (-1, WNOHANG);
1038   return 0 if $pid <= 0;
1039
1040   my $whatslot = ($slot[$proc{$pid}->{slot}]->{node}->{name}
1041                   . "."
1042                   . $slot[$proc{$pid}->{slot}]->{cpu});
1043   my $jobstepid = $proc{$pid}->{jobstep};
1044   my $elapsed = time - $proc{$pid}->{time};
1045   my $Jobstep = $jobstep[$jobstepid];
1046
1047   my $childstatus = $?;
1048   my $exitvalue = $childstatus >> 8;
1049   my $exitinfo = "exit ".exit_status_s($childstatus);
1050   $Jobstep->{'arvados_task'}->reload;
1051   my $task_success = $Jobstep->{'arvados_task'}->{success};
1052
1053   Log ($jobstepid, "child $pid on $whatslot $exitinfo success=$task_success");
1054
1055   if (!defined $task_success) {
1056     # task did not indicate one way or the other --> fail
1057     $Jobstep->{'arvados_task'}->{success} = 0;
1058     $Jobstep->{'arvados_task'}->save;
1059     $task_success = 0;
1060   }
1061
1062   if (!$task_success)
1063   {
1064     my $temporary_fail;
1065     $temporary_fail ||= $Jobstep->{node_fail};
1066     $temporary_fail ||= ($exitvalue == TASK_TEMPFAIL);
1067
1068     ++$thisround_failed;
1069     ++$thisround_failed_multiple if $Jobstep->{'failures'} >= 1;
1070
1071     # Check for signs of a failed or misconfigured node
1072     if (++$slot[$proc{$pid}->{slot}]->{node}->{losing_streak} >=
1073         2+$slot[$proc{$pid}->{slot}]->{node}->{ncpus}) {
1074       # Don't count this against jobstep failure thresholds if this
1075       # node is already suspected faulty and srun exited quickly
1076       if ($slot[$proc{$pid}->{slot}]->{node}->{hold_until} &&
1077           $elapsed < 5) {
1078         Log ($jobstepid, "blaming failure on suspect node " .
1079              $slot[$proc{$pid}->{slot}]->{node}->{name});
1080         $temporary_fail ||= 1;
1081       }
1082       ban_node_by_slot($proc{$pid}->{slot});
1083     }
1084
1085     Log ($jobstepid, sprintf('failure (#%d, %s) after %d seconds',
1086                              ++$Jobstep->{'failures'},
1087                              $temporary_fail ? 'temporary ' : 'permanent',
1088                              $elapsed));
1089
1090     if (!$temporary_fail || $Jobstep->{'failures'} >= 3) {
1091       # Give up on this task, and the whole job
1092       $main::success = 0;
1093     }
1094     # Put this task back on the todo queue
1095     push @jobstep_todo, $jobstepid;
1096     $Job->{'tasks_summary'}->{'failed'}++;
1097   }
1098   else
1099   {
1100     ++$thisround_succeeded;
1101     $slot[$proc{$pid}->{slot}]->{node}->{losing_streak} = 0;
1102     $slot[$proc{$pid}->{slot}]->{node}->{hold_until} = 0;
1103     push @jobstep_done, $jobstepid;
1104     Log ($jobstepid, "success in $elapsed seconds");
1105   }
1106   $Jobstep->{exitcode} = $childstatus;
1107   $Jobstep->{finishtime} = time;
1108   $Jobstep->{'arvados_task'}->{finished_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{finishtime});
1109   $Jobstep->{'arvados_task'}->save;
1110   process_stderr ($jobstepid, $task_success);
1111   Log ($jobstepid, sprintf("task output (%d bytes): %s",
1112                            length($Jobstep->{'arvados_task'}->{output}),
1113                            $Jobstep->{'arvados_task'}->{output}));
1114
1115   close $reader{$jobstepid};
1116   delete $reader{$jobstepid};
1117   delete $slot[$proc{$pid}->{slot}]->{pid};
1118   push @freeslot, $proc{$pid}->{slot};
1119   delete $proc{$pid};
1120
1121   if ($task_success) {
1122     # Load new tasks
1123     my $newtask_list = [];
1124     my $newtask_results;
1125     do {
1126       $newtask_results = api_call(
1127         "job_tasks/list",
1128         'where' => {
1129           'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid}
1130         },
1131         'order' => 'qsequence',
1132         'offset' => scalar(@$newtask_list),
1133       );
1134       push(@$newtask_list, @{$newtask_results->{items}});
1135     } while (@{$newtask_results->{items}});
1136     foreach my $arvados_task (@$newtask_list) {
1137       my $jobstep = {
1138         'level' => $arvados_task->{'sequence'},
1139         'failures' => 0,
1140         'arvados_task' => $arvados_task
1141       };
1142       push @jobstep, $jobstep;
1143       push @jobstep_todo, $#jobstep;
1144     }
1145   }
1146
1147   $progress_is_dirty = 1;
1148   1;
1149 }
1150
1151 sub check_refresh_wanted
1152 {
1153   my @stat = stat $ENV{"CRUNCH_REFRESH_TRIGGER"};
1154   if (@stat && $stat[9] > $latest_refresh) {
1155     $latest_refresh = scalar time;
1156     my $Job2 = api_call("jobs/get", uuid => $jobspec);
1157     for my $attr ('cancelled_at',
1158                   'cancelled_by_user_uuid',
1159                   'cancelled_by_client_uuid',
1160                   'state') {
1161       $Job->{$attr} = $Job2->{$attr};
1162     }
1163     if ($Job->{'state'} ne "Running") {
1164       if ($Job->{'state'} eq "Cancelled") {
1165         Log (undef, "Job cancelled at " . $Job->{'cancelled_at'} . " by user " . $Job->{'cancelled_by_user_uuid'});
1166       } else {
1167         Log (undef, "Job state unexpectedly changed to " . $Job->{'state'});
1168       }
1169       $main::success = 0;
1170       $main::please_freeze = 1;
1171     }
1172   }
1173 }
1174
1175 sub check_squeue
1176 {
1177   # return if the kill list was checked <4 seconds ago
1178   if (defined $squeue_kill_checked && $squeue_kill_checked > time - 4)
1179   {
1180     return;
1181   }
1182   $squeue_kill_checked = time;
1183
1184   # use killem() on procs whose killtime is reached
1185   for (keys %proc)
1186   {
1187     if (exists $proc{$_}->{killtime}
1188         && $proc{$_}->{killtime} <= time)
1189     {
1190       killem ($_);
1191     }
1192   }
1193
1194   # return if the squeue was checked <60 seconds ago
1195   if (defined $squeue_checked && $squeue_checked > time - 60)
1196   {
1197     return;
1198   }
1199   $squeue_checked = time;
1200
1201   if (!$have_slurm)
1202   {
1203     # here is an opportunity to check for mysterious problems with local procs
1204     return;
1205   }
1206
1207   # get a list of steps still running
1208   my @squeue = `squeue -s -h -o '%i %j' && echo ok`;
1209   chop @squeue;
1210   if ($squeue[-1] ne "ok")
1211   {
1212     return;
1213   }
1214   pop @squeue;
1215
1216   # which of my jobsteps are running, according to squeue?
1217   my %ok;
1218   foreach (@squeue)
1219   {
1220     if (/^(\d+)\.(\d+) (\S+)/)
1221     {
1222       if ($1 eq $ENV{SLURM_JOBID})
1223       {
1224         $ok{$3} = 1;
1225       }
1226     }
1227   }
1228
1229   # which of my active child procs (>60s old) were not mentioned by squeue?
1230   foreach (keys %proc)
1231   {
1232     if ($proc{$_}->{time} < time - 60
1233         && !exists $ok{$proc{$_}->{jobstepname}}
1234         && !exists $proc{$_}->{killtime})
1235     {
1236       # kill this proc if it hasn't exited in 30 seconds
1237       $proc{$_}->{killtime} = time + 30;
1238     }
1239   }
1240 }
1241
1242
1243 sub release_allocation
1244 {
1245   if ($have_slurm)
1246   {
1247     Log (undef, "release job allocation");
1248     system "scancel $ENV{SLURM_JOBID}";
1249   }
1250 }
1251
1252
1253 sub readfrompipes
1254 {
1255   my $gotsome = 0;
1256   foreach my $job (keys %reader)
1257   {
1258     my $buf;
1259     while (0 < sysread ($reader{$job}, $buf, 8192))
1260     {
1261       print STDERR $buf if $ENV{CRUNCH_DEBUG};
1262       $jobstep[$job]->{stderr} .= $buf;
1263       preprocess_stderr ($job);
1264       if (length ($jobstep[$job]->{stderr}) > 16384)
1265       {
1266         substr ($jobstep[$job]->{stderr}, 0, 8192) = "";
1267       }
1268       $gotsome = 1;
1269     }
1270   }
1271   return $gotsome;
1272 }
1273
1274
1275 sub preprocess_stderr
1276 {
1277   my $job = shift;
1278
1279   while ($jobstep[$job]->{stderr} =~ /^(.*?)\n/) {
1280     my $line = $1;
1281     substr $jobstep[$job]->{stderr}, 0, 1+length($line), "";
1282     Log ($job, "stderr $line");
1283     if ($line =~ /srun: error: (SLURM job $ENV{SLURM_JOB_ID} has expired|Unable to confirm allocation for job $ENV{SLURM_JOB_ID})/) {
1284       # whoa.
1285       $main::please_freeze = 1;
1286     }
1287     elsif ($line =~ /(srun: error: (Node failure on|Unable to create job step|.*: Communication connection failure))|arvados.errors.Keep/) {
1288       $jobstep[$job]->{node_fail} = 1;
1289       ban_node_by_slot($jobstep[$job]->{slotindex});
1290     }
1291   }
1292 }
1293
1294
1295 sub process_stderr
1296 {
1297   my $job = shift;
1298   my $task_success = shift;
1299   preprocess_stderr ($job);
1300
1301   map {
1302     Log ($job, "stderr $_");
1303   } split ("\n", $jobstep[$job]->{stderr});
1304 }
1305
1306 sub fetch_block
1307 {
1308   my $hash = shift;
1309   my $keep;
1310   if (!open($keep, "-|", "arv-get", "--retries", retry_count(), $hash)) {
1311     Log(undef, "fetch_block run error from arv-get $hash: $!");
1312     return undef;
1313   }
1314   my $output_block = "";
1315   while (1) {
1316     my $buf;
1317     my $bytes = sysread($keep, $buf, 1024 * 1024);
1318     if (!defined $bytes) {
1319       Log(undef, "fetch_block read error from arv-get: $!");
1320       $output_block = undef;
1321       last;
1322     } elsif ($bytes == 0) {
1323       # sysread returns 0 at the end of the pipe.
1324       last;
1325     } else {
1326       # some bytes were read into buf.
1327       $output_block .= $buf;
1328     }
1329   }
1330   close $keep;
1331   if ($?) {
1332     Log(undef, "fetch_block arv-get exited " . exit_status_s($?));
1333     $output_block = undef;
1334   }
1335   return $output_block;
1336 }
1337
1338 # Create a collection by concatenating the output of all tasks (each
1339 # task's output is either a manifest fragment, a locator for a
1340 # manifest fragment stored in Keep, or nothing at all). Return the
1341 # portable_data_hash of the new collection.
1342 sub create_output_collection
1343 {
1344   Log (undef, "collate");
1345
1346   my ($child_out, $child_in);
1347   my $pid = open2($child_out, $child_in, 'python', '-c', q{
1348 import arvados
1349 import sys
1350 print (arvados.api("v1").collections().
1351        create(body={"manifest_text": sys.stdin.read()}).
1352        execute(num_retries=int(sys.argv[1]))["portable_data_hash"])
1353 }, retry_count());
1354
1355   my $task_idx = -1;
1356   my $manifest_size = 0;
1357   for (@jobstep)
1358   {
1359     ++$task_idx;
1360     my $output = $_->{'arvados_task'}->{output};
1361     next if (!defined($output));
1362     my $next_write;
1363     if ($output =~ /^[0-9a-f]{32}(\+\S+)*$/) {
1364       $next_write = fetch_block($output);
1365     } else {
1366       $next_write = $output;
1367     }
1368     if (defined($next_write)) {
1369       if (!defined(syswrite($child_in, $next_write))) {
1370         # There's been an error writing.  Stop the loop.
1371         # We'll log details about the exit code later.
1372         last;
1373       } else {
1374         $manifest_size += length($next_write);
1375       }
1376     } else {
1377       my $uuid = $_->{'arvados_task'}->{'uuid'};
1378       Log (undef, "Error retrieving '$output' output by task $task_idx ($uuid)");
1379       $main::success = 0;
1380     }
1381   }
1382   close($child_in);
1383   Log(undef, "collated output manifest text to send to API server is $manifest_size bytes with access tokens");
1384
1385   my $joboutput;
1386   my $s = IO::Select->new($child_out);
1387   if ($s->can_read(120)) {
1388     sysread($child_out, $joboutput, 1024 * 1024);
1389     waitpid($pid, 0);
1390     if ($?) {
1391       Log(undef, "output collection creation exited " . exit_status_s($?));
1392       $joboutput = undef;
1393     } else {
1394       chomp($joboutput);
1395     }
1396   } else {
1397     Log (undef, "timed out while creating output collection");
1398     foreach my $signal (2, 2, 2, 15, 15, 9) {
1399       kill($signal, $pid);
1400       last if waitpid($pid, WNOHANG) == -1;
1401       sleep(1);
1402     }
1403   }
1404   close($child_out);
1405
1406   return $joboutput;
1407 }
1408
1409
1410 sub killem
1411 {
1412   foreach (@_)
1413   {
1414     my $sig = 2;                # SIGINT first
1415     if (exists $proc{$_}->{"sent_$sig"} &&
1416         time - $proc{$_}->{"sent_$sig"} > 4)
1417     {
1418       $sig = 15;                # SIGTERM if SIGINT doesn't work
1419     }
1420     if (exists $proc{$_}->{"sent_$sig"} &&
1421         time - $proc{$_}->{"sent_$sig"} > 4)
1422     {
1423       $sig = 9;                 # SIGKILL if SIGTERM doesn't work
1424     }
1425     if (!exists $proc{$_}->{"sent_$sig"})
1426     {
1427       Log ($proc{$_}->{jobstep}, "sending 2x signal $sig to pid $_");
1428       kill $sig, $_;
1429       select (undef, undef, undef, 0.1);
1430       if ($sig == 2)
1431       {
1432         kill $sig, $_;     # srun wants two SIGINT to really interrupt
1433       }
1434       $proc{$_}->{"sent_$sig"} = time;
1435       $proc{$_}->{"killedafter"} = time - $proc{$_}->{"time"};
1436     }
1437   }
1438 }
1439
1440
1441 sub fhbits
1442 {
1443   my($bits);
1444   for (@_) {
1445     vec($bits,fileno($_),1) = 1;
1446   }
1447   $bits;
1448 }
1449
1450
1451 # Send log output to Keep via arv-put.
1452 #
1453 # $log_pipe_in and $log_pipe_out are the input and output filehandles to the arv-put pipe.
1454 # $log_pipe_pid is the pid of the arv-put subprocess.
1455 #
1456 # The only functions that should access these variables directly are:
1457 #
1458 # log_writer_start($logfilename)
1459 #     Starts an arv-put pipe, reading data on stdin and writing it to
1460 #     a $logfilename file in an output collection.
1461 #
1462 # log_writer_send($txt)
1463 #     Writes $txt to the output log collection.
1464 #
1465 # log_writer_finish()
1466 #     Closes the arv-put pipe and returns the output that it produces.
1467 #
1468 # log_writer_is_active()
1469 #     Returns a true value if there is currently a live arv-put
1470 #     process, false otherwise.
1471 #
1472 my ($log_pipe_in, $log_pipe_out, $log_pipe_pid);
1473
1474 sub log_writer_start($)
1475 {
1476   my $logfilename = shift;
1477   $log_pipe_pid = open2($log_pipe_out, $log_pipe_in,
1478                         'arv-put',
1479                         '--portable-data-hash',
1480                         '--project-uuid', $Job->{owner_uuid},
1481                         '--retries', '3',
1482                         '--name', $logfilename,
1483                         '--filename', $logfilename,
1484                         '-');
1485 }
1486
1487 sub log_writer_send($)
1488 {
1489   my $txt = shift;
1490   print $log_pipe_in $txt;
1491 }
1492
1493 sub log_writer_finish()
1494 {
1495   return unless $log_pipe_pid;
1496
1497   close($log_pipe_in);
1498   my $arv_put_output;
1499
1500   my $s = IO::Select->new($log_pipe_out);
1501   if ($s->can_read(120)) {
1502     sysread($log_pipe_out, $arv_put_output, 1024);
1503     chomp($arv_put_output);
1504   } else {
1505     Log (undef, "timed out reading from 'arv-put'");
1506   }
1507
1508   waitpid($log_pipe_pid, 0);
1509   $log_pipe_pid = $log_pipe_in = $log_pipe_out = undef;
1510   if ($?) {
1511     Log("log_writer_finish: arv-put exited ".exit_status_s($?))
1512   }
1513
1514   return $arv_put_output;
1515 }
1516
1517 sub log_writer_is_active() {
1518   return $log_pipe_pid;
1519 }
1520
1521 sub Log                         # ($jobstep_id, $logmessage)
1522 {
1523   if ($_[1] =~ /\n/) {
1524     for my $line (split (/\n/, $_[1])) {
1525       Log ($_[0], $line);
1526     }
1527     return;
1528   }
1529   my $fh = select STDERR; $|=1; select $fh;
1530   my $message = sprintf ("%s %d %s %s", $job_id, $$, @_);
1531   $message =~ s{([^ -\176])}{"\\" . sprintf ("%03o", ord($1))}ge;
1532   $message .= "\n";
1533   my $datetime;
1534   if (log_writer_is_active() || -t STDERR) {
1535     my @gmtime = gmtime;
1536     $datetime = sprintf ("%04d-%02d-%02d_%02d:%02d:%02d",
1537                          $gmtime[5]+1900, $gmtime[4]+1, @gmtime[3,2,1,0]);
1538   }
1539   print STDERR ((-t STDERR) ? ($datetime." ".$message) : $message);
1540
1541   if (log_writer_is_active()) {
1542     log_writer_send($datetime . " " . $message);
1543   }
1544 }
1545
1546
1547 sub croak
1548 {
1549   my ($package, $file, $line) = caller;
1550   my $message = "@_ at $file line $line\n";
1551   Log (undef, $message);
1552   freeze() if @jobstep_todo;
1553   create_output_collection() if @jobstep_todo;
1554   cleanup();
1555   save_meta();
1556   die;
1557 }
1558
1559
1560 sub cleanup
1561 {
1562   return unless $Job;
1563   if ($Job->{'state'} eq 'Cancelled') {
1564     $Job->update_attributes('finished_at' => scalar gmtime);
1565   } else {
1566     $Job->update_attributes('state' => 'Failed');
1567   }
1568 }
1569
1570
1571 sub save_meta
1572 {
1573   my $justcheckpoint = shift; # false if this will be the last meta saved
1574   return if $justcheckpoint;  # checkpointing is not relevant post-Warehouse.pm
1575   return unless log_writer_is_active();
1576
1577   my $loglocator = log_writer_finish();
1578   Log (undef, "log manifest is $loglocator");
1579   $Job->{'log'} = $loglocator;
1580   $Job->update_attributes('log', $loglocator);
1581 }
1582
1583
1584 sub freeze_if_want_freeze
1585 {
1586   if ($main::please_freeze)
1587   {
1588     release_allocation();
1589     if (@_)
1590     {
1591       # kill some srun procs before freeze+stop
1592       map { $proc{$_} = {} } @_;
1593       while (%proc)
1594       {
1595         killem (keys %proc);
1596         select (undef, undef, undef, 0.1);
1597         my $died;
1598         while (($died = waitpid (-1, WNOHANG)) > 0)
1599         {
1600           delete $proc{$died};
1601         }
1602       }
1603     }
1604     freeze();
1605     create_output_collection();
1606     cleanup();
1607     save_meta();
1608     exit 1;
1609   }
1610 }
1611
1612
1613 sub freeze
1614 {
1615   Log (undef, "Freeze not implemented");
1616   return;
1617 }
1618
1619
1620 sub thaw
1621 {
1622   croak ("Thaw not implemented");
1623 }
1624
1625
1626 sub freezequote
1627 {
1628   my $s = shift;
1629   $s =~ s/\\/\\\\/g;
1630   $s =~ s/\n/\\n/g;
1631   return $s;
1632 }
1633
1634
1635 sub freezeunquote
1636 {
1637   my $s = shift;
1638   $s =~ s{\\(.)}{$1 eq "n" ? "\n" : $1}ge;
1639   return $s;
1640 }
1641
1642
1643 sub srun
1644 {
1645   my $srunargs = shift;
1646   my $execargs = shift;
1647   my $opts = shift || {};
1648   my $stdin = shift;
1649   my $args = $have_slurm ? [@$srunargs, @$execargs] : $execargs;
1650
1651   $Data::Dumper::Terse = 1;
1652   $Data::Dumper::Indent = 0;
1653   my $show_cmd = Dumper($args);
1654   $show_cmd =~ s/(TOKEN\\*=)[^\s\']+/${1}[...]/g;
1655   $show_cmd =~ s/\n/ /g;
1656   warn "starting: $show_cmd\n";
1657
1658   if (defined $stdin) {
1659     my $child = open STDIN, "-|";
1660     defined $child or die "no fork: $!";
1661     if ($child == 0) {
1662       print $stdin or die $!;
1663       close STDOUT or die $!;
1664       exit 0;
1665     }
1666   }
1667
1668   return system (@$args) if $opts->{fork};
1669
1670   exec @$args;
1671   warn "ENV size is ".length(join(" ",%ENV));
1672   die "exec failed: $!: @$args";
1673 }
1674
1675
1676 sub ban_node_by_slot {
1677   # Don't start any new jobsteps on this node for 60 seconds
1678   my $slotid = shift;
1679   $slot[$slotid]->{node}->{hold_until} = 60 + scalar time;
1680   $slot[$slotid]->{node}->{hold_count}++;
1681   Log (undef, "backing off node " . $slot[$slotid]->{node}->{name} . " for 60 seconds");
1682 }
1683
1684 sub must_lock_now
1685 {
1686   my ($lockfile, $error_message) = @_;
1687   open L, ">", $lockfile or croak("$lockfile: $!");
1688   if (!flock L, LOCK_EX|LOCK_NB) {
1689     croak("Can't lock $lockfile: $error_message\n");
1690   }
1691 }
1692
1693 sub find_docker_image {
1694   # Given a Keep locator, check to see if it contains a Docker image.
1695   # If so, return its stream name and Docker hash.
1696   # If not, return undef for both values.
1697   my $locator = shift;
1698   my ($streamname, $filename);
1699   my $image = api_call("collections/get", uuid => $locator);
1700   if ($image) {
1701     foreach my $line (split(/\n/, $image->{manifest_text})) {
1702       my @tokens = split(/\s+/, $line);
1703       next if (!@tokens);
1704       $streamname = shift(@tokens);
1705       foreach my $filedata (grep(/^\d+:\d+:/, @tokens)) {
1706         if (defined($filename)) {
1707           return (undef, undef);  # More than one file in the Collection.
1708         } else {
1709           $filename = (split(/:/, $filedata, 3))[2];
1710         }
1711       }
1712     }
1713   }
1714   if (defined($filename) and ($filename =~ /^([0-9A-Fa-f]{64})\.tar$/)) {
1715     return ($streamname, $1);
1716   } else {
1717     return (undef, undef);
1718   }
1719 }
1720
1721 sub retry_count {
1722   # Calculate the number of times an operation should be retried,
1723   # assuming exponential backoff, and that we're willing to retry as
1724   # long as tasks have been running.  Enforce a minimum of 3 retries.
1725   my ($starttime, $endtime, $timediff, $retries);
1726   if (@jobstep) {
1727     $starttime = $jobstep[0]->{starttime};
1728     $endtime = $jobstep[-1]->{finishtime};
1729   }
1730   if (!defined($starttime)) {
1731     $timediff = 0;
1732   } elsif (!defined($endtime)) {
1733     $timediff = time - $starttime;
1734   } else {
1735     $timediff = ($endtime - $starttime) - (time - $endtime);
1736   }
1737   if ($timediff > 0) {
1738     $retries = int(log($timediff) / log(2));
1739   } else {
1740     $retries = 1;  # Use the minimum.
1741   }
1742   return ($retries > 3) ? $retries : 3;
1743 }
1744
1745 sub retry_op {
1746   # Pass in two function references.
1747   # This method will be called with the remaining arguments.
1748   # If it dies, retry it with exponential backoff until it succeeds,
1749   # or until the current retry_count is exhausted.  After each failure
1750   # that can be retried, the second function will be called with
1751   # the current try count (0-based), next try time, and error message.
1752   my $operation = shift;
1753   my $retry_callback = shift;
1754   my $retries = retry_count();
1755   foreach my $try_count (0..$retries) {
1756     my $next_try = time + (2 ** $try_count);
1757     my $result = eval { $operation->(@_); };
1758     if (!$@) {
1759       return $result;
1760     } elsif ($try_count < $retries) {
1761       $retry_callback->($try_count, $next_try, $@);
1762       my $sleep_time = $next_try - time;
1763       sleep($sleep_time) if ($sleep_time > 0);
1764     }
1765   }
1766   # Ensure the error message ends in a newline, so Perl doesn't add
1767   # retry_op's line number to it.
1768   chomp($@);
1769   die($@ . "\n");
1770 }
1771
1772 sub api_call {
1773   # Pass in a /-separated API method name, and arguments for it.
1774   # This function will call that method, retrying as needed until
1775   # the current retry_count is exhausted, with a log on the first failure.
1776   my $method_name = shift;
1777   my $log_api_retry = sub {
1778     my ($try_count, $next_try_at, $errmsg) = @_;
1779     $errmsg =~ s/\s*\bat \Q$0\E line \d+\.?\s*//;
1780     $errmsg =~ s/\s/ /g;
1781     $errmsg =~ s/\s+$//;
1782     my $retry_msg;
1783     if ($next_try_at < time) {
1784       $retry_msg = "Retrying.";
1785     } else {
1786       my $next_try_fmt = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($next_try_at);
1787       $retry_msg = "Retrying at $next_try_fmt.";
1788     }
1789     Log(undef, "API method $method_name failed: $errmsg. $retry_msg");
1790   };
1791   my $method = $arv;
1792   foreach my $key (split(/\//, $method_name)) {
1793     $method = $method->{$key};
1794   }
1795   return retry_op(sub { $method->execute(@_); }, $log_api_retry, @_);
1796 }
1797
1798 sub exit_status_s {
1799   # Given a $?, return a human-readable exit code string like "0" or
1800   # "1" or "0 with signal 1" or "1 with signal 11".
1801   my $exitcode = shift;
1802   my $s = $exitcode >> 8;
1803   if ($exitcode & 0x7f) {
1804     $s .= " with signal " . ($exitcode & 0x7f);
1805   }
1806   if ($exitcode & 0x80) {
1807     $s .= " with core dump";
1808   }
1809   return $s;
1810 }
1811
1812 sub handle_readall {
1813   # Pass in a glob reference to a file handle.
1814   # Read all its contents and return them as a string.
1815   my $fh_glob_ref = shift;
1816   local $/ = undef;
1817   return <$fh_glob_ref>;
1818 }
1819
1820 sub tar_filename_n {
1821   my $n = shift;
1822   return sprintf("%s/git.%s.%d.tar", $ENV{CRUNCH_TMP}, $job_id, $n);
1823 }
1824
1825 sub add_git_archive {
1826   # Pass in a git archive command as a string or list, a la system().
1827   # This method will save its output to be included in the archive sent to the
1828   # build script.
1829   my $git_input;
1830   $git_tar_count++;
1831   if (!open(GIT_ARCHIVE, ">", tar_filename_n($git_tar_count))) {
1832     croak("Failed to save git archive: $!");
1833   }
1834   my $git_pid = open2(">&GIT_ARCHIVE", $git_input, @_);
1835   close($git_input);
1836   waitpid($git_pid, 0);
1837   close(GIT_ARCHIVE);
1838   if ($?) {
1839     croak("Failed to save git archive: git exited " . exit_status_s($?));
1840   }
1841 }
1842
1843 sub combined_git_archive {
1844   # Combine all saved tar archives into a single archive, then return its
1845   # contents in a string.  Return undef if no archives have been saved.
1846   if ($git_tar_count < 1) {
1847     return undef;
1848   }
1849   my $base_tar_name = tar_filename_n(1);
1850   foreach my $tar_to_append (map { tar_filename_n($_); } (2..$git_tar_count)) {
1851     my $tar_exit = system("tar", "-Af", $base_tar_name, $tar_to_append);
1852     if ($tar_exit != 0) {
1853       croak("Error preparing build archive: tar -A exited " .
1854             exit_status_s($tar_exit));
1855     }
1856   }
1857   if (!open(GIT_TAR, "<", $base_tar_name)) {
1858     croak("Could not open build archive: $!");
1859   }
1860   my $tar_contents = handle_readall(\*GIT_TAR);
1861   close(GIT_TAR);
1862   return $tar_contents;
1863 }
1864
1865 __DATA__
1866 #!/usr/bin/perl
1867 #
1868 # This is crunch-job's internal dispatch script.  crunch-job running on the API
1869 # server invokes this script on individual compute nodes, or localhost if we're
1870 # running a job locally.  It gets called in two modes:
1871 #
1872 # * No arguments: Installation mode.  Read a tar archive from the DATA
1873 #   file handle; it includes the Crunch script's source code, and
1874 #   maybe SDKs as well.  Those should be installed in the proper
1875 #   locations.  This runs outside of any Docker container, so don't try to
1876 #   introspect Crunch's runtime environment.
1877 #
1878 # * With arguments: Crunch script run mode.  This script should set up the
1879 #   environment, then run the command specified in the arguments.  This runs
1880 #   inside any Docker container.
1881
1882 use Fcntl ':flock';
1883 use File::Path qw( make_path remove_tree );
1884 use POSIX qw(getcwd);
1885
1886 use constant TASK_TEMPFAIL => 111;
1887
1888 # Map SDK subdirectories to the path environments they belong to.
1889 my %SDK_ENVVARS = ("perl/lib" => "PERLLIB", "ruby/lib" => "RUBYLIB");
1890
1891 my $destdir = $ENV{"CRUNCH_SRC"};
1892 my $archive_hash = $ENV{"CRUNCH_GIT_ARCHIVE_HASH"};
1893 my $repo = $ENV{"CRUNCH_SRC_URL"};
1894 my $install_dir = $ENV{"CRUNCH_INSTALL"} || (getcwd() . "/opt");
1895 my $job_work = $ENV{"JOB_WORK"};
1896 my $task_work = $ENV{"TASK_WORK"};
1897
1898 for my $dir ($destdir, $job_work, $task_work) {
1899   if ($dir) {
1900     make_path $dir;
1901     -e $dir or die "Failed to create temporary directory ($dir): $!";
1902   }
1903 }
1904
1905 if ($task_work) {
1906   remove_tree($task_work, {keep_root => 1});
1907 }
1908
1909 open(STDOUT_ORIG, ">&", STDOUT);
1910 open(STDERR_ORIG, ">&", STDERR);
1911 open(STDOUT, ">>", "$destdir.log");
1912 open(STDERR, ">&", STDOUT);
1913
1914 ### Crunch script run mode
1915 if (@ARGV) {
1916   # We want to do routine logging during task 0 only.  This gives the user
1917   # the information they need, but avoids repeating the information for every
1918   # task.
1919   my $Log;
1920   if ($ENV{TASK_SEQUENCE} eq "0") {
1921     $Log = sub {
1922       my $msg = shift;
1923       printf STDERR_ORIG "[Crunch] $msg\n", @_;
1924     };
1925   } else {
1926     $Log = sub { };
1927   }
1928
1929   my $python_src = "$install_dir/python";
1930   my $venv_dir = "$job_work/.arvados.venv";
1931   my $venv_built = -e "$venv_dir/bin/activate";
1932   if ((!$venv_built) and (-d $python_src) and can_run("virtualenv")) {
1933     shell_or_die(undef, "virtualenv", "--quiet", "--system-site-packages",
1934                  "--python=python2.7", $venv_dir);
1935     shell_or_die(TASK_TEMPFAIL, "$venv_dir/bin/pip", "--quiet", "install", "-I", $python_src);
1936     $venv_built = 1;
1937     $Log->("Built Python SDK virtualenv");
1938   }
1939
1940   my $pip_bin = "pip";
1941   if ($venv_built) {
1942     $Log->("Running in Python SDK virtualenv");
1943     $pip_bin = "$venv_dir/bin/pip";
1944     my $orig_argv = join(" ", map { quotemeta($_); } @ARGV);
1945     @ARGV = ("/bin/sh", "-ec",
1946              ". \Q$venv_dir/bin/activate\E; exec $orig_argv");
1947   } elsif (-d $python_src) {
1948     $Log->("Warning: virtualenv not found inside Docker container default " .
1949            "\$PATH. Can't install Python SDK.");
1950   }
1951
1952   my $pkgs = `(\Q$pip_bin\E freeze 2>/dev/null | grep arvados) || dpkg-query --show '*arvados*'`;
1953   if ($pkgs) {
1954     $Log->("Using Arvados SDK:");
1955     foreach my $line (split /\n/, $pkgs) {
1956       $Log->($line);
1957     }
1958   } else {
1959     $Log->("Arvados SDK packages not found");
1960   }
1961
1962   while (my ($sdk_dir, $sdk_envkey) = each(%SDK_ENVVARS)) {
1963     my $sdk_path = "$install_dir/$sdk_dir";
1964     if (-d $sdk_path) {
1965       if ($ENV{$sdk_envkey}) {
1966         $ENV{$sdk_envkey} = "$sdk_path:" . $ENV{$sdk_envkey};
1967       } else {
1968         $ENV{$sdk_envkey} = $sdk_path;
1969       }
1970       $Log->("Arvados SDK added to %s", $sdk_envkey);
1971     }
1972   }
1973
1974   close(STDOUT);
1975   close(STDERR);
1976   open(STDOUT, ">&", STDOUT_ORIG);
1977   open(STDERR, ">&", STDERR_ORIG);
1978   exec(@ARGV);
1979   die "Cannot exec `@ARGV`: $!";
1980 }
1981
1982 ### Installation mode
1983 open L, ">", "$destdir.lock" or die "$destdir.lock: $!";
1984 flock L, LOCK_EX;
1985 if (readlink ("$destdir.archive_hash") eq $archive_hash && -d $destdir) {
1986   # This exact git archive (source + arvados sdk) is already installed
1987   # here, so there's no need to reinstall it.
1988
1989   # We must consume our DATA section, though: otherwise the process
1990   # feeding it to us will get SIGPIPE.
1991   my $buf;
1992   while (read(DATA, $buf, 65536)) { }
1993
1994   exit(0);
1995 }
1996
1997 unlink "$destdir.archive_hash";
1998 mkdir $destdir;
1999
2000 if (!open(TARX, "|-", "tar", "-xC", $destdir)) {
2001   die "Error launching 'tar -xC $destdir': $!";
2002 }
2003 # If we send too much data to tar in one write (> 4-5 MiB), it stops, and we
2004 # get SIGPIPE.  We must feed it data incrementally.
2005 my $tar_input;
2006 while (read(DATA, $tar_input, 65536)) {
2007   print TARX $tar_input;
2008 }
2009 if(!close(TARX)) {
2010   die "'tar -xC $destdir' exited $?: $!";
2011 }
2012
2013 mkdir $install_dir;
2014
2015 my $sdk_root = "$destdir/.arvados.sdk/sdk";
2016 if (-d $sdk_root) {
2017   foreach my $sdk_lang (("python",
2018                          map { (split /\//, $_, 2)[0]; } keys(%SDK_ENVVARS))) {
2019     if (-d "$sdk_root/$sdk_lang") {
2020       if (!rename("$sdk_root/$sdk_lang", "$install_dir/$sdk_lang")) {
2021         die "Failed to install $sdk_lang SDK: $!";
2022       }
2023     }
2024   }
2025 }
2026
2027 my $python_dir = "$install_dir/python";
2028 if ((-d $python_dir) and can_run("python2.7") and
2029     (system("python2.7", "$python_dir/setup.py", "--quiet", "egg_info") != 0)) {
2030   # egg_info failed, probably when it asked git for a build tag.
2031   # Specify no build tag.
2032   open(my $pysdk_cfg, ">>", "$python_dir/setup.cfg");
2033   print $pysdk_cfg "\n[egg_info]\ntag_build =\n";
2034   close($pysdk_cfg);
2035 }
2036
2037 if (-e "$destdir/crunch_scripts/install") {
2038     shell_or_die (undef, "$destdir/crunch_scripts/install", $install_dir);
2039 } elsif (!-e "./install.sh" && -e "./tests/autotests.sh") {
2040     # Old version
2041     shell_or_die (undef, "./tests/autotests.sh", $install_dir);
2042 } elsif (-e "./install.sh") {
2043     shell_or_die (undef, "./install.sh", $install_dir);
2044 }
2045
2046 if ($archive_hash) {
2047     unlink "$destdir.archive_hash.new";
2048     symlink ($archive_hash, "$destdir.archive_hash.new") or die "$destdir.archive_hash.new: $!";
2049     rename ("$destdir.archive_hash.new", "$destdir.archive_hash") or die "$destdir.archive_hash: $!";
2050 }
2051
2052 close L;
2053
2054 sub can_run {
2055   my $command_name = shift;
2056   open(my $which, "-|", "which", $command_name);
2057   while (<$which>) { }
2058   close($which);
2059   return ($? == 0);
2060 }
2061
2062 sub shell_or_die
2063 {
2064   my $exitcode = shift;
2065
2066   if ($ENV{"DEBUG"}) {
2067     print STDERR "@_\n";
2068   }
2069   if (system (@_) != 0) {
2070     my $err = $!;
2071     my $code = $?;
2072     my $exitstatus = sprintf("exit %d signal %d", $code >> 8, $code & 0x7f);
2073     open STDERR, ">&STDERR_ORIG";
2074     system ("cat $destdir.log >&2");
2075     warn "@_ failed ($err): $exitstatus";
2076     if (defined($exitcode)) {
2077       exit $exitcode;
2078     }
2079     else {
2080       exit (($code >> 8) || 1);
2081     }
2082   }
2083 }
2084
2085 __DATA__