X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/6263b4bdab2869ffc52c49ab283dd39c93f25a32..b1b9ffbb5e49d04c058d11702ecd240b1ef8a958:/sdk/cli/bin/crunch-job

diff --git a/sdk/cli/bin/crunch-job b/sdk/cli/bin/crunch-job
index 87b4fbf8a5..2f8623ba66 100755
--- a/sdk/cli/bin/crunch-job
+++ b/sdk/cli/bin/crunch-job
@@ -33,6 +33,12 @@ Path to .git directory where the specified commit is found.
 
 Arvados API authorization token to use during the course of the job.
 
+=item --no-clear-tmp
+
+Do not clear per-job/task temporary directories during initial job
+setup. This can speed up development and debugging when running jobs
+locally.
+
 =back
 
 =head1 RUNNING JOBS LOCALLY
@@ -73,6 +79,8 @@ use Arvados;
 use Getopt::Long;
 use IPC::Open2;
 use IO::Select;
+use File::Temp;
+use Fcntl ':flock';
 
 $ENV{"TMPDIR"} ||= "/tmp";
 unless (defined $ENV{"CRUNCH_TMP"}) {
@@ -91,11 +99,13 @@ my $force_unlock;
 my $git_dir;
 my $jobspec;
 my $job_api_token;
+my $no_clear_tmp;
 my $resume_stash;
 GetOptions('force-unlock' => \$force_unlock,
            'git-dir=s' => \$git_dir,
            'job=s' => \$jobspec,
            'job-api-token=s' => \$job_api_token,
+           'no-clear-tmp' => \$no_clear_tmp,
            'resume-stash=s' => \$resume_stash,
     );
 
@@ -165,11 +175,8 @@ else
 }
 $job_id = $Job->{'uuid'};
 
-# $metastream = Warehouse::Stream->new(whc => new Warehouse);
-# $metastream->clear;
-# $metastream->name('.');
-# $metastream->write_start($job_id . '.log.txt');
-
+my $keep_logfile = $job_id . '.log.txt';
+my $local_logfile = File::Temp->new();
 
 $Job->{'runtime_constraints'} ||= {};
 $Job->{'runtime_constraints'}->{'max_tasks_per_node'} ||= 0;
@@ -322,6 +329,12 @@ else
 }
 
 
+if (!$have_slurm)
+{
+  must_lock_now("$ENV{CRUNCH_TMP}/.lock", "a job is already running here.");
+}
+
+
 my $build_script;
 
 
@@ -330,6 +343,11 @@ $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{script_version};
 my $skip_install = ($local_job && $Job->{script_version} =~ m{^/});
 if ($skip_install)
 {
+  if (!defined $no_clear_tmp) {
+    my $clear_tmp_cmd = 'rm -rf $JOB_WORK $CRUNCH_TMP/opt $CRUNCH_TMP/src*';
+    system($clear_tmp_cmd) == 0
+	or croak ("`$clear_tmp_cmd` failed: ".($?>>8));
+  }
   $ENV{"CRUNCH_SRC"} = $Job->{script_version};
   for my $src_path ("$ENV{CRUNCH_SRC}/arvados/sdk/python") {
     if (-d $src_path) {
@@ -350,22 +368,24 @@ else
   Log (undef, "Install revision ".$Job->{script_version});
   my $nodelist = join(",", @node);
 
-  # Clean out crunch_tmp/work, crunch_tmp/opt, crunch_tmp/src*
+  if (!defined $no_clear_tmp) {
+    # Clean out crunch_tmp/work, crunch_tmp/opt, crunch_tmp/src*
 
-  my $cleanpid = fork();
-  if ($cleanpid == 0)
-  {
-    srun (["srun", "--nodelist=$nodelist", "-D", $ENV{'TMPDIR'}],
-	  ['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*']);
-    exit (1);
-  }
-  while (1)
-  {
-    last if $cleanpid == waitpid (-1, WNOHANG);
-    freeze_if_want_freeze ($cleanpid);
-    select (undef, undef, undef, 0.1);
+    my $cleanpid = fork();
+    if ($cleanpid == 0)
+    {
+      srun (["srun", "--nodelist=$nodelist", "-D", $ENV{'TMPDIR'}],
+	    ['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*']);
+      exit (1);
+    }
+    while (1)
+    {
+      last if $cleanpid == waitpid (-1, WNOHANG);
+      freeze_if_want_freeze ($cleanpid);
+      select (undef, undef, undef, 0.1);
+    }
+    Log (undef, "Clean-work-dir exited $?");
   }
-  Log (undef, "Clean-work-dir exited $?");
 
   # Install requested code version
 
@@ -462,6 +482,12 @@ else
   Log (undef, "Install exited $?");
 }
 
+if (!$have_slurm)
+{
+  # Grab our lock again (we might have deleted and re-created CRUNCH_TMP above)
+  must_lock_now("$ENV{CRUNCH_TMP}/.lock", "a job is already running here.");
+}
+
 
 
 foreach (qw (script script_version script_parameters runtime_constraints))
@@ -553,7 +579,7 @@ for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++)
     }
     $ENV{"TASK_SLOT_NODE"} = $slot[$childslot]->{node}->{name};
     $ENV{"TASK_SLOT_NUMBER"} = $slot[$childslot]->{cpu};
-    $ENV{"TASK_WORK"} = $ENV{"JOB_WORK"}."/".$slot[$childslot]->{cpu};
+    $ENV{"TASK_WORK"} = $ENV{"JOB_WORK"}."/$id.$$";
     $ENV{"TASK_KEEPMOUNT"} = $ENV{"TASK_WORK"}."/keep";
     $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
     $ENV{"CRUNCH_NODE_SLOTS"} = $slot[$childslot]->{node}->{ncpus};
@@ -579,10 +605,6 @@ for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++)
       $command .=
 	  "&& perl -";
     }
-    $ENV{"PYTHONPATH"} =~ s{^}{:} if $ENV{"PYTHONPATH"};
-    $ENV{"PYTHONPATH"} =~ s{^}{$ENV{CRUNCH_SRC}/sdk/python}; # xxx hack
-    $ENV{"PYTHONPATH"} =~ s{^}{$ENV{CRUNCH_SRC}/arvados/sdk/python:}; # xxx hack
-    $ENV{"PYTHONPATH"} =~ s{$}{:/usr/local/arvados/src/sdk/python}; # xxx hack
     $command .=
         "&& exec arv-mount $ENV{TASK_KEEPMOUNT} --exec $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
     my @execargs = ('bash', '-c', $command);
@@ -736,7 +758,7 @@ if ($job_has_uuid) {
 if ($Job->{'output'})
 {
   eval {
-    my $manifest_text = `arv keep get $Job->{'output'}`;
+    my $manifest_text = `arv keep get \Q$Job->{'output'}\E`;
     $arv->{'collections'}->{'create'}->execute('collection' => {
       'uuid' => $Job->{'output'},
       'manifest_text' => $manifest_text,
@@ -1039,11 +1061,12 @@ sub process_stderr
 sub fetch_block
 {
   my $hash = shift;
-  my ($child_out, $child_in, $output_block);
+  my ($child_out, $output_block);
 
-  my $pid = open2($child_out, $child_in, 'arv', 'keep', 'get', $hash);
-  sysread($child_out, $output_block, 64 * 1024 * 1024);
-  waitpid($pid, 0);
+  my $cmd = "arv keep get \Q$hash\E";
+  open($keep, '-|', $cmd) or die "fetch_block: $cmd: $!"
+  sysread($keep, $output_block, 64 * 1024 * 1024);
+  close $keep;
   return $output_block;
 }
 
@@ -1077,15 +1100,20 @@ sub collate_output
     }
     else
     {
-      print $child_in "XXX fetch_block($output) failed XXX\n";
+      Log (undef, "XXX fetch_block($output) failed XXX");
       $main::success = 0;
     }
   }
+  $child_in->close;
+
   if (!defined $joboutput) {
     my $s = IO::Select->new($child_out);
-    sysread($child_out, $joboutput, 64 * 1024 * 1024) if $s->can_read(0);
+    if ($s->can_read(120)) {
+      sysread($child_out, $joboutput, 64 * 1024 * 1024);
+    } else {
+      Log (undef, "timed out reading from 'arv keep put'");
+    }
   }
-  $child_in->close;
   waitpid($pid, 0);
 
   if ($joboutput)
@@ -1162,8 +1190,9 @@ sub Log				# ($jobstep_id, $logmessage)
   }
   print STDERR ((-t STDERR) ? ($datetime." ".$message) : $message);
 
-  # return if !$metastream;
-  # $metastream->write_data ($datetime . " " . $message);
+  if ($metastream) {
+    print $metastream $datetime . " " . $message;
+  }
 }
 
 
@@ -1191,20 +1220,19 @@ sub cleanup
 
 sub save_meta
 {
-#  my $justcheckpoint = shift; # false if this will be the last meta saved
-#  my $m = $metastream;
-#  $m = $m->copy if $justcheckpoint;
-#  $m->write_finish;
-#  my $whc = Warehouse->new;
-#  my $loglocator = $whc->store_block ($m->as_string);
-#  $arv->{'collections'}->{'create'}->execute('collection' => {
-#    'uuid' => $loglocator,
-#    'manifest_text' => $m->as_string,
-#  });
-#  undef $metastream if !$justcheckpoint; # otherwise Log() will try to use it
-#  Log (undef, "log manifest is $loglocator");
-#  $Job->{'log'} = $loglocator;
-#  $Job->update_attributes('log', $loglocator) if $job_has_uuid;
+  my $justcheckpoint = shift; # false if this will be the last meta saved
+  return if $justcheckpoint;  # checkpointing is not relevant post-Warehouse.pm
+
+  $local_logfile->flush;
+  my $cmd = "arv keep put --filename \Q$keep_logfile\E "
+      . quotemeta($local_logfile->filename);
+  my $loglocator = `$cmd`;
+  die "system $cmd failed: $?" if $?;
+
+  $local_logfile = undef;   # the temp file is automatically deleted
+  Log (undef, "log manifest is $loglocator");
+  $Job->{'log'} = $loglocator;
+  $Job->update_attributes('log', $loglocator) if $job_has_uuid;
 }
 
 
@@ -1247,65 +1275,6 @@ sub freeze
 sub thaw
 {
   croak ("Thaw not implemented");
-
-  # my $whc;
-  # my $key = shift;
-  # Log (undef, "thaw from $key");
-
-  # @jobstep = ();
-  # @jobstep_done = ();
-  # @jobstep_todo = ();
-  # @jobstep_tomerge = ();
-  # $jobstep_tomerge_level = 0;
-  # my $frozenjob = {};
-
-  # my $stream = new Warehouse::Stream ( whc => $whc,
-  #				       hash => [split (",", $key)] );
-  # $stream->rewind;
-  # while (my $dataref = $stream->read_until (undef, "\n\n"))
-  # {
-  #   if ($$dataref =~ /^job /)
-  #   {
-  #     foreach (split ("\n", $$dataref))
-  #     {
-  #	my ($k, $v) = split ("=", $_, 2);
-  #	$frozenjob->{$k} = freezeunquote ($v);
-  #     }
-  #     next;
-  #   }
-
-  #   if ($$dataref =~ /^merge (\d+) (.*)/)
-  #   {
-  #     $jobstep_tomerge_level = $1;
-  #     @jobstep_tomerge
-  #	  = map { freezeunquote ($_) } split ("\n", freezeunquote($2));
-  #     next;
-  #   }
-
-  #   my $Jobstep = { };
-  #   foreach (split ("\n", $$dataref))
-  #   {
-  #     my ($k, $v) = split ("=", $_, 2);
-  #     $Jobstep->{$k} = freezeunquote ($v) if $k;
-  #   }
-  #   $Jobstep->{'failures'} = 0;
-  #   push @jobstep, $Jobstep;
-
-  #   if ($Jobstep->{exitcode} eq "0")
-  #   {
-  #     push @jobstep_done, $#jobstep;
-  #   }
-  #   else
-  #   {
-  #     push @jobstep_todo, $#jobstep;
-  #   }
-  # }
-
-  # foreach (qw (script script_version script_parameters))
-  # {
-  #   $Job->{$_} = $frozenjob->{$_};
-  # }
-  # $Job->save if $job_has_uuid;
 }
 
 
@@ -1365,6 +1334,15 @@ sub ban_node_by_slot {
   Log (undef, "backing off node " . $slot[$slotid]->{node}->{name} . " for 60 seconds");
 }
 
+sub must_lock_now
+{
+  my ($lockfile, $error_message) = @_;
+  open L, ">", $lockfile or croak("$lockfile: $!");
+  if (!flock L, LOCK_EX|LOCK_NB) {
+    croak("Can't lock $lockfile: $error_message\n");
+  }
+}
+
 __DATA__
 #!/usr/bin/perl