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