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