Changeset 959:31aef7f8fff9

Timestamp:

02/14/08 15:00:39 (1 year ago)

Author:

Nate Coraor <nate@bx.psu.edu>

branch:

default

convert_revision:

svn:9bcadc22-80f8-0310-8a53-c8f022958886/galaxy/trunk@2315

Message:

Some job runner changes, tools may now be run with a defined job
runner. Tools can choose which PBS server/queue to run on. See
changes to the ini file for hints on how this is done.

Some of the PBS/cluster options have changed, be sure to change these
when updating test/main!

Files:

• lib/galaxy/config.py (modified) (2 diffs)
• lib/galaxy/jobs/__init__.py (modified) (6 diffs)
• lib/galaxy/jobs/runners/pbs.py (modified) (11 diffs)
• lib/galaxy/tools/__init__.py (modified) (3 diffs)
• universe_wsgi.ini.sample (modified) (1 diff)

lib/galaxy/config.py

@@ -48 +48 @@
-        self.error_email_to = kwargs.get( 'error_email_to', None )
-        self.smtp_server = kwargs.get( 'smtp_server', None )
-
-
-
+
+
-        self.pbs_application_server = kwargs.get('pbs_application_server', "" )
-        self.pbs_dataset_server = kwargs.get('pbs_dataset_server', "" )
@@ -68 +67 @@
-        if global_conf and "__file__" in global_conf:
-            global_conf_parser.read(global_conf['__file__'])
+        #Store per-tool runner config
+        try:
+            self.tool_runners = global_conf_parser.items("galaxy:tool_runners")
+        except ConfigParser.NoSectionError:
+            self.tool_runners = []
-        #Store datatypes config
-        try:

lib/galaxy/jobs/__init__.py

@@ -118 +118 @@
-                pbs_job_state.job_file = "%s/database/pbs/%s.sh" % (os.getcwd(), j.id)
-                pbs_job_state.job_id = str( j.job_runner_external_id )
+                pbs_job_state.runner_url = job_wrapper.tool.job_runner
-                job_wrapper.command_line = j.command_line
-                pbs_job_state.job_wrapper = job_wrapper
@@ -125 +126 @@
-                    pbs_job_state.old_state = 'R'
-                    pbs_job_state.running = True
-pbs_job_runner
+job_runners["pbs"]
-                elif j.state == model.Job.states.QUEUED:
-                    # The job was in PBS but not yet running
@@ -132 +133 @@
-                        pbs_job_state.old_state = 'Q'
-                        pbs_job_state.running = False
-pbs_job_runner
+job_runners["pbs"]
-                    # The job had reached the pbs queue_job method, but not the PBS queue itself.
-                    else:
@@ -363 +364 @@
-        job.flush()
-
-name
+url
-        job = model.Job.get( self.job_id )
-        job.refresh()
-name
+url
-        job.job_runner_external_id = external_id
-        job.flush()
@@ -456 +457 @@
-    def cleanup( self ):
-        # remove temporary files
-
-
-
-
+
+
+
+
+
+
+
-        
-    def get_command_line( self ):
@@ -488 +492 @@
-    def __init__( self, app ):
-        self.app = app
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
-            
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
-    def shutdown( self ):
-
-
-
+
+

lib/galaxy/jobs/runners/pbs.py

@@ -53 +53 @@
-        self.ofile = None
-        self.efile = None
+        self.runner_url = None
-
-class PBSJobRunner( object ):
@@ -72 +73 @@
-        self.watched = []
-        self.queue = Queue()
-
+
+
-        self.monitor_thread = threading.Thread( target=self.monitor )
-        self.monitor_thread.start()
-        log.debug( "ready" )
-
-
+, url
-        """Determine what PBS server we are connecting to"""
-
-
-
-
-
+
+
+
+
+
+
+
+
-            raise Exception( "Could not find torque server" )
+        return server
+
+    def determine_pbs_queue( self, url ):
+        """Determine what PBS queue we are submitting to"""
+        url_split = url.split("/")
+        queue = url_split[3]
+        if queue == "":
+            # None = server's default queue
+            queue = None
+        return queue
-
-    def queue_job( self, job_wrapper ):
@@ -90 +105 @@
-        job_wrapper.prepare()
-        command_line = job_wrapper.get_command_line()
+        runner_url = job_wrapper.tool.job_runner
-        
-        # This is silly, why would we queue a job with no command line?
@@ -98 +114 @@
-        job_wrapper.change_state( 'queued' )
-        
-
+
+
+
-        if conn <= 0:
-            raise Exception( "Connection to PBS server for submit failed" )
@@ -106 +124 @@
-        efile = "%s/database/pbs/%s.e" % (os.getcwd(), job_wrapper.job_id)
-
-to get returned via scp/rcp we need this
+If an application server is set, we're staging
-        if self.app.config.pbs_application_server:
-            # separate vars because we need un-munged ofile/efile for collection/cleanup
-            pbs_ofile = self.app.config.pbs_application_server + ':' + ofile
-            pbs_efile = self.app.config.pbs_application_server + ':' + efile
@@ -124 +141 @@
-            job_attrs[4].name = pbs.ATTR_N
-            job_attrs[4].value = "%s" % job_wrapper.job_id
-self.app.config.pbs_instance_path
-    # FIXME: pbs_instance_path is broken until we munge PATH and PYTHONPATH here
+os.path.abspath( os.getcwd() )
+# If not, we're using NFS
-        else:
-            job_attrs = pbs.new_attropl(2)
@@ -150 +167 @@
-
-        galaxy_job_id = job_wrapper.job_id
-
-
-
-
+
+
+
+
+
+
+
-        pbs.pbs_disconnect(conn)
-
-        if not job_id:
+            errno, text = pbs.error()
+            log.debug( "(%s) pbs_submit failed, PBS error %d: %s" % (galaxy_job_id, errno, text) )
-            stdout = ''
-
-
-
-
-
+
+
+
+
+
+
+
+
+
-
-        # Get initial job state
-        stat_attrl = pbs.new_attrl(1)
-        stat_attrl[0].name = 'job_state'
-self.
+
-        if conn > 0:
-            jobs = pbs.pbs_statjob(conn, job_id, stat_attrl, 'NULL')
@@ -201 +227 @@
-        pbs_job_state.old_state = old_state
-        pbs_job_state.running = running
+        pbs_job_state.runner_url = runner_url
-        
-        # Add to our 'queue' of jobs to monitor
@@ -237 +264 @@
-            old_state = pbs_job_state.old_state
-            running = pbs_job_state.running
-
+
+
-            if conn <= 0:
-                log.debug("(%s/%s) connection to PBS server for state check failed" % (galaxy_job_id, job_id) )
@@ -262 +290 @@
-                        if killed and state != "E":
-                            log.debug( "(%s/%s) all output datasets deleted by user, dequeueing" % (galaxy_job_id, job_id) )
-
+, pbs_server
-                        old_state = state
-                    pbs_job_state.old_state = old_state
@@ -314 +342 @@
-        log.info( "pbs job runner stopped" )
-
-
+, pbs_server
-        """Attempts to delete a job from the PBS queue"""
-self.
+
-        if conn <= 0:
-            log.debug("(%s) connection to PBS server for job delete failed" % job_id )

lib/galaxy/tools/__init__.py

@@ -201 +201 @@
-            interpreter  = command.get("interpreter")
-            if interpreter:
-
-
-
-
-
-
-
-
+
+
+
+
+
-        else:
-            self.command = ''
-        # Short description of the tool
-        self.description = util.xml_text(root, "description")
+        # Job runner
+        if self.app.config.start_job_runners is None:
+            # Jobs are always local regardless of tool config if no additional
+            # runners are started
+            self.job_runner = "local:///"
+        else:
+            # Set job runner to the cluster default
+            self.job_runner = self.app.config.default_cluster_job_runner
+            for tup in self.app.config.tool_runners:
+                if tup[0] == self.id:
+                    self.job_runner = tup[1]
+                    break
-        # Is this a 'hidden' tool (hidden in tool menu)
-        self.hidden = util.xml_text(root, "hidden")
@@ -900 +909 @@
-            param_dict[name] = DatasetFilenameWrapper( data )
-            # Provide access to a path to store additional files
-
-
-
-
+
+
-            
-            for child_association in data.children:
@@ -913 +920 @@
-        # More convienent access to app.config.new_file_path; we don't need to wrap a string
-        # But this method of generating additional datasets should be considered DEPRECATED
-
-
-
-
+
+
-        # Return the dictionary of parameters
-        return param_dict

universe_wsgi.ini.sample

@@ -113 +113 @@
-#wiki_url=/path/to/my/local/wiki
-#bugs_email=mailto:bugmaster@this.site.com
+
+# ---- Job Runners ----------------------------------------------------------
+
+# Leave everything in this section commented unless you are running Galaxy on a
+# cluster.  Clustering Galaxy is not a straightforward process and requires a
+# lot of pre-configuration.  Please see the Galaxy Wiki before attempting to
+# set any of these options.
+
+# start_job_runners: Comma-separated list of job runners to start.  local is
+# always started.  If left commented, no jobs will be run on the cluster, even
+# if a cluster URL is explicitly defined in the [galaxy:tool_runners] section
+# below.  The only runner currently available is 'pbs'.
+#start_job_runners = pbs
+
+# default_cluster_job_runner: The URL for the default runner to use when a tool
+# doesn't explicity define a runner below.  For help on the cluster URL format,
+# see the Galaxy Wiki.  Leave commented if not using a cluster job runner.
+#default_cluster_job_runner = pbs:///
+
+# The PBS options are described in detail on the Galaxy Wiki
+#pbs_application_server = 
+#pbs_stage_path = 
+#pbs_dataset_server = 
+
+# Individual per-tool job runner overrides.  If not listed here, a tool will
+# run with the runner defined with default_cluster_job_runner.
+[galaxy:tool_runners]
+
+biomart = local:///
+encode_db1 = local:///
+encode_import_all_latest_datasets1 = local:///
+encode_import_chromatin_and_chromosomes1 = local:///
+encode_import_gencode1 = local:///
+encode_import_genes_and_transcripts1 = local:///
+encode_import_multi-species_sequence_analysis1 = local:///
+encode_import_transcription_regulation1 = local:///
+hbvar = local:///
+microbial_import1 = local:///
+ucsc_table_direct1 = local:///
+ucsc_table_direct_archaea1 = local:///
+ucsc_table_direct_test1 = local:///
+upload1 = local:///
-
-# ---- Datatypes ------------------------------------------------------------