Do away with the daemonized transfer manager and have the deferred job runner start off individual daemonized transfer workers. The workers update their own major states and can be queries for in-transfer progress via a socket, and an interface to that socket in app.transfer_manager. All socket communication is now JSON-RPC.

lib/galaxy/app.py

@@ -76,6 +76,10 @@ class UniverseApplication( object ):
             from galaxy.util import memdump
             if memdump.Memdump:
                 self.memdump = memdump.Memdump()
+        # Transfer manager client
+        if self.config.get_bool( 'enable_beta_job_managers', False ):
+            from jobs import transfer_manager
+            self.transfer_manager = transfer_manager.TransferManager( self )
         # Start the job queue
         self.job_manager = jobs.JobManager( self )
         # FIXME: These are exposed directly for backward compatibility
@@ -83,10 +87,6 @@ class UniverseApplication( object ):
         self.job_stop_queue = self.job_manager.job_stop_queue
         # Initialize the external service types
         self.external_service_types = external_service_types.ExternalServiceTypesCollection( self.config.external_service_type_config_file, self.config.external_service_type_path, self )
-        # Transfer manager client
-        if self.config.get_bool( 'enable_deferred_job_queue', False ):
-            from jobs import transfer_manager
-            self.transfer_manager = transfer_manager.TransferManager( self )
         
     def shutdown( self ):
         self.job_manager.shutdown()

lib/galaxy/config.py

@@ -80,6 +80,8 @@ class Configuration( object ):
         self.external_service_type_path = resolve_path( kwargs.get( 'external_service_type_path', 'external_service_types' ), self.root )
         # Tasked job runner.
         self.use_tasked_jobs = string_as_bool( kwargs.get( 'use_tasked_jobs', False ) )
+        # The transfer manager and deferred job queue
+        self.enable_beta_job_managers = string_as_bool( kwargs.get( 'enable_beta_job_managers', 'False' ) )
         self.local_task_queue_workers = int(kwargs.get("local_task_queue_workers", 2))
         self.default_cluster_job_runner = kwargs.get( 'default_cluster_job_runner', 'local:///' )
         self.pbs_application_server = kwargs.get('pbs_application_server', "" )
@@ -104,7 +106,6 @@ class Configuration( object ):
         self.ftp_upload_site = kwargs.get( 'ftp_upload_site', None )
         self.allow_library_path_paste = kwargs.get( 'allow_library_path_paste', False )
         self.disable_library_comptypes = kwargs.get( 'disable_library_comptypes', '' ).lower().split( ',' )
-        self.transfer_manager_port = int( kwargs.get( 'transfer_manager_port', 8163 ) )
         # Location for dependencies
         if 'tool_dependency_dir' in kwargs:
             self.tool_dependency_dir = resolve_path( kwargs.get( "tool_dependency_dir" ), self.root )

lib/galaxy/jobs/__init__.py

@@ -42,7 +42,7 @@ class JobManager( object ):
             # Queues for starting and stopping jobs
             self.job_queue = JobQueue( app, self.dispatcher )
             self.job_stop_queue = JobStopQueue( app, self.dispatcher )
-            if self.app.config.get_bool( 'enable_deferred_job_queue', False ):
+            if self.app.config.enable_beta_job_managers:
                 from galaxy.jobs.deferred import DeferredJobQueue
                 self.deferred_job_queue = DeferredJobQueue( app )
         else:

lib/galaxy/jobs/deferred/__init__.py

@@ -15,7 +15,7 @@ class DeferredJobQueue( object ):
                         INVALID = 'invalid' )
     def __init__( self, app ):
         self.app = app
-        self.sa_session = app.model.context
+        self.sa_session = app.model.context.current
         self.queue = Queue()
         self.plugins = {}
         self._load_plugins()
@@ -63,7 +63,9 @@ class DeferredJobQueue( object ):
             if 'check_interval' in dir( self.plugins[job.plugin] ):
                 job.check_interval = self.plugins[job.plugin].check_interval
             log.info( 'Recovered deferred job (id: %s) at startup' % job.id )
-            self.waiting_jobs.append( job )
+            # 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:
@@ -89,19 +91,30 @@ class DeferredJobQueue( object ):
             self.waiting_jobs.append( job )
         new_waiting = []
         for job in self.waiting_jobs:
+            try:
+                # Recovered jobs are passed in by ID
+                assert type( job ) is int
+                job = self.sa_session.query( model.DeferredJob ).get( job )
+            except:
+                pass
             if job.is_check_time:
                 try:
                     job_state = self.plugins[job.plugin].check_job( job )
-                except:
-                    raise # TODO: fail
+                except Exception, e:
+                    self.__fail_job( job )
+                    log.error( 'Set deferred job %s to error because of an exception in check_job(): %s' % ( job.id, str( e ) ) )
+                    continue
                 if job_state == self.job_states.READY:
                     try:
                         self.plugins[job.plugin].run_job( job )
-                    except:
-                        raise # TODO: fail
+                    except Exception, e:
+                        self.__fail_job( job )
+                        log.error( 'Set deferred job %s to error because of an exception in run_job(): %s' % ( job.id, str( e ) ) )
+                        continue
                 elif job_state == self.job_states.INVALID:
-                    # TODO: fail
+                    self.__fail_job( job )
                     log.error( 'Unable to run deferred job (id: %s): Plugin "%s" marked it as invalid' % ( job.id, job.plugin ) )
+                    continue
                 else:
                     new_waiting.append( job )
                 job.last_check = 'now'
@@ -113,7 +126,6 @@ class DeferredJobQueue( object ):
         if job.plugin not in self.plugins:
             log.error( 'Invalid deferred job plugin: %s' ) % job.plugin
             job.state = model.DeferredJob.states.ERROR
-            job.info = 'Invalid deferred job plugin: %s' % job.plugin
             self.sa_session.add( job )
             self.sa_session.flush()
             return False
@@ -122,6 +134,11 @@ class DeferredJobQueue( object ):
     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()

lib/galaxy/jobs/transfer_manager.py

@@ -1,8 +1,8 @@
 """
-Client interface to the Galaxy Transfer Manager, which is a standalone,
-lightweight on-demand daemon.
+Manage transfers from arbitrary URLs to temporary files.  Socket interface for
+IPC with multiple process configurations.
 """
-import subprocess, socket, logging
+import os, subprocess, socket, logging, threading
 
 from galaxy import eggs
 from galaxy.util import listify, json
@@ -10,12 +10,19 @@ from galaxy.util import listify, json
 log = logging.getLogger( __name__ )
 
 class TransferManager( object ):
+    """
+    Manage simple data transfers from URLs to temporary locations.
+    """
     def __init__( self, app ):
         self.app = app
         self.sa_session = app.model.context.current
-        self.tm_port = app.config.transfer_manager_port
-        self.tm_command = 'python transfer_manager.py'
-        self.tm_transfer_job_id_flag = '-i'
+        self.command = 'python %s' % os.path.abspath( os.path.join( os.getcwd(), 'scripts', 'transfer.py' ) )
+        if app.config.get_bool( 'enable_job_recovery', True ):
+            # Only one Galaxy server process should be able to recover jobs! (otherwise you'll have nasty race conditions)
+            self.running = True
+            self.sleeper = Sleeper()
+            self.restarter = threading.Thread( target=self.__restarter )
+            self.restarter.start()
     def new( self, path=None, **kwd ):
         if 'url' not in kwd:
             raise Exception( 'Missing required parameter "url".' )
@@ -32,48 +39,117 @@ class TransferManager( object ):
         transfer_manager will either daemonize or return after submitting to a
         running daemon, so it should be fairly quick to return.
         """
-        spaced_flag = ' %s ' % self.tm_transfer_job_id_flag
-        cmd = '%s %s %s' % ( self.tm_command, self.tm_transfer_job_id_flag, spaced_flag.join( [ str( tj.id ) for tj in transfer_jobs ] ) )
-        log.debug( 'Initiating Transfer Job(s): %s' % ', '.join( [ str( tj.id ) for tj in transfer_jobs ] ) )
-        p = subprocess.Popen( cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT )
-        p.wait()
-        return p.stdout.read()
-    def status( self, transfer_jobs ):
+        transfer_jobs = listify( transfer_jobs )
+        printable_tj_ids = ', '.join( [ str( tj.id ) for tj in transfer_jobs ] )
+        log.debug( 'Initiating transfer job(s): %s' % printable_tj_ids )
+        # Set all jobs running before spawning, or else updating the state may
+        # clobber a state change performed by the worker.
+        [ tj.__setattr__( 'state', tj.states.RUNNING ) for tj in transfer_jobs ]
+        self.sa_session.add_all( transfer_jobs )
+        self.sa_session.flush()
+        for tj in transfer_jobs:
+            # 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.
+            cmd = '%s %s' % ( self.command, tj.id )
+            log.debug( 'Transfer command is: %s' % cmd )
+            p = subprocess.Popen( cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT )
+            p.wait()
+            output = p.stdout.read( 32768 )
+            if p.returncode != 0:
+                log.error( 'Spawning transfer job failed: %s: %s' % ( tj.id, output ) )
+                tj.state = tj.states.ERROR
+                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 = []
-        sock = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
-        sock.settimeout( 10 )
-        try:
-            sock.connect( ( 'localhost', self.tm_port ) )
-        except Exception, e:
-            log.warning( 'sock.connect for status update of Transfer Jobs %s failed (this is okay if all jobs have finished): %s' % ( ', '.join( [ str( tj.id ) for tj in transfer_jobs ] ), str( e ) ) )
-            [ self.sa_session.refresh( tj ) for tj in transfer_jobs ]
-            new_jobs = filter( lambda x: x.state == self.app.model.TransferJob.states.NEW, transfer_jobs )
-            #terminal_jobs = filter( lambda x: x.state in [ self.app.model.TransferJob.states.DONE, \
-            #                                               self.app.model.TransferJob.states.ERROR ], transfer_jobs )
-            if new_jobs:
-                # This could be a bad idea if the transfer manager daemon is misbehaving.
-                output = self.run( new_jobs )
-            for tj in transfer_jobs:
-                if tj.state == tj.states.DONE:
-                    log.debug( 'Transfer Job %s is complete' % tj.id )
-                rval.append( dict( transfer_job_id=tj.id, state=tj.state ) )
-            if len( rval ) == 1:
-                return rval[0]
-            return rval
-        sock.send( json.to_json_string( dict( state_transfer_job_ids=[ t.id for t in transfer_jobs ] ) ) + '\n' )
-        resp = sock.recv( 8192 )
-        for line in resp.splitlines():
-            status = json.from_json_string( line )
-            # TODO: need a bunch for this
-            if status['state'] == 'unknown':
-                transfer_job = [ tj for tj in transfer_jobs if tj.id == int( status['transfer_job_id'] ) ][0]
-                self.sa_session.refresh( tj )
-                rval.append( dict( transfer_job_id=tj.id, state=tj.state ) )
+        for tj in transfer_jobs:
+            if via_socket and tj.state not in tj.terminal_states and tj.socket:
+                try:
+                    request = json.jsonrpc_request( method='get_state', id=True )
+                    sock = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
+                    sock.settimeout( 5 )
+                    sock.connect( ( 'localhost', tj.socket ) )
+                    sock.send( json.to_json_string( request ) )
+                    response = sock.recv( 8192 )
+                    valid, response = json.validate_jsonrpc_response( response, id=request['id'] )
+                    if not valid:
+                        # No valid response received, make some pseudo-json-rpc
+                        raise Exception( dict( code=128, message='Did not receive valid response from transfer daemon for state' ) )
+                    if 'error' in response:
+                        # Response was valid but Request resulted in an error
+                        raise Exception( error )
+                    else:
+                        # Request was valid
+                        response['result']['transfer_job_id'] = tj.id
+                        rval.append( response['result'] )
+                except Exception, e:
+                    # State checking via the transfer daemon failed, just
+                    # return the state from the database instead.  Callers can
+                    # look for the 'error' member of the response to see why
+                    # the check failed.
+                    self.sa_session.refresh( tj )
+                    error = e.args
+                    if type( error ) != dict:
+                        error = dict( code=256, message='Error connecting to transfer daemon', data=str( e ) )
+                    rval.append( dict( transfer_job_id=tj.id, state=tj.state, error=error ) )
             else:
-                if status['state'] == 'progress' and 'percent' in status:
-                    log.debug( 'Transfer Job %s is %s complete' % ( status['transfer_job_id'], status['percent'] ) )
-                rval.append( status )
+                rval.append( dict( transfer_job_id=tj.id, state=tj.state ) )
+        for tj_state in rval:
+            if tj_state['state'] in self.app.model.TransferJob.terminal_states:
+                log.debug( 'Transfer job %s is in terminal state: %s' % ( tj_state['transfer_job_id'], tj_state['state'] ) )
+            elif tj_state['state'] == self.app.model.TransferJob.states.PROGRESS and 'percent' in tj_state:
+                log.debug( 'Transfer job %s is %s%% complete' % ( tj_state['transfer_job_id'], tj_state['percent'] ) )
         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.
+            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:
+                    continue
+                # This will only succeed if the process exists and is owned by the
+                # user running Galaxy (unless that user is root, in which case it
+                # can be owned by anyone - but you're not running Galaxy as root,
+                # right?).  This is not guaranteed proof that the transfer is alive
+                # since another process may have assumed the original process' PID.
+                # But that will only cause the transfer to not restart until that
+                # process dies, which hopefully won't be too long from now...  If
+                # it becomes a problem, try to talk to the socket a few times and
+                # restart the transfer if socket communication fails repeatedly.
+                try:
+                    os.kill( tj.pid, 0 )
+                except:
+                    self.sa_session.refresh( tj )
+                    if tj.state == tj.states.RUNNING:
+                        log.error( 'Transfer job %s is marked as running but pid %s appears to be dead.' % ( tj.id, tj.pid ) )
+                        dead.append( tj )
+            if dead:
+                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*
+    the notify method is called (from a different thread).
+    """
+    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()

lib/galaxy/model/__init__.py

@@ -2173,14 +2173,22 @@ class VisualizationUserShareAssociation( object ):
         self.user = None
         
 class TransferJob( object ):
+    # These states are used both by the transfer manager's IPC and the object
+    # state in the database.  Not all states are used by both.
     states = Bunch( NEW = 'new',
+                    UNKNOWN = 'unknown',
+                    PROGRESS = 'progress',
                     RUNNING = 'running',
                     ERROR = 'error',
                     DONE = 'done' )
-    def __init__( self, state=None, path=None, info=None, params=None ):
+    terminal_states = [ states.ERROR,
+                        states.DONE ]
+    def __init__( self, state=None, path=None, info=None, pid=None, socket=None, params=None ):
         self.state = state
         self.path = path
         self.info = info
+        self.pid = pid
+        self.socket = socket
         self.params = params
 
 class Tag ( object ):

lib/galaxy/model/mapping.py

@@ -445,6 +445,8 @@ TransferJob.table = Table( "transfer_job", metadata,
     Column( "state", String( 64 ), index=True ),
     Column( "path", String( 1024 ) ),
     Column( "info", TEXT ),
+    Column( "pid", Integer ),
+    Column( "socket", Integer ),
     Column( "params", JSONType ) )
 
 Event.table = Table( "event", metadata, 

lib/galaxy/model/migrate/versions/0072_add_pid_and_socket_columns_to_transfer_job_table.py

@@ -0,0 +1,39 @@
+"""
+Migration script to add 'pid' and 'socket' columns to the transfer_job table.
+"""
+
+from sqlalchemy import *
+from sqlalchemy.orm import *
+from migrate import *
+from migrate.changeset import *
+
+import logging
+log = logging.getLogger( __name__ )
+
+metadata = MetaData( migrate_engine )
+db_session = scoped_session( sessionmaker( bind=migrate_engine, autoflush=False, autocommit=True ) )
+
+def upgrade():
+    print __doc__
+    metadata.reflect()
+    try:
+        TransferJob_table = Table( "transfer_job", metadata, autoload=True )
+        c = Column( "pid", Integer )
+        c.create( TransferJob_table )
+        assert c is TransferJob_table.c.pid
+        c = Column( "socket", Integer )
+        c.create( TransferJob_table )
+        assert c is TransferJob_table.c.socket
+    except Exception, e:
+        print "Adding columns to transfer_job table failed: %s" % str( e )
+        log.debug( "Adding columns to transfer_job table failed: %s" % str( e ) )
+
+def downgrade():
+    metadata.reflect()
+    try:
+        TransferJob_table = Table( "transfer_job", metadata, autoload=True )
+        TransferJob_table.c.pid.drop()
+        TransferJob_table.c.socket.drop()
+    except Exception, e:
+        print "Dropping columns from transfer_job table failed: %s" % str( e )
+        log.debug( "Dropping columns from transfer_job table failed: %s" % str( e ) )

lib/galaxy/util/json.py

@@ -1,5 +1,8 @@
 
-__all__ = [ "to_json_string", "from_json_string", "json_fix"]
+__all__ = [ "to_json_string", "from_json_string", "json_fix", "validate_jsonrpc_request", "validate_jsonrpc_response", "jsonrpc_request", "jsonrpc_response" ]
+
+import random, string, logging
+import socket
 
 import pkg_resources
 pkg_resources.require( "simplejson" )
@@ -9,6 +12,8 @@ import simplejson
 to_json_string = simplejson.dumps
 from_json_string = simplejson.loads
 
+log = logging.getLogger( __name__ )
+
 def json_fix( val ):
     if isinstance( val, list ):
         return [ json_fix( v ) for v in val ]
@@ -17,4 +22,90 @@ def json_fix( val ):
     elif isinstance( val, unicode ):
         return val.encode( "utf8" )
     else:
-        return val
+        return val
+
+# Methods for handling JSON-RPC
+
+def validate_jsonrpc_request( request, regular_methods, notification_methods ):
+    try:
+        request = from_json_string( request )
+    except Exception, e:
+        return False, request, jsonrpc_response( id = None, error = dict( code = -32700, message = 'Parse error', data = str( e ) ) )
+    try:
+        assert 'jsonrpc' in request, \
+            'This server requires JSON-RPC 2.0 and no "jsonrpc" member was sent with the Request object as per the JSON-RPC 2.0 Specification.'
+        assert request['jsonrpc'] == '2.0', \
+                'Requested JSON-RPC version "%s" != required version "2.0".' % request['jsonrpc']
+        assert 'method' in request, 'No "method" member was sent with the Request object'
+    except AssertionError, e:
+        return False, request, jsonrpc_response( request = request, error = dict( code = -32600, message = 'Invalid Request', data = str( e ) ) )
+    try:
+        assert request['method'] in ( regular_methods + notification_methods )
+    except AssertionError, e:
+        return False, request, jsonrpc_response( request = request,
+                                        error = dict( code = -32601,
+                                                      message = 'Method not found',
+                                                      data = 'Valid methods are: %s' % ', '.join( regular_methods + notification_methods ) ) )
+    try:
+        if request['method'] in regular_methods:
+            assert 'id' in request, 'No "id" member was sent with the Request object and the requested method "%s" is not a notification method' % request['method']
+    except AssertionError, e:
+        return False, request, jsonrpc_response( request = request, error = dict( code = -32600, message = 'Invalid Request', data = str( e ) ) )
+    return True, request, None
+
+def validate_jsonrpc_response( response, id=None ):
+    try:
+        response = from_json_string( response )
+    except Exception, e:
+        log.error( 'Response was not valid JSON: %s' % str( e ) )
+        log.debug( 'Response was: %s' % response )
+        return False, response
+    try:
+        assert 'jsonrpc' in response, \
+            'This server requires JSON-RPC 2.0 and no "jsonrpc" member was sent with the Response object as per the JSON-RPC 2.0 Specification.'
+        assert ( 'result' in response or 'error' in response ), \
+            'Neither of "result" or "error" members were sent with the Response object.'
+        if 'error' in response:
+            assert int( response['error']['code'] ), \
+                'The "code" member of the "error" object in the Response is missing or not an integer.'
+            assert 'message' in response, \
+                'The "message" member of the "error" object in the Response is missing.'
+    except Exception, e:
+        log.error( 'Response was not valid JSON-RPC: %s' % str( e ) )
+        log.debug( 'Response was: %s' % response )
+        return False, response
+    if id is not None:
+        try:
+            assert 'id' in response and response['id'] == id
+        except Exception, e:
+            log.error( 'The response id "%s" does not match the request id "%s"' % ( response['id'], id ) )
+            return False, response
+    return True, response
+
+def jsonrpc_request( method, params=None, id=None, jsonrpc='2.0' ):
+    if method is None:
+        log.error( 'jsonrpc_request(): "method" parameter cannot be None' )
+        return None
+    request = dict( jsonrpc = jsonrpc, method = method )
+    if params:
+        request['params'] = params
+    if id is not None and id is True:
+        request['id'] = ''.join( [ random.choice( string.hexdigits ) for i in range( 16 ) ] )
+    elif id is not None:
+        request['id'] = id
+    return request
+
+def jsonrpc_response( request=None, id=None, result=None, error=None, jsonrpc='2.0' ):
+    if result:
+        rval = dict( jsonrpc = jsonrpc, result = result )
+    elif error:
+        rval = dict( jsonrpc = jsonrpc, error = error )
+    else:
+        msg = 'jsonrpc_response() called with out a "result" or "error" parameter'
+        log.error( msg )
+        rval = dict( jsonrpc = jsonrpc, error = msg )
+    if id is not None:
+        rval['id'] = id
+    elif request is not None and 'id' in request:
+        rval['id'] = request['id']
+    return rval

scripts/transfer.py

@@ -0,0 +1,258 @@
+#!/usr/bin/env python
+"""
+Downloads files to temp locations.  This script is invoked by the Transfer
+Manager (galaxy.jobs.transfer_manager) and should not normally be invoked by
+hand.
+"""
+
+import os, sys, optparse, ConfigParser, socket, SocketServer, threading, logging, random
+
+import urllib2, tempfile
+
+import time
+
+log = logging.getLogger( __name__ )
+log.setLevel( logging.INFO )
+handler = logging.StreamHandler( sys.stdout )
+log.addHandler( handler )
+
+galaxy_root = os.path.abspath( os.path.join( os.path.dirname( __file__ ), '..' ) )
+sys.path.insert( 0, os.path.abspath( os.path.join( galaxy_root, 'lib' ) ) )
+
+from galaxy import eggs
+eggs.require( "SQLAlchemy >= 0.4" )
+
+from sqlalchemy import *
+from sqlalchemy.orm import *
+from galaxy.model.mapping import load_egg_for_url
+
+import galaxy.model
+from galaxy.util import json, bunch
+
+eggs.require( 'python_daemon' )
+from daemon import DaemonContext
+
+debug = False
+slow = False
+
+class ArgHandler( object ):
+    """
+    Collect command line flags.
+    """
+    def __init__( self ):
+        self.parser = optparse.OptionParser()
+        self.parser.add_option( '-c', '--config', dest='config', help='Path to Galaxy config file (universe_wsgi.ini)',
+                                                  default=os.path.abspath( os.path.join( galaxy_root, 'universe_wsgi.ini' ) ) )
+        self.parser.add_option( '-d', '--debug', action='store_true', dest='debug', help="Debug (don't detach)" )
+        self.parser.add_option( '-s', '--slow', action='store_true', dest='slow', help="Transfer slowly (for debugging)" )
+        self.opts = None
+    def parse( self ):
+        self.opts, args = self.parser.parse_args()
+        if len( args ) != 1:
+            log.error( 'usage: transfer.py [options] <transfer job id>' )
+            sys.exit( 1 )
+        try:
+            self.transfer_job_id = int( args[0] )
+        except TypeError:
+            log.error( 'The provided transfer job ID is not an integer: %s' % args[0] )
+            sys.exit( 1 )
+        if self.opts.debug:
+            global debug
+            debug = True
+            log.setLevel( logging.DEBUG )
+        if self.opts.slow:
+            global slow
+            slow = True
+
+class GalaxyApp( object ):
+    """
+    A shell Galaxy App to provide access to the Galaxy configuration and
+    model/database.
+    """
+    def __init__( self, config_file ):
+        self.config = ConfigParser.ConfigParser( dict( database_file = 'database/universe.sqlite',
+                                                       file_path = 'database/files',
+                                                       transfer_worker_port_range = '12275-12675',
+                                                       transfer_worker_log = None ) )
+        self.config.read( config_file )
+        self.model = bunch.Bunch()
+        self.connect_database()
+    def connect_database( self ):
+        # Avoid loading the entire model since doing so is exceptionally slow
+        default_dburl = 'sqlite:///%s?isolation_level=IMMEDIATE' % self.config.get( 'app:main', 'database_file' )
+        try:
+            dburl = self.config.get( 'app:main', 'database_connection' )
+        except ConfigParser.NoOptionError:
+            dburl = default_dburl
+        load_egg_for_url( dburl )
+        engine = create_engine( dburl )
+        metadata = MetaData( engine )
+        self.sa_session = scoped_session( sessionmaker( bind=engine, autoflush=False, autocommit=True ) )
+        self.model.TransferJob = galaxy.model.TransferJob
+        self.model.TransferJob.table = Table( "transfer_job", metadata, autoload=True )
+    def get_transfer_job( self, id ):
+        return self.sa_session.query( self.model.TransferJob ).get( int( id ) )
+
+class ListenerServer( SocketServer.ThreadingTCPServer ):
+    """
+    The listener will accept state requests and new transfers for as long as
+    the manager is running.
+    """
+    def __init__( self, port_range, RequestHandlerClass, app, transfer_job, state_result ):
+        self.state_result = state_result
+        # Try random ports until a free one is found
+        while True:
+            random_port = random.choice( port_range )
+            try:
+                SocketServer.ThreadingTCPServer.__init__( self, ( 'localhost', random_port ), RequestHandlerClass )
+                log.info( 'Listening on port %s' % random_port )
+                break
+            except Exception, e:
+                log.warning( 'Tried binding port %s: %s' % ( random_port, str( e ) ) )
+        transfer_job.socket = random_port
+        app.sa_session.add( transfer_job )
+        app.sa_session.flush()
+
+class ListenerRequestHandler( SocketServer.BaseRequestHandler ):
+    """
+    Handle state or transfer requests received on the socket.
+    """
+    def handle( self ):
+        request = self.request.recv( 8192 )
+        response = {}
+        valid, request, response = json.validate_jsonrpc_request( request, ( 'get_state', ), () )
+        if valid:
+            self.request.send( json.to_json_string( json.jsonrpc_response( request=request, result=self.server.state_result.result ) ) )
+        else:
+            error_msg = 'Unable to serve request: %s' % response['error']['message']
+            if 'data' in response['error']:
+                error_msg += ': %s' % response['error']['data']
+            log.error( error_msg )
+            log.debug( 'Original request was: %s' % request )
+
+class StateResult( object ):
+    """
+    A mutable container for the 'result' portion of JSON-RPC responses to state requests.
+    """
+    def __init__( self, result=None ):
+        self.result = result
+
+def transfer( app, transfer_job_id ):
+
+    transfer_job = app.get_transfer_job( transfer_job_id )
+    if transfer_job is None:
+        log.error( 'Invalid transfer job ID: %s' % transfer_job_id )
+        return False
+
+    port_range = app.config.get( 'app:main', 'transfer_worker_port_range' )
+    try:
+        port_range = [ int( p ) for p in port_range.split( '-' ) ]
+    except Exception, e:
+        log.error( 'Invalid port range set in transfer_worker_port_range: %s: %s' % ( port_range, str( e ) ) )
+        return False
+
+    protocol = transfer_job.params['url'].split( '://' )[0]
+    if protocol not in ( 'http', 'https' ):
+        log.error( 'Unsupported protocol: %s' % protocol )
+        return False
+
+    state_result = StateResult( result = dict( state = transfer_job.states.RUNNING, info='Transfer process starting up.' ) )
+
+    listener_server = ListenerServer( range( port_range[0], port_range[1] + 1 ), ListenerRequestHandler, app, transfer_job, state_result )
+
+    # daemonize here (if desired)
+    if not debug:
+        daemon_context = DaemonContext( files_preserve=[ listener_server.fileno() ], working_directory=os.getcwd() )
+        daemon_context.open()
+
+        # If this fails, it'll never be detected.  Hopefully it won't fail since it succeeded once.
+        app.connect_database() # daemon closed the database fd
+        transfer_job = app.get_transfer_job( transfer_job_id )
+
+    listener_thread = threading.Thread( target=listener_server.serve_forever )
+    listener_thread.setDaemon( True )
+    listener_thread.start()
+
+    # Store this process' pid so unhandled deaths can be handled by the restarter
+    transfer_job.pid = os.getpid()
+    app.sa_session.add( transfer_job )
+    app.sa_session.flush()
+
+    terminal_state = None
+    if protocol in ( 'http', 'https' ):
+        for state in http_transfer( transfer_job ):
+            state_result.result = state
+            if state['state'] in transfer_job.terminal_states:
+                terminal_state = state
+    if terminal_state is not None:
+        transfer_job.state = terminal_state['state']
+        for name in ( 'info', 'path' ):
+            if name in terminal_state:
+                transfer_job.__setattr__( name, terminal_state[name] )
+    else:
+        transfer_job.state = transfer_job.states.ERROR
+        transfer_job.info = 'Unknown error encountered by transfer worker.'
+    app.sa_session.add( transfer_job )
+    app.sa_session.flush()
+    return True
+
+def http_transfer( transfer_job ):
+    """
+    "Plugin" for handling http(s) transfers.
+    """
+    url = transfer_job.params['url']
+    try:
+        f = urllib2.urlopen( url )
+    except urllib2.URLError, e:
+        yield dict( state = transfer_job.states.ERROR, info = 'Unable to open URL: %s' % str( e ) )
+        return
+    size = f.info().getheader( 'Content-Length' )
+    if size is not None:
+        size = int( size )
+    chunksize = 1024 * 1024
+    if slow:
+        chunksize = 1024
+    read = 0
+    last = 0
+    try:
+        fh, fn = tempfile.mkstemp()
+    except Exception, e:
+        yield dict( state = transfer_job.states.ERROR, info = 'Unable to create temporary file for transfer: %s' % str( e ) )
+        return
+    log.debug( 'Writing %s to %s, size is %s' % ( url, fn, size or 'unknown' ) )
+    try:
+        while True:
+            chunk = f.read( chunksize )
+            if not chunk:
+                break
+            os.write( fh, chunk )
+            read += chunksize
+            if size is not None and read < size:
+                percent = int( float( read ) / size * 100 )
+                if percent != last:
+                    yield dict( state = transfer_job.states.PROGRESS, read = read, percent = '%s' % percent )
+                    last = percent
+            elif size is None:
+                yield dict( state = transfer_job.states.PROGRESS, read = read )
+            if slow:
+                time.sleep( 1 )
+        os.close( fh )
+        yield dict( state = transfer_job.states.DONE, path = fn )
+    except Exception, e:
+        yield dict( state = transfer_job.states.ERROR, info = 'Error during file transfer: %s' % str( e ) )
+        return
+    return
+
+if __name__ == '__main__':
+
+    arg_handler = ArgHandler()
+    arg_handler.parse()
+    app = GalaxyApp( arg_handler.opts.config )
+
+    log.debug( 'Initiating transfer' )
+    if transfer( app, arg_handler.transfer_job_id ):
+        log.debug( 'Finished' )
+    else:
+        log.error( 'Error in transfer process' )
+        sys.exit( 1 )
+    sys.exit( 0 )

transfer_manager.py

@@ -1,330 +0,0 @@
-#!/usr/bin/env python
-'''
-Downloads files to temp locations.
-'''
-
-import os, sys, optparse, ConfigParser, socket, SocketServer, errno, Queue, threading, subprocess
-
-import urllib2, tempfile
-
-import time
-
-sys.path.insert( 0, os.path.abspath( 'lib' ) )
-from galaxy import eggs
-import galaxy.model.mapping
-from galaxy.util import json, bunch
-
-eggs.require( 'python_daemon' )
-from daemon import DaemonContext
-
-class ArgHandler( object ):
-    """
-    Collect command line flags.
-    """
-    def __init__( self ):
-        self.parser = optparse.OptionParser()
-        self.parser.add_option( '-c', '--config', dest='config', help='Path to Galaxy config file (universe_wsgi.ini)', default='universe_wsgi.ini' )
-        self.parser.add_option( '-i', '--transfer-job-id', action='append', dest='transfer_job_ids', help='Initiate management of the specified TransferJob id' )
-        self.parser.add_option( '--do', dest='initiate_transfer_job_id', help='Used by this script when it calls itself to actually initiate the download' )
-        self.parser.add_option( '-s', '--state-transfer-job-id', action='append', dest='state_transfer_job_ids', help='Report the state of the specified TransferJob id' )
-        self.parser.add_option( '-d', '--debug', action='store_true', dest='debug', help="Debug (don't detach)" )
-        self.opts = None
-    def parse( self ):
-        self.opts, args = self.parser.parse_args()
-
-class GalaxyApp( object ):
-    """
-    A shell Galaxy App to provide access to the Galaxy configuration and
-    model/database.
-    """
-    def __init__( self, config_file='universe_wsgi.ini' ):
-        self.config = ConfigParser.ConfigParser( dict( database_file = 'database/universe.sqlite',
-                                                       file_path = 'database/files',
-                                                       transfer_manager_port = '8163',
-                                                       transfer_manager_log = 'transfer_manager.log' ) )
-        self.config.read( config_file )
-        self.model = None
-    @property
-    def sa_session( self ):
-        if not self.model:
-            default_dburl = 'sqlite:///%s?isolation_level=IMMEDIATE' % self.config.get( 'app:main', 'database_file' )
-            try:
-                dburl = self.config.get( 'app:main', 'database_connection' )
-            except ConfigParser.NoOptionError:
-                dburl = default_dburl
-            self.model = galaxy.model.mapping.init( self.config.get( 'app:main', 'file_path' ), dburl, create_tables = False )
-        return self.model.context.current
-    def get_transfer_job( self, id ):
-        return self.sa_session.query( self.model.TransferJob ).get( int( id ) )
-
-class ListenerServer( SocketServer.ThreadingTCPServer ):
-    """
-    The listener will accept state requests and new transfers for as long as
-    the manager is running.
-    """
-    def __init__( self, server_address, RequestHandlerClass, transfer_manager ):
-        SocketServer.ThreadingTCPServer.__init__( self, server_address, RequestHandlerClass )
-        self.transfer_manager = transfer_manager
-        self.app = transfer_manager.app
-
-class ListenerRequestHandler( SocketServer.BaseRequestHandler ):
-    """
-    Handle state or transfer requests received on the socket.
-    """
-    def handle( self ):
-        if not self.server.transfer_manager.accepting:
-            # TODO: does the submitter handle this condition?  i'm sure it doesn't...
-            self.request.send( 'Manager shutting down\n' )
-            return
-        data = ''
-        while len( data ) < 8 * 1024 * 1024:
-            # read data up to 8MB (overkill, but be safe)
-            chunk = self.request.recv( 1024 )
-            if not chunk:
-                break
-            data += chunk
-            if '\n' in data:
-                break
-        else:
-            self.request.send( 'Message too large\n' )
-            return
-        data = json.from_json_string( data )
-        if 'transfer_job_ids' in data:
-            # Get all of the TransferJob objects and stick them on the queue.
-            print 'Adding transfer job ids to transfer queue: %s' % data['transfer_job_ids']
-            [ self.server.transfer_manager.transfer_queue.put( self.server.app.get_transfer_job( transfer_job_id ) ) for transfer_job_id in data['transfer_job_ids'] ]
-            self.request.send( "Added jobs to transfer queue: %s\n" % ', '.join( data['transfer_job_ids'] ) )
-        elif 'state_transfer_job_ids' in data:
-            print 'Servicing state request for transfer job ids: %s' % data['state_transfer_job_ids']
-            for state_transfer_job_id in data['state_transfer_job_ids']:
-                state = self.server.transfer_manager.get_state( int( state_transfer_job_id ) )
-                state['transfer_job_id'] = state_transfer_job_id
-                print 'State of transfer job id %s is: %s' % ( state_transfer_job_id, state )
-                self.request.send( json.to_json_string( state ) )
-
-class Transfer( object ):
-    """
-    Instantiated for each transfer to track the progress of the transfer via
-    commmunication with a subprocess.
-    """
-    states = bunch.Bunch( NEW = 'new',
-                          UNKNOWN = 'unknown',
-                          STARTED = 'started',
-                          PROGRESS = 'progress',
-                          DONE = 'done',
-                          ERROR = 'error' )
-    def __init__( self, transfer_job ):
-        self.transfer_job = transfer_job
-        self.state = dict( state = self.states.NEW )
-        self.done = False
-    def run( self ):
-        cmd = '%s -u %s --do %s' % ( sys.executable, os.path.abspath( __file__ ), self.transfer_job.id )
-        print cmd
-        self.p = subprocess.Popen( cmd, bufsize=0, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT )
-        line = self.p.stdout.readline()
-        while line:
-            try:
-                self.state = json.from_json_string( line )
-                assert 'state' in self.state
-            except:
-                print 'Received unknown state from transfer (transfer job id: %s): %s' % ( self.transfer_job.id, line )
-                self.state = dict( state = self.states.UNKNOWN, info=line )
-            line = self.p.stdout.readline()
-        self.p.wait()
-        self.done = True
-
-class TransferManager( object ):
-    """
-    Manage the queue of transfers, handle the setup of new transfers and
-    completion of finished transfers.
-    """
-    def __init__( self, app, transfer_job_ids, debug=False ):
-        self.app = app
-        self.daemon_context = None
-        self.sa_session = app.sa_session
-        self.transfer_job_ids = transfer_job_ids
-        self.debug = debug
-        self.port = int( app.config.get( 'app:main', 'transfer_manager_port' ) )
-        self.log_file_name = app.config.get( 'app:main', 'transfer_manager_log' )
-        self.transfer_queue = Queue.Queue()
-        self.accepting = True
-        self.watchlist = []
-        self.listen_or_submit()
-    def listen_or_submit( self ):
-        """
-        The transfer manager is invoked the same way every time, via the
-        command line, but if a transfer manager is already running, the new
-        manager will simply submit the request to the old manager and
-        terminate.
-        """
-        try:
-            self.listener_server = ListenerServer( ( 'localhost', self.port ), ListenerRequestHandler, self )
-        except socket.error, e:
-            if e[0] == errno.EADDRINUSE:
-                self.submit()
-                sys.exit()
-
-        # Daemonize
-        if not self.debug:
-            log_file = open( self.log_file_name, 'a+' )
-            self.daemon_context = DaemonContext( files_preserve=[ self.listener_server.fileno() ],
-                                                 working_directory=os.getcwd(),
-                                                 stdout=log_file,
-                                                 stderr=log_file )
-            self.daemon_context.open()
-
-        # Listen for more stuff on the socket
-        self.listener = threading.Thread( target=self.listener_server.serve_forever )
-        self.listener.start()
-
-        # Put all the URLs into the queue, additional URLs received before this
-        # instance of the manager terminates will also be handled.
-        print 'Started up with transfer job ids: %s' % self.transfer_job_ids
-        [ self.transfer_queue.put( app.get_transfer_job( tj_id ) ) for tj_id in self.transfer_job_ids ]
-
-        while True:
-            # TODO: max transfer limit
-            try:
-                transfer_job = self.transfer_queue.get_nowait()
-                print 'Fetching transfer job:', transfer_job.id, 'URL is:', transfer_job.params['url']
-                t = Transfer( transfer_job )
-                transfer_job.state = app.model.TransferJob.states.RUNNING
-                self.sa_session.add( transfer_job )
-                self.sa_session.flush()
-                tt = threading.Thread( target=t.run )
-                tt.start()
-                self.watchlist.append( t )
-            except Queue.Empty:
-                # TODO: this will shutdown the queue as soon the queue is
-                # empty, which is almost immediately.  instead, we need to
-                # accept until all transfers are finished, then stop accepting,
-                # but finish out the queue in case one was added in the instant
-                # between empty and no longer accepting.
-                # (check the queue one more time after accepting = False)
-                if not self.watchlist:
-                    self.acecpting = False
-                    break
-            new_watchlist = []
-            for transfer in self.watchlist:
-                # TODO: handle failure
-                if transfer.done:
-                    if transfer.state['state'] == transfer.states.DONE:
-                        transfer.transfer_job.state = app.model.TransferJob.states.DONE
-                        transfer.transfer_job.path = transfer.state['path']
-                        print 'Transfer of job %s ended successfully, output path is: %s' % ( transfer.transfer_job.id, transfer.transfer_job.path )
-                    elif transfer.state['state'] == transfer.states.ERROR:
-                        transfer.transfer_job.info = transfer.state.get( 'info', None )
-                        transfer.transfer_job.state = app.model.TransferJob.states.ERROR
-                        print 'Transfer of job %s ended in error: %s' % ( transfer.transfer_job.id, transfer.transfer_job.info )
-                    else:
-                        print 'Unknown state received for transfer job %s: %s' % ( transfer.transfer_job.id, transfer.state['state'] )
-                        transfer.transfer_job.info = 'Unknown error encountered in transfer manager'
-                        transfer.transfer_job.state = app.model.TransferJob.states.ERROR
-                    self.sa_session.add( transfer.transfer_job )
-                    self.sa_session.flush()
-                else:
-                    new_watchlist.append( transfer )
-                time.sleep( 1 )
-            self.watchlist = new_watchlist
-        #except Queue.Empty:
-        #    self.accepting = False
-        self.listener_server.shutdown()
-
-    def get_state( self, transfer_job_id ):
-        rval = {}
-        for transfer in self.watchlist:
-            if transfer.transfer_job.id == transfer_job_id:
-                rval = transfer.state
-                break
-        else:
-            rval['state'] = Transfer.states.UNKNOWN # should be DONE?
-        return rval
-
-    def submit( self ):
-        # TODO: can fail if shutdown occurs between failure to bind and submission
-        # Needs error handling.
-        # This may not work at all right now.
-        print "Submitting jobs to running transfer manager: %s" % ', '.join( self.transfer_job_ids )
-        sock = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
-        sock.connect( ( 'localhost', self.port ) )
-        sock.send( json.to_json_string( dict( transfer_job_ids=self.transfer_job_ids ) ) + '\n' )
-        print sock.recv( 8192 ),
-        sock.close()
-        sys.exit()
-
-def do_transfer( transfer_job ):
-    """
-    Actually initiate a transfer (used when the transfer manager calls itself
-    with the '--do' flag).
-    """
-    protocol = transfer_job.params['url'].split( '://' )[0]
-    if protocol in ( 'http', 'https' ):
-        return do_http_transfer( transfer_job )
-    else:
-        print >>sys.stderr, 'ERROR: Unsupported protocol: %s' % protocol
-        sys.exit( 1 )
-
-def do_http_transfer( transfer_job ):
-    """
-    "Plugin" for handling http(s) transfers.
-    """
-    url = transfer_job.params['url']
-    try:
-        f = urllib2.urlopen( url )
-    except urllib2.URLError, e:
-        print json.to_json_string( dict( state=Transfer.states.ERROR,
-                                         info=str( e ) ) )
-        return
-    size = f.info().getheader( 'Content-Length' )
-    if size is not None:
-        size = int( size )
-    chunksize = 1024 * 1024
-    read = 0
-    last = 0
-    fh, fn = tempfile.mkstemp()
-    while True:
-        chunk = f.read( chunksize )
-        if not chunk:
-            break
-        os.write( fh, chunk )
-        if read == 0 and size is None:
-            print json.to_json_string( dict( state=Transfer.states.STARTED,
-                                             size=None ) ) #+ '\n'
-        elif read == 0:
-            print json.to_json_string( dict( state=Transfer.states.STARTED,
-                                             size=size ) ) #+ '\n'
-        read += chunksize
-        if size is not None and read < size:
-            percent = int( float( read ) / size * 100 )
-            if percent != last:
-                print json.to_json_string( dict( state=Transfer.states.PROGRESS,
-                                                 read=read,
-                                                 percent='%s%%' % percent ) ) #+ '\n'
-                last = percent
-        elif size is None:
-            print json.to_json_string( dict( state=Transfer.states.PROGRESS,
-                                             read=read ) )
-    os.close( fh )
-    print json.to_json_string( dict( state=Transfer.states.DONE, path=fn ) ) #+ '\n'
-
-def request_state( state_transfer_job_ids, port ):
-    sock = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
-    sock.connect( ( 'localhost', port ) )
-    sock.send( json.to_json_string( dict( state_transfer_job_ids=state_transfer_job_ids ) ) + '\n' )
-    print sock.recv( 1024 ),
-    sock.close()
-    sys.exit()
-
-if __name__ == '__main__':
-    arg_handler = ArgHandler()
-    arg_handler.parse()
-    app = GalaxyApp( config_file=arg_handler.opts.config )
-    if arg_handler.opts.initiate_transfer_job_id is not None:
-        do_transfer( app.get_transfer_job( arg_handler.opts.initiate_transfer_job_id ) )
-    elif arg_handler.opts.state_transfer_job_ids:
-        request_state( arg_handler.opts.state_transfer_job_ids, int( app.config.get( 'app:main', 'transfer_manager_port' ) ) )
-    elif arg_handler.opts.transfer_job_ids:
-        transfer_manager = TransferManager( app, arg_handler.opts.transfer_job_ids, arg_handler.opts.debug )
-    else:
-        arg_handler.parser.print_usage( sys.stderr )
-        sys.exit( 1 )