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