Bio::EnsEMBL::Hive::Scheduler Class Reference
Public Member Functions | |
| public | scheduler_say () |
| public | schedule_workers_resync_if_necessary () |
| public | suggest_analysis_to_specialize_a_worker () |
| public | schedule_workers () |
| public | sort_pairs_by_suitability () |
Detailed Description
Description
Scheduler starts with the numbers of required workers for unblocked analyses,
then goes through several kinds of restrictions (submit_limit, meadow_limits, hive_capacity, etc)
that act as limiters and may cap the original numbers in several ways.
The capped numbers are then grouped by meadow_type and rc_name and returned in a two-level hash.
Definition at line 15 of file Scheduler.pm.
Member Function Documentation
◆ schedule_workers()
| public Bio::EnsEMBL::Hive::Scheduler::schedule_workers | ( | ) |
Undocumented method
Code:
click to view
sub schedule_workers {
my ($queen, $submit_capacity, $default_meadow_type, $list_of_analyses, $meadow_capacity_limiter_hashed_by_type) = @_;
my @workers_to_submit_by_analysis = (); # The down-to-analysis "plan" that may completely change by the time the Workers are born and specialized
my %workers_to_submit_by_meadow_type_rc_name = (); # Pre-pending-adjusted per-resource breakout
my $total_extra_workers_required = 0;
my ($pairs_sorted_by_suitability, $log_buffer) = Bio::EnsEMBL::Hive::Scheduler::sort_pairs_by_suitability( $list_of_analyses );
unless( @$pairs_sorted_by_suitability ) {
unless( @$log_buffer ) {
push @$log_buffer, "Could not find any suitable analyses to start scheduling.";
}
} else {
my $submit_capacity_limiter = Bio::EnsEMBL::Hive::Limiter->new( 'Max number of Workers scheduled this time', $submit_capacity );
my $queen_capacity_limiter = Bio::EnsEMBL::Hive::Limiter->new( 'Total reciprocal capacity of the Hive', 1.0 - $queen->db->hive_pipeline->get_cached_hive_current_load() );
ANALYSIS: foreach my $pair (@$pairs_sorted_by_suitability) {
if( $submit_capacity_limiter->reached ) {
if( $meadow_capacity_limiter_hashed_by_type ) { # only add this message when scheduling and not during a Worker's specialization
push @$log_buffer, "Submission capacity (=".$submit_capacity_limiter->original_capacity.") has been reached.";
}
last ANALYSIS;
}
my ($analysis, $analysis_stats) = @$pair;
my $logic_name = $analysis->logic_name;
my $this_meadow_type = $analysis->meadow_type || $default_meadow_type;
if( $meadow_capacity_limiter_hashed_by_type && !$meadow_capacity_limiter_hashed_by_type->{$this_meadow_type} ) {
push @$log_buffer, "The Meadow '$this_meadow_type' is not reachable from here, skipping Analysis '$logic_name'.";
next ANALYSIS;
}
if( $meadow_capacity_limiter_hashed_by_type && $meadow_capacity_limiter_hashed_by_type->{$this_meadow_type}->reached ) {
push @$log_buffer, "Available capacity of '$this_meadow_type' Meadow (=".$meadow_capacity_limiter_hashed_by_type->{$this_meadow_type}->original_capacity.") has been reached, skipping Analysis '$logic_name'.";
next ANALYSIS;
}
push @$log_buffer, "Analysis '$logic_name' is ".$analysis_stats->status.", safe-synching it...";
# Do a (safe) sync to get up-to-date job-counts and status
if( $queen->safe_synchronize_AnalysisStats($analysis_stats) ) {
push @$log_buffer, "Safe-sync of Analysis '$logic_name' succeeded.";
} elsif (scalar(@{ $analysis->control_rules_collection() })) {
# The analysis is blockable and we haven't managed to sync it, so the status is unreliable
push @$log_buffer, "Safe-sync of Analysis '$logic_name' could not be run at this moment, cannot tell whether it is BLOCKED or not, skipping it.";
next ANALYSIS;
} else {
push @$log_buffer, "Safe-sync of Analysis '$logic_name' could not be run at this moment, will use old stats.";
}
if( ($analysis_stats->status eq 'BLOCKED') or (($analysis_stats->sync_lock) and scalar(@{ $analysis->control_rules_collection() }))) {
push @$log_buffer, "Analysis '$logic_name' is still ".$analysis_stats->status.", skipping it.";
next ANALYSIS;
}
# getting the initial worker requirement for this analysis (may be off if $analysis_stats has not been sync'ed recently)
my $extra_workers_this_analysis = $analysis_stats->estimate_num_required_workers;
if ($extra_workers_this_analysis <= 0) {
push @$log_buffer, "Analysis '$logic_name' doesn't require extra workers, skipping it.";
next ANALYSIS;
}
$total_extra_workers_required += $extra_workers_this_analysis; # also keep the total number required so far (if nothing required we may need a resync later)
# setting up all negotiating limiters:
$queen_capacity_limiter->multiplier( $analysis->hive_capacity );
my @limiters = (
$submit_capacity_limiter,
$queen_capacity_limiter,
$meadow_capacity_limiter_hashed_by_type
? $meadow_capacity_limiter_hashed_by_type->{$this_meadow_type}
: (),
defined($analysis->analysis_capacity)
$analysis->analysis_capacity - $analysis_stats->num_running_workers )
: (),
);
my $hit_the_limit;
# negotiations:
foreach my $limiter (@limiters) {
($extra_workers_this_analysis, $hit_the_limit) = $limiter->preliminary_offer( $extra_workers_this_analysis );
if($hit_the_limit) {
if($extra_workers_this_analysis>0) {
push @$log_buffer, "Hit the limit of *** ".$limiter->description." ***, settling for $extra_workers_this_analysis Workers.";
} else {
push @$log_buffer, "Hit the limit of *** ".$limiter->description." ***, skipping this Analysis.";
next ANALYSIS;
}
}
}
# let all parties know the final decision of negotiations:
foreach my $limiter (@limiters) {
$limiter->final_decision( $extra_workers_this_analysis );
}
push @workers_to_submit_by_analysis, [ $analysis, $extra_workers_this_analysis];
push @$log_buffer, $analysis_stats->toString;
if($meadow_capacity_limiter_hashed_by_type) {
my $this_rc_name = $analysis->resource_class->name;
$workers_to_submit_by_meadow_type_rc_name{ $this_meadow_type }{ $this_rc_name } += $extra_workers_this_analysis;
push @$log_buffer, sprintf("Before checking the Valley for pending jobs, the Scheduler allocated $extra_workers_this_analysis x $this_meadow_type:$this_rc_name extra workers for '%s' [%.4f hive_load remaining]",
$logic_name,
$queen_capacity_limiter->available_capacity,
);
}
} # /ANALYSIS : foreach my $pair (@$pairs_sorted_by_suitability)
}
return (\@workers_to_submit_by_analysis, \%workers_to_submit_by_meadow_type_rc_name, $total_extra_workers_required, $log_buffer);
}
◆ schedule_workers_resync_if_necessary()
| public Bio::EnsEMBL::Hive::Scheduler::schedule_workers_resync_if_necessary | ( | ) |
Undocumented method
Code:
click to view
my ($queen, $valley, $list_of_analyses) = @_;
my $reconciled_worker_statuses = $valley->query_worker_statuses( $queen->registered_workers_attributes );
my $submit_capacity = $valley->config_get('SubmitWorkersMax');
my $default_meadow_type = $valley->get_default_meadow()->type;
my ($valley_running_worker_count,
$meadow_capacity_limiter_hashed_by_type)= $valley->generate_limiters( $reconciled_worker_statuses );
my ($workers_to_submit_by_analysis, $workers_to_submit_by_meadow_type_rc_name, $total_extra_workers_required, $log_buffer)
= schedule_workers($queen, $submit_capacity, $default_meadow_type, $list_of_analyses, $meadow_capacity_limiter_hashed_by_type);
scheduler_say( $log_buffer );
unless( $total_extra_workers_required ) {
scheduler_say( "According to analysis_stats no workers are required... let's see if anything went out of sync." );
$queen->check_for_dead_workers($valley, 1);
scheduler_say( "re-synchronizing..." );
$queen->synchronize_hive( $list_of_analyses );
if( $queen->check_nothing_to_run_but_semaphored( $list_of_analyses ) ) { # double-check that we are really stuck
scheduler_say( "looks like we may need re-balancing semaphore_counts..." );
if( $queen->db->hive_pipeline->hive_auto_rebalance_semaphores ) { # make sure rebalancing only ever happens for the pipelines that asked for it
if( my $rebalanced_jobs_counter = $queen->db->get_AnalysisJobAdaptor->balance_semaphores( $list_of_analyses ) ) {
scheduler_say( "re-balanced $rebalanced_jobs_counter jobs, going through another re-synchronization..." );
$queen->synchronize_hive( $list_of_analyses );
} else {
scheduler_say( "hmmm... managed to re-balance 0 jobs, you may need to investigate further." );
}
} else {
scheduler_say([ "automatic re-balancing of semaphore_counts is off by default.",
"If you think your pipeline might benefit from it, set hive_auto_rebalance_semaphores => 1 in the PipeConfig's hive_meta_table.",
"You can also manually rebalance semaphores this time by running beekeeper with the '--balance_semaphores' option",
]);
}
} else {
scheduler_say( "some READY jobs still in the queue. No need to consider re-balancing at this time." );
}
($workers_to_submit_by_analysis, $workers_to_submit_by_meadow_type_rc_name, $total_extra_workers_required, $log_buffer)
= schedule_workers($queen, $submit_capacity, $default_meadow_type, $list_of_analyses, $meadow_capacity_limiter_hashed_by_type);
scheduler_say( $log_buffer );
}
# adjustment for pending workers:
my $pending_worker_counts_by_meadow_type_rc_name = $queen->get_submitted_worker_counts_by_meadow_type_rc_name_for_meadow_user(Bio::EnsEMBL::Hive::Utils::whoami());
while( my ($this_meadow_type, $partial_workers_to_submit_by_rc_name) = each %$workers_to_submit_by_meadow_type_rc_name) {
while( my ($this_rc_name, $workers_to_submit_this_group) = each %$partial_workers_to_submit_by_rc_name) {
if(my $pending_this_group = $pending_worker_counts_by_meadow_type_rc_name->{ $this_meadow_type }{ $this_rc_name }) {
scheduler_say( "The plan was to submit $workers_to_submit_this_group x $this_meadow_type:$this_rc_name workers when the Scheduler detected $pending_this_group pending in this group, " );
if( $workers_to_submit_this_group > $pending_this_group) {
$workers_to_submit_by_meadow_type_rc_name->{$this_meadow_type}{$this_rc_name} -= $pending_this_group; # adjust the hashed value
scheduler_say( "so I recommend submitting only ".$workers_to_submit_by_meadow_type_rc_name->{$this_meadow_type}{$this_rc_name}." extra" );
} else {
delete $workers_to_submit_by_meadow_type_rc_name->{$this_meadow_type}{$this_rc_name}; # avoid leaving an empty group in the hash
scheduler_say( "so I don't recommend submitting any extra" );
}
} else {
scheduler_say( "I recommend submitting $workers_to_submit_this_group x $this_meadow_type:$this_rc_name workers" );
}
}
unless(keys %{ $workers_to_submit_by_meadow_type_rc_name->{$this_meadow_type} }) { # if nothing has been scheduled for a meadow,
delete $workers_to_submit_by_meadow_type_rc_name->{$this_meadow_type}; # do not mention the meadow in the hash
}
}
return $workers_to_submit_by_meadow_type_rc_name;
}
◆ scheduler_say()
| public Bio::EnsEMBL::Hive::Scheduler::scheduler_say | ( | ) |
Undocumented method
Code:
click to view
sub scheduler_say {
my ($msgs) = @_;
$msgs = [ $msgs ] unless( ref($msgs) eq 'ARRAY' );
foreach my $msg (@$msgs) {
print "Scheduler : $msg\n";
}
}
◆ sort_pairs_by_suitability()
| public Bio::EnsEMBL::Hive::Scheduler::sort_pairs_by_suitability | ( | ) |
Undocumented method
Code:
click to view
my @sorted_stats = map { [ $_, $_->stats] } # 3. pair analyses with their stats objects
sort { $b->priority <=> $a->priority } # 2. but ordered according to their priority levels
shuffle # 1. make sure analyses are well mixed within the same priority level
@{ shift @_ };
my (@primary_candidates, @secondary_candidates, $discarded_count, @log_buffer);
foreach my $pair ( @sorted_stats ) {
my ($analysis, $stats) = @$pair;
# assuming sync() is expensive, so first trying analyses that have already been sunk:
if( ($stats->estimate_num_required_workers > 0) and ($stats->status =~/^(READY|WORKING)$/) ) {
push @primary_candidates, $pair;
} elsif( $stats->status =~ /^(LOADING|BLOCKED|ALL_CLAIMED|SYNCHING)$/ ) {
push @secondary_candidates, $pair;
} else {
$discarded_count++;
}
}
if( $discarded_count ) {
push @log_buffer, "Discarded $discarded_count analyses because they do not need any Workers.";
}
return ( [@primary_candidates, @secondary_candidates], \@log_buffer );
}
◆ suggest_analysis_to_specialize_a_worker()
| public Bio::EnsEMBL::Hive::Scheduler::suggest_analysis_to_specialize_a_worker | ( | ) |
Undocumented method
Code:
click to view
my ( $worker, $analyses_matching_pattern, $analyses_pattern ) = @_;
my $queen = $worker->adaptor;
my $worker_rc_id = $worker->resource_class_id;
my $worker_meadow_type = $worker->meadow_type;
if( ! @$analyses_matching_pattern ) {
return "Could not find any Analyses matching '$analyses_pattern' pattern";
} else {
$worker->worker_say( "Found ".scalar(@$analyses_matching_pattern)." analyses matching '$analyses_pattern' pattern" );
my @analyses_matching_worker = grep { !$worker_rc_id or $worker_rc_id==$_->resource_class_id}
grep { !$worker_meadow_type or !$_->meadow_type or ($worker_meadow_type eq $_->meadow_type) }
# if any other attributes of the worker are specifically constrained in the analysis (such as meadow_name),
# the corresponding checks should be added here
@$analyses_matching_pattern;
if( !@analyses_matching_worker ) {
return "Could not find any of the ".scalar(@$analyses_matching_pattern)." '$analyses_pattern' Analyses that would suit this Worker";
} else {
my ($workers_to_submit_by_analysis, $workers_to_submit_by_meadow_type_rc_name, $total_extra_workers_required, $log_buffer)
= schedule_workers( $queen, 1, $worker_meadow_type, \@analyses_matching_worker );
if( $worker->debug ) {
foreach my $msg (@$log_buffer) {
$worker->worker_say( $msg );
}
}
return scalar(@$workers_to_submit_by_analysis)
? $workers_to_submit_by_analysis->[0][0] # take the first analysis from the "plan" if the "plan" was not empty
: pop @$log_buffer; # or return the last line of the scheduling log
}
}
}
The documentation for this class was generated from the following file:
- modules/Bio/EnsEMBL/Hive/Scheduler.pm
