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