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 {
 839   eval {
 840     open(my $orig_manifest, '-|', 'arv-get', $collated_output)
 841         or die "failed to get collated manifest: $!";
 842     # Read the original manifest, and strip permission hints from it,
 843     # so we can put the result in a Collection.
 844     my @stripped_manifest_lines = ();
 845     my $orig_manifest_text = '';
 846     while (my $manifest_line = <$orig_manifest>) {
 847       $orig_manifest_text .= $manifest_line;
 848       my @words = split(/ /, $manifest_line, -1);
 849       foreach my $ii (0..$#words) {
 850         if ($words[$ii] =~ /^[0-9a-f]{32}\+/) {
 851           $words[$ii] =~ s/\+A[0-9a-f]{40}@[0-9a-f]{8}\b//;
 852         }
 853       }
 854       push(@stripped_manifest_lines, join(" ", @words));
 855     }
 856     my $stripped_manifest_text = join("", @stripped_manifest_lines);
 857     my $output = $arv->{'collections'}->{'create'}->execute('collection' => {
 858       'uuid' => md5_hex($stripped_manifest_text),
 859       'manifest_text' => $orig_manifest_text,
 860     });
 861     $Job->update_attributes('output' => $output->{uuid});
 862     if ($Job->{'output_is_persistent'}) {
 863       $arv->{'links'}->{'create'}->execute('link' => {
 864         'tail_kind' => 'arvados#user',
 865         'tail_uuid' => $User->{'uuid'},
 866         'head_kind' => 'arvados#collection',
 867         'head_uuid' => $Job->{'output'},
 868         'link_class' => 'resources',
 869         'name' => 'wants',
 870       });
 871     }
 872   };
 873   if ($@) {
 874     Log (undef, "Failed to register output manifest: $@");
 875   }
 876 }
 877
 878 Log (undef, "finish");
 879
 880 save_meta();
 881 exit 0;
 882
 883
 884
 885 sub update_progress_stats
 886 {
 887   $progress_stats_updated = time;
 888   return if !$progress_is_dirty;
 889   my ($todo, $done, $running) = (scalar @jobstep_todo,
 890                                  scalar @jobstep_done,
 891                                  scalar @slot - scalar @freeslot - scalar @holdslot);
 892   $Job->{'tasks_summary'} ||= {};
 893   $Job->{'tasks_summary'}->{'todo'} = $todo;
 894   $Job->{'tasks_summary'}->{'done'} = $done;
 895   $Job->{'tasks_summary'}->{'running'} = $running;
 896   if ($job_has_uuid) {
 897     $Job->update_attributes('tasks_summary' => $Job->{'tasks_summary'});
 898   }
 899   Log (undef, "status: $done done, $running running, $todo todo");
 900   $progress_is_dirty = 0;
 901 }
 902
 903
 904
 905 sub reapchildren
 906 {
 907   my $pid = waitpid (-1, WNOHANG);
 908   return 0 if $pid <= 0;
 909
 910   my $whatslot = ($slot[$proc{$pid}->{slot}]->{node}->{name}
 911                   . "."
 912                   . $slot[$proc{$pid}->{slot}]->{cpu});
 913   my $jobstepid = $proc{$pid}->{jobstep};
 914   my $elapsed = time - $proc{$pid}->{time};
 915   my $Jobstep = $jobstep[$jobstepid];
 916
 917   my $childstatus = $?;
 918   my $exitvalue = $childstatus >> 8;
 919   my $exitinfo = sprintf("exit %d signal %d%s",
 920                          $exitvalue,
 921                          $childstatus & 127,
 922                          ($childstatus & 128 ? ' core dump' : ''));
 923   $Jobstep->{'arvados_task'}->reload;
 924   my $task_success = $Jobstep->{'arvados_task'}->{success};
 925
 926   Log ($jobstepid, "child $pid on $whatslot $exitinfo success=$task_success");
 927
 928   if (!defined $task_success) {
 929     # task did not indicate one way or the other --> fail
 930     $Jobstep->{'arvados_task'}->{success} = 0;
 931     $Jobstep->{'arvados_task'}->save;
 932     $task_success = 0;
 933   }
 934
 935   if (!$task_success)
 936   {
 937     my $temporary_fail;
 938     $temporary_fail ||= $Jobstep->{node_fail};
 939     $temporary_fail ||= ($exitvalue == 111);
 940
 941     ++$thisround_failed;
 942     ++$thisround_failed_multiple if $Jobstep->{'failures'} >= 1;
 943
 944     # Check for signs of a failed or misconfigured node
 945     if (++$slot[$proc{$pid}->{slot}]->{node}->{losing_streak} >=
 946         2+$slot[$proc{$pid}->{slot}]->{node}->{ncpus}) {
 947       # Don't count this against jobstep failure thresholds if this
 948       # node is already suspected faulty and srun exited quickly
 949       if ($slot[$proc{$pid}->{slot}]->{node}->{hold_until} &&
 950           $elapsed < 5) {
 951         Log ($jobstepid, "blaming failure on suspect node " .
 952              $slot[$proc{$pid}->{slot}]->{node}->{name});
 953         $temporary_fail ||= 1;
 954       }
 955       ban_node_by_slot($proc{$pid}->{slot});
 956     }
 957
 958     Log ($jobstepid, sprintf('failure (#%d, %s) after %d seconds',
 959                              ++$Jobstep->{'failures'},
 960                              $temporary_fail ? 'temporary ' : 'permanent',
 961                              $elapsed));
 962
 963     if (!$temporary_fail || $Jobstep->{'failures'} >= 3) {
 964       # Give up on this task, and the whole job
 965       $main::success = 0;
 966       $main::please_freeze = 1;
 967     }
 968     else {
 969       # Put this task back on the todo queue
 970       push @jobstep_todo, $jobstepid;
 971     }
 972     $Job->{'tasks_summary'}->{'failed'}++;
 973   }
 974   else
 975   {
 976     ++$thisround_succeeded;
 977     $slot[$proc{$pid}->{slot}]->{node}->{losing_streak} = 0;
 978     $slot[$proc{$pid}->{slot}]->{node}->{hold_until} = 0;
 979     push @jobstep_done, $jobstepid;
 980     Log ($jobstepid, "success in $elapsed seconds");
 981   }
 982   $Jobstep->{exitcode} = $childstatus;
 983   $Jobstep->{finishtime} = time;
 984   process_stderr ($jobstepid, $task_success);
 985   Log ($jobstepid, "output " . $Jobstep->{'arvados_task'}->{output});
 986
 987   close $reader{$jobstepid};
 988   delete $reader{$jobstepid};
 989   delete $slot[$proc{$pid}->{slot}]->{pid};
 990   push @freeslot, $proc{$pid}->{slot};
 991   delete $proc{$pid};
 992
 993   if ($task_success) {
 994     # Load new tasks
 995     my $newtask_list = [];
 996     my $newtask_results;
 997     do {
 998       $newtask_results = $arv->{'job_tasks'}->{'list'}->execute(
 999         'where' => {
1000           'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid}
1001         },
1002         'order' => 'qsequence',
1003         'offset' => scalar(@$newtask_list),
1004       );
1005       push(@$newtask_list, @{$newtask_results->{items}});
1006     } while (@{$newtask_results->{items}});
1007     foreach my $arvados_task (@$newtask_list) {
1008       my $jobstep = {
1009         'level' => $arvados_task->{'sequence'},
1010         'failures' => 0,
1011         'arvados_task' => $arvados_task
1012       };
1013       push @jobstep, $jobstep;
1014       push @jobstep_todo, $#jobstep;
1015     }
1016   }
1017
1018   $progress_is_dirty = 1;
1019   1;
1020 }
1021
1022 sub check_refresh_wanted
1023 {
1024   my @stat = stat $ENV{"CRUNCH_REFRESH_TRIGGER"};
1025   if (@stat && $stat[9] > $latest_refresh) {
1026     $latest_refresh = scalar time;
1027     if ($job_has_uuid) {
1028       my $Job2 = $arv->{'jobs'}->{'get'}->execute('uuid' => $jobspec);
1029       for my $attr ('cancelled_at',
1030                     'cancelled_by_user_uuid',
1031                     'cancelled_by_client_uuid') {
1032         $Job->{$attr} = $Job2->{$attr};
1033       }
1034       if ($Job->{'cancelled_at'}) {
1035         Log (undef, "Job cancelled at " . $Job->{cancelled_at} .
1036              " by user " . $Job->{cancelled_by_user_uuid});
1037         $main::success = 0;
1038         $main::please_freeze = 1;
1039       }
1040     }
1041   }
1042 }
1043
1044 sub check_squeue
1045 {
1046   # return if the kill list was checked <4 seconds ago
1047   if (defined $squeue_kill_checked && $squeue_kill_checked > time - 4)
1048   {
1049     return;
1050   }
1051   $squeue_kill_checked = time;
1052
1053   # use killem() on procs whose killtime is reached
1054   for (keys %proc)
1055   {
1056     if (exists $proc{$_}->{killtime}
1057         && $proc{$_}->{killtime} <= time)
1058     {
1059       killem ($_);
1060     }
1061   }
1062
1063   # return if the squeue was checked <60 seconds ago
1064   if (defined $squeue_checked && $squeue_checked > time - 60)
1065   {
1066     return;
1067   }
1068   $squeue_checked = time;
1069
1070   if (!$have_slurm)
1071   {
1072     # here is an opportunity to check for mysterious problems with local procs
1073     return;
1074   }
1075
1076   # get a list of steps still running
1077   my @squeue = `squeue -s -h -o '%i %j' && echo ok`;
1078   chop @squeue;
1079   if ($squeue[-1] ne "ok")
1080   {
1081     return;
1082   }
1083   pop @squeue;
1084
1085   # which of my jobsteps are running, according to squeue?
1086   my %ok;
1087   foreach (@squeue)
1088   {
1089     if (/^(\d+)\.(\d+) (\S+)/)
1090     {
1091       if ($1 eq $ENV{SLURM_JOBID})
1092       {
1093         $ok{$3} = 1;
1094       }
1095     }
1096   }
1097
1098   # which of my active child procs (>60s old) were not mentioned by squeue?
1099   foreach (keys %proc)
1100   {
1101     if ($proc{$_}->{time} < time - 60
1102         && !exists $ok{$proc{$_}->{jobstepname}}
1103         && !exists $proc{$_}->{killtime})
1104     {
1105       # kill this proc if it hasn't exited in 30 seconds
1106       $proc{$_}->{killtime} = time + 30;
1107     }
1108   }
1109 }
1110
1111
1112 sub release_allocation
1113 {
1114   if ($have_slurm)
1115   {
1116     Log (undef, "release job allocation");
1117     system "scancel $ENV{SLURM_JOBID}";
1118   }
1119 }
1120
1121
1122 sub readfrompipes
1123 {
1124   my $gotsome = 0;
1125   foreach my $job (keys %reader)
1126   {
1127     my $buf;
1128     while (0 < sysread ($reader{$job}, $buf, 8192))
1129     {
1130       print STDERR $buf if $ENV{CRUNCH_DEBUG};
1131       $jobstep[$job]->{stderr} .= $buf;
1132       preprocess_stderr ($job);
1133       if (length ($jobstep[$job]->{stderr}) > 16384)
1134       {
1135         substr ($jobstep[$job]->{stderr}, 0, 8192) = "";
1136       }
1137       $gotsome = 1;
1138     }
1139   }
1140   return $gotsome;
1141 }
1142
1143
1144 sub preprocess_stderr
1145 {
1146   my $job = shift;
1147
1148   while ($jobstep[$job]->{stderr} =~ /^(.*?)\n/) {
1149     my $line = $1;
1150     substr $jobstep[$job]->{stderr}, 0, 1+length($line), "";
1151     Log ($job, "stderr $line");
1152     if ($line =~ /srun: error: (SLURM job $ENV{SLURM_JOB_ID} has expired|Unable to confirm allocation for job $ENV{SLURM_JOB_ID})/) {
1153       # whoa.
1154       $main::please_freeze = 1;
1155     }
1156     elsif ($line =~ /srun: error: (Node failure on|Unable to create job step) /) {
1157       $jobstep[$job]->{node_fail} = 1;
1158       ban_node_by_slot($jobstep[$job]->{slotindex});
1159     }
1160   }
1161 }
1162
1163
1164 sub process_stderr
1165 {
1166   my $job = shift;
1167   my $task_success = shift;
1168   preprocess_stderr ($job);
1169
1170   map {
1171     Log ($job, "stderr $_");
1172   } split ("\n", $jobstep[$job]->{stderr});
1173 }
1174
1175 sub fetch_block
1176 {
1177   my $hash = shift;
1178   my ($keep, $child_out, $output_block);
1179
1180   my $cmd = "arv-get \Q$hash\E";
1181   open($keep, '-|', $cmd) or die "fetch_block: $cmd: $!";
1182   sysread($keep, $output_block, 64 * 1024 * 1024);
1183   close $keep;
1184   return $output_block;
1185 }
1186
1187 sub collate_output
1188 {
1189   Log (undef, "collate");
1190
1191   my ($child_out, $child_in);
1192   my $pid = open2($child_out, $child_in, 'arv-put', '--raw');
1193   my $joboutput;
1194   for (@jobstep)
1195   {
1196     next if (!exists $_->{'arvados_task'}->{output} ||
1197              !$_->{'arvados_task'}->{'success'} ||
1198              $_->{'exitcode'} != 0);
1199     my $output = $_->{'arvados_task'}->{output};
1200     if ($output !~ /^[0-9a-f]{32}(\+\S+)*$/)
1201     {
1202       $output_in_keep ||= $output =~ / [0-9a-f]{32}\S*\+K/;
1203       print $child_in $output;
1204     }
1205     elsif (@jobstep == 1)
1206     {
1207       $joboutput = $output;
1208       last;
1209     }
1210     elsif (defined (my $outblock = fetch_block ($output)))
1211     {
1212       $output_in_keep ||= $outblock =~ / [0-9a-f]{32}\S*\+K/;
1213       print $child_in $outblock;
1214     }
1215     else
1216     {
1217       Log (undef, "XXX fetch_block($output) failed XXX");
1218       $main::success = 0;
1219     }
1220   }
1221   $child_in->close;
1222
1223   if (!defined $joboutput) {
1224     my $s = IO::Select->new($child_out);
1225     if ($s->can_read(120)) {
1226       sysread($child_out, $joboutput, 64 * 1024 * 1024);
1227       chomp($joboutput);
1228     } else {
1229       Log (undef, "timed out reading from 'arv-put'");
1230     }
1231   }
1232   waitpid($pid, 0);
1233
1234   if ($joboutput)
1235   {
1236     Log (undef, "output $joboutput");
1237     $Job->update_attributes('output' => $joboutput) if $job_has_uuid;
1238   }
1239   else
1240   {
1241     Log (undef, "output undef");
1242   }
1243   return $joboutput;
1244 }
1245
1246
1247 sub killem
1248 {
1249   foreach (@_)
1250   {
1251     my $sig = 2;                # SIGINT first
1252     if (exists $proc{$_}->{"sent_$sig"} &&
1253         time - $proc{$_}->{"sent_$sig"} > 4)
1254     {
1255       $sig = 15;                # SIGTERM if SIGINT doesn't work
1256     }
1257     if (exists $proc{$_}->{"sent_$sig"} &&
1258         time - $proc{$_}->{"sent_$sig"} > 4)
1259     {
1260       $sig = 9;                 # SIGKILL if SIGTERM doesn't work
1261     }
1262     if (!exists $proc{$_}->{"sent_$sig"})
1263     {
1264       Log ($proc{$_}->{jobstep}, "sending 2x signal $sig to pid $_");
1265       kill $sig, $_;
1266       select (undef, undef, undef, 0.1);
1267       if ($sig == 2)
1268       {
1269         kill $sig, $_;     # srun wants two SIGINT to really interrupt
1270       }
1271       $proc{$_}->{"sent_$sig"} = time;
1272       $proc{$_}->{"killedafter"} = time - $proc{$_}->{"time"};
1273     }
1274   }
1275 }
1276
1277
1278 sub fhbits
1279 {
1280   my($bits);
1281   for (@_) {
1282     vec($bits,fileno($_),1) = 1;
1283   }
1284   $bits;
1285 }
1286
1287
1288 sub Log                         # ($jobstep_id, $logmessage)
1289 {
1290   if ($_[1] =~ /\n/) {
1291     for my $line (split (/\n/, $_[1])) {
1292       Log ($_[0], $line);
1293     }
1294     return;
1295   }
1296   my $fh = select STDERR; $|=1; select $fh;
1297   my $message = sprintf ("%s %d %s %s", $job_id, $$, @_);
1298   $message =~ s{([^ -\176])}{"\\" . sprintf ("%03o", ord($1))}ge;
1299   $message .= "\n";
1300   my $datetime;
1301   if ($local_logfile || -t STDERR) {
1302     my @gmtime = gmtime;
1303     $datetime = sprintf ("%04d-%02d-%02d_%02d:%02d:%02d",
1304                          $gmtime[5]+1900, $gmtime[4]+1, @gmtime[3,2,1,0]);
1305   }
1306   print STDERR ((-t STDERR) ? ($datetime." ".$message) : $message);
1307
1308   if ($local_logfile) {
1309     print $local_logfile $datetime . " " . $message;
1310   }
1311 }
1312
1313
1314 sub croak
1315 {
1316   my ($package, $file, $line) = caller;
1317   my $message = "@_ at $file line $line\n";
1318   Log (undef, $message);
1319   freeze() if @jobstep_todo;
1320   collate_output() if @jobstep_todo;
1321   cleanup();
1322   save_meta() if $local_logfile;
1323   die;
1324 }
1325
1326
1327 sub cleanup
1328 {
1329   return if !$job_has_uuid;
1330   $Job->update_attributes('running' => 0,
1331                           'success' => 0,
1332                           'finished_at' => scalar gmtime);
1333 }
1334
1335
1336 sub save_meta
1337 {
1338   my $justcheckpoint = shift; # false if this will be the last meta saved
1339   return if $justcheckpoint;  # checkpointing is not relevant post-Warehouse.pm
1340
1341   $local_logfile->flush;
1342   my $cmd = "arv-put --filename ''\Q$keep_logfile\E "
1343       . quotemeta($local_logfile->filename);
1344   my $loglocator = `$cmd`;
1345   die "system $cmd failed: $?" if $?;
1346   chomp($loglocator);
1347
1348   $local_logfile = undef;   # the temp file is automatically deleted
1349   Log (undef, "log manifest is $loglocator");
1350   $Job->{'log'} = $loglocator;
1351   $Job->update_attributes('log', $loglocator) if $job_has_uuid;
1352 }
1353
1354
1355 sub freeze_if_want_freeze
1356 {
1357   if ($main::please_freeze)
1358   {
1359     release_allocation();
1360     if (@_)
1361     {
1362       # kill some srun procs before freeze+stop
1363       map { $proc{$_} = {} } @_;
1364       while (%proc)
1365       {
1366         killem (keys %proc);
1367         select (undef, undef, undef, 0.1);
1368         my $died;
1369         while (($died = waitpid (-1, WNOHANG)) > 0)
1370         {
1371           delete $proc{$died};
1372         }
1373       }
1374     }
1375     freeze();
1376     collate_output();
1377     cleanup();
1378     save_meta();
1379     exit 0;
1380   }
1381 }
1382
1383
1384 sub freeze
1385 {
1386   Log (undef, "Freeze not implemented");
1387   return;
1388 }
1389
1390
1391 sub thaw
1392 {
1393   croak ("Thaw not implemented");
1394 }
1395
1396
1397 sub freezequote
1398 {
1399   my $s = shift;
1400   $s =~ s/\\/\\\\/g;
1401   $s =~ s/\n/\\n/g;
1402   return $s;
1403 }
1404
1405
1406 sub freezeunquote
1407 {
1408   my $s = shift;
1409   $s =~ s{\\(.)}{$1 eq "n" ? "\n" : $1}ge;
1410   return $s;
1411 }
1412
1413
1414 sub srun
1415 {
1416   my $srunargs = shift;
1417   my $execargs = shift;
1418   my $opts = shift || {};
1419   my $stdin = shift;
1420   my $args = $have_slurm ? [@$srunargs, @$execargs] : $execargs;
1421   print STDERR (join (" ",
1422                       map { / / ? "'$_'" : $_ }
1423                       (@$args)),
1424                 "\n")
1425       if $ENV{CRUNCH_DEBUG};
1426
1427   if (defined $stdin) {
1428     my $child = open STDIN, "-|";
1429     defined $child or die "no fork: $!";
1430     if ($child == 0) {
1431       print $stdin or die $!;
1432       close STDOUT or die $!;
1433       exit 0;
1434     }
1435   }
1436
1437   return system (@$args) if $opts->{fork};
1438
1439   exec @$args;
1440   warn "ENV size is ".length(join(" ",%ENV));
1441   die "exec failed: $!: @$args";
1442 }
1443
1444
1445 sub ban_node_by_slot {
1446   # Don't start any new jobsteps on this node for 60 seconds
1447   my $slotid = shift;
1448   $slot[$slotid]->{node}->{hold_until} = 60 + scalar time;
1449   $slot[$slotid]->{node}->{hold_count}++;
1450   Log (undef, "backing off node " . $slot[$slotid]->{node}->{name} . " for 60 seconds");
1451 }
1452
1453 sub must_lock_now
1454 {
1455   my ($lockfile, $error_message) = @_;
1456   open L, ">", $lockfile or croak("$lockfile: $!");
1457   if (!flock L, LOCK_EX|LOCK_NB) {
1458     croak("Can't lock $lockfile: $error_message\n");
1459   }
1460 }
1461
1462 sub find_docker_hash {
1463   # Given a Keep locator, search for a matching link to find the Docker hash
1464   # of the stored image.
1465   my $locator = shift;
1466   my $links_result = $arv->{links}->{list}->execute(
1467     filters => [["head_uuid", "=", $locator],
1468                 ["link_class", "=", "docker_image_hash"]],
1469     limit => 1);
1470   my $docker_hash;
1471   foreach my $link (@{$links_result->{items}}) {
1472     $docker_hash = lc($link->{name});
1473   }
1474   return $docker_hash;
1475 }
1476
1477 __DATA__
1478 #!/usr/bin/perl
1479
1480 # checkout-and-build
1481
1482 use Fcntl ':flock';
1483
1484 my $destdir = $ENV{"CRUNCH_SRC"};
1485 my $commit = $ENV{"CRUNCH_SRC_COMMIT"};
1486 my $repo = $ENV{"CRUNCH_SRC_URL"};
1487
1488 open L, ">", "$destdir.lock" or die "$destdir.lock: $!";
1489 flock L, LOCK_EX;
1490 if (readlink ("$destdir.commit") eq $commit && -d $destdir) {
1491     exit 0;
1492 }
1493
1494 unlink "$destdir.commit";
1495 open STDOUT, ">", "$destdir.log";
1496 open STDERR, ">&STDOUT";
1497
1498 mkdir $destdir;
1499 my @git_archive_data = <DATA>;
1500 if (@git_archive_data) {
1501   open TARX, "|-", "tar", "-C", $destdir, "-xf", "-";
1502   print TARX @git_archive_data;
1503   if(!close(TARX)) {
1504     die "'tar -C $destdir -xf -' exited $?: $!";
1505   }
1506 }
1507
1508 my $pwd;
1509 chomp ($pwd = `pwd`);
1510 my $install_dir = $ENV{"CRUNCH_INSTALL"} || "$pwd/opt";
1511 mkdir $install_dir;
1512
1513 for my $src_path ("$destdir/arvados/sdk/python") {
1514   if (-d $src_path) {
1515     shell_or_die ("virtualenv", $install_dir);
1516     shell_or_die ("cd $src_path && ./build.sh && $install_dir/bin/python setup.py install");
1517   }
1518 }
1519
1520 if (-e "$destdir/crunch_scripts/install") {
1521     shell_or_die ("$destdir/crunch_scripts/install", $install_dir);
1522 } elsif (!-e "./install.sh" && -e "./tests/autotests.sh") {
1523     # Old version
1524     shell_or_die ("./tests/autotests.sh", $install_dir);
1525 } elsif (-e "./install.sh") {
1526     shell_or_die ("./install.sh", $install_dir);
1527 }
1528
1529 if ($commit) {
1530     unlink "$destdir.commit.new";
1531     symlink ($commit, "$destdir.commit.new") or die "$destdir.commit.new: $!";
1532     rename ("$destdir.commit.new", "$destdir.commit") or die "$destdir.commit: $!";
1533 }
1534
1535 close L;
1536
1537 exit 0;
1538
1539 sub shell_or_die
1540 {
1541   if ($ENV{"DEBUG"}) {
1542     print STDERR "@_\n";
1543   }
1544   system (@_) == 0
1545       or die "@_ failed: $! exit 0x".sprintf("%x",$?);
1546 }
1547
1548 __DATA__