#!/usr/bin/perl # Cluster interface. Hides differences between PBS and SGE. # Copyright © 2007, 2010, 2011 Dan Zeman # License: GNU GPL package cluster; use utf8; use open ":utf8"; use Carp; use cas; use dzsys; BEGIN { # Are we running PBS or SGE? A bit tricky to tell... # PBS = Portable Batch System, runs at UMD # SGE = Sun Grid Engine, runs at CUNI # man qstat padá na strojích, kde není vůbec nic (vadí, pokud pouštím něco, co obsahuje use cluster, ale v tomto případě ho nechce volat) # if(`man qstat` =~ m/pbs/) if(!exists($ENV{SGE_ROOT})) { $software = "pbs"; # module-wide global variable } else { $software = "sge"; } } #------------------------------------------------------------------------------ # qsub: submit jobs to queues # example: $jobid = qsub("memory" => "1gb", "script" => "~/script.sh"); # Parameters: # script => "path" # command => "commandline" # only if no script; temporary script will be created # command => ['command1', 'command2', ...] # alternative: array of command; temporary script will be created # name => "jobname" # only if no script; will be shown by qstat # memory => "1gb", "300mb" etc. # walltime => "24:00:00" (h:mm:ss) # priority => -200 # or -300 etc. Lowers priority of getting machine, not of running on the machine. Good if creating many jobs at a time. # deps => [123456, 123457, ...] # staging => [{dirct => ..., cpath => ..., shost => ..., spath => ...}, {...}] # only PBS # qsub -sync y[es]|n[o] ... má qsub počkat, až vzdálená úloha doběhne (zatím nepoužito v této funkci) #------------------------------------------------------------------------------ sub qsub { my %params = @_; my $params = \%params; # hash ref # Make sure we know what to submit. if(!$params->{script} && !$params->{command}) { confess("cluster::qsub() needs either 'script' or 'command'."); } my $command = "qsub"; # All jobs are bash scripts. Some clusters may default to other shells so we better ask for bash explicitly. $command .= ' -S /bin/bash'; # This would be the place to select machines (queues) we want to send the job to. # We used to do that because of incompatible behavior of two cluster in the transition phase (spring 2011). # Now we are merely stating the default, i.e. the all.q queue, containing all available computers. $command .= " -q 'all.q\@*'"; # If we want to submit a commandline that is not a script, create script first. my $script; if($params->{script} eq "" && $params->{command} ne "") { # Create temporary script. # We will remove the script at the end of this qsub() call, so we need not care about other calls to qsub() from the current process. # Try to figure out a good name for the script from the command. my $scriptname = 'script'; if($params->{name}) { $scriptname = $params->{name}; } else { my @command_tokens = split(/\s+/, ref($params->{command}) eq 'ARRAY' ? $params->{command}[0] : $params->{command}); my $command_path = $command_tokens[0]; $command_path =~ s-^.*/--; $command_path =~ s/\.pl$//; $scriptname = $command_path if($command_path); } $script = "/tmp/$scriptname.$$.sh"; open(SCRIPT, ">$script") or die("Cannot write to $script: $!\n"); # Out of habit, we declare the intended shell in the beginning of the script. # However, SGE will ignore it and we must use the -S option of qsub to override the cluster default shell! print SCRIPT ("#!/bin/bash\n"); print SCRIPT ("# A temporary script generated by cluster.pm\n"); # Make the script identify itself. print SCRIPT ("echo 'Job name =' $scriptname\n"); print SCRIPT ("echo 'Host name =' \`hostname -f\`\n"); print SCRIPT ("echo 'Current folder =' \`pwd\`\n"); print SCRIPT ("echo 'Script =' \$0\n"); # Initial security settings. print SCRIPT ("set -o pipefail\n"); print SCRIPT ("function die() { echo \"\$@\" >&2 ; exit 1 ; }\n"); print SCRIPT ("renice 10 \$\$\n"); print SCRIPT ("ulimit -c 1 \# core files limited to 1 byte\n"); # For debugging purposes, we could also print the script itself. # However, it would be only good to make sure that we have escaped special characters correctly when generating the script. # Otherwise, all commands are echoed anyway. # That's why this feature is disabled by default. if(0) { print SCRIPT ("echo ==================================================\n"); print SCRIPT ("echo Script source\n"); print SCRIPT ("echo ==================================================\n"); print SCRIPT ("cat \$0\n"); print SCRIPT ("echo ==================================================\n"); print SCRIPT ("echo End of script source\n"); print SCRIPT ("echo ==================================================\n"); } if(ref($params->{command}) eq 'ARRAY') { foreach my $c (@{$params->{command}}) { print SCRIPT (wrap_command_bash($c)); } } else { print SCRIPT (wrap_command_bash($params->{command})); } close(SCRIPT); chmod(0755, $script); $params->{script} = $script; } # priority of getting a machine (tested only on sge so far) if(exists($params->{priority})) { # example: -200, -300 $command .= " -p $params->{priority}"; # Note: In addition to this, we may consider prepending 'nice' to the script call, # to decrease the priority of the job process on the target machine. In general, # there should not be much competition for the CPU (the job should be alone there) # but if the job goes mad - having high priority - we may not even be able to kill it. } # resources not known by sge: walltime, pmem if($software eq "pbs") { $command .= " -q batch"; # time requirements if(exists($params->{walltime})) { # walltime format: h:mm:ss $command .= " -l walltime=$params->{walltime}"; } # memory requirements if(exists($params->{memory})) { # pmem examples: 1mb, 2gb... $command .= " -l pmem=$params->{memory}"; } } else # SGE { # memory requirements if(exists($params->{memory})) { # Note: '15gb' is wrong format for SGE, correct format is '15g'! # Update 7.10.2011 by Milan: the 'G' must be uppercase, i.e. '15G'! my $memory = $params->{memory}; $memory =~ s/([mg])b$/$1/; $memory = uc($memory); # mf=$memory ... want machine where this amount of memory has not been reserved by other jobs via cluster scheduler # act_mem_free=$memory ... want machine where this amount of memory is currently free (nasty jobs take more than they reserved via cluster scheduler) # h_vmem=$memory ... kill myself if my memory consumption exceeds this amount of memory (so that others can rely on what I reserved) $command .= " -hard -l mf=$memory -l act_mem_free=$memory -l h_vmem=$memory"; } # only job id on output, no crap $command .= " -terse"; # run job in current folder rather than in home folder $command .= " -cwd"; } # copy my environment variables to the job environment $command .= " -V"; # merge job's STDOUT and STDERR to STDOUT $command .= " -j yes"; # data staging if(scalar(@{$params->{staging}})) { if($software eq "pbs") { foreach my $stage (@{$params->{staging}}) { # stage keys: # dirct = in|out (in = script input, i.e. from submit node to computing node) # shost = submit host # spath = path at submit host # cpath = path at computing node $command .= " -W stage$stage->{dirct}=$stage->{cpath}\@$stage->{shost}:$stage->{spath}"; } } else # SGE { # We currently cannot stage data under SGE! # The SGE qsub command does not provide such functionality. # The possible workaround using scp does not work (problem in authentication). # Data staging may not be needed with NFS disks; however, the computing # nodes should work with local /tmp to save the network bandwidth. die("Data staging functionality not available.\n"); } } # job dependencies if($software eq "pbs") { foreach my $dep (@{$params->{deps}}) { $command .= " -W depend=afterok:$dep"; } } elsif(scalar(@{$params->{deps}})) { # Even if the array is not empty the deps can be empty strings. my $deps = join(',', @{$params->{deps}}); if($deps =~ m/\d/) { $command .= " -hold_jid $deps"; } } # script for the job $command .= " $params->{script}"; # submit the job and get the job ID print STDERR ("$command\n"); my $jobid = `$command`; $jobid =~ s/\r?\n$//; # Clean up the temporary script, if created. if($script ne "") { unlink($script); } return $jobid; } #------------------------------------------------------------------------------ # Kills jobs identified by jobids. #------------------------------------------------------------------------------ sub qdel { my @jobids = @_; my $jobids = join(',', @jobids); # Do not die if saferun() does not return true (which indicates nonzero exit status of qdel). return $jobids && dzsys::saferun("qdel $jobids"); } #------------------------------------------------------------------------------ # Reads the output of the qstat cluster command and returns the result in a # hash. This should be called qstat() and the extended function below should be # called qstat_resubmit() or something but I have been stupid and now there is # the compatibility issue. #------------------------------------------------------------------------------ sub qstat0 { # Warning: This code depends on the particular output format of qstat, as seen on ÚFAL LRC. my %qstat; open(QSTAT, "qstat -u '*' |") or die("Cannot pipe from qstat: $!\n"); while() { # Remove initial spaces. s/^\s+//; # Remove final spaces and line break. s/\r?\n$//; s/\s+$//; # If the line does not now start with a digit, it is a header line. if(m/^\d/) { # Split fields. my @f = split(/\s+/, $_); # Hash fields. my %record = ( 'job_id' => $f[0], 'priority' => $f[1], 'name' => $f[2], 'user' => $f[3], 'state' => $f[4], 'date' => $f[5], 'time' => $f[6], 'queue' => $f[7] ); $record{host} = $record{queue}; $record{host} =~ s/^.*?\@//; $record{host} =~ s/\.ufal\.hide.*//; $qstat{$record{job_id}} = \%record; } } close(QSTAT); return %qstat; } #------------------------------------------------------------------------------ # Returns the list of ids of jobs whose owner is the current user and status is # 'Eqw'. This means error during submission, without a chance for recovery. We # may want to remove them. #------------------------------------------------------------------------------ sub list_my_eqw { my %qstat = qstat0(); return sort {$a <=> $b} map {$qstat{$_}{job_id}} grep {$qstat{$_}{user} eq $ENV{USER} && $qstat{$_}{state} eq 'Eqw'} keys(%qstat); } #------------------------------------------------------------------------------ # Checks the current status of the jobs at the cluster. Gets the array of # hashes with descriptions of jobs we are monitoring (presumably, every job # corresponds to a chunk of input data we are processing paralelly; that is why # we are referring to the jobs as chunks as well; however, any set of jobs can # be described this way). Returns the number of jobs we are still waiting for. # Writes status reports to STDERR. # # If we are waiting for a single job, we can give the function the job ID (a # scalar) instead of the reference to array of hashes: # instead of: qstat([{'job_id' => $jid}]); # just call: qstat($jid); # Similarly, array of scalars (not reference to array!) means multiple job ids: # like this: qstat($jid0, $jid1, $jid2); # WARNING: If the function is called this way (without the hashes), it will not # be able to keep track between calls of what has been reported and will report # the same thing (such as "the job N is running on HOST...") over and over. # # Waiting until a job finishes: # while(qstat($jid)) {sleep(10)} # # Input hash keys: # job_id ... job identifier on the cluster as reported by qsub # number ... chunk number, used only in reports # script ... needed if the jobs are to be resubmitted on error # command ... alternative for 'script' # Intermediate and output hash keys: # is_running ... does qstat report the job as 'r' and have we reported it? # host ... name of the computer the job is running on # has_finished ... is the job unknown to qstat (does not even appear as 'qw') # and have we reported it? # The hashes may contain other keys, such as chunk descriptions. #------------------------------------------------------------------------------ sub qstat { my $chunks = shift; # ref to array of hashes my $check = shift; # optional ref to callback routine that can check whether the output of a job is ok my $memory = shift; # required cluster memory (e.g. '15g') for resubmitted jobs my $nretries = shift; # number of retries before giving up (default: 4) $nretries = 4 unless($nretries); unless(ref($chunks) eq 'ARRAY') { my @job_ids = @_; my @chunks = map {{'job_id' => $_}} (@job_ids); $chunks = \@chunks; } # Get the list of jobs reported by the cluster. my %qstat = qstat0(); # Go through all the chunks, note their jobs and hosts. my $waiting_for = scalar(@{$chunks}); foreach my $chunk (@{$chunks}) { # If the job is running and we have not reported it, report it. if(exists($qstat{$chunk->{job_id}})) { unless($chunk->{is_running}) { if($qstat{$chunk->{job_id}}{state} eq 'r') { print STDERR (get_timestamp(), " The chunk number $chunk->{number} (job $chunk->{job_id}) is running on $qstat{$chunk->{job_id}}{host} since $qstat{$chunk->{job_id}}{date} $qstat{$chunk->{job_id}}{time}.\n"); $chunk->{is_running} = 1; $chunk->{host} = $qstat{$chunk->{job_id}}{host}; } } } # If the job is unknown to qstat, it has exited. else { unless($chunk->{has_finished}) { print STDERR (get_timestamp(), " The chunk number $chunk->{number} (job $chunk->{job_id}) has finished.\n"); $chunk->{has_finished} = 1; # Check whether the job succeeded (produced the desired output) if we know how. if($check) { unless(check_and_resubmit_job($chunk, $check, $nretries, $memory)) { # The job had to be resubmitted, clear the state indicators. $chunk->{has_finished} = 0; $chunk->{is_running} = 0; } } } } # If this chunk has finished, we are no longer waiting for it. $waiting_for-- if($chunk->{has_finished}); } # Return the number of jobs still running or even waiting for a host. print STDERR (get_timestamp(), " Still waiting for $waiting_for jobs.\n"); return $waiting_for; } #------------------------------------------------------------------------------ # Uses qstat to repeatedly check status of one or more cluster jobs. This is # used for jobs other than chunks, i.e. we have only job id's and the function # wraps them in a chunk-like hash. The function does not check whether the # jobs succeeded and does not resubmit them in the case of failure. #------------------------------------------------------------------------------ sub waitfor { my $interval = shift; # seconds to wait between qstat calls my @jobids = @_; # the remaining parameters are id's of jobs to wait for my @chunks = map {{'job_id' => $_}} (@jobids); while(qstat(\@chunks)) { sleep($interval); } } #------------------------------------------------------------------------------ # Runs a system command (calls an external program) and reports it to STDERR. #------------------------------------------------------------------------------ sub runecho { my $commandline = join(" ", @_); print STDERR ($commandline); my $return = system(@_); if($return==-1) { croak("Cannot run command: $!\n"); } elsif($return!=0) { my $systemexitvalue = $?>>8; croak("Command failed: $systemexitvalue\n"); } } #------------------------------------------------------------------------------ # Gets sortable but human-readable, fixed-width string with timestamp (current # date and time). Does not care about time zone. #------------------------------------------------------------------------------ sub get_timestamp { my $time = cas::ted(); my $timestamp = sprintf('%04d-%02d-%02d %02d:%02d:%02d', $time->{rok}, $time->{mes}, $time->{den}, $time->{hod}, $time->{min}, $time->{sek}); return $timestamp; } #============================================================================== # Functions for management of parallelly processed chunks of data. #============================================================================== #------------------------------------------------------------------------------ # Figures out the number of lines in a file. The function is not portable as it # calls the unix command wc. #------------------------------------------------------------------------------ sub get_number_of_lines { my $file = shift; my $n = `wc -l $file`; # If the output does not contain any digit, the file probably could not be found. # (We avoid searching for "No such file or directory" because the error message wording may be system-dependent.) if($n !~ m/^\d/) { print STDERR ($n); die("Cannot figure out the number of lines of the file $file.\n"); } # Otherwise, make sure that we return ONLY the number. $n =~ s/\r?\n$//; $n =~ s/\D.*//; return $n; } #------------------------------------------------------------------------------ # According to number of input lines and maximum allowed number of chunks, # figures out the number of non-empty chunks and their sizes (all chunks except # the last one have the same size). #------------------------------------------------------------------------------ sub get_chunking { my $n_lines = shift; my $max_chunks = shift; my $n_chunks = $max_chunks; my $n_lines_first; my $n_lines_last; # For zero input return zero output. if($max_chunks<=0) { die("Chunking constraints unsatisfiable: max chunks = $max_chunks\n"); } if($n_lines<=0) { return (0, 0, 0); } # Round the chunk size upwards to make sure the text fits in. $n_lines_first = $n_lines/$n_chunks; if(int($n_lines_first)<$n_lines_first) { $n_lines_first = int($n_lines_first)+1; } else { $n_lines_first = int($n_lines_first); } # Since we rounded the size up, for very small inputs some chunks may remain empty. while(1) { my $projected_total = $n_chunks*$n_lines_first; my $over = $projected_total-$n_lines; if($over<$n_lines_first) { $n_lines_last = $n_lines_first-$over; return ($n_chunks, $n_lines_first, $n_lines_last); } $n_chunks--; } # We will never reach this place. } #------------------------------------------------------------------------------ # Reads the input file and writes its contents into the pre-calculated number # of chunks. #------------------------------------------------------------------------------ sub create_chunks { my $input_file = shift; my $n_chunks = shift; my $n_lines_first = shift; my $n_lines_last = shift; my $workdir = shift; my $get_script_for_chunk = shift; # reference to callback code that creates chunk script; params($workdir, $chunk_id) my $taskname = shift; # common part of names of scripts and jobs, e.g. 'parse' # Open the input file. print STDERR ("Reading input file $input_file...\n"); open(INPUT, $input_file) or die("Cannot read from the file $input_file: $!\n"); # Store all information about each chunk in an array of hashes. my @chunks; for(my $i = 0; $i<$n_chunks; $i++) { my $chunk_size = $i<$n_chunks-1 ? $n_lines_first : $n_lines_last; my $chunk_id = sprintf("%02d", $i); my $chunk_name = "chunk.$chunk_id.txt"; my $chunk_path = "$workdir/$chunk_name"; print STDERR ("Writing chunk $chunk_path...\n"); open(CHUNK, ">$chunk_path") or die("Cannot write to the chunk file $chunk_path: $!\n"); for(my $j = 0; $j<$chunk_size; $j++) { $_ = ; print CHUNK ($_); } close(CHUNK); my $chunk_script = &{$get_script_for_chunk}($workdir, $chunk_id); my $chunk_script_name = "$taskname.$chunk_id.sh"; my $chunk_script_path = "$workdir/$chunk_script_name"; print STDERR ("Writing script $chunk_script_path...\n"); open(SCRIPT, ">$chunk_script_path") or die("Cannot write the script $chunk_script_path: $!\n"); print SCRIPT ($chunk_script); close(SCRIPT); chmod(0755, $chunk_script_path) or die("Cannot change permissions of the script $chunk_script_path: $!\n"); my $chunk_output_name = "output.$chunk_id.txt"; my $chunk_output_path = "$workdir/$chunk_output_name"; push(@chunks, { 'number' => $i, 'size' => $chunk_size, 'name' => $chunk_name, 'path' => $chunk_path, 'script' => $chunk_script_name, 'script_path' => $chunk_script_path, 'output' => $chunk_output_name, 'output_path' => $chunk_output_path }); } # Check that there are no lines left. # If it happens either our chunk-planning function is buggy or the wc command (get number of input lines) has not worked properly. my $n_forgotten_lines = 0; while() { $n_forgotten_lines++; } if($n_forgotten_lines) { print STDERR ("WARNING: Probable miscomputation of chunk sizes! $n_forgotten_lines lines of input have not been chunked!\n"); } # Close the input file. close(INPUT); return \@chunks; } #------------------------------------------------------------------------------ # Submits processing of all chunks to the cluster. #------------------------------------------------------------------------------ sub submit_chunk_jobs { my $chunks = shift; # ref to array of hashes my $memory = shift; # required cluster memory (e.g. '15g') # Default is the maximum memory on smaller cluster machines, minus 1 GB for system etc. $memory = '15g' unless($memory); foreach my $chunk (@{$chunks}) { print STDERR (get_timestamp(), " Submitting chunk $chunk->{number} to the cluster...\n"); if($chunk->{script}) { $chunk->{job_id} = qsub('script' => $chunk->{script}, 'priority' => -200, 'memory' => $memory); } elsif($chunk->{command}) { $chunk->{job_id} = qsub('command' => $chunk->{command}, 'priority' => -200, 'memory' => $memory); } else { die("Cannot submit job: no 'script' or 'command' specified.\n"); } print STDERR (" job ID is $chunk->{job_id}\n"); } } #------------------------------------------------------------------------------ # Checks the output of a job for cues that the job died prematurely. # Resubmits the job if it did not succeed and the maximum number of attempts # has not been exceeded. #------------------------------------------------------------------------------ sub check_and_resubmit_job { my $chunk = shift; # ref to hash my $check_job_output = shift; # callback function; params($chunk) my $max_reruns = shift; my $memory = shift; # required cluster memory (e.g. '15g') # Default is the maximum memory on smaller cluster machines, minus 1 GB for system etc. $memory = '15g' unless($memory); my $ok = &{$check_job_output}($chunk); # In case of bad output rerun the job. # Sometimes, some jobs fail due to temporary network conditions and simple rerunning helps. # Of course, if there is a bug in the decoder, the rerunning would cause an endless loop, so we probably want to limit the number of attempts. unless($ok) { $chunk->{failed_attempts}++; if($chunk->{failed_attempts}>$max_reruns) { die("Failed to succeed with chunk number $chunk->{number}: $chunk->{failed_attempts}, giving up.\n"); } print STDERR (get_timestamp(), " Resubmitting chunk $chunk->{number} to the cluster...\n"); if($chunk->{script}) { $chunk->{job_id} = qsub('script' => $chunk->{script}, 'priority' => -200, 'memory' => $memory); } elsif($chunk->{command}) { $chunk->{job_id} = qsub('command' => $chunk->{command}, 'priority' => -200, 'memory' => $memory); } else { die("Cannot resubmit job: no 'script' or 'command' specified.\n"); } print STDERR (" job ID is $chunk->{job_id}\n"); } return $ok; } #============================================================================== # Function for on-the-fly script creation. #============================================================================== #------------------------------------------------------------------------------ # Wraps a command with some debugging and safety constructs. Assumption: the # command will be executed in tcsh shell. #------------------------------------------------------------------------------ sub wrap_command_tcsh { my $command0 = shift; my $command1; # Before the execution of every command, report current date and time. $command1 .= "echo --------------------------------------------------\n"; $command1 .= "date\n"; # List the command before executing it. # Any shell-special characters must be protected from the shell! my @parts = split(/'/, $command0); my $pcommand = join("\\'", map {"'$_'"} (@parts)); $command1 .= "echo Executing: $pcommand\n"; # Add the command itself and check its exit status. # The inner parentheses protect any STDIN|OUT|ERR redirections. # The curly brackets convert the command exit status into a boolean value. # The outer parentheses are required by the 'if' syntax. $command1 .= "if ( { ( $command0 ) } ) then\n"; $command1 .= " date\n"; $command1 .= " echo Execution succeeded.\n"; $command1 .= "else\n"; $command1 .= " date\n"; $command1 .= " echo Execution failed, status = \$status\n"; $command1 .= " exit 1\n"; $command1 .= "endif\n"; return $command1; } #------------------------------------------------------------------------------ # Wraps a command with some debugging and safety constructs. Assumption: the # command will be executed in bash-compatible shell. #------------------------------------------------------------------------------ sub wrap_command_bash { my $command0 = shift; my $command1; # Before the execution of every command, report current date and time. $command1 .= "echo --------------------------------------------------\n"; $command1 .= "date\n"; # List the command before executing it. # Any shell-special characters must be protected from the shell! my @parts = split(/'/, $command0); my $pcommand = join("\\'", map {"'$_'"} (@parts)); $command1 .= "echo Executing: $pcommand\n"; # Add the command itself and check its exit status. # The parentheses protect any STDIN|OUT|ERR redirections. $command1 .= "if ( $command0 ) ; then\n"; $command1 .= " date\n"; $command1 .= " echo Execution succeeded.\n"; $command1 .= "else\n"; $command1 .= " date\n"; $command1 .= " echo Execution failed, status = \$PIPESTATUS\n"; $command1 .= " exit 1\n"; $command1 .= "fi\n"; return $command1; } 1;