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