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