Functions
public	main ()

public	log_and_die ()

public	generate_worker_cmd ()

public	register_beekeeper ()

public	big_red_button ()

public	run_autonomously ()

Function Documentation

◆ big_red_button()

public big_red_button ( )

Undocumented method

Code:

click to view

◆ generate_worker_cmd()

public generate_worker_cmd ( )

Undocumented method

Code:

click to view

◆ log_and_die()

public log_and_die ( )

Undocumented method

Code:

click to view

◆ main()

public main ( )

Undocumented method

Code:

click to view

Undocumented method

Code:

click to view

Undocumented method

Code:

click to view

Undocumented method

Code:

click to view

Undocumented method

Code:

click to view

Undocumented method

Code:

click to view

Undocumented method

Code:

click to view

Undocumented method

Code:

click to view

Undocumented method

Code:

click to view

Undocumented method

Code:

click to view

Undocumented method

Code:

click to view

Undocumented method

Code:

click to view

Undocumented method

Code:

click to view

Undocumented method

Code:

click to view

Undocumented method

Code:

click to view

◆ register_beekeeper()

public register_beekeeper ( )

Undocumented method

Code:

click to view

◆ run_autonomously()

public run_autonomously ( )

Undocumented method

Code:

click to view

sub run_autonomously {
    my ($self, $pipeline, $max_loops, $loop_until, $valley, $list_of_analyses, $analyses_pattern, $run_job_id) = @_;
 
    my $hive_dba    = $pipeline->hive_dba;
    my $queen       = $hive_dba->get_Queen;
    my $meadow_user = $self->{'beekeeper'} && $self->{'beekeeper'}->meadow_user;
 
    my $pathless_resourceless_worker_cmd = generate_worker_cmd($self, $analyses_pattern, $run_job_id);
 
    my $iteration=0;
    my $reasons_to_exit;
 
    BKLOOP: while( ($iteration++ != $max_loops) or ($loop_until eq 'FOREVER') ) {  # NB: the order of conditions is important!
 
        print("\nBeekeeper : loop #$iteration ======================================================\n");
 
        my $pipeline_name = $pipeline->hive_pipeline_name;
        if ($pipeline_name) {
          print "\nPipeline name: $pipeline_name";
        }
        print "\nPipeline URL: ".$self->{'url'}."\n\n";
 
        $queen->check_for_dead_workers($valley, 0) unless $self->{'read_only'};
 
        # this section is where the beekeeper decides whether or not to stop looping
        $reasons_to_exit = $queen->print_status_and_return_reasons_to_exit( $list_of_analyses, $self->{'debug'});
        my @job_fail_statuses = grep({$_->{'exit_status'} eq 'JOB_FAILED'} @$reasons_to_exit);
        my @analysis_fail_statuses = grep({$_->{'exit_status'} eq 'ANALYSIS_FAILED'} @$reasons_to_exit);
        my @no_work_statuses = grep({$_->{'exit_status'} eq 'NO_WORK'} @$reasons_to_exit);
 
        my $found_reason_to_exit = 0;
 
        if (($loop_until eq 'JOB_FAILURE') &&
            (scalar(@job_fail_statuses)) > 0) {
            print "Beekeeper : last loop because at least one Job failed and loop-until mode is '$loop_until'\n";
            print "Beekeeper : details from analyses with failed Jobs:\n";
            print join("\n", map {$_->{'message'}} @job_fail_statuses) . "\n";
            $found_reason_to_exit = 1;
            last BKLOOP;
        }
 
        if (scalar(@analysis_fail_statuses > 0)) {
            # at least one analysis has hit its fault tolerance
            if (($loop_until eq 'FOREVER') ||
                ($loop_until eq 'NO_WORK')) {
                if (scalar(@no_work_statuses) == 0) {
                    print "Beekeeper : detected the following exit condition(s), but staying alive because loop-until mode is set to '$loop_until' :\n" .
                        join(", ", map {$_->{'message'}} @analysis_fail_statuses) . "\n";
                }
            } else {
                # loop_until_mode is either job_failure or analysis_failure, and both of these exit on analysis failure
                unless ($found_reason_to_exit) {
                    print "Beekeeper : last loop because at least one analysis failed and loop-until mode is '$loop_until'\n";
                    print "Beekeeper : details from analyses with failed Jobs:\n";
                    print join("\n", map {$_->{'message'}} @analysis_fail_statuses) . "\n";
                    $found_reason_to_exit = 1;
                    last BKLOOP;
                }
            }
        }
 
        if ((scalar(@no_work_statuses) > 0) &&
            ($loop_until ne 'FOREVER')) {
            print "Beekeeper : last loop because there is no more work and loop-until mode is '$loop_until'\n"
                unless ($found_reason_to_exit);
            last BKLOOP;
        }
 
        # end of testing for loop end conditions
 
        $hive_dba->get_RoleAdaptor->print_active_role_counts;
 
        my $workers_to_submit_by_meadow_type_rc_name = $self->{'read_only'} ? {} :
              Bio::EnsEMBL::Hive::Scheduler::schedule_workers_resync_if_necessary($queen, $valley, $list_of_analyses);
 
        if( keys %$workers_to_submit_by_meadow_type_rc_name ) {
 
            my $submit_log_subdir;
 
            if( $self->{'submit_log_dir'} ) {
                $submit_log_subdir = $self->{'submit_log_dir'}."/submit_bk".$self->{'beekeeper_id'}."_iter${iteration}";
                make_path( $submit_log_subdir );
            }
 
                # create an "index" over the freshly loaded RC/RD collections:
            my %meadow_type_rc_name2resource_param_list = ();
            foreach my $rd ( $pipeline->collection_of('ResourceDescription')->list ) {
                my $rc_name = $rd->resource_class->name;
                $meadow_type_rc_name2resource_param_list{ $rd->meadow_type }{ $rc_name } = [ $rd->submission_cmd_args, $rd->worker_cmd_args ];
            }
 
            foreach my $meadow_type (keys %$workers_to_submit_by_meadow_type_rc_name) {
 
                my $this_meadow = $valley->available_meadow_hash->{$meadow_type};
 
                foreach my $rc_name (keys %{ $workers_to_submit_by_meadow_type_rc_name->{$meadow_type} }) {
                    my $this_meadow_rc_worker_count = $workers_to_submit_by_meadow_type_rc_name->{$meadow_type}{$rc_name};
 
                    my $submission_message = "submitting $this_meadow_rc_worker_count workers (rc_name=$rc_name) to ".$this_meadow->signature();
                    print "\nBeekeeper : $submission_message\n";
                    $self->{'logmessage_adaptor'}->store_beekeeper_message($self->{'beekeeper_id'},
                        "loop iteration $iteration, $submission_message",
                        'INFO', 'ALIVE');
 
                    my ($submission_cmd_args, $worker_cmd_args) = @{ $meadow_type_rc_name2resource_param_list{ $meadow_type }{ $rc_name } || [] };
 
                    my $specific_worker_cmd = $this_meadow->runWorker_path
                                            . $pathless_resourceless_worker_cmd
                                            . (defined($worker_cmd_args) ? " $worker_cmd_args" : '')
                                            . ' -preregistered';
 
                    my $meadow_process_ids = [];
                    if ($this_meadow_rc_worker_count > 1 and !$this_meadow->config_get('CanSubmitJobArrays')) {
                        foreach my $i (1..$this_meadow_rc_worker_count) {
                            my $submitted_process_id = $this_meadow->submit_workers_return_meadow_pids($specific_worker_cmd, 1, $self->{'beekeeper_id'}.'_'."$iteration.$i", $rc_name, $submission_cmd_args || '', $submit_log_subdir);
                            push @$meadow_process_ids, $submitted_process_id->[0];
                        }
                    } else {
                        $meadow_process_ids = $this_meadow->submit_workers_return_meadow_pids($specific_worker_cmd, $this_meadow_rc_worker_count, $self->{'beekeeper_id'}.'_'.$iteration, $rc_name, $submission_cmd_args || '', $submit_log_subdir);
                    }
 
                    print "Submitted the following process_ids to ".$this_meadow->signature.": ".join(', ', @$meadow_process_ids)."\n";
 
                    my $resource_class  = $pipeline->collection_of('ResourceClass')->find_one_by('name', $rc_name);
                    my $meadow_name     = $this_meadow->cached_name;
 
                    my @pre_allocated_workers = map {
                            Bio::EnsEMBL::Hive::Worker->new(
                                'meadow_type'       => $meadow_type,                # non-unique key components
                                'meadow_name'       => $meadow_name,
                                'meadow_user'       => $meadow_user,
                                'process_id'        => $_,
 
                                'resource_class'    => $resource_class,             # non-key, but known at the time of pre-allocation
                                'beekeeper_id'      => $self->{'beekeeper_id'},
 
                                'status'            => 'SUBMITTED',
                            )
                    } @$meadow_process_ids;
 
                    $queen->store( \@pre_allocated_workers );
                }
            }
 
        } elsif ($self->{'read_only'}) {
            print STDERR "\n";
            print STDERR "*" x 70, "\n";
            print STDERR "* beekeeper.pl is running in read-only mode, i.e. it only\n";
            print STDERR "* prints the current status of the pipeline.\n";
            print STDERR "*\n";
            print STDERR "*" x 70, "\n";
            print STDERR "\n";
 
        } else {
            print "\nBeekeeper : not submitting any workers this iteration\n";
            $self->{'logmessage_adaptor'}->store_beekeeper_message($self->{'beekeeper_id'},
                "loop iteration $iteration, 0 workers submitted",
                'INFO', 'ALIVE');
        }
 
        if( $iteration != $max_loops ) {    # skip the last sleep
            while (1) {
                $hive_dba->dbc->disconnect_if_idle;
                printf("Beekeeper : going to sleep for %.2f minute(s). Expect next iteration at %s\n", $self->{'sleep_minutes'}, scalar localtime(time+$self->{'sleep_minutes'}*60));
                sleep($self->{'sleep_minutes'}*60);
                last if $self->{'read_only'};
                # this is a good time to check up on other beekeepers as well:
                $self->{'beekeeper'}->adaptor->bury_other_beekeepers($self->{'beekeeper'});
                if ($self->{'beekeeper'}->check_if_blocked()) {
                    print "Beekeeper : We have been blocked !\n".
                          "This can happen if a Job has explicitly required the Beekeeper to stop (have a look at log_message).\n".
                          "It may also happen if someone has set is_blocked=1 in the beekeeper table for beekeeper_id=".$self->{'beekeeper_id'}.".\n";
                } else {
                    last;
                }
            }
 
            # after waking up reload Resources and Analyses to stay current.
            unless($run_job_id) {
                # reset all the collections so that fresher data will be used at this iteration:
                $pipeline->invalidate_collections();
                $pipeline->invalidate_hive_current_load();
 
                $list_of_analyses = $pipeline->collection_of('Analysis')->find_all_by_pattern( $analyses_pattern );
            }
        }
    }
 
    # in this section, the beekeeper determines why it exited, sets an appropriate cause of death,
    # and prints/logs an appropriate message
    my @stringified_reasons_builder;
    my $beekeeper_cause_of_death;
    my $cause_of_death_is_error;
    my %exit_statuses; # keep a set of unique exit statuses seen
    if ($reasons_to_exit) {
        foreach my $reason_to_exit (@$reasons_to_exit) {
            $exit_statuses{$reason_to_exit->{'exit_status'}} = 1;
            push(@stringified_reasons_builder, $reason_to_exit->{'message'});
        }
    }
 
    my $stringified_reasons = join(", ", @stringified_reasons_builder);
 
    if (($loop_until eq 'JOB_FAILURE') &&
        (grep(/JOB_FAILED/, keys(%exit_statuses)))) {
        $beekeeper_cause_of_death = 'JOB_FAILED';
        $cause_of_death_is_error = 1;
    }
 
    if (($loop_until eq 'ANALYSIS_FAILURE') &&
        (grep(/ANALYSIS_FAILED/, keys(%exit_statuses)))) {
        $beekeeper_cause_of_death = 'ANALYSIS_FAILED';
        $cause_of_death_is_error = 1;
    }
 
    if (!$beekeeper_cause_of_death) {
        if (grep(/NO_WORK/, keys(%exit_statuses))) {
            $beekeeper_cause_of_death = 'NO_WORK';
        } else {
            $beekeeper_cause_of_death = 'LOOP_LIMIT';
        }
        $cause_of_death_is_error = 0;
    }
 
    $self->{'logmessage_adaptor'}->store_beekeeper_message($self->{'beekeeper_id'},
        "stopped looping because of $stringified_reasons",
        $cause_of_death_is_error ? 'PIPELINE_ERROR' : 'INFO',
        $beekeeper_cause_of_death) unless $self->{'read_only'};
 
    if ($reasons_to_exit and $ENV{EHIVE_SLACK_WEBHOOK}) {
        send_beekeeper_message_to_slack($ENV{EHIVE_SLACK_WEBHOOK}, $self->{'pipeline'}, $cause_of_death_is_error, 1, $stringified_reasons, $loop_until);
    }
 
    $self->{'beekeeper'}->set_cause_of_death($beekeeper_cause_of_death) unless $self->{'read_only'};
    printf("Beekeeper: dbc %d disconnect cycles\n", $hive_dba->dbc->disconnect_count);
    return $cause_of_death_is_error;
}

Functions

Function Documentation

◆ big_red_button()

◆ generate_worker_cmd()

◆ log_and_die()

◆ main()

◆ register_beekeeper()

◆ run_autonomously()