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   my $id = $jobstep_todo[$todo_ptr];
 824   my $Jobstep = $jobstep[$id];
 825   if ($Jobstep->{level} != $level)
 826   {
 827     next;
 828   }
 829
 830   pipe $reader{$id}, "writer" or croak("pipe() failed: $!");
 831   set_nonblocking($reader{$id});
 832
 833   my $childslot = $freeslot[0];
 834   my $childnode = $slot[$childslot]->{node};
 835   my $childslotname = join (".",
 836                             $slot[$childslot]->{node}->{name},
 837                             $slot[$childslot]->{cpu});
 838
 839   my $childpid = fork();
 840   if ($childpid == 0)
 841   {
 842     $SIG{'INT'} = 'DEFAULT';
 843     $SIG{'QUIT'} = 'DEFAULT';
 844     $SIG{'TERM'} = 'DEFAULT';
 845
 846     foreach (values (%reader))
 847     {
 848       close($_);
 849     }
 850     fcntl ("writer", F_SETFL, 0) or croak ($!); # no close-on-exec
 851     open(STDOUT,">&writer");
 852     open(STDERR,">&writer");
 853
 854     undef $dbh;
 855     undef $sth;
 856
 857     delete $ENV{"GNUPGHOME"};
 858     $ENV{"TASK_UUID"} = $Jobstep->{'arvados_task'}->{'uuid'};
 859     $ENV{"TASK_QSEQUENCE"} = $id;
 860     $ENV{"TASK_SEQUENCE"} = $level;
 861     $ENV{"JOB_SCRIPT"} = $Job->{script};
 862     while (my ($param, $value) = each %{$Job->{script_parameters}}) {
 863       $param =~ tr/a-z/A-Z/;
 864       $ENV{"JOB_PARAMETER_$param"} = $value;
 865     }
 866     $ENV{"TASK_SLOT_NODE"} = $slot[$childslot]->{node}->{name};
 867     $ENV{"TASK_SLOT_NUMBER"} = $slot[$childslot]->{cpu};
 868     $ENV{"TASK_WORK"} = $ENV{"CRUNCH_TMP"}."/task/$childslotname";
 869     $ENV{"HOME"} = $ENV{"TASK_WORK"};
 870     $ENV{"TASK_KEEPMOUNT"} = $ENV{"TASK_WORK"}.".keep";
 871     $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
 872     $ENV{"CRUNCH_NODE_SLOTS"} = $round_max_slots{$ENV{TASK_SLOT_NODE}};
 873     $ENV{"PATH"} = $ENV{"CRUNCH_INSTALL"} . "/bin:" . $ENV{"PATH"};
 874
 875     $ENV{"GZIP"} = "-n";
 876
 877     my @srunargs = (
 878       "srun",
 879       "--nodelist=".$childnode->{name},
 880       qw(-n1 -c1 -N1 -D), $ENV{'TMPDIR'},
 881       "--job-name=$job_id.$id.$$",
 882         );
 883
 884     my $stdbuf = " stdbuf --output=0 --error=0 ";
 885
 886     my $command =
 887         "if [ -e $ENV{TASK_WORK} ]; then rm -rf $ENV{TASK_WORK}; fi; "
 888         ."mkdir -p $ENV{CRUNCH_TMP} $ENV{JOB_WORK} $ENV{TASK_WORK} $ENV{TASK_KEEPMOUNT} "
 889         ."&& cd $ENV{CRUNCH_TMP} "
 890         # These environment variables get used explicitly later in
 891         # $command.  No tool is expected to read these values directly.
 892         .q{&& MEM=$(awk '($1 == "MemTotal:"){print $2}' </proc/meminfo) }
 893         .q{&& SWAP=$(awk '($1 == "SwapTotal:"){print $2}' </proc/meminfo) }
 894         ."&& MEMLIMIT=\$(( (\$MEM * 95) / ($ENV{CRUNCH_NODE_SLOTS} * 100) )) "
 895         ."&& let SWAPLIMIT=\$MEMLIMIT+\$SWAP ";
 896     $command .= "&& exec arv-mount --by-id --allow-other $ENV{TASK_KEEPMOUNT} --exec ";
 897     if ($docker_hash)
 898     {
 899       my $containername = "$Jobstep->{arvados_task}->{uuid}-$Jobstep->{failures}";
 900       my $cidfile = "$ENV{CRUNCH_TMP}/$containername.cid";
 901       $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -cgroup-parent=docker -cgroup-cid=$cidfile -poll=10000 ";
 902       $command .= "$docker_bin run --name=$containername --attach=stdout --attach=stderr --attach=stdin -i \Q$dockeruserarg\E --cidfile=$cidfile --sig-proxy ";
 903       # We only set memory limits if Docker lets us limit both memory and swap.
 904       # Memory limits alone have been supported longer, but subprocesses tend
 905       # to get SIGKILL if they exceed that without any swap limit set.
 906       # See #5642 for additional background.
 907       if ($docker_limitmem) {
 908         $command .= "--memory=\${MEMLIMIT}k --memory-swap=\${SWAPLIMIT}k ";
 909       }
 910
 911       # The source tree and $destdir directory (which we have
 912       # installed on the worker host) are available in the container,
 913       # under the same path.
 914       $command .= "--volume=\Q$ENV{CRUNCH_SRC}:$ENV{CRUNCH_SRC}:ro\E ";
 915       $command .= "--volume=\Q$ENV{CRUNCH_INSTALL}:$ENV{CRUNCH_INSTALL}:ro\E ";
 916
 917       # Currently, we make arv-mount's mount point appear at /keep
 918       # inside the container (instead of using the same path as the
 919       # host like we do with CRUNCH_SRC and CRUNCH_INSTALL). However,
 920       # crunch scripts and utilities must not rely on this. They must
 921       # use $TASK_KEEPMOUNT.
 922       $command .= "--volume=\Q$ENV{TASK_KEEPMOUNT}:/keep:ro\E ";
 923       $ENV{TASK_KEEPMOUNT} = "/keep";
 924
 925       # TASK_WORK is almost exactly like a docker data volume: it
 926       # starts out empty, is writable, and persists until no
 927       # containers use it any more. We don't use --volumes-from to
 928       # share it with other containers: it is only accessible to this
 929       # task, and it goes away when this task stops.
 930       #
 931       # However, a docker data volume is writable only by root unless
 932       # the mount point already happens to exist in the container with
 933       # different permissions. Therefore, we [1] assume /tmp already
 934       # exists in the image and is writable by the crunch user; [2]
 935       # avoid putting TASK_WORK inside CRUNCH_TMP (which won't be
 936       # writable if they are created by docker while setting up the
 937       # other --volumes); and [3] create $TASK_WORK inside the
 938       # container using $build_script.
 939       $command .= "--volume=/tmp ";
 940       $ENV{"TASK_WORK"} = "/tmp/crunch-job-task-work/$childslotname";
 941       $ENV{"HOME"} = $ENV{"TASK_WORK"};
 942       $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
 943
 944       # TODO: Share a single JOB_WORK volume across all task
 945       # containers on a given worker node, and delete it when the job
 946       # ends (and, in case that doesn't work, when the next job
 947       # starts).
 948       #
 949       # For now, use the same approach as TASK_WORK above.
 950       $ENV{"JOB_WORK"} = "/tmp/crunch-job-work";
 951
 952       while (my ($env_key, $env_val) = each %ENV)
 953       {
 954         if ($env_key =~ /^(ARVADOS|CRUNCH|JOB|TASK)_/) {
 955           $command .= "--env=\Q$env_key=$env_val\E ";
 956         }
 957       }
 958       $command .= "--env=\QHOME=$ENV{HOME}\E ";
 959       $command .= "\Q$docker_hash\E ";
 960
 961       if ($Job->{arvados_sdk_version}) {
 962         $command .= $stdbuf;
 963         $command .= "perl - \Q$ENV{CRUNCH_SRC}/crunch_scripts/$Job->{script}\E";
 964       } else {
 965         $command .= "/bin/sh -c \'mkdir -p \"$ENV{JOB_WORK}\" \"$ENV{TASK_WORK}\" && " .
 966             "if which stdbuf >/dev/null ; then " .
 967             "  exec $stdbuf \Q$ENV{CRUNCH_SRC}/crunch_scripts/$Job->{script}\E ;" .
 968             " else " .
 969             "  exec \Q$ENV{CRUNCH_SRC}/crunch_scripts/$Job->{script}\E ;" .
 970             " fi\'";
 971       }
 972     } else {
 973       # Non-docker run
 974       $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -poll=10000 ";
 975       $command .= $stdbuf;
 976       $command .= "perl - $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
 977     }
 978
 979     my @execargs = ('bash', '-c', $command);
 980     srun (\@srunargs, \@execargs, undef, $build_script);
 981     # exec() failed, we assume nothing happened.
 982     die "srun() failed on build script\n";
 983   }
 984   close("writer");
 985   if (!defined $childpid)
 986   {
 987     close $reader{$id};
 988     delete $reader{$id};
 989     next;
 990   }
 991   shift @freeslot;
 992   $proc{$childpid} = { jobstep => $id,
 993                        time => time,
 994                        slot => $childslot,
 995                        jobstepname => "$job_id.$id.$childpid",
 996                      };
 997   croak ("assert failed: \$slot[$childslot]->{'pid'} exists") if exists $slot[$childslot]->{pid};
 998   $slot[$childslot]->{pid} = $childpid;
 999
1000   Log ($id, "job_task ".$Jobstep->{'arvados_task'}->{'uuid'});
1001   Log ($id, "child $childpid started on $childslotname");
1002   $Jobstep->{starttime} = time;
1003   $Jobstep->{node} = $childnode->{name};
1004   $Jobstep->{slotindex} = $childslot;
1005   delete $Jobstep->{stderr};
1006   delete $Jobstep->{finishtime};
1007   delete $Jobstep->{tempfail};
1008
1009   $Jobstep->{'arvados_task'}->{started_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{starttime});
1010   $Jobstep->{'arvados_task'}->save;
1011
1012   splice @jobstep_todo, $todo_ptr, 1;
1013   --$todo_ptr;
1014
1015   $progress_is_dirty = 1;
1016
1017   while (!@freeslot
1018          ||
1019          ($round_num_freeslots > @freeslot && $todo_ptr+1 > $#jobstep_todo))
1020   {
1021     last THISROUND if $main::please_freeze || defined($main::success);
1022     if ($main::please_info)
1023     {
1024       $main::please_info = 0;
1025       freeze();
1026       create_output_collection();
1027       save_meta(1);
1028       update_progress_stats();
1029     }
1030     my $gotsome
1031         = readfrompipes ()
1032         + reapchildren ();
1033     if (!$gotsome || ($latest_refresh + 2 < scalar time))
1034     {
1035       check_refresh_wanted();
1036       check_squeue();
1037       update_progress_stats();
1038       select (undef, undef, undef, 0.1);
1039     }
1040     elsif (time - $progress_stats_updated >= 30 || $progress_is_dirty)
1041     {
1042       update_progress_stats();
1043     }
1044     $working_slot_count = scalar(grep { $_->{node}->{fail_count} == 0 &&
1045                                         $_->{node}->{hold_count} < 4 } @slot);
1046     if (($thisround_failed_multiple >= 8 && $thisround_succeeded == 0) ||
1047         ($thisround_failed_multiple >= 16 && $thisround_failed_multiple > $thisround_succeeded))
1048     {
1049       my $message = "Repeated failure rate too high ($thisround_failed_multiple/"
1050           .($thisround_failed+$thisround_succeeded)
1051           .") -- giving up on this round";
1052       Log (undef, $message);
1053       last THISROUND;
1054     }
1055
1056     # move slots from freeslot to holdslot (or back to freeslot) if necessary
1057     for (my $i=$#freeslot; $i>=0; $i--) {
1058       if ($slot[$freeslot[$i]]->{node}->{hold_until} > scalar time) {
1059         push @holdslot, (splice @freeslot, $i, 1);
1060       }
1061     }
1062     for (my $i=$#holdslot; $i>=0; $i--) {
1063       if ($slot[$holdslot[$i]]->{node}->{hold_until} <= scalar time) {
1064         push @freeslot, (splice @holdslot, $i, 1);
1065       }
1066     }
1067
1068     # give up if no nodes are succeeding
1069     if ($working_slot_count < 1) {
1070       Log(undef, "Every node has failed -- giving up");
1071       last THISROUND;
1072     }
1073   }
1074 }
1075
1076
1077 push @freeslot, splice @holdslot;
1078 map { $slot[$freeslot[$_]]->{node}->{losing_streak} = 0 } (0..$#freeslot);
1079
1080
1081 Log (undef, "wait for last ".(scalar keys %proc)." children to finish");
1082 while (%proc)
1083 {
1084   if ($main::please_continue) {
1085     $main::please_continue = 0;
1086     goto THISROUND;
1087   }
1088   $main::please_info = 0, freeze(), create_output_collection(), save_meta(1) if $main::please_info;
1089   readfrompipes ();
1090   if (!reapchildren())
1091   {
1092     check_refresh_wanted();
1093     check_squeue();
1094     update_progress_stats();
1095     select (undef, undef, undef, 0.1);
1096     killem (keys %proc) if $main::please_freeze;
1097   }
1098 }
1099
1100 update_progress_stats();
1101 freeze_if_want_freeze();
1102
1103
1104 if (!defined $main::success)
1105 {
1106   if (!@jobstep_todo) {
1107     $main::success = 1;
1108   } elsif ($working_slot_count < 1) {
1109     save_output_collection();
1110     save_meta();
1111     exit(EX_RETRY_UNLOCKED);
1112   } elsif ($thisround_succeeded == 0 &&
1113            ($thisround_failed == 0 || $thisround_failed > 4)) {
1114     my $message = "stop because $thisround_failed tasks failed and none succeeded";
1115     Log (undef, $message);
1116     $main::success = 0;
1117   }
1118 }
1119
1120 goto ONELEVEL if !defined $main::success;
1121
1122
1123 release_allocation();
1124 freeze();
1125 my $collated_output = save_output_collection();
1126 Log (undef, "finish");
1127
1128 save_meta();
1129
1130 my $final_state;
1131 if ($collated_output && $main::success) {
1132   $final_state = 'Complete';
1133 } else {
1134   $final_state = 'Failed';
1135 }
1136 $Job->update_attributes('state' => $final_state);
1137
1138 exit (($final_state eq 'Complete') ? 0 : 1);
1139
1140
1141
1142 sub update_progress_stats
1143 {
1144   $progress_stats_updated = time;
1145   return if !$progress_is_dirty;
1146   my ($todo, $done, $running) = (scalar @jobstep_todo,
1147                                  scalar @jobstep_done,
1148                                  scalar keys(%proc));
1149   $Job->{'tasks_summary'} ||= {};
1150   $Job->{'tasks_summary'}->{'todo'} = $todo;
1151   $Job->{'tasks_summary'}->{'done'} = $done;
1152   $Job->{'tasks_summary'}->{'running'} = $running;
1153   $Job->update_attributes('tasks_summary' => $Job->{'tasks_summary'});
1154   Log (undef, "status: $done done, $running running, $todo todo");
1155   $progress_is_dirty = 0;
1156 }
1157
1158
1159
1160 sub reapchildren
1161 {
1162   my $pid = waitpid (-1, WNOHANG);
1163   return 0 if $pid <= 0;
1164
1165   my $whatslot = ($slot[$proc{$pid}->{slot}]->{node}->{name}
1166                   . "."
1167                   . $slot[$proc{$pid}->{slot}]->{cpu});
1168   my $jobstepid = $proc{$pid}->{jobstep};
1169   my $elapsed = time - $proc{$pid}->{time};
1170   my $Jobstep = $jobstep[$jobstepid];
1171
1172   my $childstatus = $?;
1173   my $exitvalue = $childstatus >> 8;
1174   my $exitinfo = "exit ".exit_status_s($childstatus);
1175   $Jobstep->{'arvados_task'}->reload;
1176   my $task_success = $Jobstep->{'arvados_task'}->{success};
1177
1178   Log ($jobstepid, "child $pid on $whatslot $exitinfo success=$task_success");
1179
1180   if (!defined $task_success) {
1181     # task did not indicate one way or the other --> fail
1182     $Jobstep->{'arvados_task'}->{success} = 0;
1183     $Jobstep->{'arvados_task'}->save;
1184     $task_success = 0;
1185   }
1186
1187   if (!$task_success)
1188   {
1189     my $temporary_fail;
1190     $temporary_fail ||= $Jobstep->{tempfail};
1191     $temporary_fail ||= ($exitvalue == TASK_TEMPFAIL);
1192
1193     ++$thisround_failed;
1194     ++$thisround_failed_multiple if $Jobstep->{'failures'} >= 1;
1195
1196     # Check for signs of a failed or misconfigured node
1197     if (++$slot[$proc{$pid}->{slot}]->{node}->{losing_streak} >=
1198         2+$slot[$proc{$pid}->{slot}]->{node}->{ncpus}) {
1199       # Don't count this against jobstep failure thresholds if this
1200       # node is already suspected faulty and srun exited quickly
1201       if ($slot[$proc{$pid}->{slot}]->{node}->{hold_until} &&
1202           $elapsed < 5) {
1203         Log ($jobstepid, "blaming failure on suspect node " .
1204              $slot[$proc{$pid}->{slot}]->{node}->{name});
1205         $temporary_fail ||= 1;
1206       }
1207       ban_node_by_slot($proc{$pid}->{slot});
1208     }
1209
1210     Log ($jobstepid, sprintf('failure (#%d, %s) after %d seconds',
1211                              ++$Jobstep->{'failures'},
1212                              $temporary_fail ? 'temporary' : 'permanent',
1213                              $elapsed));
1214
1215     if (!$temporary_fail || $Jobstep->{'failures'} >= 3) {
1216       # Give up on this task, and the whole job
1217       $main::success = 0;
1218     }
1219     # Put this task back on the todo queue
1220     push @jobstep_todo, $jobstepid;
1221     $Job->{'tasks_summary'}->{'failed'}++;
1222   }
1223   else
1224   {
1225     ++$thisround_succeeded;
1226     $slot[$proc{$pid}->{slot}]->{node}->{losing_streak} = 0;
1227     $slot[$proc{$pid}->{slot}]->{node}->{hold_until} = 0;
1228     $slot[$proc{$pid}->{slot}]->{node}->{fail_count} = 0;
1229     push @jobstep_done, $jobstepid;
1230     Log ($jobstepid, "success in $elapsed seconds");
1231   }
1232   $Jobstep->{exitcode} = $childstatus;
1233   $Jobstep->{finishtime} = time;
1234   $Jobstep->{'arvados_task'}->{finished_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{finishtime});
1235   $Jobstep->{'arvados_task'}->save;
1236   process_stderr ($jobstepid, $task_success);
1237   Log ($jobstepid, sprintf("task output (%d bytes): %s",
1238                            length($Jobstep->{'arvados_task'}->{output}),
1239                            $Jobstep->{'arvados_task'}->{output}));
1240
1241   close $reader{$jobstepid};
1242   delete $reader{$jobstepid};
1243   delete $slot[$proc{$pid}->{slot}]->{pid};
1244   push @freeslot, $proc{$pid}->{slot};
1245   delete $proc{$pid};
1246
1247   if ($task_success) {
1248     # Load new tasks
1249     my $newtask_list = [];
1250     my $newtask_results;
1251     do {
1252       $newtask_results = api_call(
1253         "job_tasks/list",
1254         'where' => {
1255           'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid}
1256         },
1257         'order' => 'qsequence',
1258         'offset' => scalar(@$newtask_list),
1259       );
1260       push(@$newtask_list, @{$newtask_results->{items}});
1261     } while (@{$newtask_results->{items}});
1262     foreach my $arvados_task (@$newtask_list) {
1263       my $jobstep = {
1264         'level' => $arvados_task->{'sequence'},
1265         'failures' => 0,
1266         'arvados_task' => $arvados_task
1267       };
1268       push @jobstep, $jobstep;
1269       push @jobstep_todo, $#jobstep;
1270     }
1271   }
1272
1273   $progress_is_dirty = 1;
1274   1;
1275 }
1276
1277 sub check_refresh_wanted
1278 {
1279   my @stat = stat $ENV{"CRUNCH_REFRESH_TRIGGER"};
1280   if (@stat && $stat[9] > $latest_refresh) {
1281     $latest_refresh = scalar time;
1282     my $Job2 = api_call("jobs/get", uuid => $jobspec);
1283     for my $attr ('cancelled_at',
1284                   'cancelled_by_user_uuid',
1285                   'cancelled_by_client_uuid',
1286                   'state') {
1287       $Job->{$attr} = $Job2->{$attr};
1288     }
1289     if ($Job->{'state'} ne "Running") {
1290       if ($Job->{'state'} eq "Cancelled") {
1291         Log (undef, "Job cancelled at " . $Job->{'cancelled_at'} . " by user " . $Job->{'cancelled_by_user_uuid'});
1292       } else {
1293         Log (undef, "Job state unexpectedly changed to " . $Job->{'state'});
1294       }
1295       $main::success = 0;
1296       $main::please_freeze = 1;
1297     }
1298   }
1299 }
1300
1301 sub check_squeue
1302 {
1303   my $last_squeue_check = $squeue_checked;
1304
1305   # Do not call `squeue` or check the kill list more than once every
1306   # 15 seconds.
1307   return if $last_squeue_check > time - 15;
1308   $squeue_checked = time;
1309
1310   # Look for children from which we haven't received stderr data since
1311   # the last squeue check. If no such children exist, all procs are
1312   # alive and there's no need to even look at squeue.
1313   #
1314   # As long as the crunchstat poll interval (10s) is shorter than the
1315   # squeue check interval (15s) this should make the squeue check an
1316   # infrequent event.
1317   my $silent_procs = 0;
1318   for my $jobstep (values %proc)
1319   {
1320     if ($jobstep->{stderr_at} < $last_squeue_check)
1321     {
1322       $silent_procs++;
1323     }
1324   }
1325   return if $silent_procs == 0;
1326
1327   # use killem() on procs whose killtime is reached
1328   while (my ($pid, $jobstep) = each %proc)
1329   {
1330     if (exists $jobstep->{killtime}
1331         && $jobstep->{killtime} <= time
1332         && $jobstep->{stderr_at} < $last_squeue_check)
1333     {
1334       my $sincewhen = "";
1335       if ($jobstep->{stderr_at}) {
1336         $sincewhen = " in last " . (time - $jobstep->{stderr_at}) . "s";
1337       }
1338       Log($jobstep->{jobstep}, "killing orphaned srun process $pid (task not in slurm queue, no stderr received$sincewhen)");
1339       killem ($pid);
1340     }
1341   }
1342
1343   if (!$have_slurm)
1344   {
1345     # here is an opportunity to check for mysterious problems with local procs
1346     return;
1347   }
1348
1349   # Get a list of steps still running.  Note: squeue(1) says --steps
1350   # selects a format (which we override anyway) and allows us to
1351   # specify which steps we're interested in (which we don't).
1352   # Importantly, it also changes the meaning of %j from "job name" to
1353   # "step name" and (although this isn't mentioned explicitly in the
1354   # docs) switches from "one line per job" mode to "one line per step"
1355   # mode. Without it, we'd just get a list of one job, instead of a
1356   # list of N steps.
1357   my @squeue = `squeue --jobs=\Q$ENV{SLURM_JOB_ID}\E --steps --format='%j' --noheader`;
1358   if ($? != 0)
1359   {
1360     Log(undef, "warning: squeue exit status $? ($!)");
1361     return;
1362   }
1363   chop @squeue;
1364
1365   # which of my jobsteps are running, according to squeue?
1366   my %ok;
1367   for my $jobstepname (@squeue)
1368   {
1369     $ok{$jobstepname} = 1;
1370   }
1371
1372   # Check for child procs >60s old and not mentioned by squeue.
1373   while (my ($pid, $jobstep) = each %proc)
1374   {
1375     if ($jobstep->{time} < time - 60
1376         && $jobstep->{jobstepname}
1377         && !exists $ok{$jobstep->{jobstepname}}
1378         && !exists $jobstep->{killtime})
1379     {
1380       # According to slurm, this task has ended (successfully or not)
1381       # -- but our srun child hasn't exited. First we must wait (30
1382       # seconds) in case this is just a race between communication
1383       # channels. Then, if our srun child process still hasn't
1384       # terminated, we'll conclude some slurm communication
1385       # error/delay has caused the task to die without notifying srun,
1386       # and we'll kill srun ourselves.
1387       $jobstep->{killtime} = time + 30;
1388       Log($jobstep->{jobstep}, "notice: task is not in slurm queue but srun process $pid has not exited");
1389     }
1390   }
1391 }
1392
1393
1394 sub release_allocation
1395 {
1396   if ($have_slurm)
1397   {
1398     Log (undef, "release job allocation");
1399     system "scancel $ENV{SLURM_JOB_ID}";
1400   }
1401 }
1402
1403
1404 sub readfrompipes
1405 {
1406   my $gotsome = 0;
1407   foreach my $job (keys %reader)
1408   {
1409     my $buf;
1410     while (0 < sysread ($reader{$job}, $buf, 8192))
1411     {
1412       print STDERR $buf if $ENV{CRUNCH_DEBUG};
1413       $jobstep[$job]->{stderr_at} = time;
1414       $jobstep[$job]->{stderr} .= $buf;
1415       preprocess_stderr ($job);
1416       if (length ($jobstep[$job]->{stderr}) > 16384)
1417       {
1418         substr ($jobstep[$job]->{stderr}, 0, 8192) = "";
1419       }
1420       $gotsome = 1;
1421     }
1422   }
1423   return $gotsome;
1424 }
1425
1426
1427 sub preprocess_stderr
1428 {
1429   my $job = shift;
1430
1431   while ($jobstep[$job]->{stderr} =~ /^(.*?)\n/) {
1432     my $line = $1;
1433     substr $jobstep[$job]->{stderr}, 0, 1+length($line), "";
1434     Log ($job, "stderr $line");
1435     if ($line =~ /srun: error: (SLURM job $ENV{SLURM_JOB_ID} has expired|Unable to confirm allocation for job $ENV{SLURM_JOB_ID})/) {
1436       # whoa.
1437       $main::please_freeze = 1;
1438     }
1439     elsif ($line =~ /srun: error: Node failure on/) {
1440       my $job_slot_index = $jobstep[$job]->{slotindex};
1441       $slot[$job_slot_index]->{node}->{fail_count}++;
1442       $jobstep[$job]->{tempfail} = 1;
1443       ban_node_by_slot($job_slot_index);
1444     }
1445     elsif ($line =~ /srun: error: (Unable to create job step|.*: Communication connection failure)/) {
1446       $jobstep[$job]->{tempfail} = 1;
1447       ban_node_by_slot($jobstep[$job]->{slotindex});
1448     }
1449     elsif ($line =~ /arvados\.errors\.Keep/) {
1450       $jobstep[$job]->{tempfail} = 1;
1451     }
1452   }
1453 }
1454
1455
1456 sub process_stderr
1457 {
1458   my $job = shift;
1459   my $task_success = shift;
1460   preprocess_stderr ($job);
1461
1462   map {
1463     Log ($job, "stderr $_");
1464   } split ("\n", $jobstep[$job]->{stderr});
1465 }
1466
1467 sub fetch_block
1468 {
1469   my $hash = shift;
1470   my $keep;
1471   if (!open($keep, "-|", "arv-get", "--retries", retry_count(), $hash)) {
1472     Log(undef, "fetch_block run error from arv-get $hash: $!");
1473     return undef;
1474   }
1475   my $output_block = "";
1476   while (1) {
1477     my $buf;
1478     my $bytes = sysread($keep, $buf, 1024 * 1024);
1479     if (!defined $bytes) {
1480       Log(undef, "fetch_block read error from arv-get: $!");
1481       $output_block = undef;
1482       last;
1483     } elsif ($bytes == 0) {
1484       # sysread returns 0 at the end of the pipe.
1485       last;
1486     } else {
1487       # some bytes were read into buf.
1488       $output_block .= $buf;
1489     }
1490   }
1491   close $keep;
1492   if ($?) {
1493     Log(undef, "fetch_block arv-get exited " . exit_status_s($?));
1494     $output_block = undef;
1495   }
1496   return $output_block;
1497 }
1498
1499 # Create a collection by concatenating the output of all tasks (each
1500 # task's output is either a manifest fragment, a locator for a
1501 # manifest fragment stored in Keep, or nothing at all). Return the
1502 # portable_data_hash of the new collection.
1503 sub create_output_collection
1504 {
1505   Log (undef, "collate");
1506
1507   my ($child_out, $child_in);
1508   my $pid = open2($child_out, $child_in, 'python', '-c', q{
1509 import arvados
1510 import sys
1511 print (arvados.api("v1").collections().
1512        create(body={"manifest_text": sys.stdin.read()}).
1513        execute(num_retries=int(sys.argv[1]))["portable_data_hash"])
1514 }, retry_count());
1515
1516   my $task_idx = -1;
1517   my $manifest_size = 0;
1518   for (@jobstep)
1519   {
1520     ++$task_idx;
1521     my $output = $_->{'arvados_task'}->{output};
1522     next if (!defined($output));
1523     my $next_write;
1524     if ($output =~ /^[0-9a-f]{32}(\+\S+)*$/) {
1525       $next_write = fetch_block($output);
1526     } else {
1527       $next_write = $output;
1528     }
1529     if (defined($next_write)) {
1530       if (!defined(syswrite($child_in, $next_write))) {
1531         # There's been an error writing.  Stop the loop.
1532         # We'll log details about the exit code later.
1533         last;
1534       } else {
1535         $manifest_size += length($next_write);
1536       }
1537     } else {
1538       my $uuid = $_->{'arvados_task'}->{'uuid'};
1539       Log (undef, "Error retrieving '$output' output by task $task_idx ($uuid)");
1540       $main::success = 0;
1541     }
1542   }
1543   close($child_in);
1544   Log(undef, "collated output manifest text to send to API server is $manifest_size bytes with access tokens");
1545
1546   my $joboutput;
1547   my $s = IO::Select->new($child_out);
1548   if ($s->can_read(120)) {
1549     sysread($child_out, $joboutput, 1024 * 1024);
1550     waitpid($pid, 0);
1551     if ($?) {
1552       Log(undef, "output collection creation exited " . exit_status_s($?));
1553       $joboutput = undef;
1554     } else {
1555       chomp($joboutput);
1556     }
1557   } else {
1558     Log (undef, "timed out while creating output collection");
1559     foreach my $signal (2, 2, 2, 15, 15, 9) {
1560       kill($signal, $pid);
1561       last if waitpid($pid, WNOHANG) == -1;
1562       sleep(1);
1563     }
1564   }
1565   close($child_out);
1566
1567   return $joboutput;
1568 }
1569
1570 # Calls create_output_collection, logs the result, and returns it.
1571 # If that was successful, save that as the output in the job record.
1572 sub save_output_collection {
1573   my $collated_output = create_output_collection();
1574
1575   if (!$collated_output) {
1576     Log(undef, "Failed to write output collection");
1577   }
1578   else {
1579     Log(undef, "job output $collated_output");
1580     $Job->update_attributes('output' => $collated_output);
1581   }
1582   return $collated_output;
1583 }
1584
1585 sub killem
1586 {
1587   foreach (@_)
1588   {
1589     my $sig = 2;                # SIGINT first
1590     if (exists $proc{$_}->{"sent_$sig"} &&
1591         time - $proc{$_}->{"sent_$sig"} > 4)
1592     {
1593       $sig = 15;                # SIGTERM if SIGINT doesn't work
1594     }
1595     if (exists $proc{$_}->{"sent_$sig"} &&
1596         time - $proc{$_}->{"sent_$sig"} > 4)
1597     {
1598       $sig = 9;                 # SIGKILL if SIGTERM doesn't work
1599     }
1600     if (!exists $proc{$_}->{"sent_$sig"})
1601     {
1602       Log ($proc{$_}->{jobstep}, "sending 2x signal $sig to pid $_");
1603       kill $sig, $_;
1604       select (undef, undef, undef, 0.1);
1605       if ($sig == 2)
1606       {
1607         kill $sig, $_;     # srun wants two SIGINT to really interrupt
1608       }
1609       $proc{$_}->{"sent_$sig"} = time;
1610       $proc{$_}->{"killedafter"} = time - $proc{$_}->{"time"};
1611     }
1612   }
1613 }
1614
1615
1616 sub fhbits
1617 {
1618   my($bits);
1619   for (@_) {
1620     vec($bits,fileno($_),1) = 1;
1621   }
1622   $bits;
1623 }
1624
1625
1626 # Send log output to Keep via arv-put.
1627 #
1628 # $log_pipe_in and $log_pipe_out are the input and output filehandles to the arv-put pipe.
1629 # $log_pipe_out_buf is a string containing all output read from arv-put so far.
1630 # $log_pipe_out_select is an IO::Select object around $log_pipe_out.
1631 # $log_pipe_pid is the pid of the arv-put subprocess.
1632 #
1633 # The only functions that should access these variables directly are:
1634 #
1635 # log_writer_start($logfilename)
1636 #     Starts an arv-put pipe, reading data on stdin and writing it to
1637 #     a $logfilename file in an output collection.
1638 #
1639 # log_writer_read_output([$timeout])
1640 #     Read output from $log_pipe_out and append it to $log_pipe_out_buf.
1641 #     Passes $timeout to the select() call, with a default of 0.01.
1642 #     Returns the result of the last read() call on $log_pipe_out, or
1643 #     -1 if read() wasn't called because select() timed out.
1644 #     Only other log_writer_* functions should need to call this.
1645 #
1646 # log_writer_send($txt)
1647 #     Writes $txt to the output log collection.
1648 #
1649 # log_writer_finish()
1650 #     Closes the arv-put pipe and returns the output that it produces.
1651 #
1652 # log_writer_is_active()
1653 #     Returns a true value if there is currently a live arv-put
1654 #     process, false otherwise.
1655 #
1656 my ($log_pipe_in, $log_pipe_out, $log_pipe_out_buf, $log_pipe_out_select,
1657     $log_pipe_pid);
1658
1659 sub log_writer_start($)
1660 {
1661   my $logfilename = shift;
1662   $log_pipe_pid = open2($log_pipe_out, $log_pipe_in,
1663                         'arv-put',
1664                         '--stream',
1665                         '--retries', '3',
1666                         '--filename', $logfilename,
1667                         '-');
1668   $log_pipe_out_buf = "";
1669   $log_pipe_out_select = IO::Select->new($log_pipe_out);
1670 }
1671
1672 sub log_writer_read_output {
1673   my $timeout = shift || 0.01;
1674   my $read = -1;
1675   while ($read && $log_pipe_out_select->can_read($timeout)) {
1676     $read = read($log_pipe_out, $log_pipe_out_buf, 65536,
1677                  length($log_pipe_out_buf));
1678   }
1679   if (!defined($read)) {
1680     Log(undef, "error reading log manifest from arv-put: $!");
1681   }
1682   return $read;
1683 }
1684
1685 sub log_writer_send($)
1686 {
1687   my $txt = shift;
1688   print $log_pipe_in $txt;
1689   log_writer_read_output();
1690 }
1691
1692 sub log_writer_finish()
1693 {
1694   return unless $log_pipe_pid;
1695
1696   close($log_pipe_in);
1697
1698   my $read_result = log_writer_read_output(120);
1699   if ($read_result == -1) {
1700     Log (undef, "timed out reading from 'arv-put'");
1701   } elsif ($read_result != 0) {
1702     Log(undef, "failed to read arv-put log manifest to EOF");
1703   }
1704
1705   waitpid($log_pipe_pid, 0);
1706   if ($?) {
1707     Log(undef, "log_writer_finish: arv-put exited " . exit_status_s($?))
1708   }
1709
1710   close($log_pipe_out);
1711   my $arv_put_output = $log_pipe_out_buf;
1712   $log_pipe_pid = $log_pipe_in = $log_pipe_out = $log_pipe_out_buf =
1713       $log_pipe_out_select = undef;
1714
1715   return $arv_put_output;
1716 }
1717
1718 sub log_writer_is_active() {
1719   return $log_pipe_pid;
1720 }
1721
1722 sub Log                         # ($jobstep_id, $logmessage)
1723 {
1724   if ($_[1] =~ /\n/) {
1725     for my $line (split (/\n/, $_[1])) {
1726       Log ($_[0], $line);
1727     }
1728     return;
1729   }
1730   my $fh = select STDERR; $|=1; select $fh;
1731   my $message = sprintf ("%s %d %s %s", $job_id, $$, @_);
1732   $message =~ s{([^ -\176])}{"\\" . sprintf ("%03o", ord($1))}ge;
1733   $message .= "\n";
1734   my $datetime;
1735   if (log_writer_is_active() || -t STDERR) {
1736     my @gmtime = gmtime;
1737     $datetime = sprintf ("%04d-%02d-%02d_%02d:%02d:%02d",
1738                          $gmtime[5]+1900, $gmtime[4]+1, @gmtime[3,2,1,0]);
1739   }
1740   print STDERR ((-t STDERR) ? ($datetime." ".$message) : $message);
1741
1742   if (log_writer_is_active()) {
1743     log_writer_send($datetime . " " . $message);
1744   }
1745 }
1746
1747
1748 sub croak
1749 {
1750   my ($package, $file, $line) = caller;
1751   my $message = "@_ at $file line $line\n";
1752   Log (undef, $message);
1753   freeze() if @jobstep_todo;
1754   create_output_collection() if @jobstep_todo;
1755   cleanup();
1756   save_meta();
1757   die;
1758 }
1759
1760
1761 sub cleanup
1762 {
1763   return unless $Job;
1764   if ($Job->{'state'} eq 'Cancelled') {
1765     $Job->update_attributes('finished_at' => scalar gmtime);
1766   } else {
1767     $Job->update_attributes('state' => 'Failed');
1768   }
1769 }
1770
1771
1772 sub save_meta
1773 {
1774   my $justcheckpoint = shift; # false if this will be the last meta saved
1775   return if $justcheckpoint;  # checkpointing is not relevant post-Warehouse.pm
1776   return unless log_writer_is_active();
1777
1778   my $log_manifest = "";
1779   if ($Job->{log}) {
1780     my $prev_log_coll = api_call("collections/get", uuid => $Job->{log});
1781     $log_manifest .= $prev_log_coll->{manifest_text};
1782   }
1783   $log_manifest .= log_writer_finish();
1784
1785   my $log_coll = api_call(
1786     "collections/create", ensure_unique_name => 1, collection => {
1787       manifest_text => $log_manifest,
1788       owner_uuid => $Job->{owner_uuid},
1789       name => sprintf("Log from %s job %s", $Job->{script}, $Job->{uuid}),
1790     });
1791   Log(undef, "log collection is " . $log_coll->{portable_data_hash});
1792   $Job->update_attributes('log' => $log_coll->{portable_data_hash});
1793 }
1794
1795
1796 sub freeze_if_want_freeze
1797 {
1798   if ($main::please_freeze)
1799   {
1800     release_allocation();
1801     if (@_)
1802     {
1803       # kill some srun procs before freeze+stop
1804       map { $proc{$_} = {} } @_;
1805       while (%proc)
1806       {
1807         killem (keys %proc);
1808         select (undef, undef, undef, 0.1);
1809         my $died;
1810         while (($died = waitpid (-1, WNOHANG)) > 0)
1811         {
1812           delete $proc{$died};
1813         }
1814       }
1815     }
1816     freeze();
1817     create_output_collection();
1818     cleanup();
1819     save_meta();
1820     exit 1;
1821   }
1822 }
1823
1824
1825 sub freeze
1826 {
1827   Log (undef, "Freeze not implemented");
1828   return;
1829 }
1830
1831
1832 sub thaw
1833 {
1834   croak ("Thaw not implemented");
1835 }
1836
1837
1838 sub freezequote
1839 {
1840   my $s = shift;
1841   $s =~ s/\\/\\\\/g;
1842   $s =~ s/\n/\\n/g;
1843   return $s;
1844 }
1845
1846
1847 sub freezeunquote
1848 {
1849   my $s = shift;
1850   $s =~ s{\\(.)}{$1 eq "n" ? "\n" : $1}ge;
1851   return $s;
1852 }
1853
1854
1855 sub srun
1856 {
1857   my $srunargs = shift;
1858   my $execargs = shift;
1859   my $opts = shift || {};
1860   my $stdin = shift;
1861   my $args = $have_slurm ? [@$srunargs, @$execargs] : $execargs;
1862
1863   $Data::Dumper::Terse = 1;
1864   $Data::Dumper::Indent = 0;
1865   my $show_cmd = Dumper($args);
1866   $show_cmd =~ s/(TOKEN\\*=)[^\s\']+/${1}[...]/g;
1867   $show_cmd =~ s/\n/ /g;
1868   if ($opts->{fork}) {
1869     Log(undef, "starting: $show_cmd");
1870   } else {
1871     # This is a child process: parent is in charge of reading our
1872     # stderr and copying it to Log() if needed.
1873     warn "starting: $show_cmd\n";
1874   }
1875
1876   if (defined $stdin) {
1877     my $child = open STDIN, "-|";
1878     defined $child or die "no fork: $!";
1879     if ($child == 0) {
1880       print $stdin or die $!;
1881       close STDOUT or die $!;
1882       exit 0;
1883     }
1884   }
1885
1886   return system (@$args) if $opts->{fork};
1887
1888   exec @$args;
1889   warn "ENV size is ".length(join(" ",%ENV));
1890   die "exec failed: $!: @$args";
1891 }
1892
1893
1894 sub ban_node_by_slot {
1895   # Don't start any new jobsteps on this node for 60 seconds
1896   my $slotid = shift;
1897   $slot[$slotid]->{node}->{hold_until} = 60 + scalar time;
1898   $slot[$slotid]->{node}->{hold_count}++;
1899   Log (undef, "backing off node " . $slot[$slotid]->{node}->{name} . " for 60 seconds");
1900 }
1901
1902 sub must_lock_now
1903 {
1904   my ($lockfile, $error_message) = @_;
1905   open L, ">", $lockfile or croak("$lockfile: $!");
1906   if (!flock L, LOCK_EX|LOCK_NB) {
1907     croak("Can't lock $lockfile: $error_message\n");
1908   }
1909 }
1910
1911 sub find_docker_image {
1912   # Given a Keep locator, check to see if it contains a Docker image.
1913   # If so, return its stream name and Docker hash.
1914   # If not, return undef for both values.
1915   my $locator = shift;
1916   my ($streamname, $filename);
1917   my $image = api_call("collections/get", uuid => $locator);
1918   if ($image) {
1919     foreach my $line (split(/\n/, $image->{manifest_text})) {
1920       my @tokens = split(/\s+/, $line);
1921       next if (!@tokens);
1922       $streamname = shift(@tokens);
1923       foreach my $filedata (grep(/^\d+:\d+:/, @tokens)) {
1924         if (defined($filename)) {
1925           return (undef, undef);  # More than one file in the Collection.
1926         } else {
1927           $filename = (split(/:/, $filedata, 3))[2];
1928         }
1929       }
1930     }
1931   }
1932   if (defined($filename) and ($filename =~ /^([0-9A-Fa-f]{64})\.tar$/)) {
1933     return ($streamname, $1);
1934   } else {
1935     return (undef, undef);
1936   }
1937 }
1938
1939 sub retry_count {
1940   # Calculate the number of times an operation should be retried,
1941   # assuming exponential backoff, and that we're willing to retry as
1942   # long as tasks have been running.  Enforce a minimum of 3 retries.
1943   my ($starttime, $endtime, $timediff, $retries);
1944   if (@jobstep) {
1945     $starttime = $jobstep[0]->{starttime};
1946     $endtime = $jobstep[-1]->{finishtime};
1947   }
1948   if (!defined($starttime)) {
1949     $timediff = 0;
1950   } elsif (!defined($endtime)) {
1951     $timediff = time - $starttime;
1952   } else {
1953     $timediff = ($endtime - $starttime) - (time - $endtime);
1954   }
1955   if ($timediff > 0) {
1956     $retries = int(log($timediff) / log(2));
1957   } else {
1958     $retries = 1;  # Use the minimum.
1959   }
1960   return ($retries > 3) ? $retries : 3;
1961 }
1962
1963 sub retry_op {
1964   # Pass in two function references.
1965   # This method will be called with the remaining arguments.
1966   # If it dies, retry it with exponential backoff until it succeeds,
1967   # or until the current retry_count is exhausted.  After each failure
1968   # that can be retried, the second function will be called with
1969   # the current try count (0-based), next try time, and error message.
1970   my $operation = shift;
1971   my $retry_callback = shift;
1972   my $retries = retry_count();
1973   foreach my $try_count (0..$retries) {
1974     my $next_try = time + (2 ** $try_count);
1975     my $result = eval { $operation->(@_); };
1976     if (!$@) {
1977       return $result;
1978     } elsif ($try_count < $retries) {
1979       $retry_callback->($try_count, $next_try, $@);
1980       my $sleep_time = $next_try - time;
1981       sleep($sleep_time) if ($sleep_time > 0);
1982     }
1983   }
1984   # Ensure the error message ends in a newline, so Perl doesn't add
1985   # retry_op's line number to it.
1986   chomp($@);
1987   die($@ . "\n");
1988 }
1989
1990 sub api_call {
1991   # Pass in a /-separated API method name, and arguments for it.
1992   # This function will call that method, retrying as needed until
1993   # the current retry_count is exhausted, with a log on the first failure.
1994   my $method_name = shift;
1995   my $log_api_retry = sub {
1996     my ($try_count, $next_try_at, $errmsg) = @_;
1997     $errmsg =~ s/\s*\bat \Q$0\E line \d+\.?\s*//;
1998     $errmsg =~ s/\s/ /g;
1999     $errmsg =~ s/\s+$//;
2000     my $retry_msg;
2001     if ($next_try_at < time) {
2002       $retry_msg = "Retrying.";
2003     } else {
2004       my $next_try_fmt = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($next_try_at);
2005       $retry_msg = "Retrying at $next_try_fmt.";
2006     }
2007     Log(undef, "API method $method_name failed: $errmsg. $retry_msg");
2008   };
2009   my $method = $arv;
2010   foreach my $key (split(/\//, $method_name)) {
2011     $method = $method->{$key};
2012   }
2013   return retry_op(sub { $method->execute(@_); }, $log_api_retry, @_);
2014 }
2015
2016 sub exit_status_s {
2017   # Given a $?, return a human-readable exit code string like "0" or
2018   # "1" or "0 with signal 1" or "1 with signal 11".
2019   my $exitcode = shift;
2020   my $s = $exitcode >> 8;
2021   if ($exitcode & 0x7f) {
2022     $s .= " with signal " . ($exitcode & 0x7f);
2023   }
2024   if ($exitcode & 0x80) {
2025     $s .= " with core dump";
2026   }
2027   return $s;
2028 }
2029
2030 sub handle_readall {
2031   # Pass in a glob reference to a file handle.
2032   # Read all its contents and return them as a string.
2033   my $fh_glob_ref = shift;
2034   local $/ = undef;
2035   return <$fh_glob_ref>;
2036 }
2037
2038 sub tar_filename_n {
2039   my $n = shift;
2040   return sprintf("%s/git.%s.%d.tar", $ENV{CRUNCH_TMP}, $job_id, $n);
2041 }
2042
2043 sub add_git_archive {
2044   # Pass in a git archive command as a string or list, a la system().
2045   # This method will save its output to be included in the archive sent to the
2046   # build script.
2047   my $git_input;
2048   $git_tar_count++;
2049   if (!open(GIT_ARCHIVE, ">", tar_filename_n($git_tar_count))) {
2050     croak("Failed to save git archive: $!");
2051   }
2052   my $git_pid = open2(">&GIT_ARCHIVE", $git_input, @_);
2053   close($git_input);
2054   waitpid($git_pid, 0);
2055   close(GIT_ARCHIVE);
2056   if ($?) {
2057     croak("Failed to save git archive: git exited " . exit_status_s($?));
2058   }
2059 }
2060
2061 sub combined_git_archive {
2062   # Combine all saved tar archives into a single archive, then return its
2063   # contents in a string.  Return undef if no archives have been saved.
2064   if ($git_tar_count < 1) {
2065     return undef;
2066   }
2067   my $base_tar_name = tar_filename_n(1);
2068   foreach my $tar_to_append (map { tar_filename_n($_); } (2..$git_tar_count)) {
2069     my $tar_exit = system("tar", "-Af", $base_tar_name, $tar_to_append);
2070     if ($tar_exit != 0) {
2071       croak("Error preparing build archive: tar -A exited " .
2072             exit_status_s($tar_exit));
2073     }
2074   }
2075   if (!open(GIT_TAR, "<", $base_tar_name)) {
2076     croak("Could not open build archive: $!");
2077   }
2078   my $tar_contents = handle_readall(\*GIT_TAR);
2079   close(GIT_TAR);
2080   return $tar_contents;
2081 }
2082
2083 sub set_nonblocking {
2084   my $fh = shift;
2085   my $flags = fcntl ($fh, F_GETFL, 0) or croak ($!);
2086   fcntl ($fh, F_SETFL, $flags | O_NONBLOCK) or croak ($!);
2087 }
2088
2089 __DATA__
2090 #!/usr/bin/env perl
2091 #
2092 # This is crunch-job's internal dispatch script.  crunch-job running on the API
2093 # server invokes this script on individual compute nodes, or localhost if we're
2094 # running a job locally.  It gets called in two modes:
2095 #
2096 # * No arguments: Installation mode.  Read a tar archive from the DATA
2097 #   file handle; it includes the Crunch script's source code, and
2098 #   maybe SDKs as well.  Those should be installed in the proper
2099 #   locations.  This runs outside of any Docker container, so don't try to
2100 #   introspect Crunch's runtime environment.
2101 #
2102 # * With arguments: Crunch script run mode.  This script should set up the
2103 #   environment, then run the command specified in the arguments.  This runs
2104 #   inside any Docker container.
2105
2106 use Fcntl ':flock';
2107 use File::Path qw( make_path remove_tree );
2108 use POSIX qw(getcwd);
2109
2110 use constant TASK_TEMPFAIL => 111;
2111
2112 # Map SDK subdirectories to the path environments they belong to.
2113 my %SDK_ENVVARS = ("perl/lib" => "PERLLIB", "ruby/lib" => "RUBYLIB");
2114
2115 my $destdir = $ENV{"CRUNCH_SRC"};
2116 my $archive_hash = $ENV{"CRUNCH_GIT_ARCHIVE_HASH"};
2117 my $repo = $ENV{"CRUNCH_SRC_URL"};
2118 my $install_dir = $ENV{"CRUNCH_INSTALL"} || (getcwd() . "/opt");
2119 my $job_work = $ENV{"JOB_WORK"};
2120 my $task_work = $ENV{"TASK_WORK"};
2121
2122 open(STDOUT_ORIG, ">&", STDOUT);
2123 open(STDERR_ORIG, ">&", STDERR);
2124
2125 for my $dir ($destdir, $job_work, $task_work) {
2126   if ($dir) {
2127     make_path $dir;
2128     -e $dir or die "Failed to create temporary directory ($dir): $!";
2129   }
2130 }
2131
2132 if ($task_work) {
2133   remove_tree($task_work, {keep_root => 1});
2134 }
2135
2136 ### Crunch script run mode
2137 if (@ARGV) {
2138   # We want to do routine logging during task 0 only.  This gives the user
2139   # the information they need, but avoids repeating the information for every
2140   # task.
2141   my $Log;
2142   if ($ENV{TASK_SEQUENCE} eq "0") {
2143     $Log = sub {
2144       my $msg = shift;
2145       printf STDERR_ORIG "[Crunch] $msg\n", @_;
2146     };
2147   } else {
2148     $Log = sub { };
2149   }
2150
2151   my $python_src = "$install_dir/python";
2152   my $venv_dir = "$job_work/.arvados.venv";
2153   my $venv_built = -e "$venv_dir/bin/activate";
2154   if ((!$venv_built) and (-d $python_src) and can_run("virtualenv")) {
2155     shell_or_die(undef, "virtualenv", "--quiet", "--system-site-packages",
2156                  "--python=python2.7", $venv_dir);
2157     shell_or_die(TASK_TEMPFAIL, "$venv_dir/bin/pip", "--quiet", "install", "-I", $python_src);
2158     $venv_built = 1;
2159     $Log->("Built Python SDK virtualenv");
2160   }
2161
2162   my $pip_bin = "pip";
2163   if ($venv_built) {
2164     $Log->("Running in Python SDK virtualenv");
2165     $pip_bin = "$venv_dir/bin/pip";
2166     my $orig_argv = join(" ", map { quotemeta($_); } @ARGV);
2167     @ARGV = ("/bin/sh", "-ec",
2168              ". \Q$venv_dir/bin/activate\E; exec $orig_argv");
2169   } elsif (-d $python_src) {
2170     $Log->("Warning: virtualenv not found inside Docker container default " .
2171            "\$PATH. Can't install Python SDK.");
2172   }
2173
2174   my $pkgs = `(\Q$pip_bin\E freeze 2>/dev/null | grep arvados) || dpkg-query --show '*arvados*'`;
2175   if ($pkgs) {
2176     $Log->("Using Arvados SDK:");
2177     foreach my $line (split /\n/, $pkgs) {
2178       $Log->($line);
2179     }
2180   } else {
2181     $Log->("Arvados SDK packages not found");
2182   }
2183
2184   while (my ($sdk_dir, $sdk_envkey) = each(%SDK_ENVVARS)) {
2185     my $sdk_path = "$install_dir/$sdk_dir";
2186     if (-d $sdk_path) {
2187       if ($ENV{$sdk_envkey}) {
2188         $ENV{$sdk_envkey} = "$sdk_path:" . $ENV{$sdk_envkey};
2189       } else {
2190         $ENV{$sdk_envkey} = $sdk_path;
2191       }
2192       $Log->("Arvados SDK added to %s", $sdk_envkey);
2193     }
2194   }
2195
2196   exec(@ARGV);
2197   die "Cannot exec `@ARGV`: $!";
2198 }
2199
2200 ### Installation mode
2201 open L, ">", "$destdir.lock" or die "$destdir.lock: $!";
2202 flock L, LOCK_EX;
2203 if (readlink ("$destdir.archive_hash") eq $archive_hash && -d $destdir) {
2204   # This exact git archive (source + arvados sdk) is already installed
2205   # here, so there's no need to reinstall it.
2206
2207   # We must consume our DATA section, though: otherwise the process
2208   # feeding it to us will get SIGPIPE.
2209   my $buf;
2210   while (read(DATA, $buf, 65536)) { }
2211
2212   exit(0);
2213 }
2214
2215 unlink "$destdir.archive_hash";
2216 mkdir $destdir;
2217
2218 do {
2219   # Ignore SIGPIPE: we check retval of close() instead. See perlipc(1).
2220   local $SIG{PIPE} = "IGNORE";
2221   warn "Extracting archive: $archive_hash\n";
2222   # --ignore-zeros is necessary sometimes: depending on how much NUL
2223   # padding tar -A put on our combined archive (which in turn depends
2224   # on the length of the component archives) tar without
2225   # --ignore-zeros will exit before consuming stdin and cause close()
2226   # to fail on the resulting SIGPIPE.
2227   if (!open(TARX, "|-", "tar", "--ignore-zeros", "-xC", $destdir)) {
2228     die "Error launching 'tar -xC $destdir': $!";
2229   }
2230   # If we send too much data to tar in one write (> 4-5 MiB), it stops, and we
2231   # get SIGPIPE.  We must feed it data incrementally.
2232   my $tar_input;
2233   while (read(DATA, $tar_input, 65536)) {
2234     print TARX $tar_input;
2235   }
2236   if(!close(TARX)) {
2237     die "'tar -xC $destdir' exited $?: $!";
2238   }
2239 };
2240
2241 mkdir $install_dir;
2242
2243 my $sdk_root = "$destdir/.arvados.sdk/sdk";
2244 if (-d $sdk_root) {
2245   foreach my $sdk_lang (("python",
2246                          map { (split /\//, $_, 2)[0]; } keys(%SDK_ENVVARS))) {
2247     if (-d "$sdk_root/$sdk_lang") {
2248       if (!rename("$sdk_root/$sdk_lang", "$install_dir/$sdk_lang")) {
2249         die "Failed to install $sdk_lang SDK: $!";
2250       }
2251     }
2252   }
2253 }
2254
2255 my $python_dir = "$install_dir/python";
2256 if ((-d $python_dir) and can_run("python2.7")) {
2257   open(my $egg_info_pipe, "-|",
2258        "python2.7 \Q$python_dir/setup.py\E egg_info 2>&1 >/dev/null");
2259   my @egg_info_errors = <$egg_info_pipe>;
2260   close($egg_info_pipe);
2261
2262   if ($?) {
2263     if (@egg_info_errors and (($egg_info_errors[-1] =~ /\bgit\b/) or ($egg_info_errors[-1] =~ /\[Errno 2\]/))) {
2264       # egg_info apparently failed because it couldn't ask git for a build tag.
2265       # Specify no build tag.
2266       open(my $pysdk_cfg, ">>", "$python_dir/setup.cfg");
2267       print $pysdk_cfg "\n[egg_info]\ntag_build =\n";
2268       close($pysdk_cfg);
2269     } else {
2270       my $egg_info_exit = $? >> 8;
2271       foreach my $errline (@egg_info_errors) {
2272         warn $errline;
2273       }
2274       warn "python setup.py egg_info failed: exit $egg_info_exit";
2275       exit ($egg_info_exit || 1);
2276     }
2277   }
2278 }
2279
2280 # Hide messages from the install script (unless it fails: shell_or_die
2281 # will show $destdir.log in that case).
2282 open(STDOUT, ">>", "$destdir.log");
2283 open(STDERR, ">&", STDOUT);
2284
2285 if (-e "$destdir/crunch_scripts/install") {
2286     shell_or_die (undef, "$destdir/crunch_scripts/install", $install_dir);
2287 } elsif (!-e "./install.sh" && -e "./tests/autotests.sh") {
2288     # Old version
2289     shell_or_die (undef, "./tests/autotests.sh", $install_dir);
2290 } elsif (-e "./install.sh") {
2291     shell_or_die (undef, "./install.sh", $install_dir);
2292 }
2293
2294 if ($archive_hash) {
2295     unlink "$destdir.archive_hash.new";
2296     symlink ($archive_hash, "$destdir.archive_hash.new") or die "$destdir.archive_hash.new: $!";
2297     rename ("$destdir.archive_hash.new", "$destdir.archive_hash") or die "$destdir.archive_hash: $!";
2298 }
2299
2300 close L;
2301
2302 sub can_run {
2303   my $command_name = shift;
2304   open(my $which, "-|", "which", $command_name);
2305   while (<$which>) { }
2306   close($which);
2307   return ($? == 0);
2308 }
2309
2310 sub shell_or_die
2311 {
2312   my $exitcode = shift;
2313
2314   if ($ENV{"DEBUG"}) {
2315     print STDERR "@_\n";
2316   }
2317   if (system (@_) != 0) {
2318     my $err = $!;
2319     my $code = $?;
2320     my $exitstatus = sprintf("exit %d signal %d", $code >> 8, $code & 0x7f);
2321     open STDERR, ">&STDERR_ORIG";
2322     system ("cat $destdir.log >&2");
2323     warn "@_ failed ($err): $exitstatus";
2324     if (defined($exitcode)) {
2325       exit $exitcode;
2326     }
2327     else {
2328       exit (($code >> 8) || 1);
2329     }
2330   }
2331 }
2332
2333 __DATA__