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