| 1 |
#!/usr/bin/perl |
| 2 |
use strict; |
| 3 |
use Getopt::Std; |
| 4 |
use File::Basename; #for basename, dirname |
| 5 |
use POSIX "sys_wait_h"; |
| 6 |
use Net::Domain qw(hostname hostdomain); |
| 7 |
use Net::SMTP; # for send_mail function |
| 8 |
use Cwd qw(abs_path cwd); |
| 9 |
use Fcntl qw(:DEFAULT :seek); #mostly for the SEEK constants |
| 10 |
use FindBin; use lib "$FindBin::Bin"; |
| 11 |
|
| 12 |
my $NORMAL_ENDING=1; #set to 0 while working, to 1 upon normal ending of script (i.e. no die()) |
| 13 |
my $USER=$ENV{USER} || POSIX::cuserid(); #it may not be there for condor workers |
| 14 |
my $PWD=cwd(); #from Cwd module |
| 15 |
my $PERL_BIN='/usr/bin/perl'; |
| 16 |
my $MAX_RETRIES=3; #-- how many times to try a failed task |
| 17 |
my $F_WRKCOUNT='.wrkCount'; # count of currently running workers |
| 18 |
my $F_ALLDONE='.all.Done.'; |
| 19 |
my $F_WRKSTART='.wrkStarted'; #count of (ever) started workers |
| 20 |
my $F_LASTTASK='.lastTask'; # maximum task# ever started |
| 21 |
my $F_TASKSDONE='tasksDone'; # number of tasks successfully finished |
| 22 |
my $F_ERRTASKS='err.tasks'; #LIST of task#s which returned non-zero status |
| 23 |
# even after MAX_RETRIES |
| 24 |
my $F_RETRYTASKS='retry.tasks'; #stack of task#s which returned non-zero status |
| 25 |
#less then MAX_RETRIES times |
| 26 |
my $F_WRKRUNNING='.running-worker'; #semaphore file in each worker directory |
| 27 |
my $F_ENDCMD='epilogue'; |
| 28 |
my $F_WRKDIR='workdir'; |
| 29 |
my $F_NOTIFY='notify'; |
| 30 |
my $F_TASKDB='taskDb'; |
| 31 |
my $GRID_DEBUG; |
| 32 |
my $starting_dir; |
| 33 |
my $STARTED_GRID_TASK; |
| 34 |
my $SMPChildren=0; #SMP case: number of children running |
| 35 |
my %SMPChildren=(); #set of child PIDs |
| 36 |
my %Locks = (); # filehandle -> [lockfilelink, hostlockfile] |
| 37 |
#my $HOSTNAME = (&POSIX::uname)[1]; # can't trust HOST envvar under condor, because |
| 38 |
# # it will either inherit HOST and HOSTNAME from submitter env for getenv=TRUE, |
| 39 |
# # OR it will not set anything for getenv=FALSE |
| 40 |
my $HOSTNAME = hostname(); # from Net::Domain |
| 41 |
chomp($HOSTNAME); |
| 42 |
$HOSTNAME=lc($HOSTNAME); |
| 43 |
my $HOST=$HOSTNAME; |
| 44 |
# my ($DOMAIN)=($HOST=~m/^[\w\-]+\.(.+)/); |
| 45 |
# unless ($DOMAIN) { |
| 46 |
# if ($HOST=~m/^flicker|^wren|^ibis/) { |
| 47 |
# $DOMAIN='umiacs.umd.edu'; |
| 48 |
# } |
| 49 |
# unless ($DOMAIN) { |
| 50 |
# $DOMAIN=($PWD=~/^\/export|^\/home/) ? 'dfci.harvard.edu' : 'umiacs.umd.edu'; |
| 51 |
# } |
| 52 |
# } |
| 53 |
my $DOMAIN = hostdomain(); # from Net::Domain |
| 54 |
($HOST)=($HOST=~m/^([\w\-]+)/); |
| 55 |
$ENV{HOST}=$HOST; |
| 56 |
$ENV{MACHINE}=$HOST; |
| 57 |
$ENV{HOSTNAME}=$HOSTNAME; |
| 58 |
$ENV{USER}=$USER; |
| 59 |
#&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&! - site specific - |
| 60 |
#---------- modify the SITE-SPECIFIC paths here if needed: |
| 61 |
#&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&! |
| 62 |
# ===== PATH management ==== |
| 63 |
# default: assume architecture/PATH/LIB uniformity across nodes |
| 64 |
# you can change this by adding your own worker/server paths here |
| 65 |
my $binpath=$ENV{PATH}; |
| 66 |
my $libpath=$ENV{LD_LIBRARY_PATH}; |
| 67 |
my $perllib=$ENV{PERLLIB}; |
| 68 |
|
| 69 |
# use home directory for symbolic links to working directories |
| 70 |
my $homebase=$ENV{HOME}; |
| 71 |
$homebase=~s/(\/[^\/]+$)//; |
| 72 |
|
| 73 |
# sometimes the HOME path is not defined within the grid job |
| 74 |
# please use some globally mounted path instead |
| 75 |
|
| 76 |
# default grid engine used: |
| 77 |
my ($GRID_MONHOME, $GRID_ENGINE) = ($homebase, 'pbs'); |
| 78 |
my $sysopen_mode = O_RDWR | O_CREAT ; |
| 79 |
# $sysopen_mode |= O_DIRECT if $DOMAIN!~/umiacs/; |
| 80 |
|
| 81 |
#&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&! - |
| 82 |
# |
| 83 |
my $morehints=qq{ |
| 84 |
The following entries will be created by gridx: |
| 85 |
<GRID_JOBDIR>/epilogue - only if -e option was given, a file containing the |
| 86 |
<end_cmd> parameter |
| 87 |
<GRID_JOBDIR>/notify - a file containing the e-mail address to notify |
| 88 |
if -m was given |
| 89 |
<GRID_JOBDIR>/.lastTask - a file keeping track of the last task# scheduled |
| 90 |
<GRID_JOBDIR>/.wrkStarted - a file keeping track of the number of |
| 91 |
<GRID_JOBDIR>/.wrkCount - a file keeping track of the number of |
| 92 |
currenly running workers |
| 93 |
<GRID_JOBDIR>/err.tasks - list of all error-terminated tasks, after retry |
| 94 |
(should be zero or empty if all tasks |
| 95 |
finished successfully) |
| 96 |
<GRID_JOBDIR>/retry.tasks - list of all error-terminated tasks that will |
| 97 |
still be retried |
| 98 |
<GRID_JOBDIR>/taskDb - pseudo-fasta db with the info & status |
| 99 |
of each task |
| 100 |
<GRID_JOBDIR>/taskDb.cidx - the cdbfasta index of the above db file |
| 101 |
<GRID_JOBDIR>/locks/ - lockfiles/links are placed here |
| 102 |
<GRID_JOBDIR>/running/ - while a task is processed, a file entry |
| 103 |
called task_<GRID_TASK> will be created in |
| 104 |
here for locking purposes; such file will have |
| 105 |
a line with processing info for the current |
| 106 |
task: <hostname> <pid> <CPU#> |
| 107 |
<GRID_JOBDIR>/wrk_<CPU#>/ - working directory exclusive to each worker |
| 108 |
process, one per CPU (1 <= CPU <= maxCPUs); |
| 109 |
also contains stderr and stdout log files |
| 110 |
for each worker process (wrk_stderr.log and |
| 111 |
wrk_stdout.log) |
| 112 |
* if -i option is used, a slice db file is created for <fastadb> with the |
| 113 |
positions of all slices of <numseqs> fasta records in the <fastadb> file. |
| 114 |
This file is called <GRID_JOBDIR>/taskDb |
| 115 |
* if -b option was given: executes the <begin_script> in the current submit |
| 116 |
directory (the one above <GRID_JOBDIR>) |
| 117 |
* if all tasks completed successfully and if -e option was given |
| 118 |
the <end_script> is executed in the <GRID_JOBDIR> subdirectory |
| 119 |
}; |
| 120 |
|
| 121 |
my $usage=qq{ |
| 122 |
Submit a list of commands to the grid. Can also automatically |
| 123 |
slice a multi-fasta file into chunks and pass the slice file names as |
| 124 |
parameters to a <cmd>: |
| 125 |
|
| 126 |
Usage: |
| 127 |
|
| 128 |
jgridx \{-f <cmds_file> | -i <fastadb> | -r <num_iterations> \} -p <num_workers> |
| 129 |
[-w <#threads_per_worker>] [-M <req_mem> ]\{-U|-u <slot>\} [-n <numseqs>] |
| 130 |
[-s <skipseqs>] [-t <totalseqs>] [-g <engine>] [-d <dir_prefix>] [-a] |
| 131 |
[-L <local_wrk_dir>] [-O <log_files_dir>] [-q] [-m <e-mail>] |
| 132 |
[-S] [-T] [-v] [-x <node_exclude_list>] [-y <node_targets_list>] |
| 133 |
[-b <begin_script>] [-e <end_script>] [-c] <cmd> [-C] [<cmd_args>..] |
| 134 |
|
| 135 |
|
| 136 |
Unless -J option is given, gridx will submit the job to the grid, |
| 137 |
creating a <GRID_JOBDIR> subdirectory (which on Condor has the format: |
| 138 |
gridx-<hostname>_<job_id>). A file called 'cmdline-<numtasks>.cmd' |
| 139 |
will also be created there containing the current gridx commmand line. |
| 140 |
|
| 141 |
Options: |
| 142 |
-g grid engine to use, can be 'pbs', 'smp', 'condor' or 'sge' (default: $GRID_ENGINE) |
| 143 |
-p maximum number of worker processes to request for this job |
| 144 |
-w (pbs engine only) request number of threads (CPUs) for each worker process |
| 145 |
-M (pbs engine only) request node memory (in GB) for each worker process; |
| 146 |
MUST use this option if your program needs more than 2GB memory! |
| 147 |
-f provide a file with a list of commands to be run on the grid (one |
| 148 |
command per line per CPU) |
| 149 |
-i grid processing of slices of the given multi-fasta file <fastadb>; |
| 150 |
the total number of tasks/iterations is given by the number of slices |
| 151 |
in <fastadb> (and the -n <numseqs> option); the script will |
| 152 |
automatically create the slice file in the worker directory before |
| 153 |
<cmd> is run with these parameters: |
| 154 |
<fastaslice> <numseqs> <slice#> <is_last> <skipped> <total> <cmd_args> |
| 155 |
-n slice size: for -i option, how many fasta records from <fastadb> |
| 156 |
should be placed into a slice file for each iteration (default: 2000) |
| 157 |
-C legacy option for psx-like usage (to pass <cmd_args>) |
| 158 |
-r submit an array job with <numtasks> iterations/tasks (default: 1) |
| 159 |
|
| 160 |
-S switch to (stay in) the current directory (where gridx was launched from) |
| 161 |
before each task execution (especially useful for -f option); by default |
| 162 |
worker process are executed in their own worker subdirectory |
| 163 |
-a (psx legacy): inherit the submitter environment (default) |
| 164 |
-b prologue script to be executed BEFORE the grid job is submitted |
| 165 |
-e epilogue script to be executed AFTER all the grid tasks were completed, |
| 166 |
and only if ALL the tasks terminated successfully. |
| 167 |
-m send an e-mail to the given address when all tasks are finished |
| 168 |
-v do not try to validate <cmd> (assume it is going to be valid on the grid |
| 169 |
nodes) |
| 170 |
-x provide a list of machines (host names) that should be excluded |
| 171 |
from the condor pool for this submitted job ('condor' engine only) |
| 172 |
-y the submitted job is only allowed to run on the machines on this list |
| 173 |
('condor' engine only) |
| 174 |
-d create <dir_prefix> prefixed symbolic links to worker directories |
| 175 |
in the current directory |
| 176 |
-O place all Condor log files (condor stderr/stdout redirects) |
| 177 |
into <log_files_dir> |
| 178 |
-M parent directory for creating a "monitoring" symlink; a symbolic link to |
| 179 |
the current <GRID_JOBDIR> will be created there, under |
| 180 |
<monitoring_basedir>/$USER/ (default: home directory) |
| 181 |
-T do not exit after the job is submitted, wait around instead |
| 182 |
until the last worker terminates and prints some job completion stats |
| 183 |
-U force one worker per machine (only for 'condor' engine) |
| 184 |
-u force each worker to run only on CPU <slot#> (between 1 and 12) of each |
| 185 |
machine (only for 'condor' engine) |
| 186 |
For -r option, the provided <cmd> could make use of the environment variables |
| 187 |
GRID_TASK, GRID_TASKLAST to determine the current iteration being executed. |
| 188 |
|
| 189 |
Unless -S option was used, each <cmd> iteration will be executed in its own |
| 190 |
worker subdirectory <GRID_JOBDIR>/wrk_<CPU#> |
| 191 |
|
| 192 |
Job monitoring/resuming/stopping (-J mode): |
| 193 |
|
| 194 |
gridx [-m e-mail] [-e <end_script>] [-K|-R] -J <jobID> |
| 195 |
|
| 196 |
If -J option is provided, gridx will report the status of the job <jobID> |
| 197 |
which must have been submitted by a previous, "regular" use of gridx; it relies |
| 198 |
on the the symbolic link <monitor_basedir>/$USER/gridx-<jobID> which must |
| 199 |
be valid (<monitor_basedir> can be set by -M). |
| 200 |
|
| 201 |
Additional/alternate actions for -J mode: |
| 202 |
-e update the <end_script> for the *running* <jobID> or for the <jobID> |
| 203 |
rerun (if -R option is also given); does not work with -K option |
| 204 |
-m update the e-mail notification option for job <jobID> |
| 205 |
-K kill/abandon/terminate ALL pending tasks for grid job jobID |
| 206 |
(trying to remove all running/pending tasks on the grid) |
| 207 |
-R rerun/resubmit all the unfinished/unprocessed or unsuccessful tasks |
| 208 |
for <GRID_JOB>; this assumes -K -J <JOB_ID> was given first (so there |
| 209 |
are no pending tasks) and then will submit a new job in the same working |
| 210 |
directory, renaming the GRID_JOBDIR accordingly while workers |
| 211 |
will now *skip* all the tasks found with a "Done" status ('.') in the |
| 212 |
<GRID_JOBDIR>/taskDb file |
| 213 |
}; #' |
| 214 |
|
| 215 |
|
| 216 |
RESUME_JOBID: |
| 217 |
my @ar=@ARGV; |
| 218 |
while ($ar[0] eq '-Z') { shift(@ar); shift(@ar); } |
| 219 |
my $CMDLINE="$FindBin::Script\t".join("\t",@ar); |
| 220 |
# parse script options |
| 221 |
print STDERR "Running on $HOST (domain: $DOMAIN): $0 ".join(' ',@ARGV)."\n"; |
| 222 |
getopts('USWHM:O:NKBRDTqvaJZ:L:u:r:x:y:i:Ff:n:t:d:s:p:w:m:g:b:e:c:C:') || die($usage."\n"); |
| 223 |
umask 0002; |
| 224 |
if ($Getopt::Std::opt_H) { |
| 225 |
print STDERR $usage; |
| 226 |
die($morehints."\n"); |
| 227 |
} |
| 228 |
my $SwitchDir=$Getopt::Std::opt_S; |
| 229 |
$NORMAL_ENDING=0; |
| 230 |
$GRID_DEBUG=$Getopt::Std::opt_D; |
| 231 |
#print STDERR "Host=$HOST, Domain=$DOMAIN\n" if $GRID_DEBUG; |
| 232 |
|
| 233 |
my $GRID_DIRPREFIX=$Getopt::Std::opt_d; |
| 234 |
my ($submitJob, $removeJob); |
| 235 |
$GRID_ENGINE=lc($Getopt::Std::opt_g) if $Getopt::Std::opt_g; |
| 236 |
if ($GRID_ENGINE eq 'pbs') { |
| 237 |
($submitJob, $removeJob)=(\&submitJob_pbs, \&removeJob_sge); |
| 238 |
} |
| 239 |
elsif ($GRID_ENGINE eq 'sge') { |
| 240 |
($submitJob, $removeJob)=(\&submitJob_sge, \&removeJob_sge); |
| 241 |
} |
| 242 |
elsif ($GRID_ENGINE eq 'condor') { |
| 243 |
($submitJob, $removeJob)=(\&submitJob_condor, \&removeJob_condor); |
| 244 |
} |
| 245 |
elsif ($GRID_ENGINE eq 'smp') { |
| 246 |
($submitJob, $removeJob)=(\&submitJob_smp, \&removeJob_smp); |
| 247 |
} |
| 248 |
else { |
| 249 |
die("Error: invalid grid engine given (only 'pbs', 'condor' or 'smp' are accepted)!\n"); |
| 250 |
} |
| 251 |
my $UniqueVM=$Getopt::Std::opt_U; |
| 252 |
|
| 253 |
my $UniqVMreq=$Getopt::Std::opt_u; |
| 254 |
$UniqVMreq=int($UniqVMreq) if $UniqVMreq; |
| 255 |
die("Error: use either -U or -u option, not both!\n") if $UniqueVM && $UniqVMreq>0; |
| 256 |
my $worker_threads=$Getopt::Std::opt_w || 1; |
| 257 |
my $node_mem_req = $Getopt::Std::opt_M; |
| 258 |
|
| 259 |
# - exclude the following machines |
| 260 |
my @xmachinelist=split(/\,/,$Getopt::Std::opt_x); #do NOT allow jobs to run on these machines |
| 261 |
my @ymachinelist=split(/\,/,$Getopt::Std::opt_y); #only allow jobs to run on these machines, not others |
| 262 |
#submitJob MUST use the globals: |
| 263 |
# GRID_CMD, GRID_TASKLAST, GRID_MONHOME and update GRID_JOB |
| 264 |
|
| 265 |
#$GRID_MONHOME=$Getopt::Std::opt_M if $Getopt::Std::opt_M; |
| 266 |
$GRID_MONHOME.='/'.$USER unless $GRID_MONHOME=~m/$USER$/; |
| 267 |
# |
| 268 |
|
| 269 |
die("Error: directory $GRID_MONHOME should already be created! Aborting..\n") |
| 270 |
unless (-d $GRID_MONHOME); |
| 271 |
|
| 272 |
my $mailnotify=$Getopt::Std::opt_m; |
| 273 |
#-------- GLOBALs --------------------------- |
| 274 |
my $GRID_JOBDIR; |
| 275 |
my $GRID_JOB; |
| 276 |
my $GRID_LOCKDIR; |
| 277 |
my $GRID_TASKLAST; |
| 278 |
my $GRID_CMD; # user's command and arguments |
| 279 |
my $GRID_CMDLIST=$Getopt::Std::opt_f; # one line per run.. with arguments for command <cmd> |
| 280 |
my $GRID_USECMDLIST=1 if $GRID_CMDLIST || $Getopt::Std::opt_F; |
| 281 |
|
| 282 |
my $GRID_PSXFASTA; #if -i was used |
| 283 |
my $GRID_PSXSTEP; #if -i was used (-n value) |
| 284 |
my $GRID_PSXSKIP=0; # -s option |
| 285 |
my $GRID_PSXTOTAL=0; # -t option |
| 286 |
|
| 287 |
my $GRID_NUMPROCS=0; # -p option |
| 288 |
my $GRID_RESUME=$Getopt::Std::opt_R || $Getopt::Std::opt_Z; |
| 289 |
my $GRID_LOCAL_JOBDIR=$Getopt::Std::opt_L; |
| 290 |
#---------- worker side global vars: |
| 291 |
my $GRID_ENVSET=0; #was the environment set? |
| 292 |
my $GRID_WORKER=0; # worker# |
| 293 |
#my $GRID_NOWRKLINKS=1; |
| 294 |
my $GRID_LOGDIR=$Getopt::Std::opt_O; |
| 295 |
my $GRID_TASK; # dynamic -- task iteration# |
| 296 |
my $TASK_ERRCOUNT; # dynamic -- current task error (retry) counter |
| 297 |
my $TASK_DATA; # current task's user data as stored in taskDb |
| 298 |
my $TASK_LOCKH; #file handle for the current task lock file |
| 299 |
my $TASK_LOCKF; #file name for the current task lock file |
| 300 |
my $GRID_WRKDIR; #only for the worker case, it's the current worker's subdirectory |
| 301 |
|
| 302 |
if ($Getopt::Std::opt_J) { |
| 303 |
################################################################# |
| 304 |
# jgridx job monitoring use: |
| 305 |
#--------------------------------------------------- |
| 306 |
# gridx -J [-m <e-mail>] [-M <mondir>] [-R | -K] <jobid> |
| 307 |
#vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv |
| 308 |
my ($jobid)=shift(@ARGV); |
| 309 |
$jobid=~tr/ //d; |
| 310 |
unless ($jobid) { # list mode |
| 311 |
my $fmask="$GRID_MONHOME/gridx-*"; |
| 312 |
my @jdirs=<${fmask}>; |
| 313 |
if (@jdirs>0) { |
| 314 |
print STDOUT "The following jobs were found (in $GRID_MONHOME):\n"; |
| 315 |
foreach (@jdirs) { |
| 316 |
my ($jobid)=(m/gridx\-(\w[\w\-]+)$/); |
| 317 |
print " $jobid\n"; |
| 318 |
} |
| 319 |
} |
| 320 |
else { |
| 321 |
print STDOUT "No jobs were found (in $GRID_MONHOME).\n"; |
| 322 |
} |
| 323 |
$NORMAL_ENDING=1; |
| 324 |
exit(0); |
| 325 |
} |
| 326 |
#a valid $jobid was given, hopefully |
| 327 |
$jobid=lc($jobid); |
| 328 |
my $subdir='gridx-'.$jobid; |
| 329 |
my $jobdir="$GRID_MONHOME/gridx-$jobid"; |
| 330 |
unless (-d $jobdir) { #try current directory.. |
| 331 |
if (-d $subdir) { |
| 332 |
$jobdir=$subdir; |
| 333 |
} |
| 334 |
else { |
| 335 |
die "No such job found($jobid) - neither $jobdir nor $subdir exist!\n"; |
| 336 |
} |
| 337 |
} |
| 338 |
#print STDERR "..found jobdir='$jobdir'\n"; |
| 339 |
chdir($jobdir) || die("Error at chdir($jobdir)!\n"); |
| 340 |
#my $msg=jobSummary($mailnotify); |
| 341 |
my $msg=jobSummary(); |
| 342 |
print STDOUT $msg."\n"; |
| 343 |
if ($Getopt::Std::opt_K) { |
| 344 |
&$removeJob($jobid); |
| 345 |
} |
| 346 |
elsif ($Getopt::Std::opt_R) { #resume/rerun! |
| 347 |
#chdir($jobdir) || die("Error at chdir($jobdir)!\n"); |
| 348 |
my @cmdfile=<cmdline-*.cmd>; |
| 349 |
die "Error getting the cmdline-*.cmd from current directory!\n" |
| 350 |
unless @cmdfile; |
| 351 |
my ($numtasks)=($cmdfile[0]=~m/cmdline\-(\d+)/); |
| 352 |
die "Error parsing the number of tasks from $cmdfile[0]!\n" unless $numtasks>0; |
| 353 |
my $cmdline=readFile($cmdfile[0]); |
| 354 |
chomp($cmdline); |
| 355 |
my @args=split(/\t/,$cmdline); |
| 356 |
die("$jobdir/$F_TASKDB and/or index not valid - cannot resume!\n") |
| 357 |
unless -s $F_TASKDB && -s "$F_TASKDB.cidx"; |
| 358 |
shift(@args); #discard gridx command itself |
| 359 |
chdir('..'); #go in the original working dir |
| 360 |
$PWD=cwd(); #from Cwd module |
| 361 |
$CMDLINE="$FindBin::Script\t".join("\t",@args); |
| 362 |
@ARGV=('-Z',$jobid, @args); |
| 363 |
undef($Getopt::Std::opt_J); |
| 364 |
undef($Getopt::Std::opt_R); |
| 365 |
goto RESUME_JOBID; |
| 366 |
} |
| 367 |
$NORMAL_ENDING=1; |
| 368 |
exit(0); |
| 369 |
} |
| 370 |
elsif ($Getopt::Std::opt_B) { |
| 371 |
################################################################# |
| 372 |
# jgridx setup job: |
| 373 |
#--------------------------------------------------- |
| 374 |
# gridx -B <jobid> |
| 375 |
# prepare the directory and taskDb for <jobid> |
| 376 |
# which was submitted but placed on hold until this setup job |
| 377 |
# finishes |
| 378 |
#vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv |
| 379 |
} |
| 380 |
elsif ($Getopt::Std::opt_W) { |
| 381 |
########################################################## |
| 382 |
# jgridx Worker Mode: |
| 383 |
#--------------------------------------------------------- |
| 384 |
# This is the actual job that gets submitted |
| 385 |
# gridx -W <cmd> <cmd_args> |
| 386 |
# At runtime the environment should be set accordingly |
| 387 |
#vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv |
| 388 |
beginWorker(); #set up environment and worker directory |
| 389 |
#(wrk_NNNNN) and chdir() to it |
| 390 |
$GRID_USECMDLIST=$Getopt::Std::opt_F; |
| 391 |
my @cmd=@ARGV; # cmd and cmdargs |
| 392 |
while (my $taskId = getNextTask()) { |
| 393 |
runTask($taskId, @cmd); |
| 394 |
} |
| 395 |
my $runningworkers=endWorker(); |
| 396 |
my $exitcode=0; |
| 397 |
if ($runningworkers==0) { #this was the last worker! |
| 398 |
#run any JOB finishing/clean up code |
| 399 |
chdir($GRID_JOBDIR); |
| 400 |
my $epilogue=readFile($F_ENDCMD) if -s $F_ENDCMD; |
| 401 |
if ($epilogue) { |
| 402 |
chomp($epilogue); |
| 403 |
#only run it if ALL tasks succeeded |
| 404 |
my $tasksdone=readFile($F_TASKSDONE);chomp($tasksdone); |
| 405 |
if ($tasksdone==$GRID_TASKLAST && !(-s $F_ERRTASKS)) { |
| 406 |
runCmd($epilogue); |
| 407 |
} |
| 408 |
} |
| 409 |
my $notify=readFile($F_NOTIFY) if -s $F_NOTIFY; |
| 410 |
chomp($notify); |
| 411 |
jobSummary($notify); |
| 412 |
local *FDONE; |
| 413 |
open(FDONE, '>'.$GRID_JOBDIR.'/'.$F_ALLDONE) || die("Error creating $GRID_JOBDIR/$F_ALLDONE\n"); |
| 414 |
print FDONE "done.\n"; |
| 415 |
close(FDONE); |
| 416 |
} #last worker ending |
| 417 |
$NORMAL_ENDING=1; |
| 418 |
exit(0); |
| 419 |
}# -- worker use case |
| 420 |
|
| 421 |
|
| 422 |
########################################################## |
| 423 |
# |
| 424 |
# gridx Submit use: |
| 425 |
# |
| 426 |
########################################################## |
| 427 |
|
| 428 |
my $gridBeginCmd=$Getopt::Std::opt_b; |
| 429 |
my $gridEndCmd=$Getopt::Std::opt_e; |
| 430 |
$GRID_RESUME=$Getopt::Std::opt_Z; #caused by an initial -J -R request |
| 431 |
$GRID_TASKLAST=$Getopt::Std::opt_r; |
| 432 |
#--submit specific globals -- temporary |
| 433 |
my $JOBID; |
| 434 |
#-- |
| 435 |
unless ($GRID_CMDLIST) { |
| 436 |
$GRID_CMD=$Getopt::Std::opt_c || shift(@ARGV); |
| 437 |
die "$usage\nError: no command given!\n" unless $GRID_CMD; |
| 438 |
unless ($Getopt::Std::opt_v) { |
| 439 |
$GRID_CMD = getCmdPath($GRID_CMD) || |
| 440 |
die "Error: command $GRID_CMD not found in PATH!\n"; |
| 441 |
} |
| 442 |
$GRID_CMD.=' '.$Getopt::Std::opt_C if $Getopt::Std::opt_C; |
| 443 |
$GRID_CMD.=' '.join(' ',@ARGV) if @ARGV; |
| 444 |
} |
| 445 |
else { |
| 446 |
$GRID_CMD='.'; |
| 447 |
} |
| 448 |
$GRID_NUMPROCS=$Getopt::Std::opt_p || 1; #numer of workers requested |
| 449 |
|
| 450 |
if ($GRID_PSXFASTA=$Getopt::Std::opt_i) { #psx emulation case |
| 451 |
die "Error: -r and -i are mutually exclusive options!\n" if $GRID_TASKLAST; |
| 452 |
$GRID_PSXSTEP=$Getopt::Std::opt_n || 1; |
| 453 |
$GRID_PSXSKIP=$Getopt::Std::opt_s || 0; |
| 454 |
$GRID_PSXTOTAL=$Getopt::Std::opt_t || 0; |
| 455 |
$GRID_PSXFASTA=getFullPath($GRID_PSXFASTA, 1); |
| 456 |
} |
| 457 |
else { $GRID_TASKLAST=1 unless $GRID_TASKLAST; } #one shot run |
| 458 |
|
| 459 |
$JOBID = &$submitJob({PATH=> $binpath, |
| 460 |
LD_LIBRARY_PATH=>$libpath, PERLLIB=>$perllib} ); |
| 461 |
#-- submitJob also calls setupJobDir() to create the gridx-<JOBID> subdirectory |
| 462 |
#-- and move/rename the taskDb in there. |
| 463 |
die("Error: no job ID has been obtained!\n") unless $JOBID; |
| 464 |
if ($Getopt::Std::opt_T) { #wait for all children to finish |
| 465 |
my $wstat; |
| 466 |
chdir($GRID_JOBDIR); |
| 467 |
do { |
| 468 |
sleep(5); |
| 469 |
} until (-e $F_ALLDONE); |
| 470 |
my $msg=jobSummary($mailnotify); |
| 471 |
my $exitcode = (-s $F_ERRTASKS) ? `wc -l $F_ERRTASKS` : 0; |
| 472 |
chomp($exitcode); |
| 473 |
print STDOUT $msg."\n"; |
| 474 |
chdir($PWD); #this should be the original working directory |
| 475 |
$NORMAL_ENDING=1; |
| 476 |
exit($exitcode); |
| 477 |
} |
| 478 |
$NORMAL_ENDING=1; |
| 479 |
exit(0); |
| 480 |
|
| 481 |
#********************************************************** |
| 482 |
#******************* SUBROUTINES ************************** |
| 483 |
|
| 484 |
|
| 485 |
|
| 486 |
#******************** SUBMIT SIDE ************************* |
| 487 |
|
| 488 |
=head2 ----------- taskDbRec ----------- |
| 489 |
|
| 490 |
taskDbRec($fhdb, $jobId, $command_line..) |
| 491 |
|
| 492 |
Creates (writes) a record into the taskDb file open for writing |
| 493 |
with the file handle $fhdb. This subroutine takes care of |
| 494 |
formatting the fixed length defline (header) of the job record. |
| 495 |
|
| 496 |
This subroutine should only be used by the program which |
| 497 |
creates the taskDb. After all the records are added to the taskDb; |
| 498 |
the taskDb should be eventually indexed with cdbfasta. |
| 499 |
|
| 500 |
=cut |
| 501 |
|
| 502 |
sub taskDbRec { |
| 503 |
my ($fh, $taskId, @userdata)=@_; |
| 504 |
my $rec='>'.$taskId."\t{-|-|----|----|--------}\t{".('-' x 25).'}'; |
| 505 |
$rec.="\t".join(' ',@userdata) if @userdata; |
| 506 |
print($fh $rec."\n"); |
| 507 |
} |
| 508 |
|
| 509 |
|
| 510 |
sub prepareTaskDb { |
| 511 |
#we must be in GRID_JOBDIR already |
| 512 |
if (-s $F_TASKDB && $GRID_RESUME) { |
| 513 |
#taskDb already there, it's a Resume request |
| 514 |
my $r=`cdbyank -s $F_TASKDB.cidx`; # force reindex |
| 515 |
($GRID_TASKLAST)=($r=~m/\nNumber of records:\s+(\d+)\n/s); |
| 516 |
die "Error: couldn't determine number of records in $F_TASKDB\n" |
| 517 |
unless $GRID_TASKLAST>0; |
| 518 |
return; |
| 519 |
} |
| 520 |
if ($GRID_PSXFASTA) { #-i given, psx mode |
| 521 |
#-- iterate through the fasta file, create slice index |
| 522 |
# and the index for the slice index.. |
| 523 |
local *TASKDB; |
| 524 |
local *PSXFASTA; |
| 525 |
open(PSXFASTA, $GRID_PSXFASTA) || die "Error opening file $GRID_PSXFASTA\n"; |
| 526 |
binmode(PSXFASTA); |
| 527 |
open(TASKDB, '>'.$F_TASKDB) || die "Error creating file $F_TASKDB ($!)\n"; |
| 528 |
my $foffset=0; |
| 529 |
my $rcount=0; |
| 530 |
my $numrecs=0; |
| 531 |
$GRID_TASKLAST=0; |
| 532 |
while (<PSXFASTA>) { |
| 533 |
my $linelen=length($_); |
| 534 |
if (/^>/) { |
| 535 |
$rcount++; |
| 536 |
if ($rcount>$GRID_PSXSKIP) { |
| 537 |
$numrecs++; |
| 538 |
if (($numrecs-1) % $GRID_PSXSTEP == 0) { |
| 539 |
$GRID_TASKLAST++; |
| 540 |
taskDbRec(\*TASKDB, $GRID_TASKLAST, $foffset); |
| 541 |
} |
| 542 |
} |
| 543 |
last if ($GRID_PSXTOTAL>0 && $numrecs>$GRID_PSXTOTAL); |
| 544 |
} |
| 545 |
$foffset+=$linelen; |
| 546 |
} |
| 547 |
close(PSXFASTA); |
| 548 |
#&taskdbRec(\*TASKDB, $GRID_TASKLAST, $foffset) |
| 549 |
# unless ($numrecs-1) % $GRID_PSXSTEP == 0; |
| 550 |
close(TASKDB); |
| 551 |
} |
| 552 |
elsif ($GRID_CMDLIST) { # -f option given |
| 553 |
#return if ($GRID_RESUME); #taskDb must already be there! |
| 554 |
local *TASKDB; |
| 555 |
open(CMDFILE, $GRID_CMDLIST) || die "Error opening file $GRID_CMDLIST\n"; |
| 556 |
open(TASKDB, '>'.$F_TASKDB) || die "Error creating file $F_TASKDB ($!)\n"; |
| 557 |
my $i=1; |
| 558 |
while(<CMDFILE>) { |
| 559 |
next if m/^\s*#/; |
| 560 |
chomp; |
| 561 |
taskDbRec(\*TASKDB, $i, $_); |
| 562 |
$i++; |
| 563 |
} |
| 564 |
close(TASKDB); |
| 565 |
close(CMDFILE); |
| 566 |
$GRID_TASKLAST=$i-1; |
| 567 |
} |
| 568 |
elsif ($GRID_TASKLAST) { # -r option given |
| 569 |
#return if ($GRID_RESUME); #taskDb must already be there! |
| 570 |
local *TASKDB; |
| 571 |
open(TASKDB, '>'.$F_TASKDB) || die "Error creating file $F_TASKDB ($!)\n"; |
| 572 |
for (my $i=1;$i<=$GRID_TASKLAST;$i++) { |
| 573 |
taskDbRec(\*TASKDB, $i, $GRID_CMD); |
| 574 |
} |
| 575 |
close(TASKDB); |
| 576 |
} |
| 577 |
else { return; } |
| 578 |
runCmd("cdbfasta $F_TASKDB"); |
| 579 |
} |
| 580 |
|
| 581 |
sub getFName { |
| 582 |
return basename($_[0]); |
| 583 |
} |
| 584 |
|
| 585 |
sub getFDir { |
| 586 |
return dirname($_[0]); |
| 587 |
} |
| 588 |
|
| 589 |
sub getFullPath { |
| 590 |
my ($fname, $check)=@_; |
| 591 |
die("Error: file $fname does not exist!\n") if $check && !-r $fname; |
| 592 |
return abs_path($fname); #Cwd module |
| 593 |
} |
| 594 |
|
| 595 |
#== getCmdPath -- checks for executable in the PATH if no full path given |
| 596 |
# tests if the executable is a text file and interpreter was requested |
| 597 |
sub getCmdPath { |
| 598 |
my $cmd=$_[0]; |
| 599 |
my $fullpath; |
| 600 |
my $checkBinary=wantarray(); |
| 601 |
if ($cmd =~ m/^\//) { |
| 602 |
$fullpath = (-x $cmd) ? $cmd : ''; |
| 603 |
} |
| 604 |
elsif ($cmd =~ m/^\.+\//) { #relative path given |
| 605 |
$fullpath= (-x $cmd) ? abs_path($cmd) : ''; |
| 606 |
} |
| 607 |
else { #we search in the path.. |
| 608 |
my @paths=split(/:/, $ENV{'PATH'}); |
| 609 |
foreach my $p (@paths) { |
| 610 |
if (-x $p.'/'.$cmd) { |
| 611 |
$fullpath=$p.'/'.$cmd; |
| 612 |
last; |
| 613 |
} |
| 614 |
} |
| 615 |
}#path searching |
| 616 |
if ($checkBinary) { #asked for interpreter, if any |
| 617 |
if ($fullpath) { |
| 618 |
my $interpreter=''; |
| 619 |
if (-r $fullpath && -T $fullpath) {#readable text file, look for bang line |
| 620 |
open(TFILE, $fullpath); |
| 621 |
my $linesread=1; |
| 622 |
while ($linesread<10) {#read only the first 10 lines.. |
| 623 |
$_=<TFILE>; |
| 624 |
chomp; |
| 625 |
if (m/^\s*#\!\s*(\S.+)/) { |
| 626 |
$interpreter=$1; |
| 627 |
last; |
| 628 |
} |
| 629 |
$linesread++; |
| 630 |
} |
| 631 |
$interpreter=~s/\s+$//; |
| 632 |
} |
| 633 |
return ($fullpath, $interpreter); |
| 634 |
} |
| 635 |
else { return (); } #cmd not found; |
| 636 |
} |
| 637 |
else { return $fullpath; } |
| 638 |
} |
| 639 |
|
| 640 |
sub runCmd { |
| 641 |
my ($cmd, $jobid)=@_; |
| 642 |
my $exitstatus=system($cmd); |
| 643 |
if ($exitstatus != 0) { |
| 644 |
if ($jobid) { |
| 645 |
&$removeJob($jobid); |
| 646 |
} |
| 647 |
die("Error at running system command: $cmd\n"); |
| 648 |
} |
| 649 |
} |
| 650 |
|
| 651 |
|
| 652 |
sub removeJob_sge { |
| 653 |
my $jobid=shift(@_); |
| 654 |
runCmd("qdel $jobid"); |
| 655 |
} |
| 656 |
|
| 657 |
sub removeJob_condor { |
| 658 |
my $jobid=shift(@_); #machine+'_'+job# |
| 659 |
#must be on the same machine that submit was issuedlh |
| 660 |
my ($hostname, $job)=($jobid=~m/([\w\-]+)_(\d+)$/); |
| 661 |
die("Error parsing hostname, job# from $jobid!\n") |
| 662 |
unless $hostname && ($job>0); |
| 663 |
#print STDERR "$hostname, $HOST, $jobid, $job\n"; |
| 664 |
if (lc($hostname) eq lc($HOST)) { #local host |
| 665 |
runCmd("condor_rm $job"); |
| 666 |
} |
| 667 |
else { |
| 668 |
runCmd("condor_rm -name $hostname $job"); |
| 669 |
} |
| 670 |
} |
| 671 |
|
| 672 |
|
| 673 |
sub smpTaskReaper { # takes care of dead children $SIG{CHLD} = \&taskReaper; |
| 674 |
my $childpid; |
| 675 |
while (($childpid = waitpid(-1, WNOHANG)) > 0) { |
| 676 |
$SMPChildren --; |
| 677 |
delete $SMPChildren{$childpid}; |
| 678 |
} |
| 679 |
$SIG{CHLD}=\&smpTaskReaper; |
| 680 |
} |
| 681 |
|
| 682 |
sub smpTaskKiller { # signal handler for SIGINT |
| 683 |
local($SIG{CHLD}) = 'IGNORE'; # we're going to kill our children |
| 684 |
kill 'INT' => keys %SMPChildren; |
| 685 |
} |
| 686 |
|
| 687 |
sub removeJob_smp { |
| 688 |
taskKiller(); |
| 689 |
|
| 690 |
} |
| 691 |
|
| 692 |
sub submitJob_sge { |
| 693 |
my ($envhash)=@_; |
| 694 |
# submit array job for SGE grid |
| 695 |
my $array=''; |
| 696 |
if ($GRID_NUMPROCS > 1) { |
| 697 |
$array="-t 1-$GRID_NUMPROCS"; |
| 698 |
} |
| 699 |
#append our GRID_ environment |
| 700 |
my $envparam="-v 'GRID_ENGINE=pbs". |
| 701 |
",GRID_JOBDIR=$PWD". |
| 702 |
",GRID_TASKLAST=$GRID_TASKLAST". |
| 703 |
",GRID_MONHOME=$GRID_MONHOME". |
| 704 |
",GRID_CMD=$GRID_CMD". |
| 705 |
",GRID_DIRPREFIX=$GRID_DIRPREFIX"; |
| 706 |
$envparam.=",GRID_RESUME=$GRID_RESUME" if $GRID_RESUME; |
| 707 |
$envparam.= ",GRID_CMDLIST=$GRID_CMDLIST" if $GRID_CMDLIST; |
| 708 |
$envparam.=",GRID_PSXFASTA=$GRID_PSXFASTA". |
| 709 |
",GRID_PSXSTEP=$GRID_PSXSTEP" if $GRID_PSXFASTA; |
| 710 |
$envparam.= ",GRID_PSXSKIP=$GRID_PSXSKIP" if $GRID_PSXSKIP; |
| 711 |
$envparam.= ",GRID_PSXTOTAL=$GRID_PSXTOTAL" if $GRID_PSXTOTAL; |
| 712 |
$envparam.= ",GRID_LOCAL_JOBDIR=$GRID_LOCAL_JOBDIR" if $GRID_LOCAL_JOBDIR; |
| 713 |
$envparam.=',BLASTMAT='.$ENV{BLASTMAT} if $ENV{BLASTMAT}; |
| 714 |
$envparam.=',BOWTIE_INDEXES='.$ENV{BOWTIE_INDEXES} if $ENV{BOWTIE_INDEXES}; |
| 715 |
|
| 716 |
if (keys(%$envhash)>0) { |
| 717 |
$envparam.=','; |
| 718 |
my @envars; |
| 719 |
while (my ($env, $val)= each(%$envhash)) { |
| 720 |
push(@envars, $env.'='.$val); |
| 721 |
} |
| 722 |
$envparam.=join(',',@envars); |
| 723 |
} |
| 724 |
$envparam.="'"; |
| 725 |
#-- |
| 726 |
|
| 727 |
my $sub="qsub -cwd -b y $envparam"; |
| 728 |
#$sub.="-e $errout" if $errout; |
| 729 |
#$sub.="-o $stdout" if $stdout; |
| 730 |
my $logdir=''; |
| 731 |
if ($GRID_LOGDIR) { |
| 732 |
#separate log dir given |
| 733 |
$logdir=$GRID_LOGDIR; |
| 734 |
if (-d $GRID_LOGDIR) { |
| 735 |
print STDERR "Warning: log dir $GRID_LOGDIR exists; existing files will be overwritten!\n"; |
| 736 |
} |
| 737 |
else { |
| 738 |
mkdir($logdir) || die("Error creating log dir '$logdir'!\n"); |
| 739 |
} |
| 740 |
$logdir.='/' unless $logdir=~m/\/$/; |
| 741 |
} |
| 742 |
my $otherflags=''; |
| 743 |
$otherflags.=' -D' if $GRID_DEBUG; |
| 744 |
$otherflags.=' -F' if $GRID_USECMDLIST; |
| 745 |
$otherflags.=' -S' if $SwitchDir; |
| 746 |
my $subcmd= "$sub $array $PERL_BIN $0 $otherflags -W $GRID_CMD"; |
| 747 |
print STDERR "$subcmd\n" if $GRID_DEBUG; |
| 748 |
my $subout = `$subcmd`; |
| 749 |
my ($jobid)=($subout=~m/^(\d+)/); |
| 750 |
die "Error: No Job ID# could be parsed!\n($subout)" unless ($jobid>0); |
| 751 |
|
| 752 |
setupJobDir($jobid); |
| 753 |
|
| 754 |
return $jobid; |
| 755 |
} |
| 756 |
|
| 757 |
sub submitJob_pbs { |
| 758 |
my ($envhash)=@_; |
| 759 |
# submit array job for PBS grid |
| 760 |
my $array=''; |
| 761 |
if ($GRID_NUMPROCS > 1) { |
| 762 |
$array="-t 1-$GRID_NUMPROCS"; |
| 763 |
} |
| 764 |
#append our GRID_ environment |
| 765 |
my $envparam="-v 'GRID_ENGINE=pbs". |
| 766 |
",GRID_JOBDIR=$PWD". |
| 767 |
",GRID_TASKLAST=$GRID_TASKLAST". |
| 768 |
",GRID_MONHOME=$GRID_MONHOME". |
| 769 |
",GRID_CMD=$GRID_CMD". |
| 770 |
",GRID_DIRPREFIX=$GRID_DIRPREFIX"; |
| 771 |
$envparam.=",GRID_RESUME=$GRID_RESUME" if $GRID_RESUME; |
| 772 |
$envparam.= ",GRID_CMDLIST=$GRID_CMDLIST" if $GRID_CMDLIST; |
| 773 |
$envparam.=",GRID_PSXFASTA=$GRID_PSXFASTA". |
| 774 |
",GRID_PSXSTEP=$GRID_PSXSTEP" if $GRID_PSXFASTA; |
| 775 |
$envparam.= ",GRID_PSXSKIP=$GRID_PSXSKIP" if $GRID_PSXSKIP; |
| 776 |
$envparam.= ",GRID_PSXTOTAL=$GRID_PSXTOTAL" if $GRID_PSXTOTAL; |
| 777 |
$envparam.= ",GRID_LOCAL_JOBDIR=$GRID_LOCAL_JOBDIR" if $GRID_LOCAL_JOBDIR; |
| 778 |
$envparam.=',BLASTMAT='.$ENV{BLASTMAT} if $ENV{BLASTMAT}; |
| 779 |
$envparam.=',BOWTIE_INDEXES='.$ENV{BOWTIE_INDEXES} if $ENV{BOWTIE_INDEXES}; |
| 780 |
|
| 781 |
if (keys(%$envhash)>0) { |
| 782 |
$envparam.=','; |
| 783 |
my @envars; |
| 784 |
while (my ($env, $val)= each(%$envhash)) { |
| 785 |
push(@envars, $env.'='.$val); |
| 786 |
} |
| 787 |
$envparam.=join(',',@envars); |
| 788 |
} |
| 789 |
$envparam.="'"; |
| 790 |
#-- |
| 791 |
|
| 792 |
my $sub="qsub -cwd -b y $envparam"; |
| 793 |
#$sub.="-e $errout" if $errout; |
| 794 |
#$sub.="-o $stdout" if $stdout; |
| 795 |
my $logdir=''; |
| 796 |
if ($GRID_LOGDIR) { |
| 797 |
#separate log dir given |
| 798 |
$logdir=$GRID_LOGDIR; |
| 799 |
if (-d $GRID_LOGDIR) { |
| 800 |
print STDERR "Warning: log dir $GRID_LOGDIR exists; existing files will be overwritten!\n"; |
| 801 |
} |
| 802 |
else { |
| 803 |
mkdir($logdir) || die("Error creating log dir '$logdir'!\n"); |
| 804 |
} |
| 805 |
$logdir.='/' unless $logdir=~m/\/$/; |
| 806 |
} |
| 807 |
my $otherflags=''; |
| 808 |
$otherflags.=' -D' if $GRID_DEBUG; |
| 809 |
$otherflags.=' -F' if $GRID_USECMDLIST; |
| 810 |
$otherflags.=' -S' if $SwitchDir; |
| 811 |
my $subcmd= "$sub $array $PERL_BIN $0 $otherflags -W $GRID_CMD"; |
| 812 |
print STDERR "$subcmd\n" if $GRID_DEBUG; |
| 813 |
my $subout = `$subcmd`; |
| 814 |
my ($jobid)=($subout=~m/^(\d+)/); |
| 815 |
die "Error: No Job ID# could be parsed!\n($subout)" unless ($jobid>0); |
| 816 |
|
| 817 |
setupJobDir($jobid); |
| 818 |
|
| 819 |
return $jobid; |
| 820 |
} |
| 821 |
|
| 822 |
|
| 823 |
sub submitJob_smp { |
| 824 |
my ($envhash)=@_; |
| 825 |
my $jobid='smp_'.$$; |
| 826 |
$ENV{GRID_ENGINE}='smp'; |
| 827 |
@ENV{'GRID_JOBDIR', 'GRID_TASKLAST', 'GRID_MONHOME', 'GRID_CMD','GRID_DIRPREFIX'}= |
| 828 |
($PWD."/gridx-$jobid", $GRID_TASKLAST, $GRID_MONHOME, $GRID_CMD,$GRID_DIRPREFIX); |
| 829 |
$ENV{GRID_RESUME}=$GRID_RESUME; |
| 830 |
$ENV{GRID_JOB}=$jobid; |
| 831 |
$ENV{GRID_CMDLIST}=$GRID_CMDLIST if $GRID_CMDLIST; |
| 832 |
|
| 833 |
@ENV{'GRID_PSXFASTA','GRID_PSXSTEP'}=($GRID_PSXFASTA, $GRID_PSXSTEP) if $GRID_PSXFASTA; |
| 834 |
$ENV{GRID_PSXSKIP}=$GRID_PSXSKIP if $GRID_PSXSKIP; |
| 835 |
$ENV{GRID_PSXTOTAL}=$GRID_PSXTOTAL if $GRID_PSXTOTAL; |
| 836 |
|
| 837 |
$ENV{GRID_LOCAL_JOBDIR}=$GRID_LOCAL_JOBDIR."/gridx-$jobid" if $GRID_LOCAL_JOBDIR; |
| 838 |
if (keys(%$envhash)>0) { |
| 839 |
while (my ($env, $val)= each(%$envhash)) { |
| 840 |
$ENV{$env}=$val; |
| 841 |
} |
| 842 |
} |
| 843 |
# Fork off the children |
| 844 |
my $pid; |
| 845 |
setupJobDir($jobid); #we do this one in advance.. |
| 846 |
$SIG{CHLD}=\&smpTaskReaper; |
| 847 |
print STDERR "..forking $GRID_NUMPROCS workers..\n" if $GRID_DEBUG; |
| 848 |
my $logdir=''; |
| 849 |
if ($GRID_LOGDIR) { |
| 850 |
#separate log dir given |
| 851 |
$logdir=$GRID_LOGDIR; |
| 852 |
if (-d $GRID_LOGDIR) { |
| 853 |
print STDERR "Warning: log dir $GRID_LOGDIR exists; existing files will be overwritten!\n"; |
| 854 |
} |
| 855 |
else { |
| 856 |
mkdir($logdir) || die("Error creating log dir '$logdir'!\n"); |
| 857 |
} |
| 858 |
$logdir.='/' unless $logdir=~m/\/$/; |
| 859 |
} |
| 860 |
my $otherflags; |
| 861 |
$otherflags=' -D' if $GRID_DEBUG; |
| 862 |
#$otherflags.=' -N' if $GRID_NOWRKLINKS; |
| 863 |
$otherflags.=' -F' if $GRID_USECMDLIST; |
| 864 |
$otherflags.=' -S' if $SwitchDir; |
| 865 |
|
| 866 |
for (1 .. $GRID_NUMPROCS) { |
| 867 |
die "Error at fork: $!" unless defined ($pid = fork); |
| 868 |
if ($pid) { # Parent here |
| 869 |
$SMPChildren{$pid} = 1; |
| 870 |
$SMPChildren++; |
| 871 |
next; |
| 872 |
} else { #Child here |
| 873 |
# Child can *not* return from this subroutine. |
| 874 |
#$SIG{INT} = 'DEFAULT'; # make SIGINT kill us as it did before |
| 875 |
exec("$PERL_BIN $0 $otherflags -W $GRID_CMD"); #never returns |
| 876 |
} |
| 877 |
} |
| 878 |
$SIG{INT}=\&smpTaskKiller; |
| 879 |
$SIG{TERM}=\&smpTaskKiller; |
| 880 |
return $jobid; |
| 881 |
} |
| 882 |
|
| 883 |
sub submitJob_condor { |
| 884 |
my ($envhash)=@_; |
| 885 |
my $jobid; |
| 886 |
my $queue='queue'; |
| 887 |
$queue.=" $GRID_NUMPROCS" if ($GRID_NUMPROCS > 1); |
| 888 |
#append our GRID_ environment |
| 889 |
my $dprefix="gridx-$HOST".'_'; |
| 890 |
my $envparam="GRID_ENGINE=condor;". |
| 891 |
"GRID_JOBDIR=$PWD/$dprefix\$(Cluster);". |
| 892 |
"GRID_JOB=$HOST\_\$(Cluster);". |
| 893 |
"GRID_TASKLAST=$GRID_TASKLAST;". |
| 894 |
"GRID_MONHOME=$GRID_MONHOME;". |
| 895 |
"GRID_CMD=$GRID_CMD;". |
| 896 |
"GRID_DIRPREFIX=$GRID_DIRPREFIX"; |
| 897 |
$envparam.=";GRID_RESUME=$GRID_RESUME" if $GRID_RESUME; |
| 898 |
$envparam.= ";GRID_CMDLIST=$GRID_CMDLIST" if $GRID_CMDLIST; |
| 899 |
$envparam.=";GRID_PSXFASTA=$GRID_PSXFASTA". |
| 900 |
";GRID_PSXSTEP=$GRID_PSXSTEP" if $GRID_PSXFASTA; |
| 901 |
$envparam.= ";GRID_PSXSKIP=$GRID_PSXSKIP" if $GRID_PSXSKIP; |
| 902 |
$envparam.= ";GRID_PSXTOTAL=$GRID_PSXTOTAL" if $GRID_PSXTOTAL; |
| 903 |
$envparam.=';BLASTMAT='.$ENV{BLASTMAT} if $ENV{BLASTMAT}; |
| 904 |
$envparam.=';BOWTIE_INDEXES='.$ENV{BOWTIE_INDEXES} if $ENV{BOWTIE_INDEXES}; |
| 905 |
|
| 906 |
$envparam.= ";GRID_LOCAL_JOBDIR=$GRID_LOCAL_JOBDIR/$dprefix\$(Cluster);" if $GRID_LOCAL_JOBDIR; |
| 907 |
if (keys(%$envhash)>0) { |
| 908 |
$envparam.=';'; |
| 909 |
my @envars; |
| 910 |
while (my ($env, $val)= each(%$envhash)) { |
| 911 |
push(@envars, $env.'='.$val); |
| 912 |
} |
| 913 |
$envparam.=join(';',@envars); |
| 914 |
} |
| 915 |
my $logdir=''; |
| 916 |
if ($GRID_LOGDIR) { |
| 917 |
#separate log dir given |
| 918 |
$logdir=$GRID_LOGDIR; |
| 919 |
if (-d $GRID_LOGDIR) { |
| 920 |
print STDERR "Warning: log dir $GRID_LOGDIR exists; existing files will be overwritten!\n"; |
| 921 |
} |
| 922 |
else { |
| 923 |
mkdir($logdir) || die("Error creating log dir '$logdir'!\n"); |
| 924 |
} |
| 925 |
$logdir.='/' unless $logdir=~m/\/$/; |
| 926 |
} |
| 927 |
my $otherflags; |
| 928 |
$otherflags=' -D' if $GRID_DEBUG; |
| 929 |
#$otherflags.=' -N' if $GRID_NOWRKLINKS; |
| 930 |
$otherflags.=' -F' if $GRID_USECMDLIST; |
| 931 |
$otherflags.=' -S' if $SwitchDir; |
| 932 |
|
| 933 |
my $mtime=time(); |
| 934 |
my $cmdfile="condor.$$.t$mtime.$USER.$HOST.cmd"; |
| 935 |
local *CMDFILE; |
| 936 |
open(CMDFILE, '>'.$cmdfile) || die "Cannot create $cmdfile!\n"; |
| 937 |
my $requirements = '(OpSys == "LINUX") && (Arch == "INTEL" || Arch == "x86_64")'; |
| 938 |
my $force_slot; |
| 939 |
if ($UniqVMreq>0) { |
| 940 |
$force_slot=$UniqVMreq; |
| 941 |
} |
| 942 |
elsif ($UniqueVM) { |
| 943 |
$force_slot=3+int(rand(9)); #this is messed up, only works on >11 core machines |
| 944 |
} |
| 945 |
$requirements .= ' && (VirtualMachineId == '.$force_slot.')' if $force_slot; |
| 946 |
if (@xmachinelist>0) { |
| 947 |
#map { $_='Machine != "'.$_.'.'.$DOMAIN.'"' } @xmachinelist; |
| 948 |
#$requirements.= ' && ('.join(' && ',@xmachinelist).')'; |
| 949 |
if (@xmachinelist==1 && ($xmachinelist[0]=~tr/|^?*\\//)) { |
| 950 |
$requirements.=' && (TRUE != regexp("'.$xmachinelist[0].'", Machine, "i"))'; |
| 951 |
} |
| 952 |
else { |
| 953 |
$requirements.=' && (TRUE != regexp("^('.join('|',@xmachinelist). |
| 954 |
')\..*", Machine, "i"))' |
| 955 |
} |
| 956 |
} |
| 957 |
elsif (@ymachinelist>0) { |
| 958 |
if (@ymachinelist==1 && ($ymachinelist[0]=~tr/|^?*\\//)>0) { |
| 959 |
$requirements.=' && regexp("'.$ymachinelist[0].'", Machine, "i")'; |
| 960 |
} |
| 961 |
else { |
| 962 |
#map { $_='Machine == "'.$_.'.'.$DOMAIN.'"' } @ymachinelist; |
| 963 |
#$requirements.= ' && ('.join(' || ',@ymachinelist).')'; |
| 964 |
$requirements.=' && regexp("^('.join('|',@ymachinelist). |
| 965 |
')\..*", Machine, "i")' |
| 966 |
} |
| 967 |
} |
| 968 |
print CMDFILE qq{universe = vanilla |
| 969 |
requirements = $requirements |
| 970 |
notification = Never |
| 971 |
executable = $0 |
| 972 |
initialdir = $PWD |
| 973 |
}; |
| 974 |
if ($logdir) { |
| 975 |
print CMDFILE "error = ${logdir}log_$dprefix\$(Cluster).\$(Process).stderr\n". |
| 976 |
"output = ${logdir}log_$dprefix\$(Cluster).\$(Process).stdout\n"; |
| 977 |
} |
| 978 |
print CMDFILE "arguments = $otherflags -W $GRID_CMD\n". |
| 979 |
"environment = $envparam;\n". |
| 980 |
"$queue\n"; |
| 981 |
close(CMDFILE); |
| 982 |
my $subcmd="condor_submit $cmdfile"; |
| 983 |
my $subout = `$subcmd`; |
| 984 |
($jobid)=($subout=~/submitted\s+to\s+cluster\s+(\d+)\./s); |
| 985 |
die "Error: No Job ID# could be parsed!\n($subout)" unless ($jobid); |
| 986 |
$jobid=$HOST.'_'.$jobid; |
| 987 |
setupJobDir($jobid); |
| 988 |
#setupJobDir also chdirs() in the $GRID_JOBDIR |
| 989 |
system("mv ../$cmdfile condor_submit.cmd"); |
| 990 |
return $jobid; |
| 991 |
} |
| 992 |
|
| 993 |
sub jobDie { |
| 994 |
my $jobid=shift @_; |
| 995 |
print STDERR "Error: ".join("\n",@_)."\n"; |
| 996 |
&$removeJob($jobid); |
| 997 |
die(); |
| 998 |
} |
| 999 |
|
| 1000 |
sub setupJobDir { |
| 1001 |
my ($jobid)=@_; |
| 1002 |
my $jobdir = "gridx-$jobid"; |
| 1003 |
jobDie($jobid, "job directory $jobdir already exists!") |
| 1004 |
if (-d $jobdir); |
| 1005 |
if ($GRID_RESUME) { |
| 1006 |
my $prevjobdir='gridx-'.$GRID_RESUME; |
| 1007 |
print STDERR " ..taking over jobdir: $prevjobdir\n"; |
| 1008 |
unlink("$GRID_MONHOME/$prevjobdir") || warn(" couldn't unlink $GRID_MONHOME/$prevjobdir"); |
| 1009 |
unlink("$prevjobdir/$F_LASTTASK"); |
| 1010 |
unlink("$prevjobdir/$F_ALLDONE"); |
| 1011 |
rename("$prevjobdir/$F_ERRTASKS", "$prevjobdir/prev_$F_ERRTASKS"); |
| 1012 |
unlink("$prevjobdir/$F_RETRYTASKS"); |
| 1013 |
system("/bin/rm -rf $prevjobdir/wrk_*/$F_WRKRUNNING"); |
| 1014 |
system("/bin/rm -rf $prevjobdir/.wrk*"); |
| 1015 |
system("/bin/rm -rf $prevjobdir/locks"); |
| 1016 |
system("/bin/rm -rf $prevjobdir/running"); |
| 1017 |
system("mv $prevjobdir $jobdir") && |
| 1018 |
jobDie($jobid, "cannot 'mv $prevjobdir $jobdir' - $! - Resuming failed!"); |
| 1019 |
unlink("$jobdir/$F_TASKDB.cidx"); |
| 1020 |
} |
| 1021 |
else { |
| 1022 |
mkdir($jobdir) || jobDie($jobid, "cannot create subdirectory $jobdir"); |
| 1023 |
#runCmd("mv $TASKDB $jobdir/taskDb", $jobid); |
| 1024 |
#runCmd("mv $TASKDB.cidx $jobdir/taskDb.cidx", $jobid); |
| 1025 |
} |
| 1026 |
|
| 1027 |
mkdir("$jobdir/locks") || jobDie($jobid, |
| 1028 |
"cannot create subdirectory $jobdir/locks"); |
| 1029 |
mkdir("$jobdir/running") || jobDie($jobid, |
| 1030 |
"cannot create subdirectory $jobdir/running"); |
| 1031 |
if ($GRID_MONHOME ne $PWD) { |
| 1032 |
symlink("$PWD/$jobdir", "$GRID_MONHOME/$jobdir") |
| 1033 |
|| jobDie($jobid, "cannot symlink $GRID_MONHOME/$jobdir"); |
| 1034 |
} |
| 1035 |
#- CHDIR to the new GRID_JOBDIR |
| 1036 |
chdir($jobdir) || jobDie($jobid, "cannot chdir($jobdir) (from $PWD)!"); |
| 1037 |
readFile($F_TASKSDONE); |
| 1038 |
my $cmdfile="cmdline-$GRID_TASKLAST.cmd"; |
| 1039 |
local *FHND; |
| 1040 |
open(FHND, ">$cmdfile") || jobDie($jobid, "cannot create file $cmdfile\n"); |
| 1041 |
print FHND $CMDLINE."\n"; |
| 1042 |
close(FHND); |
| 1043 |
open(FHND, ">$F_WRKDIR"); |
| 1044 |
print FHND "$PWD/$jobdir\n"; |
| 1045 |
close(FHND); |
| 1046 |
|
| 1047 |
if ($mailnotify) { |
| 1048 |
open(FHND, ">$F_NOTIFY"); |
| 1049 |
print FHND "$mailnotify\n"; |
| 1050 |
close(FHND); |
| 1051 |
} |
| 1052 |
|
| 1053 |
if ($gridEndCmd) { |
| 1054 |
open(FHND, ">$F_ENDCMD") || die("Error creating file $jobdir/.$F_ENDCMD ($!)\n"); |
| 1055 |
print FHND $gridEndCmd."\n"; |
| 1056 |
close(FHND); |
| 1057 |
} |
| 1058 |
$GRID_JOBDIR = "$PWD/$jobdir"; |
| 1059 |
print STDOUT "Job $jobid scheduled to run with GRID_JOBDIR = $PWD/$jobdir\n"; |
| 1060 |
|
| 1061 |
} #setupJobDir |
| 1062 |
|
| 1063 |
|
| 1064 |
sub jobSummary { |
| 1065 |
my ($mail) = @_; |
| 1066 |
#assuming we're in GRID_JOBDIR directory |
| 1067 |
#(either directly or by $GRID_MONHOME)! |
| 1068 |
die "Error: cannot locate $F_TASKSDONE and $F_WRKDIR in current directory ($ENV{PWD})!\n" |
| 1069 |
unless -f $F_TASKSDONE && -f $F_WRKDIR; |
| 1070 |
my $tasksdone=readFile($F_TASKSDONE);chomp($tasksdone); |
| 1071 |
$tasksdone=0 unless $tasksdone; |
| 1072 |
my $wrkdir=readFile($F_WRKDIR);chomp($wrkdir); |
| 1073 |
my ($jobid)=($wrkdir=~m/\/gridx\-(\w[\w\-]+)$/); |
| 1074 |
die("Error: cannot parse jobid from $F_WRKDIR content ($wrkdir)\n") unless $jobid; |
| 1075 |
my @cmdfile=<cmdline-*.cmd>; |
| 1076 |
die "Error getting the cmdline-*.cmd from current directory!\n" |
| 1077 |
unless @cmdfile; |
| 1078 |
my ($numtasks)=($cmdfile[0]=~m/cmdline\-(\d+)/); |
| 1079 |
die "Error parsing the number of tasks from $cmdfile[0]!\n" unless $numtasks>0; |
| 1080 |
open(CMDFILE, $cmdfile[0]); |
| 1081 |
my $cmdline=<CMDFILE>; |
| 1082 |
chomp($cmdline); |
| 1083 |
$cmdline=~s/\t/ /g; |
| 1084 |
close(CMDFILE); |
| 1085 |
unlink($F_NOTIFY) if -s $F_NOTIFY; |
| 1086 |
my ($msg, $subj); |
| 1087 |
my $sig=''; |
| 1088 |
if ($mail) { |
| 1089 |
$sig = "\n\n-------------------------\n -= mail sent from $HOST"; |
| 1090 |
$sig.=" (worker $GRID_WORKER)" if $GRID_WORKER; |
| 1091 |
$sig.=" \nWorking directory: $wrkdir]\n" if $wrkdir; |
| 1092 |
$sig.=" \nCommand line: \n $cmdline\n" if $cmdline; |
| 1093 |
$sig.=" =-\n"; |
| 1094 |
} |
| 1095 |
if ($tasksdone!=$numtasks) { |
| 1096 |
$msg="Summary of gridx job $jobid: $tasksdone tasks done out of $numtasks\n". |
| 1097 |
"Check $wrkdir for more details.\n"; |
| 1098 |
$subj="gridx job $jobid (done $tasksdone out of $numtasks)"; |
| 1099 |
} |
| 1100 |
else{ |
| 1101 |
$msg="gridx job $jobid - done all $numtasks tasks\n"; |
| 1102 |
$subj="gridx job $jobid Done (all $numtasks tasks)"; |
| 1103 |
} |
| 1104 |
$msg.=$sig; |
| 1105 |
send_mail( { to=>$mail, subj=>$subj, body=>$msg }) if $mail; |
| 1106 |
return $msg; |
| 1107 |
} |
| 1108 |
|
| 1109 |
|
| 1110 |
|
| 1111 |
############## WORKER SIDE subroutines ##################### |
| 1112 |
|
| 1113 |
#=================== taskDB handling ================= |
| 1114 |
|
| 1115 |
=head2 taskDbStat (taskdb, taskId [,tstatus, cpu#, host, retries, exitcode]) |
| 1116 |
|
| 1117 |
taskDbStat($taskdb, $taskId, [, $taskstatus, $CPUno, |
| 1118 |
$host, $retrycount, $exitcode]) |
| 1119 |
|
| 1120 |
gets/sets the status of a task in an existing taskdb file |
| 1121 |
|
| 1122 |
If only $taskdb and $taskId parameters are given it works |
| 1123 |
as a getter and returns the whole $taskdb entry, either as a raw |
| 1124 |
string or, if wantarray(), as: |
| 1125 |
|
| 1126 |
($taskstatus, $userdata, $CPUno, $host, $lastexitcode, $startTime, $retrycount) |
| 1127 |
|
| 1128 |
..where $startTime is in minutes since the epoch (time/60) |
| 1129 |
|
| 1130 |
If more than 2 parameters are given, it is a setter for the task status |
| 1131 |
and the other task related data. |
| 1132 |
|
| 1133 |
Valid status values: |
| 1134 |
'-' = queued/idle/unprocessed |
| 1135 |
'r' = running (could be a retry) |
| 1136 |
'.' = finished successfully |
| 1137 |
'E' = finished with error ending (after max retries) |
| 1138 |
|
| 1139 |
Internal details: |
| 1140 |
-Each record is assumed locked at the time of writing |
| 1141 |
(no other processes are trying to update the same task record). |
| 1142 |
-A taskdb record format is: |
| 1143 |
|
| 1144 |
>taskId\t{S|R|xxxx|dddd|mmmmmmmm}\t{hostname}[\t<additional data..>] |
| 1145 |
|
| 1146 |
where: |
| 1147 |
S = running status ('-','r','.' or 'E') |
| 1148 |
R = retry counter (0-9) |
| 1149 |
xxxx = last exit code, in hex |
| 1150 |
dddd = last CPU number, in hex (the wrk_<CPU> subdirectory) |
| 1151 |
mmmmmmmm = start minute since the epoch (time/60) in hexadecimal |
| 1152 |
hostname = fixed 25 char length machine name |
| 1153 |
(blank padded as needed) |
| 1154 |
|
| 1155 |
Error protocol for the setter: when given $status is 'E' |
| 1156 |
the retry counter is incremented and the actual status |
| 1157 |
would be updated to 'E' if > $maxRetries, or back to '-' otherwise. |
| 1158 |
|
| 1159 |
=cut |
| 1160 |
|
| 1161 |
sub taskDbStat { |
| 1162 |
my ($taskdb, $entry, $status, $dirno, $machine, $errcount, $exitcode)=@_; |
| 1163 |
my $taskdbidx=$taskdb; |
| 1164 |
$taskdbidx.='.cidx' unless ($taskdb=~s/\.cidx$//); |
| 1165 |
my $tdberr="taskDbStat(".join(',',@_).") Error:"; |
| 1166 |
wrkDie("$tdberr db $taskdb (and/or index) not found!". |
| 1167 |
"(pwd = $ENV{PWD})") unless (-e $taskdb && -e $taskdbidx); |
| 1168 |
wrkDie("$tdberr no entry given!") unless $entry; |
| 1169 |
local *TASKDB; |
| 1170 |
my $dbpos= `cdbyank -a '$entry' -P $taskdbidx`; |
| 1171 |
chomp($dbpos); |
| 1172 |
wrkDie("$tdberr at retrieving pos of entry $entry\n") |
| 1173 |
if ($? || $dbpos!~/^\d+$/); |
| 1174 |
|
| 1175 |
my $openmode = O_RDONLY | O_SYNC ; |
| 1176 |
if ($status) { # setter code: |
| 1177 |
wrkDie("$tdberr invalid update parameters ($status)") |
| 1178 |
if (length($status)>1 || ($dirno && ($dirno>65535 || $dirno<1))); |
| 1179 |
$openmode= O_RDWR | O_SYNC; # | O_DIRECT might be needed ? |
| 1180 |
} |
| 1181 |
sysopen(TASKDB, $taskdb, $openmode) || |
| 1182 |
wrkDie("$tdberr sysopening $taskdb failed!"); |
| 1183 |
binmode(TASKDB); |
| 1184 |
unless (sysseek(TASKDB, $dbpos, SEEK_SET)) { |
| 1185 |
close(TASKDB); |
| 1186 |
wrkDie("$tdberr at sysseek() to $dbpos for $entry"); |
| 1187 |
} |
| 1188 |
#----- read the next line and check the format |
| 1189 |
my $targetstr='>'.$entry."\t{-|-|----|----|--------}\t{".('-' x 25)."}"; |
| 1190 |
local $/="\n"; |
| 1191 |
my $dbline=readline(*TASKDB); |
| 1192 |
binmode(TASKDB); |
| 1193 |
my ($pentry, $pstats, $pmachine, $userdata)= |
| 1194 |
split(/\t/,$dbline,4); |
| 1195 |
chomp($pmachine);chomp($userdata); |
| 1196 |
my $tcheck=join("\t",$pentry,$pstats,$pmachine); |
| 1197 |
if ($pentry ne '>'.$entry || |
| 1198 |
length($tcheck)!=length($targetstr)) { |
| 1199 |
close(TASKDB); |
| 1200 |
wrkDie("$tdberr invalid record format for '$entry' ". |
| 1201 |
"pos $dbpos, Found:\n'$tcheck'\n ..instead of:\n'$targetstr'\n"); |
| 1202 |
} |
| 1203 |
|
| 1204 |
$pmachine=~tr/{} //d; |
| 1205 |
$pstats=~tr/{}//d; |
| 1206 |
my ($pstatus, $pfcount, $pxcode, $pdirno, $ptime)=split(/\|/,$pstats); |
| 1207 |
$ptime=hex($ptime) || $ptime; |
| 1208 |
$pfcount=0 unless $pfcount>0; |
| 1209 |
#---- |
| 1210 |
if ($status) { #---- setter code: |
| 1211 |
$status=lc($status); |
| 1212 |
if ($status!~/^[\-r\.e]$/) { |
| 1213 |
close(TASKDB); |
| 1214 |
wrkDie("$tdberr invalid status provided ($status)!"); |
| 1215 |
} |
| 1216 |
$dbpos+=length(">$entry\t{"); |
| 1217 |
sysseek(TASKDB, $dbpos, SEEK_SET); |
| 1218 |
if (defined($exitcode)) { |
| 1219 |
$pxcode = $exitcode>0 ? sprintf('%04x', $exitcode) : $pxcode; |
| 1220 |
} |
| 1221 |
|
| 1222 |
$pfcount=$errcount if defined($errcount); |
| 1223 |
if ($status eq 'r') { #mark this entry as "running" |
| 1224 |
$ptime=sprintf('%08x',int(time()/60)); |
| 1225 |
$pxcode='----'; |
| 1226 |
} |
| 1227 |
elsif ($status eq '-') { # mark this entry as "available" (idle) |
| 1228 |
$ptime='--------'; |
| 1229 |
} |
| 1230 |
$pmachine=$machine if $machine; |
| 1231 |
$pmachine=sprintf('%25s',$pmachine); |
| 1232 |
$dirno=$pdirno unless $dirno>=1; |
| 1233 |
my $wstr=$status.'|'.int($pfcount). |
| 1234 |
'|'.$pxcode. |
| 1235 |
'|'.sprintf('%04x', $dirno). |
| 1236 |
'|'.$ptime."}\t{".$pmachine.'}'; |
| 1237 |
if (length($ptime)>8) { |
| 1238 |
die("Error writing into taskDb record, ptime='$ptime' is too long when writing:\n". |
| 1239 |
"$wstr\n"); |
| 1240 |
} |
| 1241 |
my $wlen=length($wstr); |
| 1242 |
my $w=syswrite(TASKDB, $wstr, $wlen); |
| 1243 |
binmode(TASKDB); # 'cause this flushes any pending I/O buffers |
| 1244 |
if ($w!=$wlen) { |
| 1245 |
close(TASKDB); |
| 1246 |
wrkDie("$tdberr failed writing '$wstr' for $entry!"); |
| 1247 |
} |
| 1248 |
close(TASKDB); |
| 1249 |
return 1; |
| 1250 |
} # setter |
| 1251 |
else { # getter code |
| 1252 |
close(TASKDB); |
| 1253 |
if (wantarray()) { #retrieve the parsed list of values |
| 1254 |
return ($pstatus, $userdata, hex($pdirno), $pmachine, int($pfcount), |
| 1255 |
hex($pxcode), $ptime); |
| 1256 |
} |
| 1257 |
else { #return the raw taskDb line |
| 1258 |
return($dbline); |
| 1259 |
} |
| 1260 |
} # getter |
| 1261 |
} |
| 1262 |
|
| 1263 |
|
| 1264 |
sub gridWorkerEnv { |
| 1265 |
if ($_[0]) { |
| 1266 |
$GRID_TASK=$_[0]; |
| 1267 |
$ENV{GRID_TASK}=$_[0]; |
| 1268 |
} |
| 1269 |
return if $GRID_ENVSET; |
| 1270 |
my $gridengine=lc($ENV{GRID_ENGINE}); |
| 1271 |
( $GRID_JOBDIR, $GRID_TASKLAST, $GRID_RESUME, $GRID_MONHOME, $GRID_LOCAL_JOBDIR, |
| 1272 |
$GRID_PSXFASTA, $GRID_PSXSTEP, $GRID_PSXSKIP, $GRID_PSXTOTAL, $GRID_DIRPREFIX, )= |
| 1273 |
@ENV{'GRID_JOBDIR','GRID_TASKLAST','GRID_RESUME','GRID_MONHOME', 'GRID_LOCAL_JOBDIR', |
| 1274 |
'GRID_PSXFASTA','GRID_PSXSTEP','GRID_PSXSKIP','GRID_PSXTOTAL', 'GRID_DIRPREFIX'}; |
| 1275 |
if ($gridengine eq 'pbs') { |
| 1276 |
#can't have dynamic environment variables in SGE/PBS |
| 1277 |
$GRID_JOB=$ENV{PBS_JOBID}; |
| 1278 |
#only static ones have been prepared (GRID_ENGINE, GRID_TASKLAST) |
| 1279 |
# with GRID_JOBDIR initially set to the submit working directory |
| 1280 |
$GRID_JOBDIR.='/gridx-'.$GRID_JOB; |
| 1281 |
|
| 1282 |
$GRID_LOCAL_JOBDIR.='/gridx-'.$GRID_JOB if $GRID_LOCAL_JOBDIR; |
| 1283 |
$GRID_PSXFASTA=$ENV{GRID_PSXFASTA}; |
| 1284 |
$GRID_PSXSTEP=$ENV{GRID_PSXSTEP}; |
| 1285 |
#dynamic ones are built now from SGE ones |
| 1286 |
$ENV{GRID_JOBDIR}=$GRID_JOBDIR; |
| 1287 |
$ENV{GRID_JOB}=$GRID_JOB; |
| 1288 |
$ENV{GRID_LOCAL_JOBDIR}=$GRID_LOCAL_JOBDIR; |
| 1289 |
} |
| 1290 |
elsif ($gridengine eq 'condor' || $gridengine eq 'smp') { |
| 1291 |
#condor should have all the environment in order |
| 1292 |
$GRID_JOB=$ENV{GRID_JOB}; |
| 1293 |
} |
| 1294 |
else { |
| 1295 |
die("Error: Invalid GRID_ENGINE (Is this a valid worker run?)\n"); |
| 1296 |
} |
| 1297 |
$GRID_LOCKDIR=$GRID_JOBDIR.'/locks'; |
| 1298 |
$GRID_ENVSET=1; |
| 1299 |
} |
| 1300 |
|
| 1301 |
sub beginWorker { |
| 1302 |
gridWorkerEnv(); #setup the worker environment |
| 1303 |
my $maxretries=20; #just give some time for the submit script to catch up |
| 1304 |
my $chdir=0; # (only if execution of a submitted task is extraordinarily fast) |
| 1305 |
my $retries=0; |
| 1306 |
while (!($chdir=chdir($GRID_JOBDIR))) { |
| 1307 |
sleep(1); |
| 1308 |
$retries++; |
| 1309 |
last if $retries==$maxretries; |
| 1310 |
} |
| 1311 |
die("Worker error: cannot chdir to $GRID_JOBDIR") unless $chdir; |
| 1312 |
# -- we are in GRID_JOBDIR now |
| 1313 |
my $errmsg="Worker PID $$ ($GRID_WORKER) failed on $HOST..\n"; |
| 1314 |
if ($GRID_WORKER>1) { |
| 1315 |
while (!-f $F_TASKDB.'.cidx') { |
| 1316 |
sleep(1); #wait for the first worker to build taskDB |
| 1317 |
} |
| 1318 |
} |
| 1319 |
my $fh=setXLock($F_WRKSTART,120) || die $errmsg; #update the worker counter |
| 1320 |
if ($GRID_WORKER==1 && !-f $F_TASKDB.'.cidx') { |
| 1321 |
print STDERR "..preparing $F_TASKDB\n"; |
| 1322 |
prepareTaskDb(); #builds a taskDb in the current directory |
| 1323 |
#checks global $GRID_PSXFASTA and $GRID_PSXSTEP |
| 1324 |
#sets TASKDB to the file name, and GRID_TASKLAST is updated as needed |
| 1325 |
if ($gridBeginCmd) { |
| 1326 |
system($gridBeginCmd) && die("Error: exit status $? returned by prologue command: '$gridBeginCmd'\n"); |
| 1327 |
} |
| 1328 |
} |
| 1329 |
|
| 1330 |
$GRID_WORKER=incFValue($fh, $F_WRKSTART); |
| 1331 |
endXLock($fh); |
| 1332 |
print STDERR "D: worker $GRID_WORKER assigned to host $HOST pid $$ at ".getTime()."\n" if $GRID_DEBUG; |
| 1333 |
$GRID_WRKDIR=sprintf("wrk_%04d",$GRID_WORKER); |
| 1334 |
$starting_dir=$GRID_JOBDIR; |
| 1335 |
$starting_dir=~s/[^\/]+\/?$//; |
| 1336 |
if ($GRID_LOCAL_JOBDIR) { |
| 1337 |
print STDERR "D: creating local dir on $HOST: $GRID_LOCAL_JOBDIR/$GRID_WRKDIR\n" if $GRID_DEBUG; |
| 1338 |
mkdir("$GRID_LOCAL_JOBDIR") unless -d $GRID_LOCAL_JOBDIR; |
| 1339 |
system("/bin/rm -f $GRID_LOCAL_JOBDIR/$GRID_WRKDIR"); |
| 1340 |
mkdir("$GRID_LOCAL_JOBDIR/$GRID_WRKDIR") unless -d $GRID_LOCAL_JOBDIR; |
| 1341 |
wrkDie("Error: couldn't create local worker directory $GRID_LOCAL_JOBDIR/$GRID_WRKDIR on $HOST!\n") |
| 1342 |
unless (-d "$GRID_LOCAL_JOBDIR/$GRID_WRKDIR"); |
| 1343 |
} |
| 1344 |
#-- also updates the "currently running" counter |
| 1345 |
unless (-d "$GRID_JOBDIR/$GRID_WRKDIR") { #worker directory doesn't exist |
| 1346 |
unless (mkdir("$GRID_JOBDIR/$GRID_WRKDIR")) { |
| 1347 |
die "Error at mkdir $GRID_JOBDIR/$GRID_WRKDIR ($!)!\n"; |
| 1348 |
} |
| 1349 |
} |
| 1350 |
else { #existing working directory |
| 1351 |
my $frunning= "$GRID_WRKDIR/$F_WRKRUNNING"; |
| 1352 |
if (-f $frunning) { #weird -- should never happen.. |
| 1353 |
die "Error: another process running in $GRID_WRKDIR?!\n".`cat $frunning`."$errmsg\n"; |
| 1354 |
} |
| 1355 |
else { #normal case: no worker-running semaphore there already |
| 1356 |
local *RSEM; |
| 1357 |
open(RSEM, '>'.$frunning) || die "Error creating $frunning file! ($!)\n$errmsg"; |
| 1358 |
print RSEM join(" ",int(time()/60), $HOST, $$)."\n"; |
| 1359 |
close(RSEM); |
| 1360 |
} |
| 1361 |
} |
| 1362 |
if ($GRID_DIRPREFIX) { |
| 1363 |
my $gwrkdir="../$GRID_DIRPREFIX".'_'.$GRID_WORKER; |
| 1364 |
unlink($gwrkdir); |
| 1365 |
symlink("gridx-$GRID_JOB/$GRID_WRKDIR", $gwrkdir) || |
| 1366 |
print STDERR "Warning: cannot symlink gridx-$GRID_JOB/$GRID_WRKDIR to $gwrkdir ($!)\n"; |
| 1367 |
#needed by PSX emulation |
| 1368 |
} |
| 1369 |
# we are in GRID_JOBDIR now - descend into wrk_<GRID_WORKER> |
| 1370 |
$fh=setXLock($F_WRKCOUNT,70,3) || die $errmsg; # update the count of running workers |
| 1371 |
incFValue($fh, $F_WRKCOUNT); |
| 1372 |
endXLock($fh); |
| 1373 |
chdir("$GRID_JOBDIR/$GRID_WRKDIR") || |
| 1374 |
die "Error at chdir($GRID_JOBDIR/$GRID_WRKDIR) ($!)\n"; |
| 1375 |
|
| 1376 |
open(STDERR, ">wrk_err.log"); |
| 1377 |
print STDERR "worker $GRID_WORKER fully assigned to host $HOST PID $$\n" if $GRID_DEBUG; |
| 1378 |
open(STDOUT, ">wrk_log.log"); |
| 1379 |
local *WRKGRAB; |
| 1380 |
open(WRKGRAB, ">.on_$HOST"); |
| 1381 |
print WRKGRAB join("\t",$GRID_WRKDIR, $HOST, $$, 'start: '.getTime())."\n"; |
| 1382 |
close(WRKGRAB); |
| 1383 |
$ENV{GRID_WORKER}=$GRID_WORKER; |
| 1384 |
if ($SwitchDir) { |
| 1385 |
chdir($starting_dir) || die "Error at chdir($starting_dir)!\n"; |
| 1386 |
} |
| 1387 |
elsif ($GRID_LOCAL_JOBDIR) { |
| 1388 |
chdir($GRID_LOCAL_JOBDIR/$GRID_WRKDIR) || die "Error at chdir($starting_dir)!\n"; |
| 1389 |
} |
| 1390 |
} |
| 1391 |
|
| 1392 |
sub endWorker { |
| 1393 |
return -1 unless $GRID_WORKER && $GRID_WRKDIR; |
| 1394 |
#make sure we are back in the wrk_<GRID_WORKER> directory |
| 1395 |
unless (chdir("$GRID_JOBDIR/$GRID_WRKDIR")) { |
| 1396 |
die("ERROR: endWorker() could not change to $GRID_JOBDIR/$GRID_WRKDIR\n"); |
| 1397 |
} |
| 1398 |
|
| 1399 |
my $fh=setXLock($F_WRKCOUNT,70, 3) || die "Error updating the number of running workers!\n"; |
| 1400 |
# my $v=readFile($fh,0,$F_WRKCOUNT);chomp($v); |
| 1401 |
my $v=incFValue($fh, $F_WRKCOUNT, -1); |
| 1402 |
unlink($F_WRKRUNNING); #remove the "worker here" semaphore.. |
| 1403 |
if ($v<0) { |
| 1404 |
endXLock($fh); |
| 1405 |
die("Error: invalid number of workers ($v) reported in $GRID_JOBDIR/$F_WRKCOUNT\n") |
| 1406 |
} |
| 1407 |
endXLock($fh); |
| 1408 |
print STDERR join("\t","D. worker $GRID_WRKDIR", $HOST, $$, 'finished at:',getTime())."\n" if $GRID_DEBUG; |
| 1409 |
|
| 1410 |
local *WRKGRAB; |
| 1411 |
sysopen(WRKGRAB, ".on_$HOST", O_RDWR | O_SYNC ); |
| 1412 |
sysseek(WRKGRAB,-1,SEEK_END); |
| 1413 |
print WRKGRAB "\tend: ".getTime()."\n"; |
| 1414 |
close(WRKGRAB); |
| 1415 |
if ($GRID_LOCAL_JOBDIR) { |
| 1416 |
my $r=system("cp -pr $GRID_LOCAL_JOBDIR/$GRID_WRKDIR/* $GRID_JOBDIR/$GRID_WRKDIR/"); |
| 1417 |
if ($r) { |
| 1418 |
print STDERR "Error at copying $HOST local files back to $GRID_JOBDIR/$GRID_WRKDIR!\n"; |
| 1419 |
} |
| 1420 |
system("/bin/rm -rf $GRID_LOCAL_JOBDIR/$GRID_WRKDIR"); |
| 1421 |
rmdir($GRID_LOCAL_JOBDIR); #it'll fail if there are other subdirs there, but that's OK |
| 1422 |
} |
| 1423 |
undef($GRID_WORKER); |
| 1424 |
undef($GRID_WRKDIR); |
| 1425 |
return $v; #returns the number of workers left running |
| 1426 |
} |
| 1427 |
|
| 1428 |
sub writeFValue { |
| 1429 |
my ($fh, $v)=@_; |
| 1430 |
truncate($fh, 0); |
| 1431 |
seek($fh,0,SEEK_SET); |
| 1432 |
$v=int($v); |
| 1433 |
print $fh $v."\n"; |
| 1434 |
return $v; |
| 1435 |
} |
| 1436 |
|
| 1437 |
sub writeFList { #with truncate |
| 1438 |
my ($fh, $listref, $delim)=@_; |
| 1439 |
$delim='' unless $delim; |
| 1440 |
seek($fh,0,SEEK_SET); |
| 1441 |
truncate($fh,0); |
| 1442 |
print $fh join($delim,@$listref); |
| 1443 |
#truncate($fh, tell($fh)); |
| 1444 |
} |
| 1445 |
# |
| 1446 |
# incFValue(fhandle) => reads an int value from fhandle |
| 1447 |
# increments it, writes it back |
| 1448 |
# and returns it |
| 1449 |
# |
| 1450 |
sub incFValue { |
| 1451 |
my ($fh, $finfo, $incv)=@_; |
| 1452 |
seek($fh, 0, SEEK_SET); |
| 1453 |
$incv=1 unless $incv; |
| 1454 |
my $v=readFile($fh, 0, "incFValue(,,$incv) in $finfo"); |
| 1455 |
chomp($v); |
| 1456 |
return writeFValue($fh, int($v)+$incv); |
| 1457 |
} |
| 1458 |
|
| 1459 |
|
| 1460 |
#******************** worker side: |
| 1461 |
# getNextTask() -- returns a taskId for the next task to be processed |
| 1462 |
# or 0 if no more |
| 1463 |
# -ignore any entries which are already "done" |
| 1464 |
# -look first into the "retry pool" ($F_RETRYTASKS) to get some tasks from there, |
| 1465 |
# if any -- and removes that entry |
| 1466 |
#--------------- resources used: |
| 1467 |
# update $F_RETRYTASKS |
| 1468 |
# update $F_LASTTASK |
| 1469 |
# update ENV{GRID_TASK} and $GRID_TASK |
| 1470 |
# |
| 1471 |
sub getNextTask { |
| 1472 |
#check for any tasks in the "retry" queue |
| 1473 |
my ($taskID, $retries); |
| 1474 |
my $sourcemsg; |
| 1475 |
SKIP_DONE: |
| 1476 |
$retries=0; |
| 1477 |
if (-s "$GRID_JOBDIR/$F_RETRYTASKS") { |
| 1478 |
my $hr=setXLock($F_RETRYTASKS,120,5) |
| 1479 |
|| wrkDie("Error locking $F_RETRYTASKS ($HOST, $GRID_WORKER)"); |
| 1480 |
my @errstack=readFile($hr,0,$F_RETRYTASKS); #format: taskid <space> #retries |
| 1481 |
($taskID, $retries)=split(/\s+/, shift(@errstack)); #fetch last |
| 1482 |
if ($taskID>0) { #valid taskID to retry |
| 1483 |
chomp($retries); |
| 1484 |
writeFList($hr, \@errstack); #write the list back |
| 1485 |
$sourcemsg=' from the retry pool.'; |
| 1486 |
} |
| 1487 |
endXLock($hr); |
| 1488 |
} |
| 1489 |
NEXT_TASK: |
| 1490 |
unless ($taskID) { #getting the next available task |
| 1491 |
#no retry tasks, so get the next task not processed yet |
| 1492 |
my $fh=setXLock($F_LASTTASK, 70, 3) || |
| 1493 |
wrkDie("Error locking $F_LASTTASK ($HOST, $GRID_WORKER)"); #30 /retries |
| 1494 |
my $last=readFile($fh, 0, $F_LASTTASK);chomp($last); |
| 1495 |
if ($last<$GRID_TASKLAST) { # valid one, take the next |
| 1496 |
$taskID=writeFValue($fh,int($last)+1); |
| 1497 |
} |
| 1498 |
$sourcemsg=''; |
| 1499 |
endXLock($fh); |
| 1500 |
} |
| 1501 |
return undef unless $taskID; |
| 1502 |
$GRID_TASK=$taskID; |
| 1503 |
# lock this taskID |
| 1504 |
$TASK_LOCKF="$GRID_JOBDIR/running/task_$taskID"; |
| 1505 |
#$TASK_LOCKF="running/task_$taskID"; |
| 1506 |
catchSigs(1); #install signal handler |
| 1507 |
$TASK_LOCKH=setXLock($TASK_LOCKF, 70, 3); |
| 1508 |
unless ($TASK_LOCKH) { #this SHOULD be available immediately.. |
| 1509 |
my $lockedby=`cat $TASK_LOCKF`;chomp($lockedby); |
| 1510 |
print STDERR "WARNING: lock-fail on task $taskID $sourcemsg (previously locked in $TASK_LOCKF by [$lockedby])\n"; |
| 1511 |
#wrkDie("Error: couldn't get a lock on task $taskID..\n"); |
| 1512 |
undef($taskID); |
| 1513 |
goto SKIP_DONE; # don't kill the worker, just move on |
| 1514 |
} |
| 1515 |
print $TASK_LOCKH "$HOST $$ ".sprintf("wrk_%04d",$GRID_WORKER)."\n"; |
| 1516 |
# check the status of this task: |
| 1517 |
my ($tstatus, $tuserdata, $tdirno, $thost, $terrcount, |
| 1518 |
$texcode, $tstartmin)=taskDbStat("$GRID_JOBDIR/$F_TASKDB.cidx", $taskID); |
| 1519 |
print STDERR ">task-$taskID assigned to worker $GRID_WORKER (on $HOST) $sourcemsg\n"; |
| 1520 |
if ($GRID_RESUME && $tstatus eq '.') { |
| 1521 |
#skip this one, it's finished (according to taskDb!) |
| 1522 |
print STDERR ">SKIP-done:$taskID \{$tstatus|$terrcount|$texcode|$tdirno|$tstartmin\}\t\{$thost\}\n"; |
| 1523 |
undef $taskID; |
| 1524 |
undef $GRID_TASK; |
| 1525 |
endXLock($TASK_LOCKH); |
| 1526 |
catchSigs(0); |
| 1527 |
unlink($TASK_LOCKF); |
| 1528 |
undef($TASK_LOCKH);undef($TASK_LOCKF); |
| 1529 |
goto SKIP_DONE; |
| 1530 |
} |
| 1531 |
#update status of this task to 'running' |
| 1532 |
$TASK_ERRCOUNT=$retries; |
| 1533 |
taskDbStat("$GRID_JOBDIR/$F_TASKDB.cidx", $taskID, 'r', $GRID_WORKER, $HOST, $TASK_ERRCOUNT); |
| 1534 |
#-- |
| 1535 |
$GRID_TASK=$taskID; |
| 1536 |
$TASK_DATA=$tuserdata; |
| 1537 |
$ENV{'GRID_TASK'}=$taskID; |
| 1538 |
$STARTED_GRID_TASK=$taskID; |
| 1539 |
return $taskID; |
| 1540 |
} |
| 1541 |
|
| 1542 |
############################################## |
| 1543 |
# runTask($taskID, @cmd) |
| 1544 |
#------------------------ |
| 1545 |
# *runs into a ./wrk_NNNN subdirectory of GRID_JOBDIR |
| 1546 |
# *employs $GRID_TASK, $TASK_DATA, $TASK_ERRCOUNT |
| 1547 |
# *with GRID_PSXFASTA, it uses TASK_DATA and other GRID_PSX.. |
| 1548 |
# to prepare the fasta slice and pass it to the cmd |
| 1549 |
# *on exit, it should: |
| 1550 |
# - set a lock on the ../running/task_$taskID file |
| 1551 |
# - IF error exit of cmd (non-zero exit status), use $TASK_ERRCOUNT+1 and $MAX_RETRIES |
| 1552 |
# to determine if the job should be put in the $F_RETRYTASKS file |
| 1553 |
# or if status should be set to 'E' in taskDb and the entry added |
| 1554 |
# to $F_ERRTASKS |
| 1555 |
# - IF successful exit : increment $F_TASKSDONE |
| 1556 |
# - update taskDb with the current status |
| 1557 |
# - remove the lock on ../running/task_$taskID and delete this file! |
| 1558 |
# |
| 1559 |
#--------------------------------------------- |
| 1560 |
sub runTask { |
| 1561 |
my $taskID=shift(@_); |
| 1562 |
my @cmd=@_; |
| 1563 |
my $exitstatus; |
| 1564 |
my $runcmd; #the actuall command run using system(); |
| 1565 |
|
| 1566 |
catchSigs(1); #install signal handler (should have been done already in getNextTask()) |
| 1567 |
if ($GRID_LOCAL_JOBDIR) { |
| 1568 |
wrkDie("Fatal: cannot chdir to local dir $GRID_LOCAL_JOBDIR/$GRID_WRKDIR!") |
| 1569 |
unless chdir("$GRID_LOCAL_JOBDIR/$GRID_WRKDIR"); |
| 1570 |
} |
| 1571 |
|
| 1572 |
#if ($SwitchDir) { |
| 1573 |
# we are currently in a wrk_* directory |
| 1574 |
# chdir('../..'); #change to the original directory (where gridx was launched from) |
| 1575 |
# } |
| 1576 |
if ($GRID_PSXFASTA) { #psx emulation |
| 1577 |
#prepare the fasta slice here |
| 1578 |
my $fslice=sprintf('%s.slice-%08d',getFName($GRID_PSXFASTA), $GRID_TASK); |
| 1579 |
#-- write the slicefile |
| 1580 |
local *FDB; |
| 1581 |
local *FSL; |
| 1582 |
open(FSL, '>'.$fslice) |
| 1583 |
|| wrkDie("Cannot create fasta slice $fslice"); |
| 1584 |
open(FDB, $GRID_PSXFASTA) || wrkDie("Cannot open fasta db $GRID_PSXFASTA"); |
| 1585 |
seek(FDB, $TASK_DATA, SEEK_SET); |
| 1586 |
local $/="\n"; |
| 1587 |
my $seqnext=$GRID_PSXSTEP*($GRID_TASK-1); #+GRID_PSXSKIP |
| 1588 |
my $seqcount=0; |
| 1589 |
my $seqmax=($GRID_PSXTOTAL<=0) ? 0 : $GRID_PSXTOTAL; |
| 1590 |
while (<FDB>) { |
| 1591 |
if (/^>/) { #record start |
| 1592 |
$seqnext++; |
| 1593 |
last if ($seqcount>=$GRID_PSXSTEP); |
| 1594 |
last if $seqmax && ($seqnext>$seqmax); |
| 1595 |
$seqcount++; |
| 1596 |
} |
| 1597 |
print FSL $_; |
| 1598 |
}#while input line from fastadb |
| 1599 |
close(FSL); |
| 1600 |
close(FDB); |
| 1601 |
|
| 1602 |
$runcmd=shift(@cmd); |
| 1603 |
$GRID_PSXSKIP=0 unless ($GRID_PSXSKIP); |
| 1604 |
$GRID_PSXTOTAL=-1 unless ($GRID_PSXTOTAL>0); |
| 1605 |
my $islast=($GRID_TASK==$GRID_TASKLAST) ? 1 : 0; |
| 1606 |
# 1 2 3 4 |
| 1607 |
$runcmd.=' '.join(' ',$fslice, $seqcount, $GRID_TASK, $islast, |
| 1608 |
# 5 6 7 |
| 1609 |
$GRID_PSXSKIP, $GRID_PSXTOTAL, @cmd); |
| 1610 |
|
| 1611 |
} |
| 1612 |
elsif ($GRID_USECMDLIST) { |
| 1613 |
# $TASK_DATA is the actual command to run |
| 1614 |
$runcmd=$TASK_DATA; |
| 1615 |
} |
| 1616 |
else { #normal repeat cmd -- let the cmd use the ENV accordingly |
| 1617 |
$runcmd=join(" ",@cmd); |
| 1618 |
} |
| 1619 |
|
| 1620 |
#... get $exitstatus for the system() call |
| 1621 |
print STDERR ">starting-task-$GRID_TASK by worker $GRID_WORKER (on $HOST): '$runcmd'\n" if $GRID_DEBUG; |
| 1622 |
$exitstatus=system($runcmd); |
| 1623 |
if ($SwitchDir) { |
| 1624 |
chdir("$GRID_JOBDIR/$GRID_WRKDIR"); |
| 1625 |
} |
| 1626 |
endTask($exitstatus); #taskID is taken from $GRID_TASK |
| 1627 |
} |
| 1628 |
|
| 1629 |
|
| 1630 |
sub catchSigs { # true/false |
| 1631 |
if ($_[0]) { |
| 1632 |
$SIG{INT}=\&sigHandler; |
| 1633 |
$SIG{TERM}=\&sigHandler; |
| 1634 |
} |
| 1635 |
else { |
| 1636 |
$SIG{INT}='DEFAULT'; |
| 1637 |
$SIG{TERM}='DEFAULT'; |
| 1638 |
} |
| 1639 |
} |
| 1640 |
|
| 1641 |
sub sigHandler { |
| 1642 |
my $signame=shift; |
| 1643 |
wrkDie("Signal $signame caught for worker $$ on $HOST, aborting..\n"); |
| 1644 |
} |
| 1645 |
|
| 1646 |
sub toRetry { |
| 1647 |
my ($taskID, $taskErrCount)=@_; |
| 1648 |
return unless $taskID; |
| 1649 |
$taskErrCount=1 unless $taskErrCount; |
| 1650 |
my $hr=setXLock($F_RETRYTASKS, 70, 3); |
| 1651 |
while (<$hr>) { |
| 1652 |
chomp; |
| 1653 |
my ($tid, $tec)=split(/\s+/); |
| 1654 |
if ($tid==$taskID) { #double retry, don't bother |
| 1655 |
endXLock($hr); |
| 1656 |
return; |
| 1657 |
} |
| 1658 |
} |
| 1659 |
fappend($hr, "$taskID\t$taskErrCount\n"); |
| 1660 |
endXLock($hr); |
| 1661 |
} |
| 1662 |
|
| 1663 |
sub endTask { |
| 1664 |
return unless $GRID_TASK; |
| 1665 |
# we MUST be in GRID_WRKDIR |
| 1666 |
chdir("$GRID_JOBDIR/$GRID_WRKDIR") || die("Error: failed to chdir to $GRID_JOBDIR/$GRID_WRKDIR!"); |
| 1667 |
my ($exitstatus)=@_; |
| 1668 |
|
| 1669 |
if (!$exitstatus && !$STARTED_GRID_TASK) { # could be a premature failure, like cdbyank not found, etc. |
| 1670 |
print STDERR "scheduling $GRID_TASK for retry..\n"; |
| 1671 |
toRetry($GRID_TASK, $TASK_ERRCOUNT); |
| 1672 |
endXLock($TASK_LOCKH) if $TASK_LOCKH; |
| 1673 |
unlink($TASK_LOCKF); |
| 1674 |
undef($TASK_LOCKH);undef($TASK_LOCKF); |
| 1675 |
unlink($F_WRKRUNNING); |
| 1676 |
catchSigs(0); |
| 1677 |
undef $GRID_TASK; |
| 1678 |
return; |
| 1679 |
} |
| 1680 |
my $dbstatus; |
| 1681 |
if (defined($exitstatus)) { |
| 1682 |
if ($exitstatus==0) { #success |
| 1683 |
$dbstatus='.'; |
| 1684 |
# update $F_TASKSDONE |
| 1685 |
my $fh=setXLock($F_TASKSDONE, 110, 5); |
| 1686 |
if ($fh) { |
| 1687 |
incFValue($fh, $F_TASKSDONE); |
| 1688 |
endXLock($fh); |
| 1689 |
} |
| 1690 |
} |
| 1691 |
else { #error status |
| 1692 |
$TASK_ERRCOUNT++; |
| 1693 |
print STDERR "task $GRID_TASK failed on $HOST (PID=$$, status='$exitstatus') (worker $GRID_WORKER); error count=$TASK_ERRCOUNT\n" if $GRID_DEBUG; |
| 1694 |
if ($TASK_ERRCOUNT>$MAX_RETRIES) { #trash it |
| 1695 |
my $he=setXLock($F_ERRTASKS, 70, 3); |
| 1696 |
fappend($he, join("\t",$GRID_TASK,$exitstatus,$HOST,$GRID_WRKDIR)."\n"); |
| 1697 |
endXLock($he); |
| 1698 |
$dbstatus='E'; |
| 1699 |
} |
| 1700 |
else {#give it another retry chance |
| 1701 |
toRetry($GRID_TASK, $TASK_ERRCOUNT); |
| 1702 |
$dbstatus='-'; |
| 1703 |
} |
| 1704 |
} |
| 1705 |
endXLock($TASK_LOCKH) if $TASK_LOCKH; |
| 1706 |
unlink($TASK_LOCKF) if $TASK_LOCKF; |
| 1707 |
undef($TASK_LOCKH);undef($TASK_LOCKF); |
| 1708 |
print STDERR "]task $GRID_TASK ended - updating taskDb with status code '$dbstatus'\n" if $GRID_DEBUG; |
| 1709 |
taskDbStat("$GRID_JOBDIR/$F_TASKDB.cidx", $GRID_TASK, $dbstatus, |
| 1710 |
$GRID_WORKER, $HOST, $TASK_ERRCOUNT, $exitstatus); |
| 1711 |
} # defined $exitstatus |
| 1712 |
else { #no exitstatus given, it was an interrupt signal or otherwise failed task |
| 1713 |
print STDERR "task $GRID_TASK ended with undefined exit status\n" if $GRID_DEBUG; |
| 1714 |
toRetry($GRID_TASK, $TASK_ERRCOUNT); |
| 1715 |
endXLock($TASK_LOCKH) if $TASK_LOCKH; |
| 1716 |
unlink($TASK_LOCKF) if $TASK_LOCKF; |
| 1717 |
undef($TASK_LOCKH);undef($TASK_LOCKF); |
| 1718 |
unlink($F_WRKRUNNING); |
| 1719 |
} |
| 1720 |
catchSigs(0); |
| 1721 |
undef($STARTED_GRID_TASK); |
| 1722 |
} |
| 1723 |
|
| 1724 |
sub wrkDie { |
| 1725 |
my ($msg)=@_; |
| 1726 |
endTask(); |
| 1727 |
my $grdwrk=$GRID_WORKER; |
| 1728 |
endWorker(); |
| 1729 |
#remove any other locks left in this worker.. |
| 1730 |
die("Error at worker $grdwrk on $HOST:\n$msg\n"); |
| 1731 |
} |
| 1732 |
|
| 1733 |
#--onExit, onEnd trigger |
| 1734 |
sub END { |
| 1735 |
unless ($NORMAL_ENDING) { |
| 1736 |
$NORMAL_ENDING=1; #to avoid recursion? |
| 1737 |
endTask(); |
| 1738 |
endWorker(); |
| 1739 |
#remove all locks |
| 1740 |
my @locks=keys(%Locks); |
| 1741 |
foreach my $lock (@locks) { |
| 1742 |
my $d=$Locks{$lock}; |
| 1743 |
endXLock($lock); |
| 1744 |
} |
| 1745 |
} |
| 1746 |
} |
| 1747 |
|
| 1748 |
sub fappend { |
| 1749 |
my $fh=shift(@_); |
| 1750 |
seek($fh,0,SEEK_END); |
| 1751 |
print $fh join("\n",@_); |
| 1752 |
} |
| 1753 |
|
| 1754 |
sub getLockDir { |
| 1755 |
my $fname=shift; |
| 1756 |
my $basename=basename($fname); |
| 1757 |
my $dirname=dirname($fname); |
| 1758 |
my $lockdir="$basename"; |
| 1759 |
$lockdir=$dirname.'/'.$lockdir if $dirname; |
| 1760 |
return ($lockdir, "by.$HOST.$$"); |
| 1761 |
} |
| 1762 |
|
| 1763 |
sub makeLockFile { |
| 1764 |
#this is called by a node just before a file lock is requested |
| 1765 |
my ($fname)=@_; #MUST be a path relative to GRID_JOBDIR |
| 1766 |
if (index($fname,$GRID_JOBDIR.'/')==0) { |
| 1767 |
#remove the full path, if there |
| 1768 |
$fname=substr($fname, length($GRID_JOBDIR)+1); |
| 1769 |
} |
| 1770 |
my $fnlock=$fname; |
| 1771 |
$fnlock=~tr/\/\\/--/s; |
| 1772 |
my $nodelockf="$GRID_LOCKDIR/$fnlock-lock.by.$HOST.$$"; |
| 1773 |
#-- |
| 1774 |
$nodelockf.='-'.substr(time(),-4); |
| 1775 |
#-- |
| 1776 |
my $fh; |
| 1777 |
open($fh, '>'.$nodelockf) || die("Error creating lockfile $nodelockf!\n"); |
| 1778 |
my $thismin=int(time()/60); # time when the lock was first initiated |
| 1779 |
print $fh "$HOST $$ $thismin $GRID_WORKER\n"; |
| 1780 |
close($fh); |
| 1781 |
return ($GRID_LOCKDIR.'/'.$fnlock, $nodelockf); |
| 1782 |
} |
| 1783 |
|
| 1784 |
sub getFileLock { |
| 1785 |
#attempts to make $lockfile a hard link to $nodefile which is host/process dependent |
| 1786 |
my ($lockfile, $nodefile)=@_; |
| 1787 |
if (link($nodefile, $lockfile)) { #could create link to node lock file |
| 1788 |
my @stat=stat($nodefile); |
| 1789 |
my $linkcount=$stat[3]; |
| 1790 |
if ($linkcount>2) { |
| 1791 |
print STDERR "WARNING: weird lnkcount=$linkcount ($GRID_WORKER, $HOST, task $GRID_TASK)!\n"; |
| 1792 |
} |
| 1793 |
return ($linkcount>1); #should never be more than 2.. |
| 1794 |
} |
| 1795 |
else { #cannot create link |
| 1796 |
return undef; |
| 1797 |
} |
| 1798 |
} |
| 1799 |
|
| 1800 |
#attempts to aquire an exclusive lock on a specific file |
| 1801 |
#-returns a file handle ref |
| 1802 |
sub setXLock { |
| 1803 |
my ($fname, $maxretries, $stalemin)=@_; |
| 1804 |
$maxretries=80 unless $maxretries; |
| 1805 |
# |
| 1806 |
# -- default: locks older than this are |
| 1807 |
# considered "stale" and removed! |
| 1808 |
$stalemin=7200 unless $stalemin; |
| 1809 |
my ($lockfile, $nodefile)=makeLockFile($fname); |
| 1810 |
my $retries=0; |
| 1811 |
my $startmin=int(time()/60); #the minute we started trying |
| 1812 |
#----------- try mkdir |
| 1813 |
my ($currentLocker, $lockage); |
| 1814 |
my $lock_age; |
| 1815 |
my $prevlock; |
| 1816 |
my $haveLock; |
| 1817 |
|
| 1818 |
while (!($haveLock=getFileLock($lockfile, $nodefile))) { |
| 1819 |
if ($retries>$maxretries) { |
| 1820 |
#tried too many times? |
| 1821 |
#check for stale lock forgotten here.. |
| 1822 |
if (-f $lockfile) { |
| 1823 |
#anyone ELSE holding it? |
| 1824 |
my $l=readFile($lockfile); |
| 1825 |
my ($lhost, $lpid, $ltime, $worker)=split(/\s+/,$l); |
| 1826 |
$prevlock="$lhost (pid $lpid, worker $worker)" if $lpid; |
| 1827 |
# code to check for a stale lock: |
| 1828 |
$lock_age=int(time()/60)-$ltime; |
| 1829 |
if ($lock_age>$stalemin) { #previous lock is older than $stalemin minutes |
| 1830 |
print STDERR "WARNING: removing stale $fname lock from $lhost PID $lpid (worker $worker), age $lockage minutes.\n"; |
| 1831 |
unlink($lockfile); |
| 1832 |
next; # hopefully we'll get it next time, unless another node steals it.. |
| 1833 |
} |
| 1834 |
} |
| 1835 |
last; #too many retries; |
| 1836 |
} |
| 1837 |
$retries++; |
| 1838 |
sleep(3); #pause 3 seconds between attempts |
| 1839 |
} #----- while failing at getting a lock |
| 1840 |
if ($haveLock) { |
| 1841 |
local *FH; |
| 1842 |
open(FH, ">$nodefile") |
| 1843 |
|| die "Error re-creating host lock file $nodefile ?! ($!)\n"; #should never happen, really |
| 1844 |
my $thismin= int(time()/60); |
| 1845 |
print FH "$HOST $$ $thismin $GRID_WORKER\n"; |
| 1846 |
#print FH "$HOST $$ $thismin $GRID_WORKER ".getTime()."\n"; |
| 1847 |
close(FH); |
| 1848 |
# -- now it's safe to open the actual file being locked.. |
| 1849 |
# has to be a path relative to $GRID_JOBDIR |
| 1850 |
$fname="$GRID_JOBDIR/$fname" unless $fname=~m/^[\/\\]/; |
| 1851 |
# -- |
| 1852 |
# -- add O_DIRECT only if you notice syncing/flushing issues for small files |
| 1853 |
# -- |
| 1854 |
#my $mode= O_RDWR | O_CREAT | O_SYNC | O_DIRECT; |
| 1855 |
#my $mode=(-f $fname) ? '+<' : '+>'; #create file if not there.. |
| 1856 |
my $fh; |
| 1857 |
#open($fh, $mode.$fname) || die "setXLock($fname) failed at open($mode.$fname): $!\n"; |
| 1858 |
sysopen($fh, $fname, $sysopen_mode) || die "setXLock($fname) failed at open ($fname): ($!)\n"; |
| 1859 |
$Locks{$fh}=[$lockfile, $nodefile]; |
| 1860 |
return $fh; |
| 1861 |
} |
| 1862 |
else { |
| 1863 |
print STDERR "ERROR getting a lock on $fname ($lockfile -> $nodefile) after $retries attempts!\n"; |
| 1864 |
print STDERR " (blocked by: $prevlock of $lock_age minutes)\n" if $prevlock; |
| 1865 |
unlink($nodefile); |
| 1866 |
return undef; |
| 1867 |
} |
| 1868 |
} |
| 1869 |
|
| 1870 |
sub endXLock { |
| 1871 |
my $fh=shift(@_); |
| 1872 |
close($fh); |
| 1873 |
my $d = delete($Locks{$fh}); |
| 1874 |
unless ($d) { |
| 1875 |
print STDERR "WARNING at endXLock(): no Locks entry found for file handle $fh!\n"; |
| 1876 |
return; |
| 1877 |
} |
| 1878 |
my ($locklink, $nodefile)=@$d; |
| 1879 |
unlink($locklink); |
| 1880 |
#-- keep these around for debugging purposes: |
| 1881 |
unlink($nodefile); |
| 1882 |
} |
| 1883 |
|
| 1884 |
# readFile(fname/fglobref) : read first or all lines from a given file |
| 1885 |
# or filehandle glob reference |
| 1886 |
sub readFile { |
| 1887 |
my ($f, $nonfatal, $context)=@_; |
| 1888 |
$context=" ($context)" if $context; |
| 1889 |
my ($fh, $open); |
| 1890 |
if (ref($f) eq 'GLOB') { # file handle or glob reference.. |
| 1891 |
$fh=$f; |
| 1892 |
$f='[fh]'; |
| 1893 |
} |
| 1894 |
else { #scalar: string = filename |
| 1895 |
#create if not there! |
| 1896 |
my $mode=(-f $f) ? '+<' : '+>'; #create if not exists |
| 1897 |
my $canopen=open($fh, $mode.$f); |
| 1898 |
unless ($canopen) { |
| 1899 |
return undef if $nonfatal; |
| 1900 |
die "readFile($mode $f)$context task $GRID_TASK on $HOST, open error: $!\n"; |
| 1901 |
} |
| 1902 |
$open=1; |
| 1903 |
} |
| 1904 |
local $/="\n"; |
| 1905 |
if (wantarray()) { |
| 1906 |
my @r=<$fh>; |
| 1907 |
close($fh) if $open; |
| 1908 |
return @r; |
| 1909 |
} |
| 1910 |
else { #first line only |
| 1911 |
my $line=<$fh> || ''; |
| 1912 |
close($fh) if $open; |
| 1913 |
return $line; |
| 1914 |
} |
| 1915 |
} |
| 1916 |
|
| 1917 |
#================================================================= |
| 1918 |
# Mailer subroutine - relies on Net::SMTP to be installed properly |
| 1919 |
# --> recognized hash keys: from, to, subj, file |
| 1920 |
#----------------------------------------------------------------- |
| 1921 |
sub send_mail { |
| 1922 |
my $hash=shift; |
| 1923 |
my $smtp=Net::SMTP->new(Host=>'mailhost'); |
| 1924 |
my $from=$hash->{'from'}; |
| 1925 |
if ($from) { |
| 1926 |
$from.='@'.$smtp->domain() unless $from=~m/@/; |
| 1927 |
} |
| 1928 |
else { |
| 1929 |
$from=$USER.'@'.$smtp->domain(); |
| 1930 |
} |
| 1931 |
my $to=$hash->{'to'}; |
| 1932 |
if ($to) { |
| 1933 |
$to.='@'.$DOMAIN unless $to =~ m/@/; |
| 1934 |
} |
| 1935 |
else { |
| 1936 |
$to=$USER.'@'.$DOMAIN; |
| 1937 |
} |
| 1938 |
my $subj=$hash->{'subj'}; |
| 1939 |
my $body=$hash->{'body'}; |
| 1940 |
$smtp->mail($from); |
| 1941 |
$smtp->to($to); |
| 1942 |
$smtp->data(); |
| 1943 |
$smtp->datasend("To: $to\n"); |
| 1944 |
if ($subj) { |
| 1945 |
$smtp->datasend("Subject: $subj\n\n"); |
| 1946 |
} |
| 1947 |
my $file; |
| 1948 |
if (defined($hash->{file})) { |
| 1949 |
local *ADDFILE; |
| 1950 |
#warning: assumes it's a decent, short text file! |
| 1951 |
local $/=undef; #read whole file |
| 1952 |
open(ADDFILE, '<'.$hash->{file}) || die("Error: cannot open file ".$hash->{file}."\n"); |
| 1953 |
$file=<ADDFILE>; |
| 1954 |
close ADDFILE; |
| 1955 |
$body.="\n\n".$file; |
| 1956 |
} |
| 1957 |
$smtp->datasend("$body\n"); |
| 1958 |
$smtp->dataend(); |
| 1959 |
$smtp->quit; |
| 1960 |
} |
| 1961 |
|
| 1962 |
sub getTime { |
| 1963 |
my $date=localtime(); |
| 1964 |
#get rid of the day so Sybase will accept it |
| 1965 |
(my $wday,$date)=split(/\s+/,$date,2); |
| 1966 |
return $date; |
| 1967 |
} |
