Skip to content

GitLab

Commit baa146d3 authored by Leo Gordon's avatar Leo Gordon
Browse files

big renaming of hive->worker, analysis_job->job, analysis_job_file->job_file,... 

big renaming of hive->worker, analysis_job->job, analysis_job_file->job_file, analysis_job_id->job_id, worker.beekeeper->worker.meadow_type, etc
parent 7dbe9126
Hide whitespace changes
Inline Side-by-side
Showing with 197 additions and 208 deletions
docs/long_mult_example_pipeline.txt
View file @ baa146d3
......@@ -63,16 +63,16 @@
or run it in step-by-step mode, initiating every step by separate executions of 'beekeeper.pl ... -run' command.
We will use the step-by-step mode in order to see what is going on.
5.4 Go to mysql window and check the contents of analysis_job table:
5.4 Go to mysql window and check the contents of job table:
MySQL> SELECT * FROM analysis_job;
MySQL> SELECT * FROM job;
It will only contain jobs that set up the multiplication tasks in 'READY' mode - meaning 'ready to be taken by workers and executed'.
Go to the beekeeper window and run the 'beekeeper.pl ... -run' once.
It will submit a worker to the farm that will at some point get the 'start' job(s).
5.5 Go to mysql window again and check the contents of analysis_job table. Keep checking as the worker may spend some time in 'pending' state.
5.5 Go to mysql window again and check the contents of job table. Keep checking as the worker may spend some time in 'pending' state.
After the first worker is done you will see that 'start' jobs are now done and new 'part_multiply' and 'add_together' jobs have been created.
Also check the contents of 'intermediate_result' table, it should be empty at that moment:
......@@ -82,7 +82,7 @@
Go back to the beekeeper window and run the 'beekeeper.pl ... -run' for the second time.
It will submit another worker to the farm that will at some point get the 'part_multiply' jobs.
5.6 Now check both 'analysis_job' and 'intermediate_result' tables again.
5.6 Now check both 'job' and 'intermediate_result' tables again.
At some moment 'part_multiply' jobs will have been completed and the results will go into 'intermediate_result' table;
'add_together' jobs are still to be done.
......
modules/Bio/EnsEMBL/Hive.pm
View file @ baa146d3
......@@ -25,7 +25,7 @@
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 relevant configuration information from the database, does its
work, creates the next layer of analysis_job entries by interfacing to
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 database of the next analysis.
It repeats this cycle until it has lived its lifetime or until there are no
......@@ -37,7 +37,7 @@
The Queen's primary job is to create Workers to get the work down.
As part of this, she is also responsible for summarizing the status of the
analyses by querying the analysis_jobs, summarizing, and updating the
analyses by querying the jobs, summarizing, and updating the
analysis_stats table. From this she is also responsible for monitoring and
'unblocking' analyses via the analysis_ctrl_rules.
The Queen is also responsible for freeing up jobs that were claimed by Workers
......
modules/Bio/EnsEMBL/Hive/DBSQL/AnalysisJobAdaptor.pm
View file @ baa146d3
......@@ -59,9 +59,9 @@ use base ('Bio::EnsEMBL::DBSQL::BaseAdaptor');
Args : -input_id => string of input_id which will be passed to run the job (or a Perl hash that will be automagically stringified)
-analysis => Bio::EnsEMBL::Analysis object from a database
-block => int(0,1) set blocking state of job (default = 0)
-input_job_id => (optional) analysis_job_id of job that is creating this
-input_job_id => (optional) job_id of job that is creating this
job. Used purely for book keeping.
Example : $analysis_job_id = Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor->CreateNewJob(
Example : $job_id = Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor->CreateNewJob(
-input_id => 'my input data',
-analysis => $myAnalysis);
Description: uses the analysis object to get the db connection from the adaptor to store a new
......@@ -69,7 +69,7 @@ use base ('Bio::EnsEMBL::DBSQL::BaseAdaptor');
Also updates corresponding analysis_stats by incrementing total_job_count,
unclaimed_job_count and flagging the incremental update by changing the status
to 'LOADING' (but only if the analysis is not blocked).
Returntype : int analysis_job_id on database analysis is from.
Returntype : int job_id on database analysis is from.
Exceptions : thrown if either -input_id or -analysis are not properly defined
Caller : general
......@@ -80,7 +80,7 @@ sub CreateNewJob {
return undef unless(scalar @args);
my ($input_id, $analysis, $prev_analysis_job_id, $blocked, $semaphore_count, $semaphored_job_id) =
my ($input_id, $analysis, $prev_job_id, $blocked, $semaphore_count, $semaphored_job_id) =
rearrange([qw(INPUT_ID ANALYSIS INPUT_JOB_ID BLOCK SEMAPHORE_COUNT SEMAPHORED_JOB_ID)], @args);
throw("must define input_id") unless($input_id);
......@@ -99,15 +99,15 @@ sub CreateNewJob {
$input_id = "_ext_input_analysis_data_id $input_data_id";
}
my $sql = q{INSERT ignore into analysis_job
(input_id, prev_analysis_job_id,analysis_id,status,semaphore_count,semaphored_job_id)
my $sql = q{INSERT ignore into job
(input_id, prev_job_id,analysis_id,status,semaphore_count,semaphored_job_id)
VALUES (?,?,?,?,?,?)};
my $status = $blocked ? 'BLOCKED' : 'READY';
my $dbc = $analysis->adaptor->db->dbc;
my $sth = $dbc->prepare($sql);
$sth->execute($input_id, $prev_analysis_job_id, $analysis->dbID, $status, $semaphore_count, $semaphored_job_id);
$sth->execute($input_id, $prev_job_id, $analysis->dbID, $status, $semaphore_count, $semaphored_job_id);
my $job_id = $sth->{'mysql_insertid'};
$sth->finish;
......@@ -131,7 +131,7 @@ sub CreateNewJob {
Arg [1] : int $id
the unique database identifier for the feature to be obtained
Example : $feat = $adaptor->fetch_by_dbID(1234);
Description: Returns the AnalysisJob defined by the analysis_job_id $id.
Description: Returns the AnalysisJob defined by the job_id $id.
Returntype : Bio::EnsEMBL::Hive::AnalysisJob
Exceptions : thrown if $id is not defined
Caller : general
......@@ -191,8 +191,8 @@ sub fetch_all {
sub fetch_all_failed_jobs {
my ($self,$analysis_id) = @_;
my $constraint = "a.status='FAILED'";
$constraint .= " AND a.analysis_id=$analysis_id" if($analysis_id);
my $constraint = "j.status='FAILED'";
$constraint .= " AND j.analysis_id=$analysis_id" if($analysis_id);
return $self->_generic_fetch($constraint);
}
......@@ -200,7 +200,7 @@ sub fetch_all_failed_jobs {
sub fetch_all_incomplete_jobs_by_worker_id {
my ($self, $worker_id) = @_;
my $constraint = "a.status IN ('COMPILATION','GET_INPUT','RUN','WRITE_OUTPUT') AND a.worker_id='$worker_id'";
my $constraint = "j.status IN ('COMPILATION','GET_INPUT','RUN','WRITE_OUTPUT') AND j.worker_id='$worker_id'";
return $self->_generic_fetch($constraint);
}
......@@ -281,25 +281,25 @@ sub _generic_fetch {
sub _tables {
my $self = shift;
return (['analysis_job', 'a']);
return (['job', 'j']);
}
sub _columns {
my $self = shift;
return qw (a.analysis_job_id
a.prev_analysis_job_id
a.analysis_id
a.input_id
a.worker_id
a.status
a.retry_count
a.completed
a.runtime_msec
a.query_count
a.semaphore_count
a.semaphored_job_id
return qw (j.job_id
j.prev_job_id
j.analysis_id
j.input_id
j.worker_id
j.status
j.retry_count
j.completed
j.runtime_msec
j.query_count
j.semaphore_count
j.semaphored_job_id
);
}
......@@ -330,7 +330,7 @@ sub _objs_from_sth {
: $column{'input_id'};
my $job = Bio::EnsEMBL::Hive::AnalysisJob->new(
-DBID => $column{'analysis_job_id'},
-DBID => $column{'job_id'},
-ANALYSIS_ID => $column{'analysis_id'},
-INPUT_ID => $input_id,
-WORKER_ID => $column{'worker_id'},
......@@ -362,7 +362,7 @@ sub decrease_semaphore_count_for_jobid { # used in semaphore annihilation or
my $jobid = shift @_;
my $dec = shift @_ || 1;
my $sql = "UPDATE analysis_job SET semaphore_count=semaphore_count-? WHERE analysis_job_id=?";
my $sql = "UPDATE job SET semaphore_count=semaphore_count-? WHERE job_id=?";
my $sth = $self->prepare($sql);
$sth->execute($dec, $jobid);
......@@ -374,7 +374,7 @@ sub increase_semaphore_count_for_jobid { # used in semaphore propagation
my $jobid = shift @_;
my $inc = shift @_ || 1;
my $sql = "UPDATE analysis_job SET semaphore_count=semaphore_count+? WHERE analysis_job_id=?";
my $sql = "UPDATE job SET semaphore_count=semaphore_count+? WHERE job_id=?";
my $sth = $self->prepare($sql);
$sth->execute($inc, $jobid);
......@@ -386,7 +386,7 @@ sub increase_semaphore_count_for_jobid { # used in semaphore propagation
Arg [1] : $analysis_id
Example :
Description: updates the analysis_job.status in the database
Description: updates the job.status in the database
Returntype :
Exceptions :
Caller : general
......@@ -396,7 +396,7 @@ sub increase_semaphore_count_for_jobid { # used in semaphore propagation
sub update_status {
my ($self, $job) = @_;
my $sql = "UPDATE analysis_job SET status='".$job->status."' ";
my $sql = "UPDATE job SET status='".$job->status."' ";
if($job->status eq 'DONE') {
$sql .= ",completed=now()";
......@@ -407,7 +407,7 @@ sub update_status {
} elsif($job->status eq 'READY') {
}
$sql .= " WHERE analysis_job_id='".$job->dbID."' ";
$sql .= " WHERE job_id='".$job->dbID."' ";
# This particular query is infamous for collisions and 'deadlock' situations; let's make them wait and retry.
foreach (0..3) {
......@@ -445,12 +445,12 @@ sub store_out_files {
return unless($job);
my $sql = sprintf("DELETE from analysis_job_file WHERE worker_id=%d and analysis_job_id=%d",
my $sql = sprintf("DELETE from job_file WHERE worker_id=%d and job_id=%d",
$job->worker_id, $job->dbID);
$self->dbc->do($sql);
return unless($job->stdout_file or $job->stderr_file);
$sql = "INSERT ignore INTO analysis_job_file (analysis_job_id, worker_id, retry, type, path) VALUES ";
$sql = "INSERT ignore INTO job_file (job_id, worker_id, retry, type, path) VALUES ";
if($job->stdout_file) {
$sql .= sprintf("(%d,%d,%d,'STDOUT','%s')", $job->dbID, $job->worker_id,
$job->retry_count, $job->stdout_file);
......@@ -484,10 +484,10 @@ sub grab_jobs_for_worker {
my ($self, $worker) = @_;
my $analysis_id = $worker->analysis->dbID();
my $worker_id = $worker->worker_id();
my $worker_id = $worker->dbID();
my $sql_base = qq{
UPDATE analysis_job
UPDATE job
SET worker_id='$worker_id', status='CLAIMED'
WHERE analysis_id='$analysis_id' AND status='READY' AND semaphore_count<=0
};
......@@ -504,7 +504,7 @@ sub grab_jobs_for_worker {
$claim_count = $self->dbc->do($sql_any);
}
my $constraint = "a.analysis_id='$analysis_id' AND a.worker_id='$worker_id' AND a.status='CLAIMED'";
my $constraint = "j.analysis_id='$analysis_id' AND j.worker_id='$worker_id' AND j.status='CLAIMED'";
return $self->_generic_fetch($constraint);
}
......@@ -514,16 +514,16 @@ sub reclaim_job_for_worker {
my $worker = shift or return;
my $job = shift or return;
my $worker_id = $worker->worker_id();
my $worker_id = $worker->dbID();
my $job_id = $job->dbID;
my $sql = "UPDATE analysis_job SET status='CLAIMED', worker_id=? WHERE analysis_job_id=? AND status='READY'";
my $sql = "UPDATE job SET status='CLAIMED', worker_id=? WHERE job_id=? AND status='READY'";
my $sth = $self->prepare($sql);
$sth->execute($worker_id, $job_id);
$sth->finish;
my $constraint = "a.analysis_job_id='$job_id' AND a.worker_id='$worker_id' AND a.status='CLAIMED'";
my $constraint = "j.job_id='$job_id' AND j.worker_id='$worker_id' AND j.status='CLAIMED'";
return $self->_generic_fetch($constraint);
}
......@@ -549,20 +549,20 @@ sub release_undone_jobs_from_worker {
my ($self, $worker, $msg) = @_;
my $max_retry_count = $worker->analysis->stats->max_retry_count();
my $worker_id = $worker->worker_id();
my $worker_id = $worker->dbID();
#first just reset the claimed jobs, these don't need a retry_count index increment:
# (previous worker_id does not matter, because that worker has never had a chance to run the job)
$self->dbc->do( qq{
UPDATE analysis_job
UPDATE job
SET status='READY', worker_id=NULL
WHERE status='CLAIMED'
AND worker_id='$worker_id'
} );
my $sth = $self->prepare( qq{
SELECT analysis_job_id
FROM analysis_job
SELECT job_id
FROM job
WHERE worker_id='$worker_id'
AND status in ('COMPILATION','GET_INPUT','RUN','WRITE_OUTPUT')
} );
......@@ -606,9 +606,9 @@ sub release_and_age_job {
# FIXME: would it be possible to retain worker_id for READY jobs in order to temporarily keep track of the previous (failed) worker?
#
$self->dbc->do( qq{
UPDATE analysis_job
UPDATE job
SET status=IF( $may_retry AND (retry_count<$max_retry_count), 'READY', 'FAILED'), retry_count=retry_count+1
WHERE analysis_job_id=$job_id
WHERE job_id=$job_id
AND status in ('COMPILATION','GET_INPUT','RUN','WRITE_OUTPUT')
} );
}
......@@ -655,9 +655,9 @@ sub reset_job_by_dbID {
my $job_id = shift or throw("job_id of the job to be reset is undefined");
$self->dbc->do( qq{
UPDATE analysis_job
UPDATE job
SET status='READY', retry_count=0
WHERE analysis_job_id=$job_id
WHERE job_id=$job_id
} );
}
......@@ -686,7 +686,7 @@ sub reset_all_jobs_for_analysis_id {
throw("must define analysis_id") unless($analysis_id);
my ($sql, $sth);
$sql = "UPDATE analysis_job SET status='READY' WHERE status!='BLOCKED' and analysis_id=?";
$sql = "UPDATE job SET status='READY' WHERE status!='BLOCKED' and analysis_id=?";
$sth = $self->prepare($sql);
$sth->execute($analysis_id);
$sth->finish;
......@@ -717,13 +717,13 @@ sub remove_analysis_id {
$self->dbc->do($sql);
$sql = "ANALYZE TABLE analysis_stats";
$self->dbc->do($sql);
$sql = "DELETE FROM analysis_job WHERE analysis_id=$analysis_id";
$sql = "DELETE FROM job WHERE analysis_id=$analysis_id";
$self->dbc->do($sql);
$sql = "ANALYZE TABLE analysis_job";
$sql = "ANALYZE TABLE job";
$self->dbc->do($sql);
$sql = "DELETE FROM hive WHERE analysis_id=$analysis_id";
$sql = "DELETE FROM worker WHERE analysis_id=$analysis_id";
$self->dbc->do($sql);
$sql = "ANALYZE TABLE hive";
$sql = "ANALYZE TABLE worker";
$self->dbc->do($sql);
}
......
modules/Bio/EnsEMBL/Hive/DBSQL/AnalysisStatsAdaptor.pm
View file @ baa146d3
......@@ -139,7 +139,7 @@ sub refresh {
sub get_running_worker_count {
my ($self, $stats) = @_;
my $sql = "SELECT count(*) FROM hive WHERE cause_of_death='' and analysis_id=?";
my $sql = "SELECT count(*) FROM worker WHERE cause_of_death='' and analysis_id=?";
my $sth = $self->prepare($sql);
$sth->execute($stats->analysis_id);
my ($liveCount) = $sth->fetchrow_array();
......
modules/Bio/EnsEMBL/Hive/DBSQL/DBAdaptor.pm
View file @ baa146d3
......@@ -36,17 +36,9 @@ Bio::EnsEMBL::Hive::DBSQL::DBAdaptor
package Bio::EnsEMBL::Hive::DBSQL::DBAdaptor;
use strict;
use Bio::EnsEMBL::DBSQL::DBConnection;
use base ('Bio::EnsEMBL::DBSQL::DBAdaptor');
#sub get_Queen {
# my $self = shift;
#
# return $self->get_QueenAdaptor();
#}
sub get_available_adaptors {
my %pairs = (
......
modules/Bio/EnsEMBL/Hive/DBSQL/JobMessageAdaptor.pm
View file @ baa146d3
......@@ -32,9 +32,9 @@ sub register_message {
# (the timestamp 'moment' column will be set automatically)
my $sql = qq{
REPLACE INTO job_message (analysis_job_id, worker_id, analysis_id, retry_count, status, msg, is_error)
SELECT analysis_job_id, worker_id, analysis_id, retry_count, status, ?, ?
FROM analysis_job WHERE analysis_job_id=?
REPLACE INTO job_message (job_id, worker_id, analysis_id, retry_count, status, msg, is_error)
SELECT job_id, worker_id, analysis_id, retry_count, status, ?, ?
FROM job WHERE job_id=?
};
my $sth = $self->prepare( $sql );
......
modules/Bio/EnsEMBL/Hive/DBSQL/NakedTableAdaptor.pm
View file @ baa146d3
......@@ -12,7 +12,7 @@
=head1 DESCRIPTION
This module together with its data container are used to enable dataflow into arbitrary tables (rather than just analysis_job).
This module together with its data container are used to enable dataflow into arbitrary tables (rather than just job).
NakedTable objects know *where* to dataflow, and NakedTableAdaptor knows *how* to dataflow.
......
modules/Bio/EnsEMBL/Hive/Meadow.pm
View file @ baa146d3
......@@ -78,7 +78,7 @@ sub generate_job_name {
sub responsible_for_worker {
my ($self, $worker) = @_;
return $worker->beekeeper() eq $self->type();
return $worker->meadow_type() eq $self->type();
}
sub check_worker_is_alive_and_mine {
......
modules/Bio/EnsEMBL/Hive/Params.pm
View file @ baa146d3
......@@ -37,11 +37,11 @@ By inheriting from this module you make your module able to deal with parameters
=head1 DESCRIPTION
Most of Compara RunnableDB methods work under assumption
that both analysis.parameters and analysis_job.input_id fields contain a Perl-style parameter hashref as a string.
that both analysis.parameters and job.input_id fields contain a Perl-style parameter hashref as a string.
This module implements a generic param() method that allows to set parameters according to the following parameter precedence rules:
(1) Job-Specific parameters defined in analysis_job.input_id hash, they have the highest priority and override everything else.
(1) Job-Specific parameters defined in job.input_id hash, they have the highest priority and override everything else.
(2) Analysis-Wide parameters defined in analysis.parameters hash. Can be overridden by (1).
......
modules/Bio/EnsEMBL/Hive/PipeConfig/LongMult_conf.pm
View file @ baa146d3
......@@ -67,7 +67,7 @@ use base ('Bio::EnsEMBL::Hive::PipeConfig::HiveGeneric_conf'); # All Hive datab
sub default_options {
my ($self) = @_;
return {
'ensembl_cvs_root_dir' => $ENV{'HOME'}.'/work', # some Compara developers might prefer $ENV{'HOME'}.'/ensembl_main'
'ensembl_cvs_root_dir' => $ENV{'ENSEMBL_CVS_ROOT_DIR'}, # it will make sense to set this variable if you are going to use ehive frequently
'pipeline_name' => 'long_mult', # name used by the beekeeper to prefix job names on the farm
......
modules/Bio/EnsEMBL/Hive/Process.pm
View file @ baa146d3
......@@ -112,7 +112,7 @@ sub new {
=head2 strict_hash_format
Title : strict_hash_format
Function: if a subclass wants more flexibility in parsing analysis_job.input_id and analysis.parameters,
Function: if a subclass wants more flexibility in parsing job.input_id and analysis.parameters,
it should redefine this method to return 0
=cut
......
modules/Bio/EnsEMBL/Hive/Queen.pm
View file @ baa146d3
......@@ -25,8 +25,8 @@
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 $worker->batch_size() jobs from the analysis_job table, does its
work, creates the next layer of analysis_job entries by interfacing to
gets $worker->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
......@@ -85,7 +85,7 @@ use base ('Bio::EnsEMBL::DBSQL::BaseAdaptor');
Description: If analysis_id is specified it will try to create a worker based
on that analysis. If not specified the queen will analyze the hive
and pick the analysis that has the most amount of work to be done.
It creates an entry in the hive table, and returns a Worker object
It creates an entry in the worker table, and returns a Worker object
based on that insert. This guarantees that each worker registered
in this queens hive is properly registered.
Returntype : Bio::EnsEMBL::Hive::Worker
......@@ -180,8 +180,8 @@ sub create_new_worker {
}
}
my $sql = q{INSERT INTO hive
(born, last_check_in, beekeeper, process_id, host, analysis_id)
my $sql = q{INSERT INTO worker
(born, last_check_in, meadow_type, process_id, host, analysis_id)
VALUES (NOW(), NOW(), ?,?,?,?)};
my $sth = $self->prepare($sql);
......@@ -189,7 +189,7 @@ sub create_new_worker {
my $worker_id = $sth->{'mysql_insertid'};
$sth->finish;
my $worker = $self->fetch_by_worker_id($worker_id);
my $worker = $self->fetch_by_dbID($worker_id);
$worker=undef unless($worker and $worker->analysis);
if($worker and $analysisStats) {
......@@ -242,11 +242,11 @@ sub register_worker_death {
$worker->analysis->stats->adaptor->decrease_running_workers($worker->analysis->stats->analysis_id);
}
my $sql = "UPDATE hive SET died=now(), last_check_in=now()";
my $sql = "UPDATE worker SET died=now(), last_check_in=now()";
$sql .= " ,status='DEAD'";
$sql .= " ,work_done='" . $worker->work_done . "'";
$sql .= " ,cause_of_death='$cod'";
$sql .= " WHERE worker_id='" . $worker->worker_id ."'";
$sql .= " WHERE worker_id='" . $worker->dbID ."'";
$self->dbc->do( $sql );
......@@ -332,9 +332,9 @@ sub worker_check_in {
my ($self, $worker) = @_;
return unless($worker);
my $sql = "UPDATE hive SET last_check_in=now()";
my $sql = "UPDATE worker SET last_check_in=now()";
$sql .= " ,work_done='" . $worker->work_done . "'";
$sql .= " WHERE worker_id='" . $worker->worker_id ."'";
$sql .= " WHERE worker_id='" . $worker->dbID ."'";
my $sth = $self->prepare($sql);
$sth->execute();
......@@ -346,11 +346,11 @@ sub worker_check_in {
=head2 reset_and_fetch_job_by_dbID
Arg [1]: int $analysis_job_id
Arg [1]: int $job_id
Example:
my $job = $queen->reset_and_fetch_job_by_dbID($analysis_job_id);
my $job = $queen->reset_and_fetch_job_by_dbID($job_id);
Description:
For the specified analysis_job_id it will fetch just that job,
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
......@@ -364,12 +364,12 @@ sub worker_check_in {
sub reset_and_fetch_job_by_dbID {
my $self = shift;
my $analysis_job_id = shift;
my $job_id = shift;
my $jobDBA = $self->db->get_AnalysisJobAdaptor;
$jobDBA->reset_job_by_dbID($analysis_job_id);
$jobDBA->reset_job_by_dbID($job_id);
my $job = $jobDBA->fetch_by_dbID($analysis_job_id);
my $job = $jobDBA->fetch_by_dbID($job_id);
my $stats = $self->db->get_AnalysisStatsAdaptor->fetch_by_analysis_id($job->analysis_id);
$self->synchronize_AnalysisStats($stats);
......@@ -391,25 +391,25 @@ sub fetch_overdue_workers {
$overdue_secs = 3600 unless(defined($overdue_secs));
my $constraint = "h.cause_of_death='' ".
"AND (UNIX_TIMESTAMP()-UNIX_TIMESTAMP(h.last_check_in))>$overdue_secs";
my $constraint = "w.cause_of_death='' ".
"AND (UNIX_TIMESTAMP()-UNIX_TIMESTAMP(w.last_check_in))>$overdue_secs";
return $self->_generic_fetch($constraint);
}
sub fetch_failed_workers {
my $self = shift;
my $constraint = "h.cause_of_death='FATALITY' ";
my $constraint = "w.cause_of_death='FATALITY' ";
return $self->_generic_fetch($constraint);
}
sub fetch_dead_workers_with_jobs {
my $self = shift;
# select h.worker_id from hive h, analysis_job WHERE h.worker_id=analysis_job.worker_id AND h.cause_of_death!='' AND analysis_job.status not in ('DONE', 'READY','FAILED', 'PASSED_ON') group by h.worker_id
# select w.worker_id from worker h, job WHERE w.worker_id=job.worker_id AND w.cause_of_death!='' AND job.status not in ('DONE', 'READY','FAILED', 'PASSED_ON') group by w.worker_id
my $constraint = "h.cause_of_death!='' ";
my $join = [[['analysis_job', 'j'], " h.worker_id=j.worker_id AND j.status NOT IN ('DONE', 'READY', 'FAILED', 'PASSED_ON') GROUP BY h.worker_id"]];
my $constraint = "w.cause_of_death!='' ";
my $join = [[['job', 'j'], " w.worker_id=j.worker_id AND j.status NOT IN ('DONE', 'READY', 'FAILED', 'PASSED_ON') GROUP BY w.worker_id"]];
return $self->_generic_fetch($constraint, $join);
}
......@@ -418,8 +418,8 @@ sub fetch_dead_workers_with_jobs {
Arg [1] : $filter_analysis (optional)
Example : $queen->synchronize_hive();
Description: Runs through all analyses in the system and synchronizes
the analysis_stats summary with the states in the analysis_job
and hive tables. Then follows by checking all the blocking rules
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
......@@ -495,7 +495,7 @@ sub safe_synchronize_AnalysisStats {
Arg [1] : Bio::EnsEMBL::Hive::AnalysisStats object
Example : $self->synchronize($analysisStats);
Description: Queries the analysis_job and hive tables to get summary counts
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
Returntype : newly synced Bio::EnsEMBL::Hive::AnalysisStats object
......@@ -518,7 +518,7 @@ sub synchronize_AnalysisStats {
$analysisStats->failed_job_count(0);
$analysisStats->num_required_workers(0);
my $sql = "SELECT status, count(*), semaphore_count FROM analysis_job ".
my $sql = "SELECT status, count(*), semaphore_count FROM job ".
"WHERE analysis_id=? GROUP BY status, semaphore_count";
my $sth = $self->prepare($sql);
$sth->execute($analysisStats->analysis_id);
......@@ -618,9 +618,9 @@ sub get_num_failed_analyses {
sub get_hive_current_load {
my $self = shift;
my $sql = "SELECT sum(1/analysis_stats.hive_capacity) FROM hive, analysis_stats ".
"WHERE hive.analysis_id=analysis_stats.analysis_id and cause_of_death ='' ".
"AND analysis_stats.hive_capacity>0";
my $sql = "SELECT sum(1/s.hive_capacity) FROM worker w, analysis_stats s ".
"WHERE w.analysis_id=s.analysis_id and w.cause_of_death ='' ".
"AND s.hive_capacity>0";
my $sth = $self->prepare($sql);
$sth->execute();
(my $load)=$sth->fetchrow_array();
......@@ -633,7 +633,7 @@ sub get_hive_current_load {
sub get_num_running_workers