sdk/cli/bin/crunch-job

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