James' changes allowing the server to run w/o a job runner, and

nginx-specific performance improvements.

lib/galaxy/app.py

@@ -31,9 +31,11 @@ class UniverseApplication( object ):
         #Load datatype converters
         self.datatypes_registry.load_datatype_converters( self.toolbox )
         # Start the job queue
-        job_dispatcher = jobs.DefaultJobDispatcher( self )
-        self.job_queue = jobs.JobQueue( self, job_dispatcher )
-        self.job_stop_queue = jobs.JobStopQueue( self, job_dispatcher )
+        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
         # Start the heartbeat process if configured and available
@@ -48,7 +50,6 @@ class UniverseApplication( object ):
             if memdump.Memdump:
                 self.memdump = memdump.Memdump()
     def shutdown( self ):
-        self.job_stop_queue.shutdown()
-        self.job_queue.shutdown()
+        self.job_manager.shutdown()
         if self.heartbeat:
             self.heartbeat.shutdown()

lib/galaxy/config.py

@@ -64,6 +64,11 @@ class Configuration( object ):
         self.bugs_email = kwargs.get( 'bugs_email', None )
         self.blog_url = kwargs.get( 'blog_url', None )
         self.screencasts_url = kwargs.get( 'screencasts_url', None )
+        # Configuration options for taking advantage of nginx features
+        self.nginx_x_accel_redirect_base = kwargs.get( 'nginx_x_accel_redirect_base', False )
+        self.nginx_upload_location = kwargs.get( 'nginx_upload_store', False )
+        if self.nginx_upload_location:
+            self.nginx_upload_location = os.path.abspath( self.nginx_upload_location )
         # Parse global_conf and save the parser
         global_conf = kwargs.get( 'global_conf', None )
         global_conf_parser = ConfigParser.ConfigParser()
@@ -78,6 +83,11 @@ class Configuration( object ):
         self.datatypes_config = kwargs.get( 'datatypes_config_file', 'datatypes_conf.xml' )
     def get( self, key, default ):
         return self.config_dict.get( key, default )
+    def get_bool( self, key, default ):
+        if key in self.config_dict:
+            return string_as_bool( key )
+        else:
+            return default
     def check( self ):
         # Check that required directories exist
         for path in self.root, self.file_path, self.tool_path, self.tool_data_path, self.template_path, self.job_working_directory:

lib/galaxy/jobs/__init__.py

@@ -17,6 +17,27 @@ log = logging.getLogger( __name__ )
 # States for running a job. These are NOT the same as data states
 JOB_WAIT, JOB_ERROR, JOB_INPUT_ERROR, JOB_INPUT_DELETED, JOB_OK, JOB_READY, JOB_DELETED = 'wait', 'error', 'input_error', 'input_deleted', 'ok', 'ready', 'deleted'
 
+class JobManager( object ):
+    """
+    Highest level interface to job management.
+    
+    TODO: Currently the app accesses "job_queue" and "job_stop_queue" directly.
+          This should be decoupled. 
+    """
+    def __init__( self, app ):
+        self.app = app
+        if self.app.config.get_bool( "enable_job_running", True ):
+            # The dispatcher launches the underlying job runners
+            self.dispatcher = DefaultJobDispatcher( app )
+            # Queues for starting and stopping jobs
+            self.job_queue = JobQueue( app, self.dispatcher )
+            self.job_stop_queue = JobStopQueue( app, self.dispatcher )
+        else:
+            self.job_queue = self.job_stop_queue = NoopQueue()
+    def shutdown( self ):
+        self.job_queue.shutdown()
+        self.job_stop_queue.shutdown()
+
 class Sleeper( object ):
     """
     Provides a 'sleep' method that sleeps for a number of seconds *unless*
@@ -594,50 +615,12 @@ class JobStopQueue( object ):
             pass  
 
         for job in jobs:
-            # jobs in a non queued/running/new state do not need to be stopped
-            if job.state not in [ model.Job.states.QUEUED, model.Job.states.RUNNING, model.Job.states.NEW ]:
-                return
-            # job has multiple datasets that aren't parent/child and not all of them are deleted.
-            if not self.check_if_output_datasets_deleted( job.id ):
-                return
-            self.mark_deleted( job.id )
             # job is in JobQueue or FooJobRunner, will be dequeued due to state change above
             if job.job_runner_name is None:
                 return
             # tell the dispatcher to stop the job
             self.dispatcher.stop( job )
 
-    def check_if_output_datasets_deleted( self, job_id ):
-        job = model.Job.get( job_id )
-        for dataset_assoc in job.output_datasets:
-            dataset = dataset_assoc.dataset
-            dataset.refresh()
-            #only the originator of the job can delete a dataset to cause
-            #cancellation of the job, no need to loop through history_associations
-            if not dataset.deleted:
-                return False
-        return True
-
-    def mark_deleted( self, job_id ):
-        job = model.Job.get( job_id )
-        job.refresh()
-        job.state = job.states.DELETED
-        job.info = "Job output deleted by user before job completed."
-        job.flush()
-        for dataset_assoc in job.output_datasets:
-            dataset = dataset_assoc.dataset
-            dataset.refresh()
-            dataset.deleted = True
-            dataset.state = dataset.states.DISCARDED
-            dataset.dataset.flush()
-            for dataset in dataset.dataset.history_associations:
-                #propagate info across shared datasets
-                dataset.deleted = True
-                dataset.blurb = 'deleted'
-                dataset.peek = 'Job deleted'
-                dataset.info = 'Job output deleted by user before job completed'
-                dataset.flush()
-
     def put( self, job ):
         self.queue.put( job )
 
@@ -652,3 +635,12 @@ class JobStopQueue( object ):
             self.queue.put( self.STOP_SIGNAL )
             self.sleeper.wake()
             log.info( "job stopper stopped" )
+
+class NoopQueue( object ):
+    """
+    Implements the JobQueue / JobStopQueue interface but does nothing
+    """
+    def put( self, *args ):
+        return
+    def shutdown( self ):
+        return

lib/galaxy/model/__init__.py

@@ -85,7 +85,35 @@ class Job( object ):
         tool = app.toolbox.tools_by_id[self.tool_id]
         param_dict = tool.params_from_strings( param_dict, app )
         return param_dict
-                
+    def check_if_output_datasets_deleted( self ):
+        """
+        Return true if all of the output datasets associated with this job are
+        in the deleted state
+        """
+        for dataset_assoc in self.output_datasets:
+            dataset = dataset_assoc.dataset
+            # only the originator of the job can delete a dataset to cause
+            # cancellation of the job, no need to loop through history_associations
+            if not dataset.deleted:
+                return False
+        return True
+    def mark_deleted( self ):
+        """
+        Mark this job as deleted, and mark any output datasets as discarded.
+        """
+        self.state = Job.states.DELETED
+        self.info = "Job output deleted by user before job completed."
+        for dataset_assoc in self.output_datasets:
+            dataset = dataset_assoc.dataset
+            dataset.deleted = True
+            dataset.state = dataset.states.DISCARDED
+            for dataset in dataset.dataset.history_associations:
+                # propagate info across shared datasets
+                dataset.deleted = True
+                dataset.blurb = 'deleted'
+                dataset.peek = 'Job deleted'
+                dataset.info = 'Job output deleted by user before job completed'
+
 class JobParameter( object ):
     def __init__( self, name, value ):
         self.name = name

lib/galaxy/tools/__init__.py

@@ -797,7 +797,7 @@ class Tool:
                 # Deal with the 'test' element and see if it's value changed
                 test_param_key = group_prefix + input.test_param.name
                 test_param_error = None
-                test_incoming = incoming.get( test_param_key, None )
+                test_incoming = get_incoming_value( incoming, test_param_key, None )
                 if test_param_key not in incoming \
                    and "__force_update__" + test_param_key not in incoming \
                    and update_only:
@@ -878,7 +878,7 @@ class Tool:
                         except:
                             pass
                     if not incoming_value_generated:
-                        incoming_value = incoming.get( key, None )
+                        incoming_value = get_incoming_value( incoming, key, None )
                     value, error = check_param( trans, input, incoming_value, context )
                     if input.dependent_params and state[ input.name ] != value:
                         # We need to keep track of changed dependency parametrs ( parameters
@@ -1362,3 +1362,12 @@ def json_fix( val ):
     else:
         return val
     
+def get_incoming_value( incoming, key, default ):
+    if "__" + key + "__is_composite" in incoming:
+        composite_keys = incoming["__" + key + "__keys"].split()
+        value = dict()
+        for composite_key in composite_keys:
+            value[composite_key] = incoming[key + "_" + composite_key]
+        return value
+    else:
+        return incoming.get( key, default )

lib/galaxy/tools/actions/upload.py

@@ -26,12 +26,22 @@ class UploadToolAction( object ):
         temp_name = ""
         data_list = []
 
-        if 'filename' in dir( data_file ):
+        if 'local_filename' in dir( data_file ):
+            # Use the existing file
             try:
                 file_name = data_file.filename
                 file_name = file_name.split( '\\' )[-1]
                 file_name = file_name.split( '/' )[-1]
-                data_list.append( self.add_file( trans, data_file.file, file_name, file_type, dbkey, space_to_tab=space_to_tab ) )
+                data_list.append( self.add_file( trans, data_file.local_filename, file_name, file_type, dbkey, space_to_tab=space_to_tab ) )
+            except Exception, e:
+                return self.upload_empty( trans, "Error:", str( e ) )
+        elif 'filename' in dir( data_file ):
+            try:
+                file_name = data_file.filename
+                file_name = file_name.split( '\\' )[-1]
+                file_name = file_name.split( '/' )[-1]
+                temp_name = sniff.stream_to_file( data_file.file )
+                data_list.append( self.add_file( trans, temp_name, file_name, file_type, dbkey, space_to_tab=space_to_tab ) )
             except Exception, e:
                 return self.upload_empty( trans, "Error:", str( e ) )
         if url_paste not in [ None, "" ]:
@@ -41,7 +51,8 @@ class UploadToolAction( object ):
                     line = line.rstrip( '\r\n' )
                     if line:
                         try:
-                            data_list.append( self.add_file( trans, urllib.urlopen( line ), line, file_type, dbkey, info="uploaded url", space_to_tab=space_to_tab ) )
+                            temp_name = sniff.stream_to_file( urllib.urlopen( line ) )
+                            data_list.append( self.add_file( trans, temp_name, line, file_type, dbkey, info="uploaded url", space_to_tab=space_to_tab ) )
                         except Exception, e:
                             return self.upload_empty( trans, "Error:", str( e ) )
             else:
@@ -53,7 +64,8 @@ class UploadToolAction( object ):
                         break
                 if is_valid:
                     try:
-                        data_list.append( self.add_file( trans, StringIO.StringIO( url_paste ), 'Pasted Entry', file_type, dbkey, info="pasted entry", space_to_tab=space_to_tab ) )
+                        temp_name = sniff.stream_to_file( StringIO.StringIO( url_paste ) )
+                        data_list.append( self.add_file( trans, temp_name, 'Pasted Entry', file_type, dbkey, info="pasted entry", space_to_tab=space_to_tab ) )
                     except Exception, e:
                         return self.upload_empty( trans, "Error:", str( e ) )
                 else:
@@ -77,9 +89,8 @@ class UploadToolAction( object ):
         trans.app.model.flush()
         return dict( output=data )
 
-    def add_file( self, trans, file_obj, file_name, file_type, dbkey, info=None, space_to_tab=False ):
+    def add_file( self, trans, temp_name, file_name, file_type, dbkey, info=None, space_to_tab=False ):
         data_type = None
-        temp_name = sniff.stream_to_file( file_obj )
         
         # See if we have an empty file
         if not os.path.getsize( temp_name ) > 0:

lib/galaxy/tools/parameters/__init__.py

@@ -16,9 +16,7 @@ def check_param( trans, param, incoming_value, param_values ):
     value = incoming_value
     error = None
     try:
-        if param.name == 'file_data':
-            pass
-        elif value is not None or isinstance(param, DataToolParameter):
+        if value is not None or isinstance(param, DataToolParameter):
             # Convert value from HTML representation
             value = param.from_html( value, trans, param_values )
             # Allow the value to be converted if neccesary

lib/galaxy/tools/parameters/basic.py

@@ -2,12 +2,13 @@
 Basic tool parameters.
 """
 
-import logging, string, sys, os
+import logging, string, sys, os, os.path
 
 from elementtree.ElementTree import XML, Element
 
 from galaxy import config, datatypes, util
 from galaxy.web import form_builder
+from galaxy.util.bunch import Bunch
 
 import validation, dynamic_options
 
@@ -294,6 +295,23 @@ class FileToolParameter( ToolParameter ):
         self.name = elem.get( 'name' )
     def get_html_field( self, trans=None, value=None, other_values={}  ):
         return form_builder.FileField( self.name )
+    def from_html( self, value, trans=None, other_values={} ):
+        # Middleware or proxies may encode files in special ways (TODO: this
+        # should be pluggable)
+        if type( value ) == dict:
+            upload_location = self.tool.app.config.nginx_upload_location
+            assert upload_location, \
+                "Request appears to have been processed by nginx_upload_module \
+                but Galaxy is not configured to recgonize it"
+            # Check that the file is in the right location
+            local_filename = os.path.abspath( value['path'] )
+            assert local_filename.startswith( upload_location ), \
+                "Filename provided by nginx is not in correct directory"
+            value = Bunch(
+                filename = value["name"],
+                local_filename = local_filename
+            )
+        return value
     def get_required_enctype( self ):
         """
         File upload elements require the multipart/form-data encoding

lib/galaxy/web/controllers/root.py

@@ -266,6 +266,28 @@ class RootController( BaseController ):
         return trans.fill_template( "/dataset/edit_attributes.mako", data=data,
                                     datatypes=ldatatypes, err=None )
 
+    def __delete_dataset( self, trans, id ):
+        data = self.app.model.HistoryDatasetAssociation.get( id )
+        if data:
+            # Walk up parent datasets to find the containing history
+            topmost_parent = data
+            while topmost_parent.parent:
+                topmost_parent = topmost_parent.parent
+            assert topmost_parent in trans.history.datasets, "Data does not belong to current history"
+            # Mark deleted and cleanup
+            data.mark_deleted()
+            data.clear_associated_files()
+            trans.log_event( "Dataset id %s marked as deleted" % str(id) )
+            if data.parent_id is None and len( data.creating_job_associations ) > 0:
+                # Mark associated job for deletion
+                job = data.creating_job_associations[0].job
+                if job.state not in [ model.Job.states.QUEUED, model.Job.states.RUNNING, model.Job.states.NEW ]:
+                    return
+                # Are *all* of the job's other output datasets deleted?
+                if job.check_if_output_datasets_deleted():
+                    job.mark_deleted()                
+            self.app.model.flush()
+
     @web.expose
     def delete( self, trans, id = None, show_deleted_on_refresh = False, **kwd):
         if id:
@@ -276,26 +298,10 @@ class RootController( BaseController ):
             history = trans.get_history()
             for id in dataset_ids:
                 try:
-                    int( id )
+                    id = int( id )
                 except:
                     continue
-                data = self.app.model.HistoryDatasetAssociation.get( id )
-                if data:
-                    # Walk up parent datasets to find the containing history
-                    topmost_parent = data
-                    while topmost_parent.parent:
-                        topmost_parent = topmost_parent.parent
-                    assert topmost_parent in history.datasets, "Data does not belong to current history"
-                    # Mark deleted and cleanup
-                    data.mark_deleted()
-                    data.clear_associated_files()
-                    self.app.model.flush()
-                    trans.log_event( "Dataset id %s marked as deleted" % str(id) )
-                    if data.parent_id is None:
-                        try:
-                            self.app.job_stop_queue.put( data.creating_job_associations[0].job )
-                        except IndexError:
-                            pass    # upload tool will cause this since it doesn't have a job
+                self.__delete_dataset( trans, id )
         return self.history( trans, show_deleted = show_deleted_on_refresh )
         
     @web.expose
@@ -305,24 +311,7 @@ class RootController( BaseController ):
                 int( id )
             except:
                 return "Dataset id '%s' is invalid" %str( id )
-            history = trans.get_history()
-            data = self.app.model.HistoryDatasetAssociation.get( id )
-            if data:
-                # Walk up parent datasets to find the containing history
-                topmost_parent = data
-                while topmost_parent.parent:
-                    topmost_parent = topmost_parent.parent
-                assert topmost_parent in history.datasets, "Data does not belong to current history"
-                # Mark deleted and cleanup
-                data.mark_deleted()
-                data.clear_associated_files()
-                self.app.model.flush()
-                trans.log_event( "Dataset id %s marked as deleted async" % str(id) )
-                if data.parent_id is None:
-                    try:
-                        self.app.job_stop_queue.put( data.creating_job_associations[0].job )
-                    except IndexError:
-                        pass    # upload tool will cause this since it doesn't have a job
+            self.__delete_dataset( trans, id )
         return "OK"
 
     ## ---- History management -----------------------------------------------

lib/galaxy/web/framework/base.py

@@ -5,6 +5,7 @@ A simple WSGI application/framework.
 import socket
 import types
 import logging
+import os.path
 import sys
 
 from Cookie import SimpleCookie
@@ -132,16 +133,16 @@ class WebApplication( object ):
         if callable( body ):
             # Assume the callable is another WSGI application to run
             return body( environ, start_response )
+        elif isinstance( body, types.FileType ):
+            # Stream the file back to the browser
+            return send_file( start_response, trans, body )
         else:
             start_response( trans.response.wsgi_status(), 
                             trans.response.wsgi_headeritems() )
             return self.make_body_iterable( trans, body )
         
     def make_body_iterable( self, trans, body ):
-        if isinstance( body, types.FileType ):
-            # Stream the file back to the browser
-            return iterate_file( body )
-        elif isinstance( body, ( types.GeneratorType, list, tuple ) ):
+        if isinstance( body, ( types.GeneratorType, list, tuple ) ):
             # Recursively stream the iterable
             return flatten( body )
         elif isinstance( body, basestring ):
@@ -302,6 +303,20 @@ class Response( object ):
 
 CHUNK_SIZE = 2**16
 
+def send_file( start_response, trans, body ):
+    # If configured use X-Accel-Redirect header for nginx
+    base = trans.app.config.nginx_x_accel_redirect_base
+    if base:
+        trans.response.headers['X-Accel-Redirect'] = \
+            base + os.path.abspath( body.name )
+        body = [ "" ]
+    # Fall back on sending the file in chunks
+    else:
+        body = iterate_file( body )
+    start_response( trans.response.wsgi_status(), 
+                    trans.response.wsgi_headeritems() )
+    return body
+
 def iterate_file( file ):
     """
     Progressively return chunks from `file`.

universe_wsgi.ini.sample

@@ -34,7 +34,7 @@ job_queue_cleanup_interval = 30
 # Database connection
 database_file = database/universe.sqlite
 # You may use a SQLAlchemy connection string to specify an external database instead
-## database_connection = postgres:///galaxy_test
+## database_connection = postgres:///galaxy
 ## database_engine_option_echo = true
 ## database_engine_option_echo_pool = true
 ## database_engine_option_pool_size = 10
@@ -89,12 +89,15 @@ mailing_join_addr = galaxy-user-join@bx.psu.edu
 # Write thread status periodically to 'heartbeat.log' (careful, uses disk space rapidly!)
 ## use_heartbeat = True
 
+# Enable the memory debugging interface (careful, negatively impacts server performance)
+## use_memdump = True
+
 # Profiling middleware (cProfile based)
 ## use_profile = True
 
-# Mail
-smtp_server = coltrane.bx.psu.edu
-error_email_to = galaxy-bugs@bx.psu.edu
+# For use by 'report this error' link on error-state datasets
+#smtp_server = smtp.example.org
+#error_email_to = galaxy-bugs@example.org
 
 # Use the new iframe / javascript based layout
 use_new_layout = true
@@ -120,29 +123,34 @@ static_style_dir = %(here)s/static/june_2007_style/blue
 ## wiki_url: replaces the default galaxy main wiki
 ## bugs_email: replaces the default galaxy bugs email list
 #brand = Private local mirror
-#wiki_url=/path/to/my/local/wiki
-#bugs_email=mailto:bugmaster@this.site.com
+#wiki_url = /path/to/my/local/wiki
+#bugs_email = mailto:galaxy-bugs@example.org
 
 # ---- Job Runners ----------------------------------------------------------
 
 # Clustering Galaxy is not a straightforward process and requires a lot of
-# pre-configuration.  See the ClusteringGalaxy Wiki before attempting to set any
-# of these options.  If running normally (without a cluster), do not change
-# anything in this section.
+# pre-configuration.  See the ClusteringGalaxy Wiki before attempting to set
+# any of these options:
+#
+# http://g2.trac.bx.psu.edu/wiki/ClusteringGalaxy
+#
+# If running normally (without a cluster), do not change anything in this
+# section.
 
 # 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'.
+# below.  The runners currently available are 'pbs' and 'sge'.
 #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 ClusteringGalaxy Wiki.  Leave commented if not using a cluster job runner.
+# see the ClusteringGalaxy Wiki.  Leave commented if not using a cluster job
+# runner.
 #default_cluster_job_runner = pbs:///
 
 # The PBS options are described in detail in the Galaxy Configuration section of
-# the ClusteringGalaxy Wiki
+# the ClusteringGalaxy Wiki, and are only necessary when using file staging.
 #pbs_application_server = 
 #pbs_stage_path = 
 #pbs_dataset_server = 
@@ -152,8 +160,6 @@ static_style_dir = %(here)s/static/june_2007_style/blue
 [galaxy:tool_runners]
 
 biomart = local:///
-blat2wig = pbs:///blast
-blat_wrapper = pbs:///blast
 encode_db1 = local:///
 encode_import_all_latest_datasets1 = local:///
 encode_import_chromatin_and_chromosomes1 = local:///
@@ -161,13 +167,8 @@ encode_import_gencode1 = local:///
 encode_import_genes_and_transcripts1 = local:///
 encode_import_multi-species_sequence_analysis1 = local:///
 encode_import_transcription_regulation1 = local:///
-generate_coverage_report = pbs:///blast
 hbvar = local:///
-hist_high_quality_score = pbs:///blast
-megablast_wrapper = pbs:///blast
-megablast_xml_parser = pbs:///blast
 microbial_import1 = local:///
-quality_score_distribution = pbs:///blast
 ucsc_table_direct1 = local:///
 ucsc_table_direct_archaea1 = local:///
 ucsc_table_direct_test1 = local:///