sdk/cli/bin/crunch-job

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