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