Changeset 94 for trunk/oarutils

Timestamp:

Jan 24, 2013, 7:17:46 PM (12 years ago)

Author:

g7moreau

Message:

• Add checkpointing and signal API. No test done...

File:

• trunk/oarutils/oar-dispatch (modified) (7 diffs)

trunk/oarutils/oar-dispatch

@@ -11 +11 @@
 use Coro::Semaphore;
 use Coro::Timer qw(sleep);
+use Coro::Handle;
+use IO::File;
 
 my $task = 0;
 my $overload = 1.1;
 my $file = '';
+my $logtrace;
 my $verbose;
 my $help;
+my $sig_transmit;
+my $sig_checkpoint = 'USR2';
 
 Getopt::Long::GetOptions(
@@ -22 +27 @@
    'overload=f'   => \$overload,
    'file=s'       => \$file,
+   'logtrace=s'   => \$logtrace,
    'verbose'      => \$verbose,
    'help'         => \$help,
+   'transmit'     => \$sig_transmit,
+   'kill=s'       => \$sig_checkpoint,
    ) || pod2usage(-verbose => 0);
 pod2usage(-verbose => 2) if $help;
@@ -31 +39 @@
    $task++ while <NODE_FILE>;
    close NODE_FILE;
+   }
+
+# re-run, keep trace of job already done
+my %state;
+my $log_h = IO::File->new();
+if (-e "$logtrace") {
+   $log_h->open("< $logtrace")
+      or die "error: can't read log file: $!";
+   while (<$log_h>) {
+      $state{$1} = 'start' if m/^start\s+job\s+([^\s]+)\s/;
+      $state{$1} = 'end'   if m/^end\s+job\s+([^\s]+)\s/;
+      }
+   $log_h->close();
+   }
+if ($logtrace) {
+   $log_h->open(">> $logtrace")
+      or die "error: can't append log file $logtrace: $!";
+   $log_h->autoflush;
+   $log_h = unblock $log_h;
    }
 
@@ -60 +87 @@
 my %scheduled = ();
 
+# OAR checkpoint and default signal SIGUSR2
+my $oar_checkpoint = new Coro::Semaphore 0;
+my $notify         = new Coro::Signal;
+$SIG{$sig_checkpoint} = sub {
+   print "warning: receive checkpoint at "
+      . time
+      . ", no new job, just finishing running job\n"
+      if $verbose;
+   $oar_checkpoint->up();
+   $notify->send if $sig_transmit;
+   };
+
+# asynchrone notify job
+async {
+   while () {
+      $notify->wait;
+
+      for my $job_pid (keys %scheduled) {
+         system "oardel --checkpoint --signal $sig_checkpoint $job_pid";
+         cede;
+         }
+      }
+   }
+
 # asynchrone start job block
 async {
@@ -67 +118 @@
          cede;
          }
+
+      # no more launch job when OAR checkpointing
+      last JOB if $oar_checkpoint->count() > 0;
+
       $job =~ s/^\s*oarsub//;
       print "oarsub $insert_oar_option $job" if $verbose;
@@ -107 +162 @@
 #         }
 
+      # checkpointing ! just finishing running job and quit
+      $finished->send if $oar_checkpoint->count() > 0 and scalar(keys(%scheduled)) == 0;
+
       $finished->send if $job_todo->count == 0;
       cede;
@@ -116 +174 @@
 # all job have been done
 $finished->wait;
+
+# close log trace file
+$log_h->close() if $logtrace;