Inheritance diagram for Bio::EnsEMBL::Hive::Queen:

Collaboration diagram for Bio::EnsEMBL::Hive::Queen:

Public Member Functions
public	default_table_name ()

public	default_insertion_method ()

public	object_class ()

public Bio::EnsEMBL::Hive::Worker	create_new_worker ()

public	specialize_worker ()

public	register_worker_death ()

public	meadow_type_2_name_2_users_of_running_workers ()

public	check_for_dead_workers ()

public	check_in_worker ()

public void	reset_job_by_dbID_and_sync ()

public	fetch_overdue_workers ()

public	synchronize_hive ()

public Boolean	safe_synchronize_AnalysisStats ()

public	synchronize_AnalysisStats ()

public	check_nothing_to_run_but_semaphored ()

public	print_status_and_return_reasons_to_exit ()

public	register_all_workers_dead ()

public	interval_workers_with_unknown_usage ()

public	store_resource_usage ()

Public Member Functions inherited from Bio::EnsEMBL::Hive::DBSQL::ObjectAdaptor
public	object_class ()

public	slicer ()

public	objectify ()

public	mark_stored ()

public	keys_to_columns ()

public The	refresh ()

Public Member Functions inherited from Bio::EnsEMBL::Hive::DBSQL::BaseAdaptor
public	default_table_name ()

public	default_insertion_method ()

public	default_overflow_limit ()

public	default_input_column_mapping ()

public	do_not_update_columns ()

public	new ()

public	db ()

public	dbc ()

public	prepare ()

public	overflow_limit ()

public	input_column_mapping ()

public	table_name ()

public	insertion_method ()

public	column_set ()

public	primary_key ()

public	updatable_column_list ()

public	autoinc_id ()

protected	_table_info_loader ()

public	count_all ()

public	fetch_all ()

public	primary_key_constraint ()

public	fetch_by_dbID ()

public	remove_all ()

public	remove ()

public	update ()

public	store_or_update_one ()

public	check_object_present_in_db_by_content ()

public	store ()

public	DESTROY ()

public	AUTOLOAD ()

Detailed Description

Description

    The Queen of the Hive based job control system is responsible to 'birthing' the
    correct number of workers of the right type so that they can find jobs to do.
    It will also free up jobs of Workers that died unexpectantly so that other workers
    can claim them to do.

    Hive based processing is a concept based on a more controlled version
    of an autonomous agent type system.  Each worker is not told what to do
    (like a centralized control system - like the current pipeline system)
    but rather queries a central database for jobs (give me jobs).

    Each worker is linked to an analysis_id, registers its self on creation
    into the Hive, creates a RunnableDB instance of the Analysis->module,
    gets $analysis->stats->batch_size jobs from the job table, does its work,
    creates the next layer of job entries by interfacing to
    the DataflowRuleAdaptor to determine the analyses it needs to pass its
    output data to and creates jobs on the next analysis database.
    It repeats this cycle until it has lived its lifetime or until there are no
    more jobs left.
    The lifetime limit is just a safety limit to prevent these from 'infecting'
    a system.

    The Queens job is to simply birth Workers of the correct analysis_id to get the
    work down.  The only other thing the Queen does is free up jobs that were
    claimed by Workers that died unexpectantly so that other workers can take
    over the work.

    The Beekeeper is in charge of interfacing between the Queen and a compute resource
    or 'compute farm'.  Its job is to query Queens if they need any workers and to
    send the requested number of workers to open machines via the runWorker.pl script.
    It is also responsible for interfacing with the Queen to identify worker which died
    unexpectantly.

Member Function Documentation

◆ check_for_dead_workers()

public Bio::EnsEMBL::Hive::Queen::check_for_dead_workers ( )

Undocumented method

Code:

click to view

sub check_for_dead_workers {    # scans the whole Valley for lost Workers (but ignores unreachable ones)
    my ($self, $valley, $check_buried_in_haste, $bury_unkwn_workers) = @_;
    my $last_few_seconds            = 5;    # FIXME: It is probably a good idea to expose this parameter for easier tuning.
    warn "GarbageCollector:\tChecking for lost Workers...\n";
    my $meadow_type_2_name_2_users      = $self->meadow_type_2_name_2_users_of_running_workers();
    my %signature_and_pid_to_worker_status = ();
    while(my ($meadow_type, $level2) = each %$meadow_type_2_name_2_users) {
        if(my $meadow = $valley->available_meadow_hash->{$meadow_type}) {   # if this Valley supports $meadow_type at all...
            while(my ($meadow_name, $level3) = each %$level2) {
                if($meadow->cached_name eq $meadow_name) {  # and we can reach the same $meadow_name from this Valley...
                    my $meadow_users_of_interest    = [ keys %$level3 ];
                    my $meadow_signature            = $meadow_type.'/'.$meadow_name;
                    $signature_and_pid_to_worker_status{$meadow_signature} ||= $meadow->status_of_all_our_workers( $meadow_users_of_interest );
                }
            }
        }
    }
    my $queen_overdue_workers       = $self->fetch_overdue_workers( $last_few_seconds );    # check the workers we have not seen active during the $last_few_seconds
    warn "GarbageCollector:\t[Queen:] out of ".scalar(@$queen_overdue_workers)." Workers that haven't checked in during the last $last_few_seconds seconds...\n";
    my $update_when_seen_sql = "UPDATE worker SET when_seen=CURRENT_TIMESTAMP WHERE worker_id=?";
    my $update_when_seen_sth;
    my %meadow_status_counts        = ();
    my %mt_and_pid_to_lost_worker   = ();
    foreach my $worker (@$queen_overdue_workers) {
        my $meadow_signature    = $worker->meadow_type.'/'.$worker->meadow_name;
        if(my $pid_to_worker_status = $signature_and_pid_to_worker_status{$meadow_signature}) {   # the whole Meadow subhash is either present or the Meadow is unreachable
            my $meadow_type = $worker->meadow_type;
            my $process_id  = $worker->process_id;
            my $status = $pid_to_worker_status->{$process_id};
            if($bury_unkwn_workers and ($status eq 'UNKWN')) {
                if( my $meadow = $valley->find_available_meadow_responsible_for_worker( $worker ) ) {
                    if($meadow->can('kill_worker')) {
                        if($worker->meadow_user eq $ENV{'USER'}) {  # if I'm actually allowed to kill the worker...
                            warn "GarbageCollector:\tKilling/forgetting the UNKWN worker by process_id $process_id";
                            $meadow->kill_worker($worker, 1);
                            $status = ''; # make it look like LOST
                        }
                    }
                }
            }
            if($status) {  # can be RUN|PEND|xSUSP
                $meadow_status_counts{$meadow_signature}{$status}++;
                    # only prepare once at most:
                $update_when_seen_sth ||= $self->prepare( $update_when_seen_sql );
                $update_when_seen_sth->execute( $worker->dbID );
            } else {
                $meadow_status_counts{$meadow_signature}{'LOST'}++;
                $mt_and_pid_to_lost_worker{$meadow_type}{$process_id} = $worker;
            }
        } else {
            $meadow_status_counts{$meadow_signature}{'UNREACHABLE'}++;   # Worker is unreachable from this Valley
        }
    }
    $update_when_seen_sth->finish() if $update_when_seen_sth;
        # print a quick summary report:
    while(my ($meadow_signature, $status_count) = each %meadow_status_counts) {
        warn "GarbageCollector:\t[$meadow_signature Meadow:]\t".join(', ', map { "$_:$status_count->{$_}" } keys %$status_count )."\n\n";
    }
    while(my ($meadow_type, $pid_to_lost_worker) = each %mt_and_pid_to_lost_worker) {
        my $this_meadow = $valley->available_meadow_hash->{$meadow_type};
        if(my $lost_this_meadow = scalar(keys %$pid_to_lost_worker) ) {
            warn "GarbageCollector:\tDiscovered $lost_this_meadow lost $meadow_type Workers\n";
            my $report_entries;
            if($this_meadow->can('find_out_causes')) {
                die "Your Meadow::$meadow_type driver now has to support get_report_entries_for_process_ids() method instead of find_out_causes(). Please update it.\n";
            } else {
                if ($report_entries = $this_meadow->get_report_entries_for_process_ids( keys %$pid_to_lost_worker )) {
                    my $lost_with_known_cod = scalar( grep { $_->{'cause_of_death'} } values %$report_entries);
                    warn "GarbageCollector:\tFound why $lost_with_known_cod of $meadow_type Workers died\n";
                }
            }
            warn "GarbageCollector:\tReleasing the jobs\n";
            while(my ($process_id, $worker) = each %$pid_to_lost_worker) {
                $worker->when_died(         $report_entries->{$process_id}{'when_died'} );
                $worker->cause_of_death(    $report_entries->{$process_id}{'cause_of_death'} );
                $self->register_worker_death( $worker );
            }
            if( %$report_entries ) {    # use the opportunity to also store resource usage of the buried workers:
                my $processid_2_workerid = { map { $_ => $pid_to_lost_worker->{$_}->dbID } keys %$pid_to_lost_worker };
                $self->store_resource_usage( $report_entries, $processid_2_workerid );
            }
        }
    }
        # the following bit is completely Meadow-agnostic and only restores database integrity:
    if($check_buried_in_haste) {
        warn "GarbageCollector:\tChecking for orphan roles...\n";
        my $orphan_roles = $self->db->get_RoleAdaptor->fetch_all_unfinished_roles_of_dead_workers();
        if(my $orphan_role_number = scalar @$orphan_roles) {
            warn "GarbageCollector:\tfound $orphan_role_number orphan roles, finalizing...\n\n";
            foreach my $orphan_role (@$orphan_roles) {
                $self->db->get_RoleAdaptor->finalize_role( $orphan_role );
            }
        }
        warn "GarbageCollector:\tChecking for orphan jobs...\n";
        my $buried_in_haste_roles = $self->db->get_RoleAdaptor->fetch_all_finished_roles_with_unfinished_jobs();
        if(my $bih_number = scalar @$buried_in_haste_roles) {
            warn "GarbageCollector:\tfound $bih_number buried roles with orphan jobs, reclaiming.\n\n";
            my $job_adaptor = $self->db->get_AnalysisJobAdaptor;
            foreach my $role (@$buried_in_haste_roles) {
                $job_adaptor->release_undone_jobs_from_role( $role );
            }
        } else {
            warn "GarbageCollector:\tfound none\n";
        }
    }
}

◆ check_in_worker()

public Bio::EnsEMBL::Hive::Queen::check_in_worker ( )

Undocumented method

Code:

click to view

◆ check_nothing_to_run_but_semaphored()

public Bio::EnsEMBL::Hive::Queen::check_nothing_to_run_but_semaphored ( )

  Arg [1]    : $list_of_analyses
  Example    :

$self->check_nothing_to_run_but_semaphored( [ $analysis_A, $analysis_B ] );

  Description: Counts the number of immediately runnable jobs in the given analyses.
  Exceptions : none
  Caller     : Scheduler

Code:

click to view

◆ create_new_worker()

public Bio::EnsEMBL::Hive::Worker Bio::EnsEMBL::Hive::Queen::create_new_worker ( )

  Description: Creates an entry in the worker table,
               populates some non-storable attributes
               and returns a Worker object based on that insert.
               This guarantees that each worker registered in this Queen's hive is properly registered.
  Returntype : Bio::EnsEMBL::Hive::Worker
  Caller     : runWorker.pl

Code:

click to view

sub create_new_worker {
    my $self    = shift @_;
    my %flags   = @_;
    my ($meadow_type, $meadow_name, $process_id, $meadow_host, $meadow_user, $resource_class_id, $resource_class_name,
        $no_write, $debug, $worker_log_dir, $hive_log_dir, $job_limit, $life_span, $no_cleanup, $retry_throwing_jobs, $can_respecialize)
     = @flags{qw(-meadow_type -meadow_name -process_id -meadow_host -meadow_user -resource_class_id -resource_class_name
            -no_write -debug -worker_log_dir -hive_log_dir -job_limit -life_span -no_cleanup -retry_throwing_jobs -can_respecialize)};
    foreach my $prev_worker_incarnation (@{ $self->fetch_all( "status!='DEAD' AND meadow_type='$meadow_type' AND meadow_name='$meadow_name' AND process_id='$process_id'" ) }) {
            # so far 'RELOCATED events' has been detected on LSF 9.0 in response to sending signal #99 or #100
            # Since I don't know how to avoid them, I am trying to register them when they happen.
            # The following snippet buries the previous incarnation of the Worker before starting a new one.
            #
            # FIXME: if GarabageCollector (beekeeper -dead) gets to these processes first, it will register them as DEAD/UNKNOWN.
            #       LSF 9.0 does not report "rescheduling" events in the output of 'bacct', but does mention them in 'bhist'.
            #       So parsing 'bhist' output would probably yield the most accurate & confident registration of these events.
        $prev_worker_incarnation->cause_of_death( 'RELOCATED' );
        $self->register_worker_death( $prev_worker_incarnation );
    }
    my $resource_class;
    if( defined($resource_class_name) ) {
        $resource_class = $self->db->get_ResourceClassAdaptor->fetch_by_name($resource_class_name)
            or die "resource_class with name='$resource_class_name' could not be fetched from the database";
    } elsif( defined($resource_class_id) ) {
        $resource_class = $self->db->get_ResourceClassAdaptor->fetch_by_dbID($resource_class_id)
            or die "resource_class with dbID='$resource_class_id' could not be fetched from the database";
    }
    my $worker = Bio::EnsEMBL::Hive::Worker->new(
        'meadow_type'       => $meadow_type,
        'meadow_name'       => $meadow_name,
        'meadow_host'       => $meadow_host,
        'meadow_user'       => $meadow_user,
        'process_id'        => $process_id,
        'resource_class'    => $resource_class,
    );
    $self->store( $worker );
    my $worker_id = $worker->dbID;
    $worker = $self->fetch_by_dbID( $worker_id )    # refresh the object to get the fields initialized at SQL level (timestamps in this case)
        or die "Could not fetch worker with dbID=$worker_id";
    if($hive_log_dir or $worker_log_dir) {
        my $dir_revhash = dir_revhash($worker_id);
        $worker_log_dir ||= $hive_log_dir .'/'. ($dir_revhash ? "$dir_revhash/" : '') .'worker_id_'.$worker_id;
        eval {
            make_path( $worker_log_dir );
            1;

◆ default_insertion_method()

public Bio::EnsEMBL::Hive::Queen::default_insertion_method ( )

Undocumented method

Code:

click to view

◆ default_table_name()

public Bio::EnsEMBL::Hive::Queen::default_table_name ( )

Undocumented method

Code:

click to view

◆ fetch_overdue_workers()

public Bio::EnsEMBL::Hive::Queen::fetch_overdue_workers ( )

Undocumented method

Code:

click to view

◆ interval_workers_with_unknown_usage()

public Bio::EnsEMBL::Hive::Queen::interval_workers_with_unknown_usage ( )

Undocumented method

Code:

click to view

◆ meadow_type_2_name_2_users_of_running_workers()

public Bio::EnsEMBL::Hive::Queen::meadow_type_2_name_2_users_of_running_workers ( )

Undocumented method

Code:

click to view

◆ object_class()

public Bio::EnsEMBL::Hive::Queen::object_class ( )

Undocumented method

Code:

click to view

◆ print_status_and_return_reasons_to_exit()

public Bio::EnsEMBL::Hive::Queen::print_status_and_return_reasons_to_exit ( )

  Arg [1]    : $list_of_analyses
  Example    :

my $reasons_to_exit = $queen->print_status_and_return_reasons_to_exit( [ $analysis_A, $analysis_B ] );

  Description: Runs through all analyses in the given list, reports failed analyses, computes some totals, prints a combined status line
                and returns a pair of ($failed_analyses_counter, $total_jobs_to_do)
  Exceptions : none
  Caller     : beekeeper.pl

Code:

click to view

◆ register_all_workers_dead()

public Bio::EnsEMBL::Hive::Queen::register_all_workers_dead ( )

  Example    :

$queen->register_all_workers_dead();

  Description: Registers all workers dead
  Exceptions : none
  Caller     : beekeeper.pl

Code:

click to view

◆ register_worker_death()

public Bio::EnsEMBL::Hive::Queen::register_worker_death ( )

Undocumented method

Code:

click to view

◆ reset_job_by_dbID_and_sync()

public void Bio::EnsEMBL::Hive::Queen::reset_job_by_dbID_and_sync ( )

  Arg [1]: int $job_id
  Example: 
    my $job = $queen->reset_job_by_dbID_and_sync($job_id);
  Description: 
    For the specified job_id it will fetch just that job, 
    reset it completely as if it has never run, and return it.  
    Specifying a specific job bypasses the safety checks, 
    thus multiple workers could be running the 
    same job simultaneously (use only for debugging).
  Returntype : none
  Exceptions :
  Caller     : beekeeper.pl

Code:

click to view

◆ safe_synchronize_AnalysisStats()

public Boolean Bio::EnsEMBL::Hive::Queen::safe_synchronize_AnalysisStats ( )

  Arg [1]    : Bio::EnsEMBL::Hive::AnalysisStats object
  Example    :

$self->safe_synchronize_AnalysisStats($stats);

  Description: Prewrapper around synchronize_AnalysisStats that does
               checks and grabs sync_lock before proceeding with sync.
               Used by distributed worker sync system to avoid contention.
               Returns 1 on success and 0 if the lock could not have been obtained,
               and so no sync was attempted.
  Returntype : boolean
  Caller     : general

Code:

click to view

◆ specialize_worker()

public Bio::EnsEMBL::Hive::Queen::specialize_worker ( )

  Description: If analysis_id or logic_name is specified it will try to specialize the Worker into this analysis.
               If not specified the Queen will analyze the hive and pick the most suitable analysis.
  Caller     : Bio::EnsEMBL::Hive::Worker

Code:

click to view

sub specialize_worker {
    my $self    = shift @_;
    my $worker  = shift @_;
    my $flags   = shift @_;
    my ($analyses_pattern, $job_id, $force)
     = @$flags{qw(-analyses_pattern -job_id -force)};
    if( $analyses_pattern and $job_id ) {
        die "At most one of the options {-analyses_pattern, -job_id} can be set to pre-specialize a Worker";
    }
    my $analysis;
    if( $job_id ) {
        warn "resetting and fetching job for job_id '$job_id'\n";
        my $job_adaptor = $self->db->get_AnalysisJobAdaptor;
        my $job = $job_adaptor->fetch_by_dbID( $job_id )
            or die "Could not fetch job with dbID='$job_id'";
        my $job_status = $job->status();
        if($job_status =~/(CLAIMED|PRE_CLEANUP|FETCH_INPUT|RUN|WRITE_OUTPUT|POST_CLEANUP)/ ) {
            die "Job with dbID='$job_id' is already in progress, cannot run";   # FIXME: try GC first, then complain
        } elsif($job_status =~/(DONE|SEMAPHORED)/ and !$force) {
            die "Job with dbID='$job_id' is $job_status, please use -force 1 to override";
        }
        $analysis = $job->analysis;
        if(($analysis->stats->status eq 'BLOCKED') and !$force) {
            die "Analysis is BLOCKED, can't specialize a worker. Please use -force 1 to override";
        }
        if(($job_status eq 'DONE') and $job->semaphored_job_id) {
            warn "Increasing the semaphore count of the dependent job";
            $job_adaptor->increase_semaphore_count_for_jobid( $job->semaphored_job_id );
        }
        my %status2counter = ('FAILED' => 'failed_job_count', 'READY' => 'ready_job_count', 'DONE' => 'done_job_count', 'PASSED_ON' => 'done_job_count', 'SEMAPHORED' => 'semaphored_job_count');
        $analysis->stats->adaptor->increment_a_counter( $status2counter{$job->status}, -1, $job->analysis_id );
    } else {
        $analyses_pattern //= '%';  # for printing
        my $analyses_matching_pattern   = Bio::EnsEMBL::Hive::Analysis->collection()->find_all_by_pattern( $analyses_pattern );
            # refresh the stats of matching analyses before re-specialization:
        foreach my $analysis ( @$analyses_matching_pattern ) {
            $analysis->stats->refresh();
        }
        $analysis = Bio::EnsEMBL::Hive::Scheduler::suggest_analysis_to_specialize_a_worker($worker, $analyses_matching_pattern, $analyses_pattern);
        unless( ref($analysis) ) {
            $worker->cause_of_death('NO_ROLE');
            my $msg = $analysis // "No analysis suitable for the worker was found";
            die "$msg\n";
        }
    }
    my $new_role = Bio::EnsEMBL::Hive::Role->new(
        'worker'        => $worker,
        'analysis'      => $analysis,
    );
    $self->db->get_RoleAdaptor->store( $new_role );
    $worker->current_role( $new_role );
    my $analysis_stats_adaptor = $self->db->get_AnalysisStatsAdaptor;
    if($job_id) {
        my $role_id = $new_role->dbID;
        if( my $job = $self->db->get_AnalysisJobAdaptor->reset_or_grab_job_by_dbID($job_id, $role_id) ) {
            $worker->special_batch( [ $job ] );
        } else {
            die "Could not claim job with dbID='$job_id' for Role with dbID='$role_id'";
        }
    } else {    # Note: special batch Workers should avoid flipping the status to 'WORKING' in case the analysis is still 'BLOCKED'
        $analysis_stats_adaptor->update_status($analysis->dbID, 'WORKING');
    }
        # The following increment used to be done only when no specific task was given to the worker,
        # thereby excluding such "special task" workers from being counted in num_running_workers.
        #
        # However this may be tricky to emulate by triggers that know nothing about "special tasks",
        # so I am (temporarily?) simplifying the accounting algorithm.
        #
    $analysis_stats_adaptor->increment_a_counter( 'num_running_workers', 1, $analysis->dbID );
}

◆ store_resource_usage()

public Bio::EnsEMBL::Hive::Queen::store_resource_usage ( )

Undocumented method

Code:

click to view

◆ synchronize_AnalysisStats()

public Bio::EnsEMBL::Hive::Queen::synchronize_AnalysisStats ( )

  Arg [1]    : Bio::EnsEMBL::Hive::AnalysisStats object
  Example    :

$self->synchronize_AnalysisStats( $stats );

  Description: Queries the job and worker tables to get summary counts
               and rebuilds the AnalysisStats object.
               Then updates the analysis_stats table with the new summary info.
  Exceptions : none
  Caller     : general

Code:

click to view

◆ synchronize_hive()

public Bio::EnsEMBL::Hive::Queen::synchronize_hive ( )

  Arg [1]    : $list_of_analyses
  Example    :

$queen->synchronize_hive( [ $analysis_A, $analysis_B ] );

  Description: Runs through all analyses in the given list and synchronizes
              the analysis_stats summary with the states in the job and worker tables.
              Then follows by checking all the blocking rules and blocks/unblocks analyses as needed.
  Exceptions : none
  Caller     : general

Code:

click to view

The documentation for this class was generated from the following file:

modules/Bio/EnsEMBL/Hive/Queen.pm

Public Member Functions

Detailed Description

Description

Member Function Documentation

◆ check_for_dead_workers()

◆ check_in_worker()

◆ check_nothing_to_run_but_semaphored()

◆ create_new_worker()

◆ default_insertion_method()

◆ default_table_name()

◆ fetch_overdue_workers()

◆ interval_workers_with_unknown_usage()

◆ meadow_type_2_name_2_users_of_running_workers()

◆ object_class()

◆ print_status_and_return_reasons_to_exit()

◆ register_all_workers_dead()

◆ register_worker_death()

◆ reset_job_by_dbID_and_sync()

◆ safe_synchronize_AnalysisStats()

◆ specialize_worker()

◆ store_resource_usage()

◆ synchronize_AnalysisStats()

◆ synchronize_hive()