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