opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Linting lib/galaxy/jobs

Browse Source
This commit is contained in:
Remi Marenco
2015-07-05 18:17:36 +01:00
committed by Nicola Soranzo
parent 8c7a906d17
commit 4931b5ef91
26 changed files with 225 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.ci/pep8_sources.txt
+1 -1
View File
@@ -7,7 +7,7 @@ lib/galaxy/datatypes/{assembly,binary,checkers,chrominfo,coverage,data,proteomic
lib/galaxy/datatypes/converters/{bedgraph_to_array_tree_converter,bgzip,wiggle_to_array_tree_converter}.py
lib/galaxy/exceptions/
lib/galaxy/external_services/
lib/galaxy/jobs/{__init__,error_level,manager,stock_rules}.py
lib/galaxy/jobs/
lib/galaxy/managers/
lib/galaxy/model/orm/{engine_factory,now}.py
lib/galaxy/model/{base,custom_types,search,util}.py
lib/galaxy/jobs/deferred/__init__.py
+25 -5
View File
@@ -1,18 +1,22 @@
"""
Queue for running deferred code via plugins.
"""
import os, sys, logging, threading
from Queue import Queue, Empty
import logging
import os
import threading
from Queue import Queue
from galaxy import model
from galaxy.util.bunch import Bunch
log = logging.getLogger( __name__ )
class DeferredJobQueue( object ):
job_states = Bunch( READY = 'ready',
WAIT = 'wait',
INVALID = 'invalid' )
job_states = Bunch( READY='ready',
WAIT='wait',
INVALID='invalid' )
def __init__( self, app ):
self.app = app
self.sa_session = app.model.context.current
@@ -26,6 +30,7 @@ class DeferredJobQueue( object ):
self.monitor_thread = threading.Thread( target=self.__monitor )
self.monitor_thread.start()
log.info( 'Deferred job queue started' )
def _load_plugins( self ):
for fname in os.listdir( os.path.dirname( __file__ ) ):
if not fname.startswith( '_' ) and fname.endswith( '.py' ):
@@ -52,6 +57,7 @@ class DeferredJobQueue( object ):
self.plugins[obj] = plugin( self.app )
self.plugins[obj].job_states = self.job_states
log.debug( 'Loaded deferred job plugin: %s' % display_name )
def __check_jobs_at_startup( self ):
waiting_jobs = self.sa_session.query( model.DeferredJob ) \
.filter( model.DeferredJob.state == model.DeferredJob.states.WAITING ).all()
@@ -64,6 +70,7 @@ class DeferredJobQueue( object ):
# Pass the job ID as opposed to the job, since the monitor thread
# needs to load it in its own threadlocal scoped session.
self.waiting_jobs.append( job.id )
def __monitor( self ):
while self.running:
try:
@@ -72,6 +79,7 @@ class DeferredJobQueue( object ):
log.exception( 'Exception in monitor_step' )
self.sleeper.sleep( 1 )
log.info( 'job queue stopped' )
def __monitor_step( self ):
# TODO: Querying the database with this frequency is bad, we need message passing
new_jobs = self.sa_session.query( model.DeferredJob ) \
@@ -117,6 +125,7 @@ class DeferredJobQueue( object ):
else:
new_waiting.append( job )
self.waiting_jobs = new_waiting
def __check_job_plugin( self, job ):
if job.plugin not in self.plugins:
log.error( 'Invalid deferred job plugin: %s' ) % job.plugin
@@ -125,16 +134,20 @@ class DeferredJobQueue( object ):
self.sa_session.flush()
return False
return True
def __check_if_ready_to_run( self, job ):
return self.plugins[job.plugin].check_job( job )
def __fail_job( self, job ):
job.state = model.DeferredJob.states.ERROR
self.sa_session.add( job )
self.sa_session.flush()
def shutdown( self ):
self.running = False
self.sleeper.wake()
class Sleeper( object ):
"""
Provides a 'sleep' method that sleeps for a number of seconds *unless*
@@ -142,15 +155,18 @@ class Sleeper( object ):
"""
def __init__( self ):
self.condition = threading.Condition()
def sleep( self, seconds ):
self.condition.acquire()
self.condition.wait( seconds )
self.condition.release()
def wake( self ):
self.condition.acquire()
self.condition.notify()
self.condition.release()
class FakeTrans( object ):
"""A fake trans for calling the external set metadata tool"""
def __init__( self, app, history=None, user=None):
@@ -169,15 +185,19 @@ class FakeTrans( object ):
else:
self.user = user
self.model = app.model
def get_galaxy_session( self ):
return self.dummy
def log_event( self, message, tool_id=None ):
pass
def get_current_user_roles( self ):
if self.user:
return self.user.all_roles()
else:
return []
def db_dataset_for( self, dbkey ):
if self.history is None:
return None
lib/galaxy/jobs/deferred/data_transfer.py
+32 -32
View File
@@ -1,35 +1,41 @@
"""
Module for managing data transfer jobs.
"""
import logging, urllib2, re, shutil
import logging
import re
import shutil
from galaxy import eggs
from sqlalchemy import and_
from galaxy.util.odict import odict
from galaxy.workflow.modules import module_factory
from galaxy.datatypes import sniff
from galaxy.jobs.actions.post import ActionBox
from galaxy.jobs.deferred import FakeTrans
from galaxy.tools.parameters import visit_input_values
from galaxy.tools.parameters.basic import DataToolParameter
from galaxy.datatypes import sniff
from galaxy.util.odict import odict
from galaxy.workflow.modules import module_factory
log = logging.getLogger( __name__ )
__all__ = [ 'DataTransfer' ]
class DataTransfer( object ):
check_interval = 15
dataset_name_re = re.compile( '(dataset\d+)_(name)' )
dataset_datatype_re = re.compile( '(dataset\d+)_(datatype)' )
def __init__( self, app ):
self.app = app
self.sa_session = app.model.context.current
def create_job( self, trans, **kwd ):
raise Exception( "Unimplemented Method" )
def check_job( self, job ):
raise Exception( "Unimplemented Method" )
def run_job( self, job ):
if job.params[ 'type' ] == 'init_transfer':
# TODO: don't create new downloads on restart.
@@ -115,16 +121,16 @@ class DataTransfer( object ):
self.sa_session.add( ld )
self.sa_session.flush()
self.app.security_agent.copy_library_permissions( FakeTrans( self.app ), sample.folder, ld )
ldda = self.app.model.LibraryDatasetDatasetAssociation( name = library_dataset_name,
extension = extension,
dbkey = '?',
library_dataset = ld,
create_dataset = True,
sa_session = self.sa_session )
ldda = self.app.model.LibraryDatasetDatasetAssociation( name=library_dataset_name,
extension=extension,
dbkey='?',
library_dataset=ld,
create_dataset=True,
sa_session=self.sa_session )
ldda.message = 'Transferred by the Data Transfer Plugin'
self.sa_session.add( ldda )
self.sa_session.flush()
ldda.state = ldda.states.QUEUED # flushed in the set property
ldda.state = ldda.states.QUEUED # flushed in the set property
ld.library_dataset_dataset_association_id = ldda.id
self.sa_session.add( ld )
self.sa_session.flush()
@@ -140,13 +146,13 @@ class DataTransfer( object ):
FakeTrans( self.app,
history=sample.history,
user=sample.request.user ),
incoming = { 'input1':ldda } )
incoming={ 'input1': ldda } )
ldda.state = ldda.states.OK
# TODO: not sure if this flush is necessary
self.sa_session.add( ldda )
self.sa_session.flush()
except Exception, e:
log.exception( 'Failure preparing library dataset for finished transfer job (id: %s) via deferred job (id: %s):' % \
log.exception( 'Failure preparing library dataset for finished transfer job (id: %s) via deferred job (id: %s):' %
( str( job.transfer_job.id ), str( job.id ) ) )
ldda.state = ldda.states.ERROR
if sample.workflow:
@@ -157,7 +163,7 @@ class DataTransfer( object ):
sub_done = True
rep_done = False
for k, v in sample.workflow[ 'mappings' ].iteritems():
if not 'hda' in v and v[ 'ds_tag' ].startswith( 'hi|' ):
if 'hda' not in v and v[ 'ds_tag' ].startswith( 'hi|' ):
sample.workflow[ 'mappings' ][ k ][ 'hda' ] = self.app.security.decode_id( v[ 'ds_tag' ][3:] )
for key, value in sample.workflow[ 'mappings' ].iteritems():
if 'url' in value and value[ 'url' ] == job.params[ 'result' ][ 'url' ]:
@@ -166,7 +172,7 @@ class DataTransfer( object ):
sample.workflow[ 'mappings' ][ key ][ 'ldda' ] = ldda.id
rep_done = True
# DBTODO replace the hi| mappings with the hda here. Just rip off the first three chars.
elif not 'ldda' in value and not 'hda' in value:
elif 'ldda' not in value and 'hda' not in value:
# We're not done if some mappings still don't have ldda or hda mappings.
sub_done = False
if sub_done and rep_done:
@@ -203,17 +209,18 @@ class DataTransfer( object ):
self._update_sample_state( sample.id, new_state )
# Update the state of the request, if possible
self._update_request_state( sample.request.id )
def _missing_params( self, params, required_params ):
missing_params = filter( lambda x: x not in params, required_params )
if missing_params:
log.error( 'Job parameters missing required keys: %s' % ', '.join( missing_params ) )
return True
return False
def _update_sample_dataset_status( self, protocol, sample_id, result_dict, new_status, error_msg=None ):
# result_dict looks something like:
# {'url': '127.0.0.1/data/filtered_subreads.fa', 'name': 'Filtered reads'}
# Check if the new status is a valid transfer status
valid_statuses = [ v[1] for v in self.app.model.SampleDataset.transfer_status.items() ]
# TODO: error checking on valid new_status value
if protocol in [ 'http', 'https' ]:
sample_dataset = self.sa_session.query( self.app.model.SampleDataset ) \
@@ -227,6 +234,7 @@ class DataTransfer( object ):
sample_dataset.error_msg = error_msg
self.sa_session.add( sample_dataset )
self.sa_session.flush()
def _update_sample_state( self, sample_id, new_state, comment=None ):
sample = self.sa_session.query( self.app.model.Sample ).get( sample_id )
if comment is None:
@@ -234,6 +242,7 @@ class DataTransfer( object ):
event = self.app.model.SampleEvent( sample, new_state, comment )
self.sa_session.add( event )
self.sa_session.flush()
def _update_request_state( self, request_id ):
request = self.sa_session.query( self.app.model.Request ).get( request_id )
# Make sure all the samples of the current request have the same state
@@ -247,13 +256,11 @@ class DataTransfer( object ):
self.sa_session.add( event )
self.sa_session.flush()
else:
final_state = False
request_type_state = request.type.final_sample_state
if common_state.id == request_type_state.id:
# Since all the samples are in the final state, change the request state to 'Complete'
comment = "All samples of this sequencing request are in the final sample state (%s). " % request_type_state.name
state = request.states.COMPLETE
final_state = True
else:
comment = "All samples of this sequencing request are in the (%s) sample state. " % common_state.name
state = request.states.SUBMITTED
@@ -261,6 +268,7 @@ class DataTransfer( object ):
self.sa_session.add( event )
self.sa_session.flush()
# TODO: handle email notification if it is configured to be sent when the samples are in this state.
def _execute_workflow( self, sample):
for key, value in sample.workflow['mappings'].iteritems():
if 'hda' not in value and 'ldda' in value:
@@ -280,7 +288,7 @@ class DataTransfer( object ):
if not isinstance(key, int):
new_wf_dict['mappings'][int(key)] = workflow_dict['mappings'][key]
workflow_dict = new_wf_dict
fk_trans = FakeTrans(self.app, history = sample.history, user=sample.request.user)
fk_trans = FakeTrans(self.app, history=sample.history, user=sample.request.user)
workflow = self.sa_session.query(self.app.model.Workflow).get(workflow_dict['id'])
if not workflow:
log.error("Workflow mapping failure.")
@@ -296,8 +304,6 @@ class DataTransfer( object ):
return
# Build the state for each step
errors = {}
has_upgrade_messages = False
has_errors = False
# Build a fake dictionary prior to execution.
# Prepare each step
for step in workflow.steps:
@@ -308,8 +314,6 @@ class DataTransfer( object ):
step.module = module_factory.from_workflow_step( fk_trans, step )
# Fix any missing parameters
step.upgrade_messages = step.module.check_and_update_state()
if step.upgrade_messages:
has_upgrade_messages = True
# Any connected input needs to have value DummyDataset (these
# are not persisted so we need to do it every time)
step.module.add_dummy_datasets( connections=step.input_connections )
@@ -317,10 +321,9 @@ class DataTransfer( object ):
step.state = step.module.state
# Error dict
if step.tool_errors:
has_errors = True
errors[step.id] = step.tool_errors
else:
## Non-tool specific stuff?
# Non-tool specific stuff?
step.module = module_factory.from_workflow_step( fk_trans, step )
step.state = step.module.get_runtime_state()
# Connections by input name
@@ -336,8 +339,6 @@ class DataTransfer( object ):
module = module_factory.from_workflow_step( fk_trans, step )
# Fix any missing parameters
step.upgrade_messages = module.check_and_update_state()
if step.upgrade_messages:
has_upgrade_messages = True
# Any connected input needs to have value DummyDataset (these
# are not persisted so we need to do it every time)
module.add_dummy_datasets( connections=step.input_connections )
@@ -345,8 +346,6 @@ class DataTransfer( object ):
tool = module.tool
# Get the state
step.state = state = module.state
# Get old errors
old_errors = state.inputs.pop( "__errors__", {} )
if step_errors:
errors[step.id] = state.inputs["__errors__"] = step_errors
# Run each step, connecting outputs to inputs
@@ -357,6 +356,7 @@ class DataTransfer( object ):
job = None
if step.type == 'tool' or step.type is None:
tool = self.app.toolbox.get_tool( step.tool_id )
def callback( input, value, prefixed_name, prefixed_label ):
if isinstance( input, DataToolParameter ):
if prefixed_name in step.input_connections_by_name:
lib/galaxy/jobs/deferred/manual_data_transfer.py
+13 -10
View File
@@ -2,16 +2,18 @@
Generic module for managing manual data transfer jobs using Galaxy's built-in file browser.
This module can be used by various external services that are configured to transfer data manually.
"""
import logging, urllib2, re, shutil
from data_transfer import *
import logging
from data_transfer import DataTransfer
log = logging.getLogger( __name__ )
__all__ = [ 'ManualDataTransferPlugin' ]
class ManualDataTransferPlugin( DataTransfer ):
def __init__( self, app ):
super( ManualDataTransferPlugin, self ).__init__( app )
def create_job( self, trans, **kwd ):
if 'sample' in kwd and 'sample_datasets' in kwd and 'external_service' in kwd and 'external_service_type' in kwd:
sample = kwd[ 'sample' ]
@@ -29,7 +31,7 @@ class ManualDataTransferPlugin( DataTransfer ):
# but without a way for the user to associate stuff it's useless. However, allowing the user this ability may
# open a can of worms, so maybe we shouldn't do it???
#
#for run_result_file_name, run_result_file_datatype in external_service_type.run_details[ 'results' ].items():
# for run_result_file_name, run_result_file_datatype in external_service_type.run_details[ 'results' ].items():
# # external_service_type.run_details[ 'results' ] looks something like: {'dataset1_name': 'dataset1_datatype'}
# if run_result_file_datatype in external_service.form_values.content:
# datatype = external_service.form_values.content[ run_result_file_datatype ]
@@ -41,13 +43,13 @@ class ManualDataTransferPlugin( DataTransfer ):
sample_datasets_dict = {}
for sample_dataset in sample_datasets:
sample_dataset_id = sample_dataset.id
sample_dataset_dict = dict( sample_id = sample_dataset.sample.id,
name = sample_dataset.name,
file_path = sample_dataset.file_path,
status = sample_dataset.status,
error_msg = sample_dataset.error_msg,
size = sample_dataset.size,
external_service_id = sample_dataset.external_service.id )
sample_dataset_dict = dict( sample_id=sample_dataset.sample.id,
name=sample_dataset.name,
file_path=sample_dataset.file_path,
status=sample_dataset.status,
error_msg=sample_dataset.error_msg,
size=sample_dataset.size,
external_service_id=sample_dataset.external_service.id )
sample_datasets_dict[ sample_dataset_id ] = sample_dataset_dict
params = { 'type' : 'init_transfer',
'sample_id' : sample.id,
@@ -72,6 +74,7 @@ class ManualDataTransferPlugin( DataTransfer ):
self.sa_session.flush()
log.debug( 'Created a deferred job in the ManualDataTransferPlugin of type: %s' % params[ 'type' ] )
# TODO: error reporting to caller (if possible?)
def check_job( self, job ):
if self._missing_params( job.params, [ 'type' ] ):
return self.job_states.INVALID
lib/galaxy/jobs/deferred/pacific_biosciences_smrt_portal.py
+9 -3
View File
@@ -2,20 +2,24 @@
Module for managing jobs in Pacific Bioscience's SMRT Portal and automatically transferring files
produced by SMRT Portal.
"""
import logging, urllib2, re, shutil
import logging
import urllib2
from string import Template
from galaxy.util import json
from data_transfer import *
from data_transfer import DataTransfer
log = logging.getLogger( __name__ )
__all__ = [ 'SMRTPortalPlugin' ]
class SMRTPortalPlugin( DataTransfer ):
api_path = '/smrtportal/api'
def __init__( self, app ):
super( SMRTPortalPlugin, self ).__init__( app )
def create_job( self, trans, **kwd ):
if 'secondary_analysis_job_id' in kwd:
sample = kwd[ 'sample' ]
@@ -42,7 +46,7 @@ class SMRTPortalPlugin( DataTransfer ):
results[ id ] = { field : v }
for id, attrs in results.items():
url_template = external_service_type.run_details[ 'results_urls' ].get( id + '_name' )
url = Template( url_template ).substitute( host = smrt_host, secondary_analysis_job_id = kwd[ 'secondary_analysis_job_id' ] )
url = Template( url_template ).substitute( host=smrt_host, secondary_analysis_job_id=kwd[ 'secondary_analysis_job_id' ] )
results[ id ][ 'url' ] = url
if sample.workflow:
# DBTODO Make sure all ds| mappings get the URL of the dataset, for linking to later.
@@ -75,6 +79,7 @@ class SMRTPortalPlugin( DataTransfer ):
self.sa_session.flush()
log.debug( 'Created a deferred job in the SMRTPortalPlugin of type: %s' % params[ 'type' ] )
# TODO: error reporting to caller (if possible?)
def check_job( self, job ):
if self._missing_params( job.params, [ 'type' ] ):
return self.job_states.INVALID
@@ -106,6 +111,7 @@ class SMRTPortalPlugin( DataTransfer ):
else:
log.error( 'Unknown job type for SMRTPortalPlugin: %s' % str( job.params[ 'type' ] ) )
return self.job_states.INVALID
def _associate_untransferred_datasets_with_sample( self, sample, external_service, results_dict ):
# results_dict looks something like:
# {'dataset2': {'datatype': 'fasta', 'url': '127.0.0.1:8080/data/filtered_subreads.fa', 'name': 'Filtered reads'} }
lib/galaxy/jobs/handler.py
+44 -44
View File
@@ -109,14 +109,14 @@ class JobHandlerQueue( object ):
model.Job.states.RUNNING )
if self.app.config.user_activation_on:
jobs_at_startup = self.sa_session.query( model.Job ).enable_eagerloads( False ) \
.outerjoin( model.User ) \
.filter( model.Job.state.in_( in_list ) \
& ( model.Job.handler == self.app.config.server_name ) \
& or_( ( model.Job.user_id == None ), ( model.User.active == True ) ) ).all()
.outerjoin( model.User ) \
.filter( model.Job.state.in_( in_list )
& ( model.Job.handler == self.app.config.server_name )
& or_( ( model.Job.user_id is None ), ( model.User.active is True ) ) ).all()
else:
jobs_at_startup = self.sa_session.query( model.Job ).enable_eagerloads( False ) \
.filter( model.Job.state.in_( in_list ) \
& ( model.Job.handler == self.app.config.server_name ) ).all()
.filter( model.Job.state.in_( in_list )
& ( model.Job.handler == self.app.config.server_name ) ).all()
for job in jobs_at_startup:
if not self.app.toolbox.has_tool( job.tool_id, job.tool_version, exact=True ):
@@ -199,32 +199,32 @@ class JobHandlerQueue( object ):
self.sa_session.expunge_all()
# Fetch all new jobs
hda_not_ready = self.sa_session.query(model.Job.id).enable_eagerloads(False) \
.join(model.JobToInputDatasetAssociation) \
.join(model.HistoryDatasetAssociation) \
.join(model.Dataset) \
.filter(and_( (model.Job.state == model.Job.states.NEW ),
or_( ( model.HistoryDatasetAssociation._state == model.HistoryDatasetAssociation.states.FAILED_METADATA ),
( model.HistoryDatasetAssociation.deleted == True ),
( model.Dataset.state != model.Dataset.states.OK ),
( model.Dataset.deleted == True) ) ) ).subquery()
.join(model.JobToInputDatasetAssociation) \
.join(model.HistoryDatasetAssociation) \
.join(model.Dataset) \
.filter(and_( (model.Job.state == model.Job.states.NEW ),
or_( ( model.HistoryDatasetAssociation._state == model.HistoryDatasetAssociation.states.FAILED_METADATA ),
( model.HistoryDatasetAssociation.deleted is True ),
( model.Dataset.state != model.Dataset.states.OK ),
( model.Dataset.deleted is True) ) ) ).subquery()
ldda_not_ready = self.sa_session.query(model.Job.id).enable_eagerloads(False) \
.join(model.JobToInputLibraryDatasetAssociation) \
.join(model.LibraryDatasetDatasetAssociation) \
.join(model.Dataset) \
.filter(and_((model.Job.state == model.Job.states.NEW),
or_((model.LibraryDatasetDatasetAssociation._state != None),
(model.LibraryDatasetDatasetAssociation.deleted == True),
(model.Dataset.state != model.Dataset.states.OK),
(model.Dataset.deleted == True)))).subquery()
.join(model.JobToInputLibraryDatasetAssociation) \
.join(model.LibraryDatasetDatasetAssociation) \
.join(model.Dataset) \
.filter(and_((model.Job.state == model.Job.states.NEW),
or_((model.LibraryDatasetDatasetAssociation._state is not None),
(model.LibraryDatasetDatasetAssociation.deleted is True),
(model.Dataset.state != model.Dataset.states.OK),
(model.Dataset.deleted is True)))).subquery()
if self.app.config.user_activation_on:
jobs_to_check = self.sa_session.query(model.Job).enable_eagerloads(False) \
.outerjoin( model.User ) \
.filter(and_((model.Job.state == model.Job.states.NEW),
or_((model.Job.user_id == None), (model.User.active == True)),
(model.Job.handler == self.app.config.server_name),
~model.Job.table.c.id.in_(hda_not_ready),
~model.Job.table.c.id.in_(ldda_not_ready))) \
.order_by(model.Job.id).all()
.outerjoin( model.User ) \
.filter(and_((model.Job.state == model.Job.states.NEW),
or_((model.Job.user_id is None), (model.User.active is True)),
(model.Job.handler == self.app.config.server_name),
~model.Job.table.c.id.in_(hda_not_ready),
~model.Job.table.c.id.in_(ldda_not_ready))) \
.order_by(model.Job.id).all()
else:
jobs_to_check = self.sa_session.query(model.Job).enable_eagerloads(False) \
.filter(and_((model.Job.state == model.Job.states.NEW),
@@ -234,9 +234,9 @@ class JobHandlerQueue( object ):
.order_by(model.Job.id).all()
# Fetch all "resubmit" jobs
resubmit_jobs = self.sa_session.query(model.Job).enable_eagerloads(False) \
.filter(and_((model.Job.state == model.Job.states.RESUBMITTED),
(model.Job.handler == self.app.config.server_name))) \
.order_by(model.Job.id).all()
.filter(and_((model.Job.state == model.Job.states.RESUBMITTED),
(model.Job.handler == self.app.config.server_name))) \
.order_by(model.Job.id).all()
else:
# Get job objects and append to watch queue for any which were
# previously waiting
@@ -428,7 +428,7 @@ class JobHandlerQueue( object ):
# This could have been incremented by a previous job dispatched on this iteration, even if we're not caching
rval = self.user_job_count.get(user_id, 0)
if not self.app.config.cache_user_job_count:
result = self.sa_session.execute(select([func.count(model.Job.table.c.id)]) \
result = self.sa_session.execute(select([func.count(model.Job.table.c.id)])
.where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED,
model.Job.states.RUNNING,
model.Job.states.RESUBMITTED)),
@@ -442,12 +442,12 @@ class JobHandlerQueue( object ):
# Cache the job count if necessary
if self.user_job_count is None and self.app.config.cache_user_job_count:
self.user_job_count = {}
query = self.sa_session.execute(select([model.Job.table.c.user_id, func.count(model.Job.table.c.user_id)]) \
query = self.sa_session.execute(select([model.Job.table.c.user_id, func.count(model.Job.table.c.user_id)])
.where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED,
model.Job.states.RUNNING,
model.Job.states.RESUBMITTED)),
(model.Job.table.c.user_id is not None))) \
.group_by(model.Job.table.c.user_id))
(model.Job.table.c.user_id is not None)))
.group_by(model.Job.table.c.user_id))
for row in query:
self.user_job_count[row[0]] = row[1]
elif self.user_job_count is None:
@@ -465,8 +465,8 @@ class JobHandlerQueue( object ):
# queue.
rval = {}
rval.update(cached)
result = self.sa_session.execute(select([model.Job.table.c.destination_id, func.count(model.Job.table.c.destination_id).label('job_count')]) \
.where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED, model.Job.states.RUNNING)), (model.Job.table.c.user_id == user_id))) \
result = self.sa_session.execute(select([model.Job.table.c.destination_id, func.count(model.Job.table.c.destination_id).label('job_count')])
.where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED, model.Job.states.RUNNING)), (model.Job.table.c.user_id == user_id)))
.group_by(model.Job.table.c.destination_id))
for row in result:
# Add the count from the database to the cached count
@@ -477,8 +477,8 @@ class JobHandlerQueue( object ):
# Cache the job count if necessary
if self.user_job_count_per_destination is None and self.app.config.cache_user_job_count:
self.user_job_count_per_destination = {}
result = self.sa_session.execute(select([model.Job.table.c.user_id, model.Job.table.c.destination_id, func.count(model.Job.table.c.user_id).label('job_count')]) \
.where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED, model.Job.states.RUNNING)))) \
result = self.sa_session.execute(select([model.Job.table.c.user_id, model.Job.table.c.destination_id, func.count(model.Job.table.c.user_id).label('job_count')])
.where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED, model.Job.states.RUNNING))))
.group_by(model.Job.table.c.user_id, model.Job.table.c.destination_id))
for row in result:
if row['user_id'] not in self.user_job_count_per_destination:
@@ -552,8 +552,8 @@ class JobHandlerQueue( object ):
# Cache the job count if necessary
if self.total_job_count_per_destination is None:
self.total_job_count_per_destination = {}
result = self.sa_session.execute(select([model.Job.table.c.destination_id, func.count(model.Job.table.c.destination_id).label('job_count')]) \
.where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED, model.Job.states.RUNNING)))) \
result = self.sa_session.execute(select([model.Job.table.c.destination_id, func.count(model.Job.table.c.destination_id).label('job_count')])
.where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED, model.Job.states.RUNNING))))
.group_by(model.Job.table.c.destination_id))
for row in result:
self.total_job_count_per_destination[row['destination_id']] = row['job_count']
@@ -663,7 +663,7 @@ class JobHandlerStopQueue( object ):
self.sa_session.expunge_all()
# Fetch all new jobs
newly_deleted_jobs = self.sa_session.query( model.Job ).enable_eagerloads( False ) \
.filter( ( model.Job.state == model.Job.states.DELETED_NEW ) \
.filter( ( model.Job.state == model.Job.states.DELETED_NEW )
& ( model.Job.handler == self.app.config.server_name ) ).all()
for job in newly_deleted_jobs:
jobs_to_check.append( ( job, job.stderr ) )
@@ -750,7 +750,7 @@ class DefaultJobDispatcher( object ):
runner_name = self.__get_runner_name( job_wrapper )
try:
if isinstance(job_wrapper, TaskWrapper):
#DBTODO Refactor
# DBTODO Refactor
log.debug( "(%s) Dispatching task %s to %s runner" % ( job_wrapper.job_id, job_wrapper.task_id, runner_name ) )
else:
log.debug( "(%s) Dispatching to %s runner" % ( job_wrapper.job_id, runner_name ) )
lib/galaxy/jobs/mapper.py
+2 -2
View File
@@ -57,8 +57,8 @@ class JobRunnerMapper( object ):
def __get_rule_modules( self ):
unsorted_module_names = self.__get_rule_module_names( )
## Load modules in reverse order to allow hierarchical overrides
## i.e. 000_galaxy_rules.py, 100_site_rules.py, 200_instance_rules.py
# Load modules in reverse order to allow hierarchical overrides
# i.e. 000_galaxy_rules.py, 100_site_rules.py, 200_instance_rules.py
module_names = sorted( unsorted_module_names, reverse=True )
modules = []
for rule_module_name in module_names:
lib/galaxy/jobs/metrics/formatting.py
+1 -1
View File
@@ -7,7 +7,7 @@ class JobMetricFormatter( object ):
return ( str( key ), str( value ) )
## Formatting utilities
# Formatting utilities
def seconds_to_str( value ):
if value < 60:
lib/galaxy/jobs/metrics/instrumenters/collectl.py
+1 -2
View File
@@ -14,8 +14,7 @@ import logging
log = logging.getLogger( __name__ )
DEFAULT_PROCFILT_ON = "username" # By default, only grab statistics for user
# processes (as identifiers by username).
DEFAULT_SUBSYSTEMS = "process"
DEFAULT_SUBSYSTEMS = "process" # processes (as identifiers by username).
DEFAULT_FLUSH_INTERVAL = "0" # Set to zero to flush every collection.
FORMATTED_RESOURCE_TITLES = {
lib/galaxy/jobs/output_checker.py
+1 -1
View File
@@ -39,7 +39,7 @@ def check_output( tool, stdout, stderr, tool_exit_code, job ):
# that range, then apply the error level and add a message.
# If we've reached a fatal error rule, then stop.
max_error_level = StdioErrorLevel.NO_ERROR
if tool_exit_code != None:
if tool_exit_code is not None:
for stdio_exit_code in tool.stdio_exit_codes:
if ( tool_exit_code >= stdio_exit_code.range_start and
tool_exit_code <= stdio_exit_code.range_end ):
lib/galaxy/jobs/rule_helper.py
+1 -4
View File
@@ -2,10 +2,7 @@ from datetime import datetime
import hashlib
import random
from sqlalchemy import (
and_,
func
)
from sqlalchemy import func
from galaxy import model
from galaxy import util
lib/galaxy/jobs/runners/__init__.py
+11 -10
View File
@@ -251,15 +251,15 @@ class BaseJobRunner( object ):
Set metadata externally. Used by the local and lwr job runners where this
shouldn't be attached to command-line to execute.
"""
#run the metadata setting script here
#this is terminate-able when output dataset/job is deleted
#so that long running set_meta()s can be canceled without having to reboot the server
# run the metadata setting script here
# this is terminate-able when output dataset/job is deleted
# so that long running set_meta()s can be canceled without having to reboot the server
if job_wrapper.get_state() not in [ model.Job.states.ERROR, model.Job.states.DELETED ] and job_wrapper.output_paths:
lib_adjust = GALAXY_LIB_ADJUST_TEMPLATE % job_wrapper.galaxy_lib_dir
external_metadata_script = job_wrapper.setup_external_metadata( output_fnames=job_wrapper.get_output_fnames(),
set_extension=True,
tmp_dir=job_wrapper.working_directory,
#we don't want to overwrite metadata that was copied over in init_meta(), as per established behavior
# We don't want to overwrite metadata that was copied over in init_meta(), as per established behavior
kwds={ 'overwrite' : False } )
external_metadata_script = "%s %s" % (lib_adjust, external_metadata_script)
if resolve_requirements:
@@ -304,8 +304,8 @@ class BaseJobRunner( object ):
working_directory=os.path.abspath( job_wrapper.working_directory ),
command=command_line,
)
## Additional logging to enable if debugging from_work_dir handling, metadata
## commands, etc... (or just peak in the job script.)
# Additional logging to enable if debugging from_work_dir handling, metadata
# commands, etc... (or just peak in the job script.)
job_id = job_wrapper.job_id
log.debug( '(%s) command is: %s' % ( job_id, command_line ) )
options.update(**kwds)
@@ -361,11 +361,12 @@ class JobState( object ):
Encapsulate state of jobs.
"""
runner_states = Bunch(
WALLTIME_REACHED = 'walltime_reached',
MEMORY_LIMIT_REACHED = 'memory_limit_reached',
GLOBAL_WALLTIME_REACHED = 'global_walltime_reached',
OUTPUT_SIZE_LIMIT = 'output_size_limit'
WALLTIME_REACHED='walltime_reached',
MEMORY_LIMIT_REACHED='memory_limit_reached',
GLOBAL_WALLTIME_REACHED='global_walltime_reached',
OUTPUT_SIZE_LIMIT='output_size_limit'
)
def __init__( self ):
self.runner_state_handled = False
lib/galaxy/jobs/runners/cli.py
+1 -4
View File
@@ -53,9 +53,6 @@ class ShellJobRunner( AsynchronousJobRunner ):
if not self.prepare_job( job_wrapper, include_metadata=True ):
return
# command line has been added to the wrapper by prepare_job()
command_line = job_wrapper.runner_command_line
# Get shell and job execution interface
job_destination = job_wrapper.job_destination
shell_params, job_params = self.parse_destination_params(job_destination.params)
@@ -98,7 +95,7 @@ class ShellJobRunner( AsynchronousJobRunner ):
log.error('(%s) submission failed (stderr): %s' % (galaxy_id_tag, cmd_out.stderr))
job_wrapper.fail("failure submitting job")
return
# Some job runners return something like 'Submitted batch job XXXX'
# Some job runners return something like 'Submitted batch job XXXX'
# Strip and split to get job ID.
external_job_id = cmd_out.stdout.strip().split()[-1]
if not external_job_id:
lib/galaxy/jobs/runners/condor.py
+1 -4
View File
@@ -49,9 +49,6 @@ class CondorJobRunner( AsynchronousJobRunner ):
if not self.prepare_job( job_wrapper, include_metadata=include_metadata):
return
# command line has been added to the wrapper by prepare_job()
command_line = job_wrapper.runner_command_line
# get configured job destination
job_destination = job_wrapper.job_destination
@@ -177,7 +174,7 @@ class CondorJobRunner( AsynchronousJobRunner ):
if not job_running and cjs.running:
log.debug( "(%s/%s) job has stopped running" % ( galaxy_id_tag, job_id ) )
# Will switching from RUNNING to QUEUED confuse Galaxy?
#cjs.job_wrapper.change_state( model.Job.states.QUEUED )
# cjs.job_wrapper.change_state( model.Job.states.QUEUED )
if job_complete:
if cjs.job_wrapper.get_state() != model.Job.states.DELETED:
external_metadata = not asbool( cjs.job_wrapper.job_destination.params.get( "embed_metadata_in_job", True) )
lib/galaxy/jobs/runners/drmaa.py
+9 -13
View File
@@ -7,7 +7,6 @@ import logging
import os
import string
import subprocess
import sys
import time
from galaxy import eggs
@@ -41,11 +40,11 @@ class DRMAAJobRunner( AsynchronousJobRunner ):
global drmaa
runner_param_specs = dict(
drmaa_library_path = dict( map = str, default = os.environ.get( 'DRMAA_LIBRARY_PATH', None ) ),
invalidjobexception_state = dict( map = str, valid = lambda x: x in ( model.Job.states.OK, model.Job.states.ERROR ), default = model.Job.states.OK ),
invalidjobexception_retries = dict( map = int, valid = lambda x: int >= 0, default = 0 ),
internalexception_state = dict( map = str, valid = lambda x: x in ( model.Job.states.OK, model.Job.states.ERROR ), default = model.Job.states.OK ),
internalexception_retries = dict( map = int, valid = lambda x: int >= 0, default = 0 ) )
drmaa_library_path=dict( map=str, default=os.environ.get( 'DRMAA_LIBRARY_PATH', None ) ),
invalidjobexception_state=dict( map=str, valid=lambda x: x in ( model.Job.states.OK, model.Job.states.ERROR ), default=model.Job.states.OK ),
invalidjobexception_retries=dict( map=int, valid=lambda x: int >= 0, default=0 ),
internalexception_state=dict( map=str, valid=lambda x: x in ( model.Job.states.OK, model.Job.states.ERROR ), default=model.Job.states.OK ),
internalexception_retries=dict( map=int, valid=lambda x: int >= 0, default=0 ) )
if 'runner_param_specs' not in kwargs:
kwargs[ 'runner_param_specs' ] = dict()
@@ -208,7 +207,7 @@ class DRMAAJobRunner( AsynchronousJobRunner ):
pwent = job_wrapper.galaxy_system_pwent
log.debug( '(%s) submitting with credentials: %s [uid: %s]' % ( galaxy_id_tag, pwent[0], pwent[2] ) )
filename = self.store_jobtemplate(job_wrapper, jt)
self.userid = pwent[2]
self.userid = pwent[2]
external_job_id = self.external_runjob(filename, pwent[2]).strip()
log.info( "(%s) queued as %s" % ( galaxy_id_tag, external_job_id ) )
@@ -296,7 +295,7 @@ class DRMAAJobRunner( AsynchronousJobRunner ):
ajs.running = True
ajs.job_wrapper.change_state( model.Job.states.RUNNING )
if state in ( drmaa.JobState.FAILED, drmaa.JobState.DONE ):
self._complete_terminal_job( ajs, drmaa_state = state )
self._complete_terminal_job( ajs, drmaa_state=state )
continue
if ajs.check_limits():
self.work_queue.put( ( self.fail_job, ajs ) )
@@ -357,7 +356,6 @@ class DRMAAJobRunner( AsynchronousJobRunner ):
ajs.exit_code_file = "%s.drmec" % os.path.join(os.getcwd(), ajs.job_wrapper.working_directory, ajs.job_wrapper.get_id_tag())
ajs.job_file = job_file
def store_jobtemplate(self, job_wrapper, jt):
""" Stores the content of a DRMAA JobTemplate object in a file as a JSON string.
Path is hard-coded, but it's no worse than other path in this module.
@@ -370,7 +368,7 @@ class DRMAAJobRunner( AsynchronousJobRunner ):
except:
pass
s = json.dumps(data)
f = open(filename,'w+')
f = open(filename, 'w+')
f.write(s)
f.close()
log.debug( '(%s) Job script for external submission is: %s' % ( job_wrapper.job_id, filename ) )
@@ -393,7 +391,7 @@ class DRMAAJobRunner( AsynchronousJobRunner ):
shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(stdoutdata, stderrdata) = p.communicate()
exitcode = p.returncode
#os.unlink(jobtemplate_filename)
# os.unlink(jobtemplate_filename)
if exitcode != 0:
# There was an error in the child process
raise RuntimeError("External_runjob failed (exit code %s)\nChild process reported error:\n%s" % (str(exitcode), stderrdata))
@@ -404,5 +402,3 @@ class DRMAAJobRunner( AsynchronousJobRunner ):
# the DRMAA job-ID. If not the case, will throw an error.
jobId = stdoutdata
return jobId
lib/galaxy/jobs/runners/local.py
+5 -5
View File
@@ -35,10 +35,10 @@ class LocalJobRunner( BaseJobRunner ):
def __init__( self, app, nworkers ):
"""Start the job runner """
#create a local copy of os.environ to use as env for subprocess.Popen
# create a local copy of os.environ to use as env for subprocess.Popen
self._environ = os.environ.copy()
#Set TEMP if a valid temp value is not already set
# Set TEMP if a valid temp value is not already set
if not ( 'TMPDIR' in self._environ or 'TEMP' in self._environ or 'TMP' in self._environ ):
self._environ[ 'TEMP' ] = os.path.abspath(tempfile.gettempdir())
@@ -50,8 +50,8 @@ class LocalJobRunner( BaseJobRunner ):
"""
command_line = job_wrapper.runner_command_line
## slots would be cleaner name, but don't want deployers to see examples and think it
## is going to work with other job runners.
# slots would be cleaner name, but don't want deployers to see examples and think it
# is going to work with other job runners.
slots = job_wrapper.job_destination.params.get( "local_slots", None )
if slots:
slots_statement = 'GALAXY_SLOTS="%d"; export GALAXY_SLOTS; GALAXY_SLOTS_CONFIGURED="1"; export GALAXY_SLOTS_CONFIGURED;' % ( int( slots ) )
@@ -132,7 +132,7 @@ class LocalJobRunner( BaseJobRunner ):
job_wrapper.fail("Unable to finish job", exception=True)
def stop_job( self, job ):
#if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished
# if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished
job_ext_output_metadata = job.get_external_output_metadata()
try:
pid = job_ext_output_metadata[0].job_runner_external_pid # every JobExternalOutputMetadata has a pid set, we just need to take from one of them
lib/galaxy/jobs/runners/lwr.py
+2 -2
View File
@@ -320,7 +320,7 @@ class LwrJobRunner( AsynchronousJobRunner ):
def get_client( self, job_destination_params, job_id, env=[] ):
# Cannot use url_for outside of web thread.
#files_endpoint = url_for( controller="job_files", job_id=encoded_job_id )
# files_endpoint = url_for( controller="job_files", job_id=encoded_job_id )
encoded_job_id = self.app.security.encode_id(job_id)
job_key = self.app.security.encode_id( job_id, kind="jobs_files" )
@@ -400,7 +400,7 @@ class LwrJobRunner( AsynchronousJobRunner ):
return False
def stop_job( self, job ):
#if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished
# if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished
job_ext_output_metadata = job.get_external_output_metadata()
if job_ext_output_metadata:
pid = job_ext_output_metadata[0].job_runner_external_pid # every JobExternalOutputMetadata has a pid set, we just need to take from one of them
lib/galaxy/jobs/runners/pbs.py
+9 -13
View File
@@ -189,10 +189,10 @@ class PBSJobRunner( AsynchronousJobRunner ):
if arg == 'l':
resource_attrs = value.split(',')
for j, ( res, val ) in enumerate( [ a.split('=', 1) for a in resource_attrs ] ):
rval.append( dict( name = pbs.ATTR_l, value = val, resource = res ) )
rval.append( dict( name=pbs.ATTR_l, value=val, resource=res ) )
else:
try:
rval.append( dict( name = getattr( pbs, 'ATTR_' + arg ), value = value ) )
rval.append( dict( name=getattr( pbs, 'ATTR_' + arg ), value=value ) )
except AttributeError, e:
raise Exception("Invalid parameter '%s': %s" % (arg, e))
return rval
@@ -208,9 +208,6 @@ class PBSJobRunner( AsynchronousJobRunner ):
if not self.prepare_job( job_wrapper, include_metadata=not( self.app.config.pbs_stage_path ) ):
return
# command line has been added to the wrapper by prepare_job()
command_line = job_wrapper.runner_command_line
job_destination = job_wrapper.job_destination
# Determine the job's PBS destination (server/queue) and options from the job destination definition
@@ -260,20 +257,20 @@ class PBSJobRunner( AsynchronousJobRunner ):
stagein = self.get_stage_in_out( job_wrapper.get_input_fnames() + output_files, symlink=True )
stageout = self.get_stage_in_out( output_files )
attrs = [
dict( name = pbs.ATTR_o, value = pbs_ofile ),
dict( name = pbs.ATTR_e, value = pbs_efile ),
dict( name = pbs.ATTR_stagein, value = stagein ),
dict( name = pbs.ATTR_stageout, value = stageout ),
dict( name=pbs.ATTR_o, value=pbs_ofile ),
dict( name=pbs.ATTR_e, value=pbs_efile ),
dict( name=pbs.ATTR_stagein, value=stagein ),
dict( name=pbs.ATTR_stageout, value=stageout ),
]
# If not, we're using NFS
else:
attrs = [
dict( name = pbs.ATTR_o, value = ofile ),
dict( name = pbs.ATTR_e, value = efile ),
dict( name=pbs.ATTR_o, value=ofile ),
dict( name=pbs.ATTR_e, value=efile ),
]
# define PBS job options
attrs.append( dict( name = pbs.ATTR_N, value = str( "%s_%s_%s" % ( job_wrapper.job_id, job_wrapper.tool.id, job_wrapper.user ) ) ) )
attrs.append( dict( name=pbs.ATTR_N, value=str( "%s_%s_%s" % ( job_wrapper.job_id, job_wrapper.tool.id, job_wrapper.user ) ) ) )
job_attrs = pbs.new_attropl( len( attrs ) + len( pbs_options ) )
for i, attr in enumerate( attrs + pbs_options ):
job_attrs[i].name = attr['name']
@@ -366,7 +363,6 @@ class PBSJobRunner( AsynchronousJobRunner ):
( failures, statuses ) = self.check_all_jobs()
for pbs_job_state in self.watched:
job_id = pbs_job_state.job_id
#galaxy_job_id = pbs_job_state.job_wrapper.job_id
galaxy_job_id = pbs_job_state.job_wrapper.get_id_tag()
old_state = pbs_job_state.old_state
pbs_server_name = self.__get_pbs_server(pbs_job_state.job_destination.params)
lib/galaxy/jobs/runners/pulsar.py
+3 -3
View File
@@ -301,7 +301,7 @@ class PulsarJobRunner( AsynchronousJobRunner ):
if value is PARAMETER_SPECIFICATION_IGNORED:
log.warn( "Pulsar runner in selected configuration ignores parameter %s" % key )
continue
#if self.runner_params.get( key, None ):
# if self.runner_params.get( key, None ):
# # Let plugin define defaults for some parameters -
# # for instance that way jobs_directory can be
# # configured next to AMQP url (where it belongs).
@@ -345,7 +345,7 @@ class PulsarJobRunner( AsynchronousJobRunner ):
def get_client( self, job_destination_params, job_id, env=[] ):
# Cannot use url_for outside of web thread.
#files_endpoint = url_for( controller="job_files", job_id=encoded_job_id )
# files_endpoint = url_for( controller="job_files", job_id=encoded_job_id )
encoded_job_id = self.app.security.encode_id(job_id)
job_key = self.app.security.encode_id( job_id, kind="jobs_files" )
@@ -429,7 +429,7 @@ class PulsarJobRunner( AsynchronousJobRunner ):
return False
def stop_job( self, job ):
#if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished
# if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished
client = self.get_client( job.destination_params, job.job_runner_external_id )
job_ext_output_metadata = job.get_external_output_metadata()
if not PulsarJobRunner.__remote_metadata( client ) and job_ext_output_metadata:
lib/galaxy/jobs/runners/slurm.py
+3 -3
View File
@@ -57,7 +57,7 @@ class SlurmJobRunner( DRMAAJobRunner ):
ajs.runner_state = ajs.runner_states.WALLTIME_REACHED
elif job_info['JobState'] == 'NODE_FAIL':
log.warning( '(%s/%s) Job failed due to node failure, attempting resubmission', ajs.job_wrapper.get_id_tag(), ajs.job_id )
ajs.job_wrapper.change_state( model.Job.states.QUEUED, info = 'Job was resubmitted due to node failure' )
ajs.job_wrapper.change_state( model.Job.states.QUEUED, info='Job was resubmitted due to node failure' )
try:
self.queue_job( ajs.job_wrapper )
return
@@ -82,9 +82,9 @@ class SlurmJobRunner( DRMAAJobRunner ):
return
except Exception, e:
log.exception( '(%s/%s) Unable to inspect failed slurm job using scontrol, job will be unconditionally failed: %s', ajs.job_wrapper.get_id_tag(), ajs.job_id, e )
super( SlurmJobRunner, self )._complete_terminal_job( ajs, drmaa_state = drmaa_state )
super( SlurmJobRunner, self )._complete_terminal_job( ajs, drmaa_state=drmaa_state )
# by default, finish as if the job was successful.
super( SlurmJobRunner, self )._complete_terminal_job( ajs, drmaa_state = drmaa_state )
super( SlurmJobRunner, self )._complete_terminal_job( ajs, drmaa_state=drmaa_state )
def __check_memory_limit( self, efile_path ):
"""
lib/galaxy/jobs/runners/state_handler_factory.py
+2 -2
View File
@@ -1,4 +1,3 @@
import os
import logging
from galaxy.util.submodules import submodules
@@ -8,9 +7,11 @@ import galaxy.jobs.runners.state_handlers
log = logging.getLogger(__name__)
def build_state_handlers():
return _get_state_handlers_dict()
def _get_state_handlers_dict():
state_handlers = {}
for module in submodules(galaxy.jobs.runners.state_handlers):
@@ -20,4 +21,3 @@ def _get_state_handlers_dict():
state_handlers[func].append(getattr(module, func))
log.debug("Loaded '%s' state handler from module %s", func, module.__name__)
return state_handlers
lib/galaxy/jobs/runners/state_handlers/resubmit.py
+2 -5
View File
@@ -1,15 +1,12 @@
import logging
from galaxy import model
__all__ = ['failure']
log = logging.getLogger(__name__)
MESSAGES = dict(
walltime_reached = 'it reached the walltime',
memory_limit_reached = 'it exceeded the amount of allocated memory'
walltime_reached='it reached the walltime',
memory_limit_reached='it exceeded the amount of allocated memory'
)
lib/galaxy/jobs/runners/tasks.py
+9 -10
View File
@@ -1,5 +1,4 @@
import logging
import subprocess
from galaxy import model
@@ -54,7 +53,7 @@ class TaskedJobRunner( BaseJobRunner ):
# Split with the defined method.
parallelism = job_wrapper.get_parallelism()
try:
splitter = getattr(__import__('galaxy.jobs.splitters', globals(), locals(), [parallelism.method]), parallelism.method)
splitter = getattr(__import__('galaxy.jobs.splitters', globals(), locals(), [parallelism.method]), parallelism.method)
except:
job_wrapper.change_state( model.Job.states.ERROR )
job_wrapper.fail("Job Splitting Failed, no match for '%s'" % parallelism)
@@ -78,8 +77,8 @@ class TaskedJobRunner( BaseJobRunner ):
sleep_time = 1
# sleep/loop until no more progress can be made. That is when
# all tasks are one of { OK, ERROR, DELETED }. If a task
completed_states = [ model.Task.states.OK, \
model.Task.states.ERROR, \
completed_states = [ model.Task.states.OK,
model.Task.states.ERROR,
model.Task.states.DELETED ]
# TODO: Should we report an error (and not merge outputs) if
@@ -104,7 +103,7 @@ class TaskedJobRunner( BaseJobRunner ):
self._cancel_job( job_wrapper, task_wrappers )
tasks_complete = True
break
elif not task_state in completed_states:
elif task_state not in completed_states:
tasks_complete = False
else:
job_exit_code = tw.get_exit_code()
@@ -115,15 +114,15 @@ class TaskedJobRunner( BaseJobRunner ):
sleep_time *= 2
job_wrapper.reclaim_ownership() # if running as the actual user, change ownership before merging.
log.debug('execution finished - beginning merge: %s' % command_line)
stdout, stderr = splitter.do_merge(job_wrapper, task_wrappers)
stdout, stderr = splitter.do_merge(job_wrapper, task_wrappers)
except Exception:
job_wrapper.fail( "failure running job", exception=True )
log.exception("failure running job %d" % job_wrapper.job_id)
return
#run the metadata setting script here
#this is terminate-able when output dataset/job is deleted
#so that long running set_meta()s can be canceled without having to reboot the server
# run the metadata setting script here
# this is terminate-able when output dataset/job is deleted
# so that long running set_meta()s can be canceled without having to reboot the server
self._handle_metadata_externally(job_wrapper, resolve_requirements=True )
# Finish the job
try:
@@ -147,7 +146,7 @@ class TaskedJobRunner( BaseJobRunner ):
# this if the tasks runner is used but the tool does not use
# parallelism.
else:
#if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished
# if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished
if job.external_output_metadata:
pid = job.external_output_metadata[0].job_runner_external_pid # every JobExternalOutputMetadata has a pid set, we just need to take from one of them
else:
lib/galaxy/jobs/splitters/basic.py
+7 -5
View File
@@ -3,22 +3,24 @@ import multi
log = logging.getLogger( __name__ )
def set_basic_defaults(job_wrapper):
parent_job = job_wrapper.get_job()
parallelism = job_wrapper.get_parallelism()
parallelism.attributes['split_inputs'] = parent_job.input_datasets[0].name
parallelism.attributes['merge_outputs'] = job_wrapper.get_output_hdas_and_fnames().keys()[0]
def do_split (job_wrapper):
def do_split(job_wrapper):
if len(job_wrapper.get_input_fnames()) > 1 or len(job_wrapper.get_output_fnames()) > 1:
log.error("The basic splitter is not capable of handling jobs with multiple inputs or outputs.")
raise Exception, "Job Splitting Failed, the basic splitter only handles tools with one input and one output"
raise Exception( "Job Splitting Failed, the basic splitter only handles tools with one input and one output" )
# add in the missing information for splitting the one input and merging the one output
set_basic_defaults(job_wrapper)
return multi.do_split(job_wrapper)
def do_merge( job_wrapper, task_wrappers):
def do_merge( job_wrapper, task_wrappers):
# add in the missing information for splitting the one input and merging the one output
set_basic_defaults(job_wrapper)
return multi.do_merge(job_wrapper, task_wrappers)
return multi.do_merge(job_wrapper, task_wrappers)
lib/galaxy/jobs/splitters/multi.py
+16 -15
View File
@@ -8,20 +8,21 @@ from galaxy import model, util
log = logging.getLogger( __name__ )
def do_split (job_wrapper):
def do_split(job_wrapper):
parent_job = job_wrapper.get_job()
working_directory = os.path.abspath(job_wrapper.working_directory)
parallel_settings = job_wrapper.get_parallelism().attributes
# Syntax: split_inputs="input1,input2" shared_inputs="genome"
# Designates inputs to be split or shared
split_inputs=parallel_settings.get("split_inputs")
split_inputs = parallel_settings.get("split_inputs")
if split_inputs is None:
split_inputs = []
else:
split_inputs = [x.strip() for x in split_inputs.split(",")]
shared_inputs=parallel_settings.get("shared_inputs")
shared_inputs = parallel_settings.get("shared_inputs")
auto_shared_inputs = False
if shared_inputs is None:
shared_inputs = []
@@ -32,10 +33,11 @@ def do_split (job_wrapper):
if len(illegal_inputs) > 0:
raise Exception("Inputs have conflicting parallelism attributes: %s" % str( illegal_inputs ))
subdir_index = [0] # use a list to get around Python 2.x lame closure support
subdir_index = [0] # use a list to get around Python 2.x lame closure support
task_dirs = []
def get_new_working_directory_name():
dir=os.path.join(working_directory, 'task_%d' % subdir_index[0])
dir = os.path.join(working_directory, 'task_%d' % subdir_index[0])
subdir_index[0] = subdir_index[0] + 1
if not os.path.exists(dir):
os.makedirs(dir)
@@ -55,9 +57,9 @@ def do_split (job_wrapper):
pass
else:
if input.name in split_inputs:
type_to_input_map.setdefault(input.dataset.datatype, []).append(input.name)
type_to_input_map.setdefault(input.dataset.datatype, []).append(input.name)
elif input.name in shared_inputs:
pass # pass original file name
pass # pass original file name
elif auto_shared_inputs:
shared_inputs.append(input.name)
else:
@@ -97,7 +99,7 @@ def do_split (job_wrapper):
names = job_wrapper.get_input_dataset_fnames(input.dataset)
for dir in task_dirs:
for file in names:
os.symlink(file, os.path.join(dir, os.path.basename(file)))
os.symlink(file, os.path.join(dir, os.path.basename(file)))
tasks = []
prepare_files = os.path.join(util.galaxy_directory(), 'extract_dataset_parts.sh') + ' %s'
for dir in task_dirs:
@@ -106,7 +108,7 @@ def do_split (job_wrapper):
return tasks
def do_merge( job_wrapper, task_wrappers):
def do_merge( job_wrapper, task_wrappers):
parallel_settings = job_wrapper.get_parallelism().attributes
# Syntax: merge_outputs="export" pickone_outputs="genomesize"
# Designates outputs to be merged, or selected from as a representative
@@ -137,14 +139,14 @@ def do_merge( job_wrapper, task_wrappers):
output_paths = job_wrapper.get_output_fnames()
pickone_done = []
task_dirs = [os.path.join(working_directory, x) for x in os.listdir(working_directory) if x.startswith('task_')]
task_dirs.sort(key = lambda x: int(x.split('task_')[-1]))
task_dirs.sort(key=lambda x: int(x.split('task_')[-1]))
for index, output in enumerate( outputs ):
output_file_name = str( output_paths[ index ] ) # Use false_path if set, else real path.
base_output_name = os.path.basename(output_file_name)
if output in merge_outputs:
output_dataset = outputs[output][0]
output_type = output_dataset.datatype
output_files = [os.path.join(dir,base_output_name) for dir in task_dirs]
output_files = [os.path.join(dir, base_output_name) for dir in task_dirs]
# Just include those files f in the output list for which the
# file f exists; some files may not exist if a task fails.
output_files = [ f for f in output_files if os.path.exists(f) ]
@@ -170,14 +172,14 @@ def do_merge( job_wrapper, task_wrappers):
# just pick one of them
if output not in pickone_done:
task_file_name = os.path.join(task_dirs[0], base_output_name)
shutil.move( task_file_name, output_file_name )
shutil.move( task_file_name, output_file_name )
pickone_done.append(output)
else:
log_error = "The output '%s' does not define a method for implementing parallelism" % output
log.exception(log_error)
raise Exception(log_error)
except Exception, e:
stdout = 'Error merging files';
stdout = 'Error merging files'
log.exception( stdout )
stderr = str(e)
@@ -190,5 +192,4 @@ def do_merge( job_wrapper, task_wrappers):
stdout += "\n" + tw.working_directory + ':\n' + out
if len(err) > 0:
stderr += "\n" + tw.working_directory + ':\n' + err
return (stdout, stderr)
return (stdout, stderr)
lib/galaxy/jobs/transfer_manager.py
+15 -5
View File
@@ -2,13 +2,17 @@
Manage transfers from arbitrary URLs to temporary files. Socket interface for
IPC with multiple process configurations.
"""
import os, subprocess, socket, logging, threading
import logging
import os
import subprocess
import socket
import threading
from galaxy import eggs
from galaxy.util import listify, json
log = logging.getLogger( __name__ )
class TransferManager( object ):
"""
Manage simple data transfers from URLs to temporary locations.
@@ -23,6 +27,7 @@ class TransferManager( object ):
self.sleeper = Sleeper()
self.restarter = threading.Thread( target=self.__restarter )
self.restarter.start()
def new( self, path=None, **kwd ):
if 'protocol' not in kwd:
raise Exception( 'Missing required parameter "protocol".' )
@@ -40,6 +45,7 @@ class TransferManager( object ):
self.sa_session.add( transfer_job )
self.sa_session.flush()
return transfer_job
def run( self, transfer_jobs ):
"""
This method blocks, so if invoking the transfer manager ever starts
@@ -56,8 +62,6 @@ class TransferManager( object ):
self.sa_session.add_all( transfer_jobs )
self.sa_session.flush()
for tj in transfer_jobs:
params_dict = tj.params
protocol = params_dict[ 'protocol' ]
# The transfer script should daemonize fairly quickly - if this is
# not the case, this process will need to be moved to a
# non-blocking method.
@@ -72,6 +76,7 @@ class TransferManager( object ):
tj.info = 'Spawning transfer job failed: %s' % output.splitlines()[-1]
self.sa_session.add( tj )
self.sa_session.flush()
def get_state( self, transfer_jobs, via_socket=False ):
transfer_jobs = listify( transfer_jobs )
rval = []
@@ -116,11 +121,12 @@ class TransferManager( object ):
if len( rval ) == 1:
return rval[0]
return rval
def __restarter( self ):
log.info( 'Transfer job restarter starting up...' )
while self.running:
dead = []
self.sa_session.expunge_all() # our session is threadlocal so this is safe.
self.sa_session.expunge_all() # our session is threadlocal so this is safe.
for tj in self.sa_session.query( self.app.model.TransferJob ) \
.filter( self.app.model.TransferJob.state == self.app.model.TransferJob.states.RUNNING ):
if not tj.pid:
@@ -145,10 +151,12 @@ class TransferManager( object ):
self.run( dead )
self.sleeper.sleep( 30 )
log.info( 'Transfer job restarter shutting down...' )
def shutdown( self ):
self.running = False
self.sleeper.wake()
class Sleeper( object ):
"""
Provides a 'sleep' method that sleeps for a number of seconds *unless*
@@ -156,10 +164,12 @@ class Sleeper( object ):
"""
def __init__( self ):
self.condition = threading.Condition()
def sleep( self, seconds ):
self.condition.acquire()
self.condition.wait( seconds )
self.condition.release()
def wake( self ):
self.condition.acquire()
self.condition.notify()