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 my $final_state;
 881 if ($collated_output && $main::success) {
 882   $final_state = 'Complete';
 883 } else {
 884   $final_state = 'Failed';
 885 }
 886 $Job->update_attributes('state' => $final_state)
 887
 888 exit (($final_state eq 'Complete') ? 0 : 1);
 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     # Put this task back on the todo queue
 976     push @jobstep_todo, $jobstepid;
 977     $Job->{'tasks_summary'}->{'failed'}++;
 978   }
 979   else
 980   {
 981     ++$thisround_succeeded;
 982     $slot[$proc{$pid}->{slot}]->{node}->{losing_streak} = 0;
 983     $slot[$proc{$pid}->{slot}]->{node}->{hold_until} = 0;
 984     push @jobstep_done, $jobstepid;
 985     Log ($jobstepid, "success in $elapsed seconds");
 986   }
 987   $Jobstep->{exitcode} = $childstatus;
 988   $Jobstep->{finishtime} = time;
 989   $Jobstep->{'arvados_task'}->{finished_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{finishtime});
 990   $Jobstep->{'arvados_task'}->save;
 991   process_stderr ($jobstepid, $task_success);
 992   Log ($jobstepid, "output " . $Jobstep->{'arvados_task'}->{output});
 993
 994   close $reader{$jobstepid};
 995   delete $reader{$jobstepid};
 996   delete $slot[$proc{$pid}->{slot}]->{pid};
 997   push @freeslot, $proc{$pid}->{slot};
 998   delete $proc{$pid};
 999
1000   if ($task_success) {
1001     # Load new tasks
1002     my $newtask_list = [];
1003     my $newtask_results;
1004     do {
1005       $newtask_results = $arv->{'job_tasks'}->{'list'}->execute(
1006         'where' => {
1007           'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid}
1008         },
1009         'order' => 'qsequence',
1010         'offset' => scalar(@$newtask_list),
1011       );
1012       push(@$newtask_list, @{$newtask_results->{items}});
1013     } while (@{$newtask_results->{items}});
1014     foreach my $arvados_task (@$newtask_list) {
1015       my $jobstep = {
1016         'level' => $arvados_task->{'sequence'},
1017         'failures' => 0,
1018         'arvados_task' => $arvados_task
1019       };
1020       push @jobstep, $jobstep;
1021       push @jobstep_todo, $#jobstep;
1022     }
1023   }
1024
1025   $progress_is_dirty = 1;
1026   1;
1027 }
1028
1029 sub check_refresh_wanted
1030 {
1031   my @stat = stat $ENV{"CRUNCH_REFRESH_TRIGGER"};
1032   if (@stat && $stat[9] > $latest_refresh) {
1033     $latest_refresh = scalar time;
1034     if ($job_has_uuid) {
1035       my $Job2 = $arv->{'jobs'}->{'get'}->execute('uuid' => $jobspec);
1036       for my $attr ('cancelled_at',
1037                     'cancelled_by_user_uuid',
1038                     'cancelled_by_client_uuid',
1039                     'state') {
1040         $Job->{$attr} = $Job2->{$attr};
1041       }
1042       if ($Job->{'state'} ne "Running") {
1043         if ($Job->{'state'} eq "Cancelled") {
1044           Log (undef, "Job cancelled at " . $Job->{'cancelled_at'} . " by user " . $Job->{'cancelled_by_user_uuid'});
1045         } else {
1046           Log (undef, "Job state unexpectedly changed to " . $Job->{'state'});
1047         }
1048         $main::success = 0;
1049         $main::please_freeze = 1;
1050       }
1051     }
1052   }
1053 }
1054
1055 sub check_squeue
1056 {
1057   # return if the kill list was checked <4 seconds ago
1058   if (defined $squeue_kill_checked && $squeue_kill_checked > time - 4)
1059   {
1060     return;
1061   }
1062   $squeue_kill_checked = time;
1063
1064   # use killem() on procs whose killtime is reached
1065   for (keys %proc)
1066   {
1067     if (exists $proc{$_}->{killtime}
1068         && $proc{$_}->{killtime} <= time)
1069     {
1070       killem ($_);
1071     }
1072   }
1073
1074   # return if the squeue was checked <60 seconds ago
1075   if (defined $squeue_checked && $squeue_checked > time - 60)
1076   {
1077     return;
1078   }
1079   $squeue_checked = time;
1080
1081   if (!$have_slurm)
1082   {
1083     # here is an opportunity to check for mysterious problems with local procs
1084     return;
1085   }
1086
1087   # get a list of steps still running
1088   my @squeue = `squeue -s -h -o '%i %j' && echo ok`;
1089   chop @squeue;
1090   if ($squeue[-1] ne "ok")
1091   {
1092     return;
1093   }
1094   pop @squeue;
1095
1096   # which of my jobsteps are running, according to squeue?
1097   my %ok;
1098   foreach (@squeue)
1099   {
1100     if (/^(\d+)\.(\d+) (\S+)/)
1101     {
1102       if ($1 eq $ENV{SLURM_JOBID})
1103       {
1104         $ok{$3} = 1;
1105       }
1106     }
1107   }
1108
1109   # which of my active child procs (>60s old) were not mentioned by squeue?
1110   foreach (keys %proc)
1111   {
1112     if ($proc{$_}->{time} < time - 60
1113         && !exists $ok{$proc{$_}->{jobstepname}}
1114         && !exists $proc{$_}->{killtime})
1115     {
1116       # kill this proc if it hasn't exited in 30 seconds
1117       $proc{$_}->{killtime} = time + 30;
1118     }
1119   }
1120 }
1121
1122
1123 sub release_allocation
1124 {
1125   if ($have_slurm)
1126   {
1127     Log (undef, "release job allocation");
1128     system "scancel $ENV{SLURM_JOBID}";
1129   }
1130 }
1131
1132
1133 sub readfrompipes
1134 {
1135   my $gotsome = 0;
1136   foreach my $job (keys %reader)
1137   {
1138     my $buf;
1139     while (0 < sysread ($reader{$job}, $buf, 8192))
1140     {
1141       print STDERR $buf if $ENV{CRUNCH_DEBUG};
1142       $jobstep[$job]->{stderr} .= $buf;
1143       preprocess_stderr ($job);
1144       if (length ($jobstep[$job]->{stderr}) > 16384)
1145       {
1146         substr ($jobstep[$job]->{stderr}, 0, 8192) = "";
1147       }
1148       $gotsome = 1;
1149     }
1150   }
1151   return $gotsome;
1152 }
1153
1154
1155 sub preprocess_stderr
1156 {
1157   my $job = shift;
1158
1159   while ($jobstep[$job]->{stderr} =~ /^(.*?)\n/) {
1160     my $line = $1;
1161     substr $jobstep[$job]->{stderr}, 0, 1+length($line), "";
1162     Log ($job, "stderr $line");
1163     if ($line =~ /srun: error: (SLURM job $ENV{SLURM_JOB_ID} has expired|Unable to confirm allocation for job $ENV{SLURM_JOB_ID})/) {
1164       # whoa.
1165       $main::please_freeze = 1;
1166     }
1167     elsif ($line =~ /srun: error: (Node failure on|Unable to create job step) /) {
1168       $jobstep[$job]->{node_fail} = 1;
1169       ban_node_by_slot($jobstep[$job]->{slotindex});
1170     }
1171   }
1172 }
1173
1174
1175 sub process_stderr
1176 {
1177   my $job = shift;
1178   my $task_success = shift;
1179   preprocess_stderr ($job);
1180
1181   map {
1182     Log ($job, "stderr $_");
1183   } split ("\n", $jobstep[$job]->{stderr});
1184 }
1185
1186 sub fetch_block
1187 {
1188   my $hash = shift;
1189   my ($keep, $child_out, $output_block);
1190
1191   my $cmd = "arv-get \Q$hash\E";
1192   open($keep, '-|', $cmd) or die "fetch_block: $cmd: $!";
1193   $output_block = '';
1194   while (1) {
1195     my $buf;
1196     my $bytes = sysread($keep, $buf, 1024 * 1024);
1197     if (!defined $bytes) {
1198       die "reading from arv-get: $!";
1199     } elsif ($bytes == 0) {
1200       # sysread returns 0 at the end of the pipe.
1201       last;
1202     } else {
1203       # some bytes were read into buf.
1204       $output_block .= $buf;
1205     }
1206   }
1207   close $keep;
1208   return $output_block;
1209 }
1210
1211 sub collate_output
1212 {
1213   Log (undef, "collate");
1214
1215   my ($child_out, $child_in);
1216   my $pid = open2($child_out, $child_in, 'arv-put', '--raw',
1217                   '--retries', put_retry_count());
1218   my $joboutput;
1219   for (@jobstep)
1220   {
1221     next if (!exists $_->{'arvados_task'}->{'output'} ||
1222              !$_->{'arvados_task'}->{'success'});
1223     my $output = $_->{'arvados_task'}->{output};
1224     if ($output !~ /^[0-9a-f]{32}(\+\S+)*$/)
1225     {
1226       $output_in_keep ||= $output =~ / [0-9a-f]{32}\S*\+K/;
1227       print $child_in $output;
1228     }
1229     elsif (@jobstep == 1)
1230     {
1231       $joboutput = $output;
1232       last;
1233     }
1234     elsif (defined (my $outblock = fetch_block ($output)))
1235     {
1236       $output_in_keep ||= $outblock =~ / [0-9a-f]{32}\S*\+K/;
1237       print $child_in $outblock;
1238     }
1239     else
1240     {
1241       Log (undef, "XXX fetch_block($output) failed XXX");
1242       $main::success = 0;
1243     }
1244   }
1245   $child_in->close;
1246
1247   if (!defined $joboutput) {
1248     my $s = IO::Select->new($child_out);
1249     if ($s->can_read(120)) {
1250       sysread($child_out, $joboutput, 64 * 1024 * 1024);
1251       chomp($joboutput);
1252     } else {
1253       Log (undef, "timed out reading from 'arv-put'");
1254     }
1255   }
1256   waitpid($pid, 0);
1257
1258   return $joboutput;
1259 }
1260
1261
1262 sub killem
1263 {
1264   foreach (@_)
1265   {
1266     my $sig = 2;                # SIGINT first
1267     if (exists $proc{$_}->{"sent_$sig"} &&
1268         time - $proc{$_}->{"sent_$sig"} > 4)
1269     {
1270       $sig = 15;                # SIGTERM if SIGINT doesn't work
1271     }
1272     if (exists $proc{$_}->{"sent_$sig"} &&
1273         time - $proc{$_}->{"sent_$sig"} > 4)
1274     {
1275       $sig = 9;                 # SIGKILL if SIGTERM doesn't work
1276     }
1277     if (!exists $proc{$_}->{"sent_$sig"})
1278     {
1279       Log ($proc{$_}->{jobstep}, "sending 2x signal $sig to pid $_");
1280       kill $sig, $_;
1281       select (undef, undef, undef, 0.1);
1282       if ($sig == 2)
1283       {
1284         kill $sig, $_;     # srun wants two SIGINT to really interrupt
1285       }
1286       $proc{$_}->{"sent_$sig"} = time;
1287       $proc{$_}->{"killedafter"} = time - $proc{$_}->{"time"};
1288     }
1289   }
1290 }
1291
1292
1293 sub fhbits
1294 {
1295   my($bits);
1296   for (@_) {
1297     vec($bits,fileno($_),1) = 1;
1298   }
1299   $bits;
1300 }
1301
1302
1303 sub Log                         # ($jobstep_id, $logmessage)
1304 {
1305   if ($_[1] =~ /\n/) {
1306     for my $line (split (/\n/, $_[1])) {
1307       Log ($_[0], $line);
1308     }
1309     return;
1310   }
1311   my $fh = select STDERR; $|=1; select $fh;
1312   my $message = sprintf ("%s %d %s %s", $job_id, $$, @_);
1313   $message =~ s{([^ -\176])}{"\\" . sprintf ("%03o", ord($1))}ge;
1314   $message .= "\n";
1315   my $datetime;
1316   if ($local_logfile || -t STDERR) {
1317     my @gmtime = gmtime;
1318     $datetime = sprintf ("%04d-%02d-%02d_%02d:%02d:%02d",
1319                          $gmtime[5]+1900, $gmtime[4]+1, @gmtime[3,2,1,0]);
1320   }
1321   print STDERR ((-t STDERR) ? ($datetime." ".$message) : $message);
1322
1323   if ($local_logfile) {
1324     print $local_logfile $datetime . " " . $message;
1325   }
1326 }
1327
1328
1329 sub croak
1330 {
1331   my ($package, $file, $line) = caller;
1332   my $message = "@_ at $file line $line\n";
1333   Log (undef, $message);
1334   freeze() if @jobstep_todo;
1335   collate_output() if @jobstep_todo;
1336   cleanup();
1337   save_meta() if $local_logfile;
1338   die;
1339 }
1340
1341
1342 sub cleanup
1343 {
1344   return if !$job_has_uuid;
1345   if ($Job->{'state'} eq 'Cancelled') {
1346     $Job->update_attributes('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 1;
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__