November 2008 - galaxy-dev - lists.galaxyproject.org

[hg] galaxy 1628: Fixes for the incorrect_gops_jobs script.
by Greg Von Kuster 12 Nov '08

12 Nov '08

details: http://www.bx.psu.edu/hg/galaxy/rev/54f12bb6cb29 changeset: 1628:54f12bb6cb29 user: Greg Von Kuster <greg(a)bx.psu.edu> date: Wed Nov 12 15:19:27 2008 -0500 description: Fixes for the incorrect_gops_jobs script. 2 file(s) affected in this change: scripts/others/incorrect_gops_jobs.py scripts/others/incorrect_gops_jobs.sh diffs (106 lines): diff -r 2cf304a0c1c6 -r 54f12bb6cb29 scripts/others/incorrect_gops_jobs.py --- a/scripts/others/incorrect_gops_jobs.py Wed Nov 12 10:16:04 2008 -0500 +++ b/scripts/others/incorrect_gops_jobs.py Wed Nov 12 15:19:27 2008 -0500 @@ -1,6 +1,7 @@ #!/usr/bin/env python """ -Fetch jobs using gops_intersect, gops_merge, gops_subtract, gops_complement, gops_coverage wherein the second dataset doesn't have chr, start and end in standard columns 1,2 and 3. +Fetch jobs using gops_intersect, gops_merge, gops_subtract, gops_complement, gops_coverage +wherein the second dataset doesn't have chr, start and end in standard columns 1, 2 and 3. """ from galaxy import eggs @@ -38,39 +39,53 @@ file_path = configuration['file_path'] app = TestApplication( database_connection=database_connection, file_path=file_path ) jobs = {} - for job in app.model.Job.filter( - sa.and_( - app.model.Job.table.c.create_time.between('2008-05-23','2008-11-29'), - app.model.Job.table.c.state == 'ok', - sa.or_(app.model.Job.table.c.tool_id == 'gops_intersect_1', - app.model.Job.table.c.tool_id == 'gops_subtract_1', - app.model.Job.table.c.tool_id == 'gops_merge_1', - app.model.Job.table.c.tool_id == 'gops_coverage_1', - app.model.Job.table.c.tool_id == 'gops_complement_1', - ), - sa.not_(app.model.Job.table.c.command_line.like('%-2 1,2,3%')) - )).all(): - for od in job.output_datasets: - ds = app.model.Dataset.get(od.dataset_id) - if not ds.deleted: - for hda in ds.history_associations: - hist = app.model.History.get(hda.history_id) - if hist.user_id: - user = app.model.User.get(hist.user_id) - jobs[job.id] = {} - jobs[job.id]['dataset_id'] = ds.id - jobs[job.id]['dataset_create_time'] = ds.create_time - jobs[job.id]['dataset_name'] = hda.name - jobs[job.id]['hda_id'] = hda.id - jobs[job.id]['hist_id'] = hist.id - jobs[job.id]['hist_name'] = hist.name - jobs[job.id]['hist_modified_time'] = hist.update_time - jobs[job.id]['user_email'] = user.email + try: + for job in app.model.Job.filter( sa.and_( app.model.Job.table.c.create_time.between( '2008-05-23', '2008-11-29' ), + app.model.Job.table.c.state == 'ok', + sa.or_( app.model.Job.table.c.tool_id == 'gops_intersect_1', + app.model.Job.table.c.tool_id == 'gops_subtract_1', + app.model.Job.table.c.tool_id == 'gops_merge_1', + app.model.Job.table.c.tool_id == 'gops_coverage_1', + app.model.Job.table.c.tool_id == 'gops_complement_1', + ), + sa.not_( app.model.Job.table.c.command_line.like( '%-2 1,2,3%' ) ) + ) + ).all(): + print "# processing job id %s" % str( job.id ) + for jtoda in job.output_datasets: + print "# --> processing JobToOutputDatasetAssociation id %s" % str( jtoda.id ) + hda = app.model.HistoryDatasetAssociation.get( jtoda.dataset_id ) + print "# ----> processing HistoryDatasetAssociation id %s" % str( hda.id ) + if not hda.deleted: + # Probably don't need this check, since the job state should suffice, but... + if hda.dataset.state == 'ok': + history = app.model.History.get( hda.history_id ) + print "# ------> processing history id %s" % str( history.id ) + if history.user_id: + user = app.model.User.get( history.user_id ) + jobs[ job.id ] = {} + jobs[ job.id ][ 'hda_id' ] = hda.id + jobs[ job.id ][ 'hda_name' ] = hda.name + jobs[ job.id ][ 'hda_info' ] = hda.info + jobs[ job.id ][ 'history_id' ] = history.id + jobs[ job.id ][ 'history_name' ] = history.name + jobs[ job.id ][ 'history_update_time' ] = history.update_time + jobs[ job.id ][ 'user_email' ] = user.email + except Exception, e: + print "# caught exception: %s" % str( e ) - print "Number of Incorrect Jobs: %d\n"%(len(jobs)) - print "#job_id\tdataset_id\tdataset_create_time\thda_id\thistory_id\thistory_name\thistory_modified_time\tuser_email" + print "\n\n# Number of incorrect Jobs: %d\n\n" % ( len( jobs ) ) + print "#job_id\thda_id\thda_name\thda_info\thistory_id\thistory_name\thistory_update_time\tuser_email" for jid in jobs: - print "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s" %(jid,jobs[jid]['dataset_id'],jobs[jid]['dataset_create_time'],jobs[jid]['dataset_name'],jobs[jid]['hda_id'],jobs[jid]['hist_id'],jobs[jid]['hist_name'],jobs[jid]['hist_modified_time'],jobs[jid]['user_email'] ) + print "%s\t%s\t%s\t%s\t%s\t%s\t%s" % \ + ( jid, jobs[ jid ][ 'hda_id' ], + jobs[ jid ][ 'hda_name' ], + jobs[ jid ][ 'hda_info' ], + jobs[ jid ][ 'history_id' ], + jobs[ jid ][ 'history_name' ], + jobs[ jid ][ 'history_update_time' ], + jobs[ jid ][ 'user_email' ] + ) sys.exit(0) if __name__ == "__main__": diff -r 2cf304a0c1c6 -r 54f12bb6cb29 scripts/others/incorrect_gops_jobs.sh --- a/scripts/others/incorrect_gops_jobs.sh Wed Nov 12 10:16:04 2008 -0500 +++ b/scripts/others/incorrect_gops_jobs.sh Wed Nov 12 15:19:27 2008 -0500 @@ -1,4 +1,4 @@ #!/bin/sh cd `dirname $0`/../.. -python ./scripts/others/incorrect_gops_jobs.py ./universe_wsgi.ini +python ./scripts/others/incorrect_gops_jobs.py ./universe_wsgi.ini >> ./scripts/others/incorrect_gops_jobs.log

1 0

[hg] galaxy 1627: Fixed an indentation bug in incorrect_gops_job...
by gua110＠scofield.bx.psu.edu 12 Nov '08

12 Nov '08

details: http://www.bx.psu.edu/hg/galaxy/rev/2cf304a0c1c6 changeset: 1627:2cf304a0c1c6 user: guru date: Wed Nov 12 10:16:04 2008 -0500 description: Fixed an indentation bug in incorrect_gops_jobs script. 1 file(s) affected in this change: scripts/others/incorrect_gops_jobs.py diffs (30 lines): diff -r ddad87f713cc -r 2cf304a0c1c6 scripts/others/incorrect_gops_jobs.py --- a/scripts/others/incorrect_gops_jobs.py Tue Nov 11 16:33:50 2008 -0500 +++ b/scripts/others/incorrect_gops_jobs.py Wed Nov 12 10:16:04 2008 -0500 @@ -6,14 +6,11 @@ from galaxy import eggs import sys, os, ConfigParser import galaxy.app -#from galaxy.util.bunch import Bunch - import galaxy.model.mapping import pkg_resources pkg_resources.require( "SQLAlchemy >= 0.4" ) import sqlalchemy as sa -#from sqlalchemy.orm import * assert sys.version_info[:2] >= ( 2, 4 ) @@ -53,10 +50,9 @@ ), sa.not_(app.model.Job.table.c.command_line.like('%-2 1,2,3%')) )).all(): - for od in job.output_datasets: + for od in job.output_datasets: ds = app.model.Dataset.get(od.dataset_id) if not ds.deleted: - #hda = app.model.HistoryDatasetAssociation.filter(app.model.HistoryDatasetAssociation.table.c.dataset_id == ds.id)[0] for hda in ds.history_associations: hist = app.model.History.get(hda.history_id) if hist.user_id:

1 0

[hg] galaxy 1625: small modification to incorrect_gops_jobs scri...
by Nate Coraor 11 Nov '08

11 Nov '08

details: http://www.bx.psu.edu/hg/galaxy/rev/82cf78640bc1 changeset: 1625:82cf78640bc1 user: guru date: Tue Nov 11 16:20:12 2008 -0500 description: small modification to incorrect_gops_jobs script to include all HDAs associated with a dataset. 1 file(s) affected in this change: scripts/others/incorrect_gops_jobs.py diffs (37 lines): diff -r bffe6becd372 -r 82cf78640bc1 scripts/others/incorrect_gops_jobs.py --- a/scripts/others/incorrect_gops_jobs.py Tue Nov 11 15:20:36 2008 -0500 +++ b/scripts/others/incorrect_gops_jobs.py Tue Nov 11 16:20:12 2008 -0500 @@ -56,19 +56,20 @@ for od in job.output_datasets: ds = app.model.Dataset.get(od.dataset_id) if not ds.deleted: - hda = app.model.HistoryDatasetAssociation.filter(app.model.HistoryDatasetAssociation.table.c.dataset_id == ds.id)[0] - hist = app.model.History.get(hda.history_id) - if hist.user_id: - user = app.model.User.get(hist.user_id) - jobs[job.id] = {} - jobs[job.id]['dataset_id'] = ds.id - jobs[job.id]['dataset_create_time'] = ds.create_time - jobs[job.id]['dataset_name'] = hda.name - jobs[job.id]['hda_id'] = hda.id - jobs[job.id]['hist_id'] = hist.id - jobs[job.id]['hist_name'] = hist.name - jobs[job.id]['hist_modified_time'] = hist.update_time - jobs[job.id]['user_email'] = user.email + #hda = app.model.HistoryDatasetAssociation.filter(app.model.HistoryDatasetAssociation.table.c.dataset_id == ds.id)[0] + for hda in ds.history_associations: + hist = app.model.History.get(hda.history_id) + if hist.user_id: + user = app.model.User.get(hist.user_id) + jobs[job.id] = {} + jobs[job.id]['dataset_id'] = ds.id + jobs[job.id]['dataset_create_time'] = ds.create_time + jobs[job.id]['dataset_name'] = hda.name + jobs[job.id]['hda_id'] = hda.id + jobs[job.id]['hist_id'] = hist.id + jobs[job.id]['hist_name'] = hist.name + jobs[job.id]['hist_modified_time'] = hist.update_time + jobs[job.id]['user_email'] = user.email print "Number of Incorrect Jobs: %d\n"%(len(jobs)) print "#job_id\tdataset_id\tdataset_create_time\thda_id\thistory_id\thistory_name\thistory_modified_time\tuser_email"

1 0

[hg] galaxy 1624: Script to fetch gops jobs that might have prod...
by Nate Coraor 11 Nov '08

11 Nov '08

details: http://www.bx.psu.edu/hg/galaxy/rev/bffe6becd372 changeset: 1624:bffe6becd372 user: guru date: Tue Nov 11 15:20:36 2008 -0500 description: Script to fetch gops jobs that might have produced incorrect/empty oUtput due to the bug in BitsetSafeReaderWrapper. 2 file(s) affected in this change: scripts/others/incorrect_gops_jobs.py scripts/others/incorrect_gops_jobs.sh diffs (92 lines): diff -r 41a4d178466d -r bffe6becd372 scripts/others/incorrect_gops_jobs.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/scripts/others/incorrect_gops_jobs.py Tue Nov 11 15:20:36 2008 -0500 @@ -0,0 +1,80 @@ +#!/usr/bin/env python +""" +Fetch jobs using gops_intersect, gops_merge, gops_subtract, gops_complement, gops_coverage wherein the second dataset doesn't have chr, start and end in standard columns 1,2 and 3. +""" + +from galaxy import eggs +import sys, os, ConfigParser +import galaxy.app +#from galaxy.util.bunch import Bunch + +import galaxy.model.mapping +import pkg_resources + +pkg_resources.require( "SQLAlchemy >= 0.4" ) +import sqlalchemy as sa +#from sqlalchemy.orm import * + +assert sys.version_info[:2] >= ( 2, 4 ) + +class TestApplication( object ): + """Encapsulates the state of a Universe application""" + def __init__( self, database_connection=None, file_path=None ): + print >> sys.stderr, "python path is: " + ", ".join( sys.path ) + if database_connection is None: + raise Exception( "CleanupDatasetsApplication requires a database_connection value" ) + if file_path is None: + raise Exception( "CleanupDatasetsApplication requires a file_path value" ) + self.database_connection = database_connection + self.file_path = file_path + # Setup the database engine and ORM + self.model = galaxy.model.mapping.init( self.file_path, self.database_connection, engine_options={}, create_tables=False ) + +def main(): + ini_file = sys.argv[1] + conf_parser = ConfigParser.ConfigParser( {'here':os.getcwd()} ) + conf_parser.read( ini_file ) + configuration = {} + for key, value in conf_parser.items( "app:main" ): + configuration[key] = value + database_connection = configuration['database_connection'] + file_path = configuration['file_path'] + app = TestApplication( database_connection=database_connection, file_path=file_path ) + jobs = {} + for job in app.model.Job.filter( + sa.and_( + app.model.Job.table.c.create_time.between('2008-05-23','2008-11-29'), + app.model.Job.table.c.state == 'ok', + sa.or_(app.model.Job.table.c.tool_id == 'gops_intersect_1', + app.model.Job.table.c.tool_id == 'gops_subtract_1', + app.model.Job.table.c.tool_id == 'gops_merge_1', + app.model.Job.table.c.tool_id == 'gops_coverage_1', + app.model.Job.table.c.tool_id == 'gops_complement_1', + ), + sa.not_(app.model.Job.table.c.command_line.like('%-2 1,2,3%')) + )).all(): + for od in job.output_datasets: + ds = app.model.Dataset.get(od.dataset_id) + if not ds.deleted: + hda = app.model.HistoryDatasetAssociation.filter(app.model.HistoryDatasetAssociation.table.c.dataset_id == ds.id)[0] + hist = app.model.History.get(hda.history_id) + if hist.user_id: + user = app.model.User.get(hist.user_id) + jobs[job.id] = {} + jobs[job.id]['dataset_id'] = ds.id + jobs[job.id]['dataset_create_time'] = ds.create_time + jobs[job.id]['dataset_name'] = hda.name + jobs[job.id]['hda_id'] = hda.id + jobs[job.id]['hist_id'] = hist.id + jobs[job.id]['hist_name'] = hist.name + jobs[job.id]['hist_modified_time'] = hist.update_time + jobs[job.id]['user_email'] = user.email + + print "Number of Incorrect Jobs: %d\n"%(len(jobs)) + print "#job_id\tdataset_id\tdataset_create_time\thda_id\thistory_id\thistory_name\thistory_modified_time\tuser_email" + for jid in jobs: + print "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s" %(jid,jobs[jid]['dataset_id'],jobs[jid]['dataset_create_time'],jobs[jid]['dataset_name'],jobs[jid]['hda_id'],jobs[jid]['hist_id'],jobs[jid]['hist_name'],jobs[jid]['hist_modified_time'],jobs[jid]['user_email'] ) + sys.exit(0) + +if __name__ == "__main__": + main() diff -r 41a4d178466d -r bffe6becd372 scripts/others/incorrect_gops_jobs.sh --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/scripts/others/incorrect_gops_jobs.sh Tue Nov 11 15:20:36 2008 -0500 @@ -0,0 +1,4 @@ +#!/bin/sh + +cd `dirname $0`/../.. +python ./scripts/others/incorrect_gops_jobs.py ./universe_wsgi.ini

1 0

[hg] galaxy 1626: Change state to running before thread handoff.
by Nate Coraor 11 Nov '08

11 Nov '08

details: http://www.bx.psu.edu/hg/galaxy/rev/ddad87f713cc changeset: 1626:ddad87f713cc user: Nate Coraor <nate(a)bx.psu.edu> date: Tue Nov 11 16:33:50 2008 -0500 description: Change state to running before thread handoff. 1 file(s) affected in this change: lib/galaxy/jobs/runners/local.py diffs (20 lines): diff -r 82cf78640bc1 -r ddad87f713cc lib/galaxy/jobs/runners/local.py --- a/lib/galaxy/jobs/runners/local.py Tue Nov 11 16:20:12 2008 -0500 +++ b/lib/galaxy/jobs/runners/local.py Tue Nov 11 16:33:50 2008 -0500 @@ -40,7 +40,6 @@ log.exception( "Uncaught exception running job" ) def run_job( self, job_wrapper ): - job_wrapper.change_state( 'running' ) job_wrapper.set_runner( 'local:///', None ) stderr = stdout = command_line = '' # Prepare the job to run @@ -86,6 +85,8 @@ def put( self, job_wrapper ): """Add a job to the queue (by job identifier)""" + # Change to queued state before handing to worker thread so the runner won't pick it up again + job_wrapper.change_state( 'running' ) self.queue.put( job_wrapper ) def shutdown( self ):

1 0

[hg] galaxy 1623: Whoops, forgot to fix changed config param in ...
by Nate Coraor 11 Nov '08

11 Nov '08

details: http://www.bx.psu.edu/hg/galaxy/rev/41a4d178466d changeset: 1623:41a4d178466d user: Nate Coraor <nate(a)bx.psu.edu> date: Tue Nov 11 14:54:24 2008 -0500 description: Whoops, forgot to fix changed config param in the local runner. 1 file(s) affected in this change: lib/galaxy/jobs/runners/local.py diffs (12 lines): diff -r fb44755b4c56 -r 41a4d178466d lib/galaxy/jobs/runners/local.py --- a/lib/galaxy/jobs/runners/local.py Tue Nov 11 14:51:13 2008 -0500 +++ b/lib/galaxy/jobs/runners/local.py Tue Nov 11 14:54:24 2008 -0500 @@ -20,7 +20,7 @@ self.app = app self.queue = Queue() self.threads = [] - nworkers = app.config.job_queue_workers + nworkers = app.config.local_job_queue_workers log.info( "starting workers" ) for i in range( nworkers ): worker = threading.Thread( target=self.run_next )

1 0

[hg] galaxy 1622: Start the job queue after the utils.
by Nate Coraor 11 Nov '08

11 Nov '08

details: http://www.bx.psu.edu/hg/galaxy/rev/fb44755b4c56 changeset: 1622:fb44755b4c56 user: Nate Coraor <nate(a)bx.psu.edu> date: Tue Nov 11 14:51:13 2008 -0500 description: Start the job queue after the utils. 1 file(s) affected in this change: lib/galaxy/app.py diffs (27 lines): diff -r 45067d8114eb -r fb44755b4c56 lib/galaxy/app.py --- a/lib/galaxy/app.py Tue Nov 11 14:48:25 2008 -0500 +++ b/lib/galaxy/app.py Tue Nov 11 14:51:13 2008 -0500 @@ -30,11 +30,6 @@ self.toolbox = tools.ToolBox( self.config.tool_config, self.config.tool_path, self ) #Load datatype converters self.datatypes_registry.load_datatype_converters( self.toolbox ) - # Start the job queue - self.job_manager = jobs.JobManager( self ) - # FIXME: These are exposed directly for backward compatibility - self.job_queue = self.job_manager.job_queue - self.job_stop_queue = self.job_manager.job_stop_queue # Heartbeat and memdump for thread / heap profiling self.heartbeat = None self.memdump = None @@ -54,6 +49,11 @@ if self.config.log_memory_usage: from galaxy.util import memory_usage self.memory_usage = memory_usage + # Start the job queue + self.job_manager = jobs.JobManager( self ) + # FIXME: These are exposed directly for backward compatibility + self.job_queue = self.job_manager.job_queue + self.job_stop_queue = self.job_manager.job_stop_queue def shutdown( self ): self.job_manager.shutdown() if self.heartbeat:

1 0

[hg] galaxy 1621: Add a thread pool for the PBS runner to use fo...
by Nate Coraor 11 Nov '08

11 Nov '08

details: http://www.bx.psu.edu/hg/galaxy/rev/45067d8114eb changeset: 1621:45067d8114eb user: Nate Coraor <nate(a)bx.psu.edu> date: Tue Nov 11 14:48:25 2008 -0500 description: Add a thread pool for the PBS runner to use for queue/finish methods. 3 file(s) affected in this change: lib/galaxy/config.py lib/galaxy/jobs/runners/pbs.py universe_wsgi.ini.sample diffs (156 lines): diff -r bc74a32f875e -r 45067d8114eb lib/galaxy/config.py --- a/lib/galaxy/config.py Tue Nov 11 10:19:21 2008 -0500 +++ b/lib/galaxy/config.py Tue Nov 11 14:48:25 2008 -0500 @@ -40,7 +40,8 @@ self.remote_user_maildomain = kwargs.get( "remote_user_maildomain", None ) self.template_path = resolve_path( kwargs.get( "template_path", "templates" ), self.root ) self.template_cache = resolve_path( kwargs.get( "template_cache_path", "database/compiled_templates" ), self.root ) - self.job_queue_workers = int( kwargs.get( "job_queue_workers", "10" ) ) + self.local_job_queue_workers = int( kwargs.get( "local_job_queue_workers", "5" ) ) + self.cluster_job_queue_workers = int( kwargs.get( "cluster_job_queue_workers", "5" ) ) self.job_scheduler_policy = kwargs.get("job_scheduler_policy", "FIFO") self.job_queue_cleanup_interval = int( kwargs.get("job_queue_cleanup_interval", "5") ) self.job_working_directory = resolve_path( kwargs.get( "job_working_directory", "database/job_working_directory" ), self.root ) diff -r bc74a32f875e -r 45067d8114eb lib/galaxy/jobs/runners/pbs.py --- a/lib/galaxy/jobs/runners/pbs.py Tue Nov 11 10:19:21 2008 -0500 +++ b/lib/galaxy/jobs/runners/pbs.py Tue Nov 11 14:48:25 2008 -0500 @@ -78,13 +78,20 @@ # be modified by the monitor thread, which will move items from 'queue' # to 'watched' and then manage the watched jobs. self.watched = [] - self.queue = Queue() + self.monitor_queue = Queue() # set the default server during startup self.default_pbs_server = None self.determine_pbs_server( 'pbs:///' ) self.monitor_thread = threading.Thread( target=self.monitor ) self.monitor_thread.start() - log.debug( "ready" ) + self.work_queue = Queue() + self.work_threads = [] + nworkers = app.config.cluster_job_queue_workers + for i in range( nworkers ): + worker = threading.Thread( target=self.run_next ) + worker.start() + self.work_threads.append( worker ) + log.debug( "%d workers ready" % nworkers ) def determine_pbs_server( self, url, rewrite = False ): """Determine what PBS server we are connecting to""" @@ -112,6 +119,22 @@ queue = None return queue + def run_next( self ): + """ + Run the next item in the queue (a job waiting to run or finish ) + """ + while 1: + ( op, obj ) = self.work_queue.get() + if op is self.STOP_SIGNAL: + return + try: + if op == 'queue': + self.queue_job( obj ) + elif op == 'finish': + self.finish_job( obj ) + except: + log.exception( "Uncaught exception %sing job" % op ) + def queue_job( self, job_wrapper ): """Create PBS script for a job and submit it to the PBS queue""" @@ -136,9 +159,6 @@ job_wrapper.cleanup() return - # Change to queued state immediately - job_wrapper.change_state( 'queued' ) - ( pbs_server_name, runner_url ) = self.determine_pbs_server( runner_url, rewrite = True ) pbs_queue_name = self.determine_pbs_queue( runner_url ) c = pbs.pbs_connect( pbs_server_name ) @@ -228,7 +248,7 @@ pbs_job_state.runner_url = runner_url # Add to our 'queue' of jobs to monitor - self.queue.put( pbs_job_state ) + self.monitor_queue.put( pbs_job_state ) def monitor( self ): """ @@ -239,7 +259,7 @@ # Take any new watched jobs and put them on the monitor list try: while 1: - pbs_job_state = self.queue.get_nowait() + pbs_job_state = self.monitor_queue.get_nowait() if pbs_job_state is self.STOP_SIGNAL: # TODO: This is where any cleanup would occur return @@ -290,7 +310,7 @@ new_watched.append( pbs_job_state ) else: log.debug("(%s/%s) job has left queue" % (galaxy_job_id, job_id) ) - self.finish_job( pbs_job_state ) + self.work_queue.put( ( 'finish', pbs_job_state ) ) # Replace the watch list with the updated version self.watched = new_watched @@ -371,12 +391,17 @@ def put( self, job_wrapper ): """Add a job to the queue (by job identifier)""" - self.queue_job( job_wrapper ) + #self.queue_job( job_wrapper ) + # Change to queued state before handing to worker thread so the runner won't pick it up again + job_wrapper.change_state( 'queued' ) + self.work_queue.put( ( 'queue', job_wrapper ) ) def shutdown( self ): """Attempts to gracefully shut down the monitor thread""" log.info( "sending stop signal to worker threads" ) - self.queue.put( self.STOP_SIGNAL ) + self.monitor_queue.put( self.STOP_SIGNAL ) + for i in range( len( self.work_threads ) ): + self.work_queue.put( ( self.STOP_SIGNAL, None ) ) log.info( "pbs job runner stopped" ) def get_stage_in_out( self, fnames, symlink=False ): @@ -419,9 +444,9 @@ log.debug( "(%s/%s) is still in running state, adding to the PBS queue" % ( job.id, job.job_runner_external_id ) ) pbs_job_state.old_state = 'R' pbs_job_state.running = True - self.queue.put( pbs_job_state ) + self.monitor_queue.put( pbs_job_state ) elif job.state == model.Job.states.QUEUED: log.debug( "(%s/%s) is still in PBS queued state, adding to the PBS queue" % ( job.id, job.job_runner_external_id ) ) pbs_job_state.old_state = 'Q' pbs_job_state.running = False - self.queue.put( pbs_job_state ) + self.monitor_queue.put( pbs_job_state ) diff -r bc74a32f875e -r 45067d8114eb universe_wsgi.ini.sample --- a/universe_wsgi.ini.sample Tue Nov 11 10:19:21 2008 -0500 +++ b/universe_wsgi.ini.sample Tue Nov 11 14:48:25 2008 -0500 @@ -26,7 +26,7 @@ ## track_jobs_in_database = true # Number of concurrent jobs to run (local runner) -job_queue_workers = 5 +local_job_queue_workers = 5 # Job scheduling policy to be used. # module/package name and classname should be in "module:classname" format. @@ -155,6 +155,12 @@ # runner. #default_cluster_job_runner = pbs:/// +# cluster_job_queue_workers: The cluster runners have their own thread pools +# used to prepare and finish jobs (so that these operations do not block normal +# queue operation). The value here is the number of worker threads available +# to each runner. +#cluster_job_queue_workers = 5 + # The PBS options are described in detail in the Galaxy Configuration section of # the ClusteringGalaxy Wiki, and are only necessary when using file staging. #pbs_application_server =

1 0

Tool with several outputs -- in a directory?
by Sebastian J. Schultheiss 11 Nov '08

11 Nov '08

Dear Galaxy Team, we have developed a machine learning tool for gene finding that uses a lot of small files to pass on its parameters learned on the training data set. It would be cool if we could put them in a directory in the Galaxy dataset store, e.g. under the same name as the actual output file. What do you think is the best idea here? It would be important to be able to delete the datasets again, as they will become huge. So here is how it works right now (please correct me if there is a misconception): - the user executes a tool that creates an $output file - galaxy creates the empty file "dataset_###.dat" in the database dir - the executed program gets this file passed as a parameter and can write its info there - galaxy will display the file upon completion of the tool as soon as the user clicks on the eye logo In our case, we would like to have a directory instead of dataset_###.dat and delete it as soon as the purge_datasets.sh is run, and the user has deleted it from their history. One idea would be to replace the file dataset_###.dat with a directory of the same name. What would Galaxy do when you click on the eye? Would it be correctly deleted? Another idea we had was to create a directory without the .dat extension and somehow modify the purge_datasets.sh script to include such directories in the deletion process. What would be easiest for you? Thanks a lot for your time! -- Sebastian

1 0

[hg] galaxy 1620: Add the "upload" job state to prevent job runn...
by Greg Von Kuster 11 Nov '08

11 Nov '08

details: http://www.bx.psu.edu/hg/galaxy/rev/bc74a32f875e changeset: 1620:bc74a32f875e user: Nate Coraor <nate(a)bx.psu.edu> date: Tue Nov 11 10:19:21 2008 -0500 description: Add the "upload" job state to prevent job runner from grabbing upload jobs when tracking in the database. 2 file(s) affected in this change: lib/galaxy/model/__init__.py lib/galaxy/tools/actions/upload.py diffs (22 lines): diff -r 9ce784bcab86 -r bc74a32f875e lib/galaxy/model/__init__.py --- a/lib/galaxy/model/__init__.py Tue Nov 11 09:52:20 2008 -0500 +++ b/lib/galaxy/model/__init__.py Tue Nov 11 10:19:21 2008 -0500 @@ -46,6 +46,7 @@ parameters, and output datasets. """ states = Bunch( NEW = 'new', + UPLOAD = 'upload', WAITING = 'waiting', QUEUED = 'queued', RUNNING = 'running', diff -r 9ce784bcab86 -r bc74a32f875e lib/galaxy/tools/actions/upload.py --- a/lib/galaxy/tools/actions/upload.py Tue Nov 11 09:52:20 2008 -0500 +++ b/lib/galaxy/tools/actions/upload.py Tue Nov 11 10:19:21 2008 -0500 @@ -38,6 +38,7 @@ job.tool_version = tool.version except: job.tool_version = "1.0.0" + job.state = trans.app.model.Job.states.UPLOAD job.flush() log.info( 'tool %s created job id %d' % ( tool.id, job.id ) ) trans.log_event( 'created job id %d' % job.id, tool_id=tool.id )

1 0