sdk/cli/bin/crunch-job

   1 #!/usr/bin/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
  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/tree","script":"scriptname",...}'
  19
  20 =head1 OPTIONS
  21
  22 =over
  23
  24 =item --force-unlock
  25
  26 If the job is already locked, steal the lock and run it anyway.
  27
  28 =item --git-dir
  29
  30 Path to .git directory where the specified commit is found.
  31
  32 =item --job-api-token
  33
  34 Arvados API authorization token to use during the course of the job.
  35
  36 =item --no-clear-tmp
  37
  38 Do not clear per-job/task temporary directories during initial job
  39 setup. This can speed up development and debugging when running jobs
  40 locally.
  41
  42 =back
  43
  44 =head1 RUNNING JOBS LOCALLY
  45
  46 crunch-job's log messages appear on stderr along with the job tasks'
  47 stderr streams. The log is saved in Keep at each checkpoint and when
  48 the job finishes.
  49
  50 If the job succeeds, the job's output locator is printed on stdout.
  51
  52 While the job is running, the following signals are accepted:
  53
  54 =over
  55
  56 =item control-C, SIGINT, SIGQUIT
  57
  58 Save a checkpoint, terminate any job tasks that are running, and stop.
  59
  60 =item SIGALRM
  61
  62 Save a checkpoint and continue.
  63
  64 =item SIGHUP
  65
  66 Refresh node allocation (i.e., check whether any nodes have been added
  67 or unallocated) and attributes of the Job record that should affect
  68 behavior (e.g., cancel job if cancelled_at becomes non-nil).
  69
  70 =back
  71
  72 =cut
  73
  74
  75 use strict;
  76 use POSIX ':sys_wait_h';
  77 use POSIX qw(strftime);
  78 use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
  79 use Arvados;
  80 use Digest::MD5 qw(md5_hex);
  81 use Getopt::Long;
  82 use IPC::Open2;
  83 use IO::Select;
  84 use File::Temp;
  85 use Fcntl ':flock';
  86 use File::Path qw( make_path remove_tree );
  87
  88 use constant EX_TEMPFAIL => 75;
  89
  90 $ENV{"TMPDIR"} ||= "/tmp";
  91 unless (defined $ENV{"CRUNCH_TMP"}) {
  92   $ENV{"CRUNCH_TMP"} = $ENV{"TMPDIR"} . "/crunch-job";
  93   if ($ENV{"USER"} ne "crunch" && $< != 0) {
  94     # use a tmp dir unique for my uid
  95     $ENV{"CRUNCH_TMP"} .= "-$<";
  96   }
  97 }
  98
  99 # Create the tmp directory if it does not exist
 100 if ( ! -d $ENV{"CRUNCH_TMP"} ) {
 101   make_path $ENV{"CRUNCH_TMP"} or die "Failed to create temporary working directory: " . $ENV{"CRUNCH_TMP"};
 102 }
 103
 104 $ENV{"JOB_WORK"} = $ENV{"CRUNCH_TMP"} . "/work";
 105 $ENV{"CRUNCH_INSTALL"} = "$ENV{CRUNCH_TMP}/opt";
 106 $ENV{"CRUNCH_WORK"} = $ENV{"JOB_WORK"}; # deprecated
 107 mkdir ($ENV{"JOB_WORK"});
 108
 109 my $force_unlock;
 110 my $git_dir;
 111 my $jobspec;
 112 my $job_api_token;
 113 my $no_clear_tmp;
 114 my $resume_stash;
 115 GetOptions('force-unlock' => \$force_unlock,
 116            'git-dir=s' => \$git_dir,
 117            'job=s' => \$jobspec,
 118            'job-api-token=s' => \$job_api_token,
 119            'no-clear-tmp' => \$no_clear_tmp,
 120            'resume-stash=s' => \$resume_stash,
 121     );
 122
 123 if (defined $job_api_token) {
 124   $ENV{ARVADOS_API_TOKEN} = $job_api_token;
 125 }
 126
 127 my $have_slurm = exists $ENV{SLURM_JOBID} && exists $ENV{SLURM_NODELIST};
 128 my $local_job = 0;
 129
 130
 131 $SIG{'USR1'} = sub
 132 {
 133   $main::ENV{CRUNCH_DEBUG} = 1;
 134 };
 135 $SIG{'USR2'} = sub
 136 {
 137   $main::ENV{CRUNCH_DEBUG} = 0;
 138 };
 139
 140
 141
 142 my $arv = Arvados->new('apiVersion' => 'v1');
 143 my $local_logfile;
 144
 145 my $User = $arv->{'users'}->{'current'}->execute;
 146
 147 my $Job;
 148 my $job_id;
 149 my $dbh;
 150 my $sth;
 151 if ($jobspec =~ /^[-a-z\d]+$/)
 152 {
 153   # $jobspec is an Arvados UUID, not a JSON job specification
 154   $Job = $arv->{'jobs'}->{'get'}->execute('uuid' => $jobspec);
 155   if (!$force_unlock) {
 156     # If some other crunch-job process has grabbed this job (or we see
 157     # other evidence that the job is already underway) we exit
 158     # EX_TEMPFAIL so crunch-dispatch (our parent process) doesn't
 159     # mark the job as failed.
 160     if ($Job->{'is_locked_by_uuid'}) {
 161       Log(undef, "Job is locked by " . $Job->{'is_locked_by_uuid'});
 162       exit EX_TEMPFAIL;
 163     }
 164     if ($Job->{'state'} ne 'Queued') {
 165       Log(undef, "Job state is " . $Job->{'state'} . ", but I can only start queued jobs.");
 166       exit EX_TEMPFAIL;
 167     }
 168     if ($Job->{'success'} ne undef) {
 169       Log(undef, "Job 'success' flag (" . $Job->{'success'} . ") is not null");
 170       exit EX_TEMPFAIL;
 171     }
 172     if ($Job->{'running'}) {
 173       Log(undef, "Job 'running' flag is already set");
 174       exit EX_TEMPFAIL;
 175     }
 176     if ($Job->{'started_at'}) {
 177       Log(undef, "Job 'started_at' time is already set (" . $Job->{'started_at'} . ")");
 178       exit EX_TEMPFAIL;
 179     }
 180   }
 181 }
 182 else
 183 {
 184   $Job = JSON::decode_json($jobspec);
 185
 186   if (!$resume_stash)
 187   {
 188     map { croak ("No $_ specified") unless $Job->{$_} }
 189     qw(script script_version script_parameters);
 190   }
 191
 192   $Job->{'is_locked_by_uuid'} = $User->{'uuid'};
 193   $Job->{'started_at'} = gmtime;
 194
 195   $Job = $arv->{'jobs'}->{'create'}->execute('job' => $Job);
 196 }
 197 $job_id = $Job->{'uuid'};
 198
 199 my $keep_logfile = $job_id . '.log.txt';
 200 $local_logfile = File::Temp->new();
 201
 202 $Job->{'runtime_constraints'} ||= {};
 203 $Job->{'runtime_constraints'}->{'max_tasks_per_node'} ||= 0;
 204 my $max_ncpus = $Job->{'runtime_constraints'}->{'max_tasks_per_node'};
 205
 206
 207 Log (undef, "check slurm allocation");
 208 my @slot;
 209 my @node;
 210 # Should use $ENV{SLURM_TASKS_PER_NODE} instead of sinfo? (eg. "4(x3),2,4(x2)")
 211 my @sinfo;
 212 if (!$have_slurm)
 213 {
 214   my $localcpus = 0 + `grep -cw ^processor /proc/cpuinfo` || 1;
 215   push @sinfo, "$localcpus localhost";
 216 }
 217 if (exists $ENV{SLURM_NODELIST})
 218 {
 219   push @sinfo, `sinfo -h --format='%c %N' --nodes=\Q$ENV{SLURM_NODELIST}\E`;
 220 }
 221 foreach (@sinfo)
 222 {
 223   my ($ncpus, $slurm_nodelist) = split;
 224   $ncpus = $max_ncpus if $max_ncpus && $ncpus > $max_ncpus;
 225
 226   my @nodelist;
 227   while ($slurm_nodelist =~ s/^([^\[,]+?(\[.*?\])?)(,|$)//)
 228   {
 229     my $nodelist = $1;
 230     if ($nodelist =~ /\[((\d+)(-(\d+))?(,(\d+)(-(\d+))?)*)\]/)
 231     {
 232       my $ranges = $1;
 233       foreach (split (",", $ranges))
 234       {
 235         my ($a, $b);
 236         if (/(\d+)-(\d+)/)
 237         {
 238           $a = $1;
 239           $b = $2;
 240         }
 241         else
 242         {
 243           $a = $_;
 244           $b = $_;
 245         }
 246         push @nodelist, map {
 247           my $n = $nodelist;
 248           $n =~ s/\[[-,\d]+\]/$_/;
 249           $n;
 250         } ($a..$b);
 251       }
 252     }
 253     else
 254     {
 255       push @nodelist, $nodelist;
 256     }
 257   }
 258   foreach my $nodename (@nodelist)
 259   {
 260     Log (undef, "node $nodename - $ncpus slots");
 261     my $node = { name => $nodename,
 262                  ncpus => $ncpus,
 263                  losing_streak => 0,
 264                  hold_until => 0 };
 265     foreach my $cpu (1..$ncpus)
 266     {
 267       push @slot, { node => $node,
 268                     cpu => $cpu };
 269     }
 270   }
 271   push @node, @nodelist;
 272 }
 273
 274
 275
 276 # Ensure that we get one jobstep running on each allocated node before
 277 # we start overloading nodes with concurrent steps
 278
 279 @slot = sort { $a->{cpu} <=> $b->{cpu} } @slot;
 280
 281
 282
 283 my $jobmanager_id;
 284 # Claim this job, and make sure nobody else does
 285 unless ($Job->update_attributes('is_locked_by_uuid' => $User->{'uuid'}) &&
 286         $Job->{'is_locked_by_uuid'} == $User->{'uuid'}) {
 287   Log(undef, "Error while updating / locking job, exiting ".EX_TEMPFAIL);
 288   exit EX_TEMPFAIL;
 289 }
 290 $Job->update_attributes('state' => 'Running',
 291                         'tasks_summary' => { 'failed' => 0,
 292                                              'todo' => 1,
 293                                              'running' => 0,
 294                                              'done' => 0 });
 295
 296
 297 Log (undef, "start");
 298 $SIG{'INT'} = sub { $main::please_freeze = 1; };
 299 $SIG{'QUIT'} = sub { $main::please_freeze = 1; };
 300 $SIG{'TERM'} = \&croak;
 301 $SIG{'TSTP'} = sub { $main::please_freeze = 1; };
 302 $SIG{'ALRM'} = sub { $main::please_info = 1; };
 303 $SIG{'CONT'} = sub { $main::please_continue = 1; };
 304 $SIG{'HUP'} = sub { $main::please_refresh = 1; };
 305
 306 $main::please_freeze = 0;
 307 $main::please_info = 0;
 308 $main::please_continue = 0;
 309 $main::please_refresh = 0;
 310 my $jobsteps_must_output_keys = 0;      # becomes 1 when any task outputs a key
 311
 312 grep { $ENV{$1} = $2 if /^(NOCACHE.*?)=(.*)/ } split ("\n", $$Job{knobs});
 313 $ENV{"CRUNCH_JOB_UUID"} = $job_id;
 314 $ENV{"JOB_UUID"} = $job_id;
 315
 316
 317 my @jobstep;
 318 my @jobstep_todo = ();
 319 my @jobstep_done = ();
 320 my @jobstep_tomerge = ();
 321 my $jobstep_tomerge_level = 0;
 322 my $squeue_checked;
 323 my $squeue_kill_checked;
 324 my $output_in_keep = 0;
 325 my $latest_refresh = scalar time;
 326
 327
 328
 329 if (defined $Job->{thawedfromkey})
 330 {
 331   thaw ($Job->{thawedfromkey});
 332 }
 333 else
 334 {
 335   my $first_task = $arv->{'job_tasks'}->{'create'}->execute('job_task' => {
 336     'job_uuid' => $Job->{'uuid'},
 337     'sequence' => 0,
 338     'qsequence' => 0,
 339     'parameters' => {},
 340                                                           });
 341   push @jobstep, { 'level' => 0,
 342                    'failures' => 0,
 343                    'arvados_task' => $first_task,
 344                  };
 345   push @jobstep_todo, 0;
 346 }
 347
 348
 349 if (!$have_slurm)
 350 {
 351   must_lock_now("$ENV{CRUNCH_TMP}/.lock", "a job is already running here.");
 352 }
 353
 354
 355 my $build_script;
 356 do {
 357   local $/ = undef;
 358   $build_script = <DATA>;
 359 };
 360 my $nodelist = join(",", @node);
 361
 362 if (!defined $no_clear_tmp) {
 363   # Clean out crunch_tmp/work, crunch_tmp/opt, crunch_tmp/src*
 364   Log (undef, "Clean work dirs");
 365
 366   my $cleanpid = fork();
 367   if ($cleanpid == 0)
 368   {
 369     srun (["srun", "--nodelist=$nodelist", "-D", $ENV{'TMPDIR'}],
 370           ['bash', '-c', 'if mount | grep -q $JOB_WORK/; then for i in $JOB_WORK/*keep; do /bin/fusermount -z -u $i; done; fi; sleep 1; rm -rf $JOB_WORK $CRUNCH_TMP/opt $CRUNCH_TMP/src*']);
 371     exit (1);
 372   }
 373   while (1)
 374   {
 375     last if $cleanpid == waitpid (-1, WNOHANG);
 376     freeze_if_want_freeze ($cleanpid);
 377     select (undef, undef, undef, 0.1);
 378   }
 379   Log (undef, "Cleanup command exited $?");
 380 }
 381
 382
 383 my $git_archive;
 384 if (!defined $git_dir && $Job->{'script_version'} =~ m{^/}) {
 385   # If we're in user-land (i.e., not called from crunch-dispatch)
 386   # script_version can be an absolute directory path, signifying we
 387   # should work straight out of that directory instead of using a git
 388   # commit.
 389   $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{'script_version'};
 390   $ENV{"CRUNCH_SRC"} = $Job->{'script_version'};
 391 }
 392 else {
 393   $ENV{"CRUNCH_SRC"} = "$ENV{CRUNCH_TMP}/src";
 394
 395   # Install requested code version
 396   Log (undef, "Looking for version ".$Job->{script_version}." from repository ".$Job->{repository});
 397
 398   $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{script_version};
 399
 400   # If we're running under crunch-dispatch, it will have already
 401   # pulled the appropriate source tree into its own repository, and
 402   # given us that repo's path as $git_dir.
 403   #
 404   # If we're running a "local" job, we might have to fetch content
 405   # from a remote repository.
 406   #
 407   # (Currently crunch-dispatch gives a local path with --git-dir, but
 408   # we might as well accept URLs there too in case it changes its
 409   # mind.)
 410   my $repo = $git_dir || $Job->{'repository'};
 411
 412   # Repository can be remote or local. If remote, we'll need to fetch it
 413   # to a local dir before doing `git log` et al.
 414   my $repo_location;
 415
 416   if ($repo =~ m{://|\@.*:}) {
 417     # $repo is a git url we can clone, like git:// or https:// or
 418     # file:/// or git@host:repo.git
 419     $repo_location = 'remote';
 420   } elsif ($repo =~ m{^\.*/}) {
 421     # $repo is a local path to a git index. We'll also resolve ../foo
 422     # to ../foo/.git if the latter is a directory.
 423     if (-d "$repo/.git") {
 424       $repo = "$repo/.git";
 425     }
 426     $repo_location = 'local';
 427     Log(undef, "Using local repository '$repo'");
 428   } else {
 429     # $repo is none of the above. It must be the name of a hosted
 430     # repository.
 431     my $arv_repo_list = $arv->{'repositories'}->{'list'}->execute(
 432       'filters' => [['name','=',$repo]]
 433         )->{'items'};
 434     my $n_found = scalar @{$arv_repo_list};
 435     if ($n_found > 0) {
 436       Log(undef, "Repository '$repo' -> "
 437           . join(", ", map { $_->{'uuid'} } @{$arv_repo_list}));
 438     }
 439     if ($n_found != 1) {
 440       croak("Error: Found $n_found repositories with name '$repo'.");
 441     }
 442     $repo = $arv_repo_list->[0]->{'fetch_url'};
 443     $repo_location = 'remote';
 444   }
 445   Log(undef, "Using $repo_location repository '$repo'");
 446   $ENV{"CRUNCH_SRC_URL"} = $repo;
 447
 448   # Resolve given script_version (we'll call that $treeish here) to a
 449   # commit sha1 ($commit).
 450   my $treeish = $Job->{'script_version'};
 451   my $commit;
 452   if ($repo_location eq 'remote') {
 453     # We minimize excess object-fetching by re-using the same bare
 454     # repository in CRUNCH_TMP/.git for multiple crunch-jobs -- we
 455     # just keep adding remotes to it as needed.
 456     my $local_repo = $ENV{'CRUNCH_TMP'}."/.git";
 457     my $gitcmd = "git --git-dir=\Q$local_repo\E";
 458
 459     # Set up our local repo for caching remote objects, making
 460     # archives, etc.
 461     if (!-d $local_repo) {
 462       make_path($local_repo) or croak("Error: could not create $local_repo");
 463     }
 464     # This works (exits 0 and doesn't delete fetched objects) even
 465     # if $local_repo is already initialized:
 466     `$gitcmd init --bare`;
 467     if ($?) {
 468       croak("Error: $gitcmd init --bare exited $?");
 469     }
 470
 471     # If $treeish looks like a hash (or abbrev hash) we look it up in
 472     # our local cache first, since that's cheaper. (We don't want to
 473     # do that with tags/branches though -- those change over time, so
 474     # they should always be resolved by the remote repo.)
 475     if ($treeish =~ /^[0-9a-f]{3,40}$/s) {
 476       my $sha1 = `$gitcmd rev-list -n1 ''\Q$treeish\E`;
 477       if ($? == 0 &&
 478           $sha1 =~ /^$treeish/ && # Don't use commit 123 @ branch abc!
 479           $sha1 =~ /^([0-9a-f]{40})$/) {
 480         $commit = $1;
 481         Log(undef, "Commit $commit already present in $local_repo");
 482       }
 483     }
 484
 485     if (!defined $commit) {
 486       # If $treeish isn't just a hash or abbrev hash, or isn't here
 487       # yet, we need to fetch the remote to resolve it correctly.
 488
 489       # First, remove all local heads. This prevents a name that does
 490       # not exist on the remote from resolving to (or colliding with)
 491       # a previously fetched branch or tag (possibly from a different
 492       # remote).
 493       remove_tree("$local_repo/refs/heads", {keep_root => 1});
 494
 495       Log(undef, "Fetching objects from $repo to $local_repo");
 496       `$gitcmd fetch --no-progress --tags ''\Q$repo\E \Q+refs/heads/*:refs/heads/*\E`;
 497       if ($?) {
 498         croak("Error: `$gitcmd fetch` exited $?");
 499       }
 500     }
 501
 502     # Now that the data is all here, we will use our local repo for
 503     # the rest of our git activities.
 504     $repo = $local_repo;
 505   }
 506
 507   my $gitcmd = "git --git-dir=\Q$repo\E";
 508   my $sha1 = `$gitcmd rev-list -n1 ''\Q$treeish\E`;
 509   unless ($? == 0 && $sha1 =~ /^([0-9a-f]{40})$/) {
 510     croak("`$gitcmd rev-list` exited $?, '$treeish' not found. Giving up.");
 511   }
 512   $commit = $1;
 513   Log(undef, "Version $treeish is commit $commit");
 514
 515   if ($commit ne $Job->{'script_version'}) {
 516     # Record the real commit id in the database, frozentokey, logs,
 517     # etc. -- instead of an abbreviation or a branch name which can
 518     # become ambiguous or point to a different commit in the future.
 519     if (!$Job->update_attributes('script_version' => $commit)) {
 520       croak("Error: failed to update job's script_version attribute");
 521     }
 522   }
 523
 524   $ENV{"CRUNCH_SRC_COMMIT"} = $commit;
 525   $git_archive = `$gitcmd archive ''\Q$commit\E`;
 526   if ($?) {
 527     croak("Error: $gitcmd archive exited $?");
 528   }
 529 }
 530
 531 if (!defined $git_archive) {
 532   Log(undef, "Skip install phase (no git archive)");
 533   if ($have_slurm) {
 534     Log(undef, "Warning: This probably means workers have no source tree!");
 535   }
 536 }
 537 else {
 538   Log(undef, "Run install script on all workers");
 539
 540   my @srunargs = ("srun",
 541                   "--nodelist=$nodelist",
 542                   "-D", $ENV{'TMPDIR'}, "--job-name=$job_id");
 543   my @execargs = ("sh", "-c",
 544                   "mkdir -p $ENV{CRUNCH_INSTALL} && cd $ENV{CRUNCH_TMP} && perl -");
 545
 546   # Note: this section is almost certainly unnecessary if we're
 547   # running tasks in docker containers.
 548   my $installpid = fork();
 549   if ($installpid == 0)
 550   {
 551     srun (\@srunargs, \@execargs, {}, $build_script . $git_archive);
 552     exit (1);
 553   }
 554   while (1)
 555   {
 556     last if $installpid == waitpid (-1, WNOHANG);
 557     freeze_if_want_freeze ($installpid);
 558     select (undef, undef, undef, 0.1);
 559   }
 560   Log (undef, "Install script exited $?");
 561 }
 562
 563 if (!$have_slurm)
 564 {
 565   # Grab our lock again (we might have deleted and re-created CRUNCH_TMP above)
 566   must_lock_now("$ENV{CRUNCH_TMP}/.lock", "a job is already running here.");
 567 }
 568
 569 # If this job requires a Docker image, install that.
 570 my $docker_bin = "/usr/bin/docker.io";
 571 my ($docker_locator, $docker_stream, $docker_hash);
 572 if ($docker_locator = $Job->{docker_image_locator}) {
 573   ($docker_stream, $docker_hash) = find_docker_image($docker_locator);
 574   if (!$docker_hash)
 575   {
 576     croak("No Docker image hash found from locator $docker_locator");
 577   }
 578   $docker_stream =~ s/^\.//;
 579   my $docker_install_script = qq{
 580 if ! $docker_bin images -q --no-trunc | grep -qxF \Q$docker_hash\E; then
 581     arv-get \Q$docker_locator$docker_stream/$docker_hash.tar\E | $docker_bin load
 582 fi
 583 };
 584   my $docker_pid = fork();
 585   if ($docker_pid == 0)
 586   {
 587     srun (["srun", "--nodelist=" . join(',', @node)],
 588           ["/bin/sh", "-ec", $docker_install_script]);
 589     exit ($?);
 590   }
 591   while (1)
 592   {
 593     last if $docker_pid == waitpid (-1, WNOHANG);
 594     freeze_if_want_freeze ($docker_pid);
 595     select (undef, undef, undef, 0.1);
 596   }
 597   if ($? != 0)
 598   {
 599     croak("Installing Docker image from $docker_locator returned exit code $?");
 600   }
 601 }
 602
 603 foreach (qw (script script_version script_parameters runtime_constraints))
 604 {
 605   Log (undef,
 606        "$_ " .
 607        (ref($Job->{$_}) ? JSON::encode_json($Job->{$_}) : $Job->{$_}));
 608 }
 609 foreach (split (/\n/, $Job->{knobs}))
 610 {
 611   Log (undef, "knob " . $_);
 612 }
 613
 614
 615
 616 $main::success = undef;
 617
 618
 619
 620 ONELEVEL:
 621
 622 my $thisround_succeeded = 0;
 623 my $thisround_failed = 0;
 624 my $thisround_failed_multiple = 0;
 625
 626 @jobstep_todo = sort { $jobstep[$a]->{level} <=> $jobstep[$b]->{level}
 627                        or $a <=> $b } @jobstep_todo;
 628 my $level = $jobstep[$jobstep_todo[0]]->{level};
 629 Log (undef, "start level $level");
 630
 631
 632
 633 my %proc;
 634 my @freeslot = (0..$#slot);
 635 my @holdslot;
 636 my %reader;
 637 my $progress_is_dirty = 1;
 638 my $progress_stats_updated = 0;
 639
 640 update_progress_stats();
 641
 642
 643
 644 THISROUND:
 645 for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++)
 646 {
 647   my $id = $jobstep_todo[$todo_ptr];
 648   my $Jobstep = $jobstep[$id];
 649   if ($Jobstep->{level} != $level)
 650   {
 651     next;
 652   }
 653
 654   pipe $reader{$id}, "writer" or croak ($!);
 655   my $flags = fcntl ($reader{$id}, F_GETFL, 0) or croak ($!);
 656   fcntl ($reader{$id}, F_SETFL, $flags | O_NONBLOCK) or croak ($!);
 657
 658   my $childslot = $freeslot[0];
 659   my $childnode = $slot[$childslot]->{node};
 660   my $childslotname = join (".",
 661                             $slot[$childslot]->{node}->{name},
 662                             $slot[$childslot]->{cpu});
 663   my $childpid = fork();
 664   if ($childpid == 0)
 665   {
 666     $SIG{'INT'} = 'DEFAULT';
 667     $SIG{'QUIT'} = 'DEFAULT';
 668     $SIG{'TERM'} = 'DEFAULT';
 669
 670     foreach (values (%reader))
 671     {
 672       close($_);
 673     }
 674     fcntl ("writer", F_SETFL, 0) or croak ($!); # no close-on-exec
 675     open(STDOUT,">&writer");
 676     open(STDERR,">&writer");
 677
 678     undef $dbh;
 679     undef $sth;
 680
 681     delete $ENV{"GNUPGHOME"};
 682     $ENV{"TASK_UUID"} = $Jobstep->{'arvados_task'}->{'uuid'};
 683     $ENV{"TASK_QSEQUENCE"} = $id;
 684     $ENV{"TASK_SEQUENCE"} = $level;
 685     $ENV{"JOB_SCRIPT"} = $Job->{script};
 686     while (my ($param, $value) = each %{$Job->{script_parameters}}) {
 687       $param =~ tr/a-z/A-Z/;
 688       $ENV{"JOB_PARAMETER_$param"} = $value;
 689     }
 690     $ENV{"TASK_SLOT_NODE"} = $slot[$childslot]->{node}->{name};
 691     $ENV{"TASK_SLOT_NUMBER"} = $slot[$childslot]->{cpu};
 692     $ENV{"TASK_WORK"} = $ENV{"JOB_WORK"}."/$id.$$";
 693     $ENV{"HOME"} = $ENV{"TASK_WORK"};
 694     $ENV{"TASK_KEEPMOUNT"} = $ENV{"TASK_WORK"}.".keep";
 695     $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
 696     $ENV{"CRUNCH_NODE_SLOTS"} = $slot[$childslot]->{node}->{ncpus};
 697     $ENV{"PATH"} = $ENV{"CRUNCH_INSTALL"} . "/bin:" . $ENV{"PATH"};
 698
 699     $ENV{"GZIP"} = "-n";
 700
 701     my @srunargs = (
 702       "srun",
 703       "--nodelist=".$childnode->{name},
 704       qw(-n1 -c1 -N1 -D), $ENV{'TMPDIR'},
 705       "--job-name=$job_id.$id.$$",
 706         );
 707     my $build_script_to_send = "";
 708     my $command =
 709         "if [ -e $ENV{TASK_WORK} ]; then rm -rf $ENV{TASK_WORK}; fi; "
 710         ."mkdir -p $ENV{CRUNCH_TMP} $ENV{JOB_WORK} $ENV{TASK_WORK} $ENV{TASK_KEEPMOUNT} "
 711         ."&& cd $ENV{CRUNCH_TMP} ";
 712     if ($build_script)
 713     {
 714       $build_script_to_send = $build_script;
 715       $command .=
 716           "&& perl -";
 717     }
 718     $command .= "&& exec arv-mount --by-id --allow-other $ENV{TASK_KEEPMOUNT} --exec ";
 719     if ($docker_hash)
 720     {
 721       $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -cgroup-parent=docker -cgroup-cid=$ENV{TASK_WORK}/docker.cid -poll=10000 ";
 722       $command .= "$docker_bin run --rm=true --attach=stdout --attach=stderr --user=crunch --cidfile=$ENV{TASK_WORK}/docker.cid ";
 723       # Dynamically configure the container to use the host system as its
 724       # DNS server.  Get the host's global addresses from the ip command,
 725       # and turn them into docker --dns options using gawk.
 726       $command .=
 727           q{$(ip -o address show scope global |
 728               gawk 'match($4, /^([0-9\.:]+)\//, x){print "--dns", x[1]}') };
 729       $command .= "--volume=\Q$ENV{CRUNCH_SRC}:/tmp/crunch-src:ro\E ";
 730       $command .= "--volume=\Q$ENV{TASK_KEEPMOUNT}:/keep:ro\E ";
 731       $command .= "--env=\QHOME=/home/crunch\E ";
 732       while (my ($env_key, $env_val) = each %ENV)
 733       {
 734         if ($env_key =~ /^(ARVADOS|JOB|TASK)_/) {
 735           if ($env_key eq "TASK_WORK") {
 736             $command .= "--env=\QTASK_WORK=/tmp/crunch-job\E ";
 737           }
 738           elsif ($env_key eq "TASK_KEEPMOUNT") {
 739             $command .= "--env=\QTASK_KEEPMOUNT=/keep\E ";
 740           }
 741           else {
 742             $command .= "--env=\Q$env_key=$env_val\E ";
 743           }
 744         }
 745       }
 746       $command .= "--env=\QCRUNCH_NODE_SLOTS=$ENV{CRUNCH_NODE_SLOTS}\E ";
 747       $command .= "--env=\QCRUNCH_SRC=/tmp/crunch-src\E ";
 748       $command .= "\Q$docker_hash\E ";
 749       $command .= "stdbuf --output=0 --error=0 ";
 750       $command .= "/tmp/crunch-src/crunch_scripts/" . $Job->{"script"};
 751     } else {
 752       # Non-docker run
 753       $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -poll=10000 ";
 754       $command .= "stdbuf --output=0 --error=0 ";
 755       $command .= "$ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
 756     }
 757
 758     my @execargs = ('bash', '-c', $command);
 759     srun (\@srunargs, \@execargs, undef, $build_script_to_send);
 760     # exec() failed, we assume nothing happened.
 761     Log(undef, "srun() failed on build script");
 762     die;
 763   }
 764   close("writer");
 765   if (!defined $childpid)
 766   {
 767     close $reader{$id};
 768     delete $reader{$id};
 769     next;
 770   }
 771   shift @freeslot;
 772   $proc{$childpid} = { jobstep => $id,
 773                        time => time,
 774                        slot => $childslot,
 775                        jobstepname => "$job_id.$id.$childpid",
 776                      };
 777   croak ("assert failed: \$slot[$childslot]->{'pid'} exists") if exists $slot[$childslot]->{pid};
 778   $slot[$childslot]->{pid} = $childpid;
 779
 780   Log ($id, "job_task ".$Jobstep->{'arvados_task'}->{'uuid'});
 781   Log ($id, "child $childpid started on $childslotname");
 782   $Jobstep->{starttime} = time;
 783   $Jobstep->{node} = $childnode->{name};
 784   $Jobstep->{slotindex} = $childslot;
 785   delete $Jobstep->{stderr};
 786   delete $Jobstep->{finishtime};
 787
 788   $Jobstep->{'arvados_task'}->{started_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{starttime});
 789   $Jobstep->{'arvados_task'}->save;
 790
 791   splice @jobstep_todo, $todo_ptr, 1;
 792   --$todo_ptr;
 793
 794   $progress_is_dirty = 1;
 795
 796   while (!@freeslot
 797          ||
 798          (@slot > @freeslot && $todo_ptr+1 > $#jobstep_todo))
 799   {
 800     last THISROUND if $main::please_freeze;
 801     if ($main::please_info)
 802     {
 803       $main::please_info = 0;
 804       freeze();
 805       collate_output();
 806       save_meta(1);
 807       update_progress_stats();
 808     }
 809     my $gotsome
 810         = readfrompipes ()
 811         + reapchildren ();
 812     if (!$gotsome)
 813     {
 814       check_refresh_wanted();
 815       check_squeue();
 816       update_progress_stats();
 817       select (undef, undef, undef, 0.1);
 818     }
 819     elsif (time - $progress_stats_updated >= 30)
 820     {
 821       update_progress_stats();
 822     }
 823     if (($thisround_failed_multiple >= 8 && $thisround_succeeded == 0) ||
 824         ($thisround_failed_multiple >= 16 && $thisround_failed_multiple > $thisround_succeeded))
 825     {
 826       my $message = "Repeated failure rate too high ($thisround_failed_multiple/"
 827           .($thisround_failed+$thisround_succeeded)
 828           .") -- giving up on this round";
 829       Log (undef, $message);
 830       last THISROUND;
 831     }
 832
 833     # move slots from freeslot to holdslot (or back to freeslot) if necessary
 834     for (my $i=$#freeslot; $i>=0; $i--) {
 835       if ($slot[$freeslot[$i]]->{node}->{hold_until} > scalar time) {
 836         push @holdslot, (splice @freeslot, $i, 1);
 837       }
 838     }
 839     for (my $i=$#holdslot; $i>=0; $i--) {
 840       if ($slot[$holdslot[$i]]->{node}->{hold_until} <= scalar time) {
 841         push @freeslot, (splice @holdslot, $i, 1);
 842       }
 843     }
 844
 845     # give up if no nodes are succeeding
 846     if (!grep { $_->{node}->{losing_streak} == 0 &&
 847                     $_->{node}->{hold_count} < 4 } @slot) {
 848       my $message = "Every node has failed -- giving up on this round";
 849       Log (undef, $message);
 850       last THISROUND;
 851     }
 852   }
 853 }
 854
 855
 856 push @freeslot, splice @holdslot;
 857 map { $slot[$freeslot[$_]]->{node}->{losing_streak} = 0 } (0..$#freeslot);
 858
 859
 860 Log (undef, "wait for last ".(scalar keys %proc)." children to finish");
 861 while (%proc)
 862 {
 863   if ($main::please_continue) {
 864     $main::please_continue = 0;
 865     goto THISROUND;
 866   }
 867   $main::please_info = 0, freeze(), collate_output(), save_meta(1) if $main::please_info;
 868   readfrompipes ();
 869   if (!reapchildren())
 870   {
 871     check_refresh_wanted();
 872     check_squeue();
 873     update_progress_stats();
 874     select (undef, undef, undef, 0.1);
 875     killem (keys %proc) if $main::please_freeze;
 876   }
 877 }
 878
 879 update_progress_stats();
 880 freeze_if_want_freeze();
 881
 882
 883 if (!defined $main::success)
 884 {
 885   if (@jobstep_todo &&
 886       $thisround_succeeded == 0 &&
 887       ($thisround_failed == 0 || $thisround_failed > 4))
 888   {
 889     my $message = "stop because $thisround_failed tasks failed and none succeeded";
 890     Log (undef, $message);
 891     $main::success = 0;
 892   }
 893   if (!@jobstep_todo)
 894   {
 895     $main::success = 1;
 896   }
 897 }
 898
 899 goto ONELEVEL if !defined $main::success;
 900
 901
 902 release_allocation();
 903 freeze();
 904 my $collated_output = &collate_output();
 905
 906 if (!$collated_output) {
 907   Log(undef, "output undef");
 908 }
 909 else {
 910   eval {
 911     open(my $orig_manifest, '-|', 'arv-get', $collated_output)
 912         or die "failed to get collated manifest: $!";
 913     my $orig_manifest_text = '';
 914     while (my $manifest_line = <$orig_manifest>) {
 915       $orig_manifest_text .= $manifest_line;
 916     }
 917     my $output = $arv->{'collections'}->{'create'}->execute('collection' => {
 918       'manifest_text' => $orig_manifest_text,
 919     });
 920     Log(undef, "output uuid " . $output->{uuid});
 921     Log(undef, "output hash " . $output->{portable_data_hash});
 922     $Job->update_attributes('output' => $output->{portable_data_hash});
 923   };
 924   if ($@) {
 925     Log (undef, "Failed to register output manifest: $@");
 926   }
 927 }
 928
 929 Log (undef, "finish");
 930
 931 save_meta();
 932
 933 my $final_state;
 934 if ($collated_output && $main::success) {
 935   $final_state = 'Complete';
 936 } else {
 937   $final_state = 'Failed';
 938 }
 939 $Job->update_attributes('state' => $final_state);
 940
 941 exit (($final_state eq 'Complete') ? 0 : 1);
 942
 943
 944
 945 sub update_progress_stats
 946 {
 947   $progress_stats_updated = time;
 948   return if !$progress_is_dirty;
 949   my ($todo, $done, $running) = (scalar @jobstep_todo,
 950                                  scalar @jobstep_done,
 951                                  scalar @slot - scalar @freeslot - scalar @holdslot);
 952   $Job->{'tasks_summary'} ||= {};
 953   $Job->{'tasks_summary'}->{'todo'} = $todo;
 954   $Job->{'tasks_summary'}->{'done'} = $done;
 955   $Job->{'tasks_summary'}->{'running'} = $running;
 956   $Job->update_attributes('tasks_summary' => $Job->{'tasks_summary'});
 957   Log (undef, "status: $done done, $running running, $todo todo");
 958   $progress_is_dirty = 0;
 959 }
 960
 961
 962
 963 sub reapchildren
 964 {
 965   my $pid = waitpid (-1, WNOHANG);
 966   return 0 if $pid <= 0;
 967
 968   my $whatslot = ($slot[$proc{$pid}->{slot}]->{node}->{name}
 969                   . "."
 970                   . $slot[$proc{$pid}->{slot}]->{cpu});
 971   my $jobstepid = $proc{$pid}->{jobstep};
 972   my $elapsed = time - $proc{$pid}->{time};
 973   my $Jobstep = $jobstep[$jobstepid];
 974
 975   my $childstatus = $?;
 976   my $exitvalue = $childstatus >> 8;
 977   my $exitinfo = sprintf("exit %d signal %d%s",
 978                          $exitvalue,
 979                          $childstatus & 127,
 980                          ($childstatus & 128 ? ' core dump' : ''));
 981   $Jobstep->{'arvados_task'}->reload;
 982   my $task_success = $Jobstep->{'arvados_task'}->{success};
 983
 984   Log ($jobstepid, "child $pid on $whatslot $exitinfo success=$task_success");
 985
 986   if (!defined $task_success) {
 987     # task did not indicate one way or the other --> fail
 988     $Jobstep->{'arvados_task'}->{success} = 0;
 989     $Jobstep->{'arvados_task'}->save;
 990     $task_success = 0;
 991   }
 992
 993   if (!$task_success)
 994   {
 995     my $temporary_fail;
 996     $temporary_fail ||= $Jobstep->{node_fail};
 997     $temporary_fail ||= ($exitvalue == 111);
 998
 999     ++$thisround_failed;
1000     ++$thisround_failed_multiple if $Jobstep->{'failures'} >= 1;
1001
1002     # Check for signs of a failed or misconfigured node
1003     if (++$slot[$proc{$pid}->{slot}]->{node}->{losing_streak} >=
1004         2+$slot[$proc{$pid}->{slot}]->{node}->{ncpus}) {
1005       # Don't count this against jobstep failure thresholds if this
1006       # node is already suspected faulty and srun exited quickly
1007       if ($slot[$proc{$pid}->{slot}]->{node}->{hold_until} &&
1008           $elapsed < 5) {
1009         Log ($jobstepid, "blaming failure on suspect node " .
1010              $slot[$proc{$pid}->{slot}]->{node}->{name});
1011         $temporary_fail ||= 1;
1012       }
1013       ban_node_by_slot($proc{$pid}->{slot});
1014     }
1015
1016     Log ($jobstepid, sprintf('failure (#%d, %s) after %d seconds',
1017                              ++$Jobstep->{'failures'},
1018                              $temporary_fail ? 'temporary ' : 'permanent',
1019                              $elapsed));
1020
1021     if (!$temporary_fail || $Jobstep->{'failures'} >= 3) {
1022       # Give up on this task, and the whole job
1023       $main::success = 0;
1024       $main::please_freeze = 1;
1025     }
1026     # Put this task back on the todo queue
1027     push @jobstep_todo, $jobstepid;
1028     $Job->{'tasks_summary'}->{'failed'}++;
1029   }
1030   else
1031   {
1032     ++$thisround_succeeded;
1033     $slot[$proc{$pid}->{slot}]->{node}->{losing_streak} = 0;
1034     $slot[$proc{$pid}->{slot}]->{node}->{hold_until} = 0;
1035     push @jobstep_done, $jobstepid;
1036     Log ($jobstepid, "success in $elapsed seconds");
1037   }
1038   $Jobstep->{exitcode} = $childstatus;
1039   $Jobstep->{finishtime} = time;
1040   $Jobstep->{'arvados_task'}->{finished_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{finishtime});
1041   $Jobstep->{'arvados_task'}->save;
1042   process_stderr ($jobstepid, $task_success);
1043   Log ($jobstepid, "output " . $Jobstep->{'arvados_task'}->{output});
1044
1045   close $reader{$jobstepid};
1046   delete $reader{$jobstepid};
1047   delete $slot[$proc{$pid}->{slot}]->{pid};
1048   push @freeslot, $proc{$pid}->{slot};
1049   delete $proc{$pid};
1050
1051   if ($task_success) {
1052     # Load new tasks
1053     my $newtask_list = [];
1054     my $newtask_results;
1055     do {
1056       $newtask_results = $arv->{'job_tasks'}->{'list'}->execute(
1057         'where' => {
1058           'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid}
1059         },
1060         'order' => 'qsequence',
1061         'offset' => scalar(@$newtask_list),
1062       );
1063       push(@$newtask_list, @{$newtask_results->{items}});
1064     } while (@{$newtask_results->{items}});
1065     foreach my $arvados_task (@$newtask_list) {
1066       my $jobstep = {
1067         'level' => $arvados_task->{'sequence'},
1068         'failures' => 0,
1069         'arvados_task' => $arvados_task
1070       };
1071       push @jobstep, $jobstep;
1072       push @jobstep_todo, $#jobstep;
1073     }
1074   }
1075
1076   $progress_is_dirty = 1;
1077   1;
1078 }
1079
1080 sub check_refresh_wanted
1081 {
1082   my @stat = stat $ENV{"CRUNCH_REFRESH_TRIGGER"};
1083   if (@stat && $stat[9] > $latest_refresh) {
1084     $latest_refresh = scalar time;
1085     my $Job2 = $arv->{'jobs'}->{'get'}->execute('uuid' => $jobspec);
1086     for my $attr ('cancelled_at',
1087                   'cancelled_by_user_uuid',
1088                   'cancelled_by_client_uuid',
1089                   'state') {
1090       $Job->{$attr} = $Job2->{$attr};
1091     }
1092     if ($Job->{'state'} ne "Running") {
1093       if ($Job->{'state'} eq "Cancelled") {
1094         Log (undef, "Job cancelled at " . $Job->{'cancelled_at'} . " by user " . $Job->{'cancelled_by_user_uuid'});
1095       } else {
1096         Log (undef, "Job state unexpectedly changed to " . $Job->{'state'});
1097       }
1098       $main::success = 0;
1099       $main::please_freeze = 1;
1100     }
1101   }
1102 }
1103
1104 sub check_squeue
1105 {
1106   # return if the kill list was checked <4 seconds ago
1107   if (defined $squeue_kill_checked && $squeue_kill_checked > time - 4)
1108   {
1109     return;
1110   }
1111   $squeue_kill_checked = time;
1112
1113   # use killem() on procs whose killtime is reached
1114   for (keys %proc)
1115   {
1116     if (exists $proc{$_}->{killtime}
1117         && $proc{$_}->{killtime} <= time)
1118     {
1119       killem ($_);
1120     }
1121   }
1122
1123   # return if the squeue was checked <60 seconds ago
1124   if (defined $squeue_checked && $squeue_checked > time - 60)
1125   {
1126     return;
1127   }
1128   $squeue_checked = time;
1129
1130   if (!$have_slurm)
1131   {
1132     # here is an opportunity to check for mysterious problems with local procs
1133     return;
1134   }
1135
1136   # get a list of steps still running
1137   my @squeue = `squeue -s -h -o '%i %j' && echo ok`;
1138   chop @squeue;
1139   if ($squeue[-1] ne "ok")
1140   {
1141     return;
1142   }
1143   pop @squeue;
1144
1145   # which of my jobsteps are running, according to squeue?
1146   my %ok;
1147   foreach (@squeue)
1148   {
1149     if (/^(\d+)\.(\d+) (\S+)/)
1150     {
1151       if ($1 eq $ENV{SLURM_JOBID})
1152       {
1153         $ok{$3} = 1;
1154       }
1155     }
1156   }
1157
1158   # which of my active child procs (>60s old) were not mentioned by squeue?
1159   foreach (keys %proc)
1160   {
1161     if ($proc{$_}->{time} < time - 60
1162         && !exists $ok{$proc{$_}->{jobstepname}}
1163         && !exists $proc{$_}->{killtime})
1164     {
1165       # kill this proc if it hasn't exited in 30 seconds
1166       $proc{$_}->{killtime} = time + 30;
1167     }
1168   }
1169 }
1170
1171
1172 sub release_allocation
1173 {
1174   if ($have_slurm)
1175   {
1176     Log (undef, "release job allocation");
1177     system "scancel $ENV{SLURM_JOBID}";
1178   }
1179 }
1180
1181
1182 sub readfrompipes
1183 {
1184   my $gotsome = 0;
1185   foreach my $job (keys %reader)
1186   {
1187     my $buf;
1188     while (0 < sysread ($reader{$job}, $buf, 8192))
1189     {
1190       print STDERR $buf if $ENV{CRUNCH_DEBUG};
1191       $jobstep[$job]->{stderr} .= $buf;
1192       preprocess_stderr ($job);
1193       if (length ($jobstep[$job]->{stderr}) > 16384)
1194       {
1195         substr ($jobstep[$job]->{stderr}, 0, 8192) = "";
1196       }
1197       $gotsome = 1;
1198     }
1199   }
1200   return $gotsome;
1201 }
1202
1203
1204 sub preprocess_stderr
1205 {
1206   my $job = shift;
1207
1208   while ($jobstep[$job]->{stderr} =~ /^(.*?)\n/) {
1209     my $line = $1;
1210     substr $jobstep[$job]->{stderr}, 0, 1+length($line), "";
1211     Log ($job, "stderr $line");
1212     if ($line =~ /srun: error: (SLURM job $ENV{SLURM_JOB_ID} has expired|Unable to confirm allocation for job $ENV{SLURM_JOB_ID})/) {
1213       # whoa.
1214       $main::please_freeze = 1;
1215     }
1216     elsif ($line =~ /srun: error: (Node failure on|Unable to create job step) /) {
1217       $jobstep[$job]->{node_fail} = 1;
1218       ban_node_by_slot($jobstep[$job]->{slotindex});
1219     }
1220   }
1221 }
1222
1223
1224 sub process_stderr
1225 {
1226   my $job = shift;
1227   my $task_success = shift;
1228   preprocess_stderr ($job);
1229
1230   map {
1231     Log ($job, "stderr $_");
1232   } split ("\n", $jobstep[$job]->{stderr});
1233 }
1234
1235 sub fetch_block
1236 {
1237   my $hash = shift;
1238   my ($keep, $child_out, $output_block);
1239
1240   my $cmd = "arv-get \Q$hash\E";
1241   open($keep, '-|', $cmd) or die "fetch_block: $cmd: $!";
1242   $output_block = '';
1243   while (1) {
1244     my $buf;
1245     my $bytes = sysread($keep, $buf, 1024 * 1024);
1246     if (!defined $bytes) {
1247       die "reading from arv-get: $!";
1248     } elsif ($bytes == 0) {
1249       # sysread returns 0 at the end of the pipe.
1250       last;
1251     } else {
1252       # some bytes were read into buf.
1253       $output_block .= $buf;
1254     }
1255   }
1256   close $keep;
1257   return $output_block;
1258 }
1259
1260 sub collate_output
1261 {
1262   Log (undef, "collate");
1263
1264   my ($child_out, $child_in);
1265   my $pid = open2($child_out, $child_in, 'arv-put', '--raw',
1266                   '--retries', put_retry_count());
1267   my $joboutput;
1268   for (@jobstep)
1269   {
1270     next if (!exists $_->{'arvados_task'}->{'output'} ||
1271              !$_->{'arvados_task'}->{'success'});
1272     my $output = $_->{'arvados_task'}->{output};
1273     if ($output !~ /^[0-9a-f]{32}(\+\S+)*$/)
1274     {
1275       $output_in_keep ||= $output =~ / [0-9a-f]{32}\S*\+K/;
1276       print $child_in $output;
1277     }
1278     elsif (@jobstep == 1)
1279     {
1280       $joboutput = $output;
1281       last;
1282     }
1283     elsif (defined (my $outblock = fetch_block ($output)))
1284     {
1285       $output_in_keep ||= $outblock =~ / [0-9a-f]{32}\S*\+K/;
1286       print $child_in $outblock;
1287     }
1288     else
1289     {
1290       Log (undef, "XXX fetch_block($output) failed XXX");
1291       $main::success = 0;
1292     }
1293   }
1294   $child_in->close;
1295
1296   if (!defined $joboutput) {
1297     my $s = IO::Select->new($child_out);
1298     if ($s->can_read(120)) {
1299       sysread($child_out, $joboutput, 64 * 1024 * 1024);
1300       chomp($joboutput);
1301     } else {
1302       Log (undef, "timed out reading from 'arv-put'");
1303     }
1304   }
1305   waitpid($pid, 0);
1306
1307   return $joboutput;
1308 }
1309
1310
1311 sub killem
1312 {
1313   foreach (@_)
1314   {
1315     my $sig = 2;                # SIGINT first
1316     if (exists $proc{$_}->{"sent_$sig"} &&
1317         time - $proc{$_}->{"sent_$sig"} > 4)
1318     {
1319       $sig = 15;                # SIGTERM if SIGINT doesn't work
1320     }
1321     if (exists $proc{$_}->{"sent_$sig"} &&
1322         time - $proc{$_}->{"sent_$sig"} > 4)
1323     {
1324       $sig = 9;                 # SIGKILL if SIGTERM doesn't work
1325     }
1326     if (!exists $proc{$_}->{"sent_$sig"})
1327     {
1328       Log ($proc{$_}->{jobstep}, "sending 2x signal $sig to pid $_");
1329       kill $sig, $_;
1330       select (undef, undef, undef, 0.1);
1331       if ($sig == 2)
1332       {
1333         kill $sig, $_;     # srun wants two SIGINT to really interrupt
1334       }
1335       $proc{$_}->{"sent_$sig"} = time;
1336       $proc{$_}->{"killedafter"} = time - $proc{$_}->{"time"};
1337     }
1338   }
1339 }
1340
1341
1342 sub fhbits
1343 {
1344   my($bits);
1345   for (@_) {
1346     vec($bits,fileno($_),1) = 1;
1347   }
1348   $bits;
1349 }
1350
1351
1352 sub Log                         # ($jobstep_id, $logmessage)
1353 {
1354   if ($_[1] =~ /\n/) {
1355     for my $line (split (/\n/, $_[1])) {
1356       Log ($_[0], $line);
1357     }
1358     return;
1359   }
1360   my $fh = select STDERR; $|=1; select $fh;
1361   my $message = sprintf ("%s %d %s %s", $job_id, $$, @_);
1362   $message =~ s{([^ -\176])}{"\\" . sprintf ("%03o", ord($1))}ge;
1363   $message .= "\n";
1364   my $datetime;
1365   if ($local_logfile || -t STDERR) {
1366     my @gmtime = gmtime;
1367     $datetime = sprintf ("%04d-%02d-%02d_%02d:%02d:%02d",
1368                          $gmtime[5]+1900, $gmtime[4]+1, @gmtime[3,2,1,0]);
1369   }
1370   print STDERR ((-t STDERR) ? ($datetime." ".$message) : $message);
1371
1372   if ($local_logfile) {
1373     print $local_logfile $datetime . " " . $message;
1374   }
1375 }
1376
1377
1378 sub croak
1379 {
1380   my ($package, $file, $line) = caller;
1381   my $message = "@_ at $file line $line\n";
1382   Log (undef, $message);
1383   freeze() if @jobstep_todo;
1384   collate_output() if @jobstep_todo;
1385   cleanup() if $Job;
1386   save_meta() if $local_logfile;
1387   die;
1388 }
1389
1390
1391 sub cleanup
1392 {
1393   if ($Job->{'state'} eq 'Cancelled') {
1394     $Job->update_attributes('finished_at' => scalar gmtime);
1395   } else {
1396     $Job->update_attributes('state' => 'Failed');
1397   }
1398 }
1399
1400
1401 sub save_meta
1402 {
1403   my $justcheckpoint = shift; # false if this will be the last meta saved
1404   return if $justcheckpoint;  # checkpointing is not relevant post-Warehouse.pm
1405
1406   $local_logfile->flush;
1407   my $retry_count = put_retry_count();
1408   my $cmd = "arv-put --portable-data-hash --retries $retry_count " .
1409       "--filename ''\Q$keep_logfile\E " . quotemeta($local_logfile->filename);
1410   my $loglocator = `$cmd`;
1411   die "system $cmd failed: $?" if $?;
1412   chomp($loglocator);
1413
1414   $local_logfile = undef;   # the temp file is automatically deleted
1415   Log (undef, "log manifest is $loglocator");
1416   $Job->{'log'} = $loglocator;
1417   $Job->update_attributes('log', $loglocator);
1418 }
1419
1420
1421 sub freeze_if_want_freeze
1422 {
1423   if ($main::please_freeze)
1424   {
1425     release_allocation();
1426     if (@_)
1427     {
1428       # kill some srun procs before freeze+stop
1429       map { $proc{$_} = {} } @_;
1430       while (%proc)
1431       {
1432         killem (keys %proc);
1433         select (undef, undef, undef, 0.1);
1434         my $died;
1435         while (($died = waitpid (-1, WNOHANG)) > 0)
1436         {
1437           delete $proc{$died};
1438         }
1439       }
1440     }
1441     freeze();
1442     collate_output();
1443     cleanup();
1444     save_meta();
1445     exit 1;
1446   }
1447 }
1448
1449
1450 sub freeze
1451 {
1452   Log (undef, "Freeze not implemented");
1453   return;
1454 }
1455
1456
1457 sub thaw
1458 {
1459   croak ("Thaw not implemented");
1460 }
1461
1462
1463 sub freezequote
1464 {
1465   my $s = shift;
1466   $s =~ s/\\/\\\\/g;
1467   $s =~ s/\n/\\n/g;
1468   return $s;
1469 }
1470
1471
1472 sub freezeunquote
1473 {
1474   my $s = shift;
1475   $s =~ s{\\(.)}{$1 eq "n" ? "\n" : $1}ge;
1476   return $s;
1477 }
1478
1479
1480 sub srun
1481 {
1482   my $srunargs = shift;
1483   my $execargs = shift;
1484   my $opts = shift || {};
1485   my $stdin = shift;
1486   my $args = $have_slurm ? [@$srunargs, @$execargs] : $execargs;
1487   print STDERR (join (" ",
1488                       map { / / ? "'$_'" : $_ }
1489                       (@$args)),
1490                 "\n")
1491       if $ENV{CRUNCH_DEBUG};
1492
1493   if (defined $stdin) {
1494     my $child = open STDIN, "-|";
1495     defined $child or die "no fork: $!";
1496     if ($child == 0) {
1497       print $stdin or die $!;
1498       close STDOUT or die $!;
1499       exit 0;
1500     }
1501   }
1502
1503   return system (@$args) if $opts->{fork};
1504
1505   exec @$args;
1506   warn "ENV size is ".length(join(" ",%ENV));
1507   die "exec failed: $!: @$args";
1508 }
1509
1510
1511 sub ban_node_by_slot {
1512   # Don't start any new jobsteps on this node for 60 seconds
1513   my $slotid = shift;
1514   $slot[$slotid]->{node}->{hold_until} = 60 + scalar time;
1515   $slot[$slotid]->{node}->{hold_count}++;
1516   Log (undef, "backing off node " . $slot[$slotid]->{node}->{name} . " for 60 seconds");
1517 }
1518
1519 sub must_lock_now
1520 {
1521   my ($lockfile, $error_message) = @_;
1522   open L, ">", $lockfile or croak("$lockfile: $!");
1523   if (!flock L, LOCK_EX|LOCK_NB) {
1524     croak("Can't lock $lockfile: $error_message\n");
1525   }
1526 }
1527
1528 sub find_docker_image {
1529   # Given a Keep locator, check to see if it contains a Docker image.
1530   # If so, return its stream name and Docker hash.
1531   # If not, return undef for both values.
1532   my $locator = shift;
1533   my ($streamname, $filename);
1534   if (my $image = $arv->{collections}->{get}->execute(uuid => $locator)) {
1535     foreach my $line (split(/\n/, $image->{manifest_text})) {
1536       my @tokens = split(/\s+/, $line);
1537       next if (!@tokens);
1538       $streamname = shift(@tokens);
1539       foreach my $filedata (grep(/^\d+:\d+:/, @tokens)) {
1540         if (defined($filename)) {
1541           return (undef, undef);  # More than one file in the Collection.
1542         } else {
1543           $filename = (split(/:/, $filedata, 3))[2];
1544         }
1545       }
1546     }
1547   }
1548   if (defined($filename) and ($filename =~ /^([0-9A-Fa-f]{64})\.tar$/)) {
1549     return ($streamname, $1);
1550   } else {
1551     return (undef, undef);
1552   }
1553 }
1554
1555 sub put_retry_count {
1556   # Calculate a --retries argument for arv-put that will have it try
1557   # approximately as long as this Job has been running.
1558   my $stoptime = shift || time;
1559   my $starttime = $jobstep[0]->{starttime};
1560   my $timediff = defined($starttime) ? ($stoptime - $starttime) : 1;
1561   my $retries = 0;
1562   while ($timediff >= 2) {
1563     $retries++;
1564     $timediff /= 2;
1565   }
1566   return ($retries > 3) ? $retries : 3;
1567 }
1568
1569 __DATA__
1570 #!/usr/bin/perl
1571
1572 # checkout-and-build
1573
1574 use Fcntl ':flock';
1575 use File::Path qw( make_path );
1576
1577 my $destdir = $ENV{"CRUNCH_SRC"};
1578 my $commit = $ENV{"CRUNCH_SRC_COMMIT"};
1579 my $repo = $ENV{"CRUNCH_SRC_URL"};
1580 my $task_work = $ENV{"TASK_WORK"};
1581
1582 for my $dir ($destdir, $task_work) {
1583     if ($dir) {
1584         make_path $dir;
1585         -e $dir or die "Failed to create temporary directory ($dir): $!";
1586     }
1587 }
1588
1589 open L, ">", "$destdir.lock" or die "$destdir.lock: $!";
1590 flock L, LOCK_EX;
1591 if (readlink ("$destdir.commit") eq $commit && -d $destdir) {
1592     if (@ARGV) {
1593         exec(@ARGV);
1594         die "Cannot exec `@ARGV`: $!";
1595     } else {
1596         exit 0;
1597     }
1598 }
1599
1600 unlink "$destdir.commit";
1601 open STDOUT, ">", "$destdir.log";
1602 open STDERR, ">&STDOUT";
1603
1604 mkdir $destdir;
1605 my @git_archive_data = <DATA>;
1606 if (@git_archive_data) {
1607   open TARX, "|-", "tar", "-C", $destdir, "-xf", "-";
1608   print TARX @git_archive_data;
1609   if(!close(TARX)) {
1610     die "'tar -C $destdir -xf -' exited $?: $!";
1611   }
1612 }
1613
1614 my $pwd;
1615 chomp ($pwd = `pwd`);
1616 my $install_dir = $ENV{"CRUNCH_INSTALL"} || "$pwd/opt";
1617 mkdir $install_dir;
1618
1619 for my $src_path ("$destdir/arvados/sdk/python") {
1620   if (-d $src_path) {
1621     shell_or_die ("virtualenv", $install_dir);
1622     shell_or_die ("cd $src_path && ./build.sh && $install_dir/bin/python setup.py install");
1623   }
1624 }
1625
1626 if (-e "$destdir/crunch_scripts/install") {
1627     shell_or_die ("$destdir/crunch_scripts/install", $install_dir);
1628 } elsif (!-e "./install.sh" && -e "./tests/autotests.sh") {
1629     # Old version
1630     shell_or_die ("./tests/autotests.sh", $install_dir);
1631 } elsif (-e "./install.sh") {
1632     shell_or_die ("./install.sh", $install_dir);
1633 }
1634
1635 if ($commit) {
1636     unlink "$destdir.commit.new";
1637     symlink ($commit, "$destdir.commit.new") or die "$destdir.commit.new: $!";
1638     rename ("$destdir.commit.new", "$destdir.commit") or die "$destdir.commit: $!";
1639 }
1640
1641 close L;
1642
1643 if (@ARGV) {
1644     exec(@ARGV);
1645     die "Cannot exec `@ARGV`: $!";
1646 } else {
1647     exit 0;
1648 }
1649
1650 sub shell_or_die
1651 {
1652   if ($ENV{"DEBUG"}) {
1653     print STDERR "@_\n";
1654   }
1655   system (@_) == 0
1656       or die "@_ failed: $! exit 0x".sprintf("%x",$?);
1657 }
1658
1659 __DATA__