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