diff --git a/config/galaxy.ini.sample b/config/galaxy.ini.sample index de7a82b8d99..b2237f89427 100644 --- a/config/galaxy.ini.sample +++ b/config/galaxy.ini.sample @@ -1228,6 +1228,9 @@ use_interactive = True # be configured through the job configuration file. #job_config_file = config/job_conf.xml +### TODO: These options will move to a stack config section in the YAML config +### file + # Some application stacks that Galaxy run under (such as uWSGI) can start # arbitrary Galaxy server processes to perform functions asynchronously from # the Galaxy web server process(es). Such stacks may give you the option to diff --git a/lib/galaxy/app.py b/lib/galaxy/app.py index 8b27caf7e5f..51574c4421f 100644 --- a/lib/galaxy/app.py +++ b/lib/galaxy/app.py @@ -191,9 +191,6 @@ class UniverseApplication(object, config.ConfiguresGalaxyMixin): from galaxy.jobs import manager self.job_manager = manager.JobManager(self) self.application_stack.register_postfork_function(self.job_manager.start) - # 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 self.proxy_manager = ProxyManager(self.config) # Initialize the external service types self.external_service_types = external_service_types.ExternalServiceTypesCollection( diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py index 7b2b3d2d714..851a3762085 100644 --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -166,15 +166,6 @@ class JobConfiguration(object, ConfiguresHandlers): except Exception as e: raise config_exception(e, job_config_file) - def __config_default_handlers(self): - """This is a stop-gap solution until the job conf loading is rewritten - for YAML job conf support. - """ - # FIXME: this will not support the case where uWSGI is being used w/o handler mules - if self.app.application_stack.handle_jobs: - self.handlers[self.app.config.server_name] = (self.app.config.server_name,) - self.default_handler_id = self.app.config.server_name - def __parse_job_conf_xml(self, tree): """Loads the new-style job configuration from options in the job config file (by default, job_conf.xml). @@ -212,13 +203,18 @@ class JobConfiguration(object, ConfiguresHandlers): handlers_conf = root.find('handlers') self._init_handlers(handlers_conf) - # Must define at least one handler to have a default. - if not self.handlers: - self.__config_default_handlers() - log.info("No handlers defined or empty handlers group, will use default handlers: %s", self.handlers) - else: - # Determine the default handler(s) - self.default_handler_id = self.__get_default(handlers, list(self.handlers.keys())) + # Determine the default handler(s) + try: + self.default_handler_id = self._get_default(self.app.config, handlers_conf, list(self.handlers.keys())) + except: + log.info('No default handler specified in job config, will check for job handlers managed by application stack') + if ((self.default_handler_id is None + or (len(self.handlers) == 1 and self.app.config.server_name.startswith(self.handlers.keys()[0]))) + and self.app.application_stack.has_pool(self.app.application_stack.pools.JOB_HANDLERS)): + # Shortcut for compatibility with existing job confs that use the default handlers block + # There are no defined handlers or there's only one handler and it's this server, and the stack has a handler pool + self.handlers = {} + self.default_handler_id = None # Parse destinations destinations = root.find('destinations') @@ -352,7 +348,14 @@ class JobConfiguration(object, ConfiguresHandlers): self.runner_plugins.append(dict(id=runner, load=runner, workers=self.app.config.cluster_job_queue_workers)) # Set the handlers - self.__config_default_handlers() + if self.app.application_stack.has_pool(self.app.application_stack.pools.JOB_HANDLERS): + self.default_handler_id = None + else: + for id in self.app.config.job_handlers: + self.handlers[id] = (id,) + + self.handlers['default_job_handlers'] = self.app.config.default_job_handlers + self.default_handler_id = 'default_job_handlers' # Set tool handler configs for id, tool_handlers in self.app.config.tool_handlers.items(): diff --git a/lib/galaxy/jobs/manager.py b/lib/galaxy/jobs/manager.py index c68ee8f6974..6805ca4ba1f 100644 --- a/lib/galaxy/jobs/manager.py +++ b/lib/galaxy/jobs/manager.py @@ -23,27 +23,19 @@ class JobManager(object): def __init__(self, app): self.app = app - self.job_handler = NoopHandler() - self.job_stop_queue = NoopQueue() - if app.application_stack.setup_jobs_with_msg: - # defer setup to postfork - log.debug('######### registering manager init function') - self.app.application_stack.register_postfork_function(self.init) - else: - self.init() - - def init(self): - log.debug("Initializing job manager interface") + self.job_lock = False if self.app.is_job_handler(): log.debug("Starting job handler") - self.job_handler = handler.JobHandler(self.app) + self.job_handler = handler.JobHandler(app) self.job_stop_queue = self.job_handler.job_stop_queue - elif self.app.application_stack.setup_jobs_with_msg: - # not a handler, but notification is via the application stack - self.job_handler = MessageJobHandler( self.app ) + elif app.application_stack.has_pool(app.application_stack.pools.JOB_HANDLERS): + log.debug("Initializing job handler messaging interface") + self.job_handler = MessageJobHandler(app) + self.job_stop_queue = NoopQueue() + else: + self.job_handler = NoopHandler() self.job_stop_queue = NoopQueue() self.job_queue = self.job_handler.job_queue - self.job_lock = False def start(self): self.job_handler.start() @@ -64,21 +56,19 @@ class NoopHandler(object): pass -class MessageJobHandler( object ): +class MessageJobHandler(NoopHandler): """ Implements the JobHandler interface but just to send setup messages on startup TODO: It should be documented that starting two Galaxy uWSGI master processes simultaneously would result in a race condition that *could* cause two handlers to pick up the same job. - The recommended config for now will be webless/main handlers if running more than one uWSGI (master) process + The recommended config for now will be webless handlers if running more than one uWSGI (master) process """ def __init__(self, app): + # This runs in the web (main) process pre-fork self.app = app self.job_queue = MessageJobQueue(app) self.job_stop_queue = NoopQueue() - - def start(self): - # This runs in the web (main) process pre-fork jobs_at_startup = self.app.model.context.query(Job).enable_eagerloads(False) \ .filter((Job.state == Job.states.NEW) & (Job.handler == null())).all() if jobs_at_startup: @@ -86,11 +76,8 @@ class MessageJobHandler( object ): for job in jobs_at_startup: self.job_queue.put(job.id, job.tool_id) - def shutdown(self, *args): - pass - -class MessageJobQueue(object): +class MessageJobQueue(NoopQueue): """ Implements the JobQueue / JobStopQueue interface but only sends messages to the actual job queue """ @@ -99,11 +86,4 @@ class MessageJobQueue(object): def put(self, job_id, tool_id): msg = JobHandlerMessage(task='setup', job_id=job_id) - # TODO: send to a specific pool - self.app.application_stack.send_message(self.app.application_stack.purposes.JOB_HANDLER, msg) - - def put_stop(self, *args): - pass - - def shutdown(self): - pass + self.app.application_stack.send_message(self.app.application_stack.pools.JOB_HANDLERS, msg) diff --git a/lib/galaxy/util/configdict.py b/lib/galaxy/util/configdict.py deleted file mode 100644 index ffe062a55e1..00000000000 --- a/lib/galaxy/util/configdict.py +++ /dev/null @@ -1,39 +0,0 @@ -"""Generic config parsing into dictionary. -""" -import errno -import logging - -try: - import yaml -except ImportError: - yaml = None - - -log = logging.getLogger(__name__) - - -def parse_config(config_file, root, default=None): - if default: - conf = default.copy() - else: - conf = {} - try: - load_func = __load_func(config_file) - with open(config_file) as fh: - c = load_func(fh) - if not c: - c = {} - conf.update(c.get(root, {})) - except (OSError, IOError) as exc: - if exc.errno == errno.ENOENT: - log.warning("config file '%s' does not exist, running with default config", config_file) - else: - raise - return conf - - -def __load_func(path): - if path.endswith('yaml') or path.endswith('.yml'): - if not yaml: - raise RuntimeError("The 'yaml' module could not be imported, please install PyYAML to read '%s'" % path) - return yaml.load diff --git a/lib/galaxy/util/handlers.py b/lib/galaxy/util/handlers.py index 8f7a7ff7969..96087ecce56 100644 --- a/lib/galaxy/util/handlers.py +++ b/lib/galaxy/util/handlers.py @@ -101,12 +101,9 @@ class ConfiguresHandlers: :return: bool """ - if self.app.application_stack.has_purpose(self.app.application_stack.purposes.JOB_HANDLER): + if self.app.application_stack.in_pool(self.app.application_stack.pools.JOB_HANDLERS): # Handlers started as uWSGI mules do not require configuration in the job conf return True - #if stack_handles_jobs is not None: - # # Handlers started as uWSGI mules do not require configuration in the job conf - # return stack_handles_jobs for collection in self.handlers.values(): if server_name in collection: return True diff --git a/lib/galaxy/web/stack/__init__.py b/lib/galaxy/web/stack/__init__.py index a97328c9a81..f4824cbbf92 100644 --- a/lib/galaxy/web/stack/__init__.py +++ b/lib/galaxy/web/stack/__init__.py @@ -17,63 +17,17 @@ try: except ImportError: uwsgi = None -#try: -# from uwsgidecorators import postfork as uwsgi_postfork -#except: -# uwsgi_postfork = lambda x: x # noqa: E731 -# if uwsgi is not None and hasattr(uwsgi, 'numproc'): -# print("WARNING: This is a uwsgi process but the uwsgidecorators library" -# " is unavailable. This is likely due to using an external (not" -# " in Galaxy's virtualenv) uwsgi and you may experience errors. " -# "HINT:\n {venv}/bin/pip install uwsgidecorators".format( -# venv=os.environ.get('VIRTUAL_ENV', '/path/to/venv'))) - from six import string_types -from .message import ApplicationStackMessage, JobHandlerMessage, decode +from .message import ApplicationStackMessage, ApplicationStackMessageDispatcher, JobHandlerMessage, decode from .transport import ApplicationStackTransport, UWSGIFarmMessageTransport from galaxy.util.bunch import Bunch -from galaxy.util.configdict import parse_config log = logging.getLogger(__name__) -class ApplicationStackMessageDispatcher(object): - def __init__(self): - self.__funcs = {} - - def __func_name(self, func, name): - if not name: - name = func.__name__ - return name - - def register_func(self, func, name=None): - name = self.__func_name(func, name) - self.__funcs[name] = func - - def deregister_func(self, func=None, name=None): - name = self.__func_name(func, name) - del self.__funcs[name] - - @property - def handler_count(self): - return len(self.__funcs) - - def dispatch(self, msg_str): - msg = decode(msg_str) - try: - msg.validate() - except AssertionError as exc: - log.error('######## Invalid message received: %s, error: %s', msg_str, exc) - return - if msg.target not in self.__funcs: - log.error("######## Received message with target '%s' but no functions were registered with that name. Params were: %s", msg.target, msg.params) - else: - self.__funcs[msg.target](msg) - - class ApplicationStackLogFilter(logging.Filter): def filter(self, record): record.worker_id = None @@ -91,97 +45,47 @@ class UWSGILogFilter(logging.Filter): class ApplicationStack(object): name = None prohibited_middleware = frozenset() - # TODO: this is a fairly clunky way of handling these cases - setup_jobs_with_msg = False # used in galaxy.jobs.manager to determine whether jobs should be sent via message - handle_jobs = False # used by galaxy.jobs to determine whether this process handles jobs - transport_class = ApplicationStackTransport log_filter_class = ApplicationStackLogFilter + # TODO: this belongs in the pool configuration + server_name_template = '{server_name}' - purposes = Bunch( - WEB_WORKER = 'web-worker', - JOB_HANDLER = 'job-handler', + # used both to route jobs to a pool with this name and indicate whether or + # not a stack is using messaging for handler assignment + pools = Bunch( + JOB_HANDLERS = 'job-handlers', ) - default_conf = { - 'processes': 1, - 'threads': 4, - 'server_name': '{server_name}', - 'workers': [], - 'pools': [], - } - - web_worker_pool = { - 'name': 'web-workers', - 'purpose': 'web-worker', - } - @classmethod def register_postfork_function(cls, f, *args, **kwargs): f(*args, **kwargs) def __init__(self, app=None): self.app = app - # FIXME: hardcoded path - self._conf = parse_config('config/stack_conf.yml', 'stack', default=self.default_conf) - self.server_name_template = '{server_name}' - self.pools = [] - self.pool_names = [] - self._purposes = [] - - def _postfork_init(self): - self.pools = [] - self.pool_names = [] - self._purposes = [] - for pool in self._conf.get('pools', []) + [ApplicationStack.web_worker_pool]: - if self.in_pool(pool['name']): - self.pools.append(pool) - self.pool_names.append(pool['name']) - self._purposes.append(pool['purpose']) - for pool in self.pools: - if pool.get('server_name', None): - self.server_name_template = pool['server_name'] - break - else: - if self._conf.get('server_name', None): - # default if the process is not in a pool and there is an unpooled server name set in the conf - self.server_name_template = self._conf['server_name'] - if self.pools: - # default if the process is in a pool - self.server_name_template = '{server_name}.{pool_name}.{process_num}' def start(self): - self._postfork_init() - - @property - def pool_name(self): - # TODO: in the future, allow for mapping job handlers in the job conf by something other than server_name, such - # as pool name and process number. but for now, the job-handlers pool should be the first one specified under a - # worker set's 'pools' list, if more than one pool is specified for a given worker set - try: - return self.pool_names[0] - except: - return None - - def in_pool(self, pool_name): - return None - - def has_purpose(self, purpose): - return purpose in self._purposes - - def workers(self): - return [] - - - - - + # TODO: with a stack config the pools could be parsed here + pass def allowed_middleware(self, middleware): if hasattr(middleware, '__name__'): middleware = middleware.__name__ return middleware not in self.prohibited_middleware + def workers(self): + return [] + + @property + def pool_name(self): + # TODO: ideally jobs would be mappable to handlers by pool name + return None + + def has_pool(self, pool_name): + return False + + def in_pool(self, pool_name): + return False + def set_server_name(self, app, caller_tmpl_dict): tmpl_dict = { 'server_name': app.config.server_name, @@ -193,10 +97,10 @@ class ApplicationStack(object): } tmpl_dict.update(caller_tmpl_dict) app.config.server_name = self.server_name_template.format(**tmpl_dict) - log.debug('######## server_name is: %s', app.config.server_name) + log.debug('server_name set to: %s', app.config.server_name) def set_postfork_server_name(self, app): - self.set_server_name(app, tmpl_dict, {}) + self.set_server_name(app, {}) def register_message_handler(self, func, name=None): pass @@ -207,13 +111,6 @@ class ApplicationStack(object): def send_message(self, dest, msg=None, target=None, params=None, **kwargs): pass - def send_purpose_message(self, purpose, **kwargs): - for pool in self.pools: - if purpose == pool['purpose']: - return self.send_message(pool['name'], **kwargs) - else: - raise RuntimeError('No pools defined for purpose: %s', purpose) - def shutdown(self): pass @@ -223,9 +120,6 @@ class MessageApplicationStack(ApplicationStack): super(MessageApplicationStack, self).__init__(app=app) self.dispatcher = ApplicationStackMessageDispatcher() self.transport = self.transport_class(app, stack=self, dispatcher=self.dispatcher) - #if app: - # log.debug('######## registering self.start') - # self.register_postfork_function(self.start) def start(self): super(MessageApplicationStack, self).start() @@ -262,28 +156,11 @@ class UWSGIApplicationStack(MessageApplicationStack): 'wrap_in_static', 'EvalException', ]) - setup_jobs_with_msg = True - transport_class = UWSGIFarmMessageTransport log_filter_class = UWSGILogFilter - # FIXME: this is copied into UWSGIFarmMessageTransport - shutdown_msg = '__SHUTDOWN__' - postfork_functions = [] + server_name_template = '{server_name}.{id}' - default_conf = { - 'processes': 1, - 'threads': 4, - 'server_name': '{server_name}.{process_num}', - 'workers': [{ - 'load': 'lib/galaxy/main.py', - 'processes': 1, - 'pools': ['job-handlers'] - }], - 'pools': [{ - 'name': 'job-handlers', - 'purpose': 'job-handler', - }] - } + postfork_functions = [] @classmethod def register_postfork_function(cls, f, *args, **kwargs): @@ -291,50 +168,42 @@ class UWSGIApplicationStack(MessageApplicationStack): cls.postfork_functions.append((f, args, kwargs)) else: # mules are forked from the master and run the master's postfork functions immediately before the forked - # process is replaced. that is prevented in the _do_uwsgi_postfork function, and because mules are - # standalone non-forking processes, they should run postfork functions immediately + # process is replaced. that is prevented in the _do_uwsgi_postfork function, and because programmed mules + # are standalone non-forking processes, they should run postfork functions immediately f(*args, **kwargs) def __init__(self, app=None): - super(UWSGIApplicationStack, self).__init__(app=app) self._farms_dict = None self._mules_list = None - self._is_mule = None + super(UWSGIApplicationStack, self).__init__(app=app) def __register_signal_handlers(self): for name in ('TERM', 'INT', 'HUP'): - log.debug('######## registered signal handler for SIG%s', name) sig = getattr(signal, 'SIG%s' % name) signal.signal(sig, self._handle_signal) def _handle_signal(self, signum, frame): - if signum == signal.SIGTERM: - log.info('######## Mule %s received SIGTERM, shutting down gracefully', uwsgi.mule_id()) - self.shutdown() - elif signum == signal.SIGINT: - log.info('######## Mule %s received SIGINT, shutting down immediately', uwsgi.mule_id()) - self.shutdown() - # This terminates the application loop in the handler entrypoint - self.app.exit = True + # uWSGI always sends SIGINT even if the master received SIGTERM + if signum in (signal.SIGTERM, signal.SIGINT): + log.info('Received SIGTERM/SIGINT, shutting down gracefully') elif signum == signal.SIGHUP: - log.debug('######## Mule %s received SIGHUP, restarting', uwsgi.mule_id()) - self.shutdown() - # uWSGI master will restart us - self.app.exit = True + log.debug('Received SIGHUP, restarting') + self.shutdown() + # this terminates the application loop in the mule script, in the case of HUP, uWSGI will restart the mule + self.app.exit = True - def in_pool(self, pool_name): - if uwsgi.worker_id() > 0 and pool_name == ApplicationStack.web_worker_pool['name']: - return True - else: - return self._in_farm(pool_name) - #return self.transport._farm_name == pool_name - - def workers(self): - return uwsgi.workers() - - # FIXME: these are copied into UWSGIFarmMessageTransport @property - def _farms(self): + def _configured_mules(self): + if self._mules_list is None: + self._mules_list = _uwsgi_configured_mules() + return self._mules_list + + @property + def _is_mule(self): + return uwsgi.mule_id() > 0 + + @property + def _configured_farms(self): if self._farms_dict is None: self._farms_dict = {} farms = uwsgi.opt.get('farm', []) @@ -345,57 +214,54 @@ class UWSGIApplicationStack(MessageApplicationStack): return self._farms_dict @property - def _mules(self): - if self._mules_list is None: - self._mules_list = [] - mules = uwsgi.opt.get('mule', []) - self._mules_list = [mules] if isinstance(mules, string_types) or mules is True else mules - return self._mules_list + def _farms(self): + farms = [] + for farm, mules in self._configured_farms.items(): + if uwsgi.mule_id() in mules: + farms.append(farm) + return farms - def _in_farm(self, farm_name): - return uwsgi.mule_id() in self._farms.get(farm_name, []) + @property + def _farm_name(self): + try: + return self._farms[0] + except IndexError: + return None def start(self): - self._is_mule = uwsgi.mule_id() > 0 + # Does a generalized `is_worker` attribute make sense? Hard to say w/o other stack paradigms. if self._is_mule: self.__register_signal_handlers() super(UWSGIApplicationStack, self).start() + def has_pool(self, pool_name): + return pool_name in self._configured_farms + + def in_pool(self, pool_name): + if not self._is_mule: + return False + else: + return pool_name in self._farms + + def workers(self): + return uwsgi.workers() + def set_postfork_server_name(self, app): tmpl_dict = { 'id': 'worker%s' % uwsgi.worker_id() if uwsgi.mule_id() == 0 else 'mule%s' % uwsgi.mule_id(), } - if uwsgi.mule_id() == 0: - tmpl_dict['process_num'] = uwsgi.worker_id() - elif self.pool_name in self._farms: - tmpl_dict['process_num'] = self._farms[self.pool_name].index(uwsgi.mule_id()) + 1 - log.debug('######## self._farms[self.pool_name]: %s', self._farms[self.pool_name]) self.set_server_name(app, tmpl_dict) - #@property - #def pool_name(self): - # # could get this from self._conf, or could get it from uwsgi.opts - # return self.transport._farm_name - def shutdown(self): - log.debug('######## STACK SHUTDOWN CALLED') super(UWSGIApplicationStack, self).shutdown() - # FIXME: blech - if not self._is_mule: - for farm in self._farms: - for mule in self._mules: - # This will possibly generate more than we need, but that's ok - self.transport.send_message(self.shutdown_msg, farm) class PasteApplicationStack(ApplicationStack): name = 'Python Paste' - handle_jobs = True class WeblessApplicationStack(ApplicationStack): name = 'Webless' - handle_jobs = True def application_stack_class(): @@ -425,22 +291,19 @@ def register_postfork_function(f, *args, **kwargs): application_stack_class().register_postfork_function(f, *args, **kwargs) -def _mules(): +def _uwsgi_configured_mules(): mules = uwsgi.opt.get('mule', []) return [mules] if isinstance(mules, string_types) or mules is True else mules -#@uwsgi_postfork def _do_uwsgi_postfork(): - import os - # FIXME: _mules duplicated again - for i, mule in enumerate(_mules()): + for i, mule in enumerate(_uwsgi_configured_mules()): if mule is not True and i + 1 == uwsgi.mule_id(): # mules will inherit the postfork function list and call them immediately upon fork, but programmed mules # should not do that (they will call the postfork functions in-place as they start up after exec()) UWSGIApplicationStack.postfork_functions = [] - log.debug('######## postfork called, pid %s mule %s functions are: %s' % (os.getpid(), uwsgi.mule_id(), UWSGIApplicationStack.postfork_functions)) for f, args, kwargs in [t for t in UWSGIApplicationStack.postfork_functions]: + log.debug('Calling postfork function: %s', f) f(*args, **kwargs) diff --git a/lib/galaxy/web/stack/message.py b/lib/galaxy/web/stack/message.py index ba5a8516f59..111794689c3 100644 --- a/lib/galaxy/web/stack/message.py +++ b/lib/galaxy/web/stack/message.py @@ -9,45 +9,52 @@ import logging log = logging.getLogger(__name__) -class ApplicationStackMessageEncoder(json.JSONEncoder): - def encode(self, o): - if isinstance(o, AttributeDict): - return o.serialize(self) - return json.JSONEncoder.encode(self, o) +class ApplicationStackMessageDispatcher(object): + def __init__(self): + self.__funcs = {} + + def __func_name(self, func, name): + if not name: + name = func.__name__ + return name + + def register_func(self, func, name=None): + name = self.__func_name(func, name) + self.__funcs[name] = func + + def deregister_func(self, func=None, name=None): + name = self.__func_name(func, name) + del self.__funcs[name] + + @property + def handler_count(self): + return len(self.__funcs) + + def dispatch(self, msg_str): + msg = decode(msg_str) + try: + msg.validate() + except AssertionError as exc: + log.error('Invalid message received: %s, error: %s', msg_str, exc) + return + if msg.target not in self.__funcs: + log.error("Received message with target '%s' but no functions were registered with that name. Params were: %s", msg.target, msg.params) + else: + self.__funcs[msg.target](msg) class ApplicationStackMessage(dict): + target = None + def __init__(self, target=None, params=None, **kwargs): - """Any extra kwargs override values in params - """ - self['target'] = target + self['target'] = target or self.__class__.target self['params'] = params or {} for k, v in kwargs.items(): self['params'][k] = v - #@classmethod - #def from_string(cls, s) - # #kwargs['object_hook'] = ApplicationStackMessage - # d = json.loads(s) - # return cls(json.loads(s, *args, **kwargs)) - def validate(self): assert self['target'] is not None, "Missing 'target' parameter" - def serialize(self, encoder): - #return json.JSONEncoder.encode(encoder, ApplicationStackMessage.clean(self)) - self.validate() - return json.JSONEncoder.encode(encoder, self) - - def dumps(self, *args, **kwargs): - #kwargs['cls'] = AttributeStackMessageEncoder - return json.dumps(self, *args, **kwargs) - - def dump(self, fp, *args, **kwargs): - #raise Exception("This isn't using the Encoder, what gives?") - #kwargs['cls'] = AttributeStackMessageEncoder - return json.dump(self, fp, *args, **kwargs) - @property def target(self): return self['target'] @@ -64,23 +71,15 @@ class ApplicationStackMessage(dict): def set_params(self, params): self['params'] = params - #property - #def class(self): - # return globals()[self['cls']] - def encode(self): self['__classname__'] = self.__class__.__name__ return json.dumps(self) -# when we add additional messages this should become generalized and subclassed +# TODO: when additional messages are added we should refactor and improve validation and param/task separation (but not all msgs may use task) class JobHandlerMessage(ApplicationStackMessage): target = 'job_handler' - def __init__(self, target=None, params=None, **kwargs): - super(JobHandlerMessage, self).__init__(params=params, **kwargs) - self['target'] = self.target - def validate(self): super(JobHandlerMessage, self).validate() for param in ('task', 'job_id'): diff --git a/lib/galaxy/web/stack/transport.py b/lib/galaxy/web/stack/transport.py index 216537c9489..82ef27db5a0 100644 --- a/lib/galaxy/web/stack/transport.py +++ b/lib/galaxy/web/stack/transport.py @@ -4,28 +4,18 @@ from __future__ import absolute_import import logging import threading -try: - from queue import Empty, Queue -except ImportError: - from Queue import Empty, Queue try: import uwsgi except ImportError: uwsgi = None -from six import string_types - log = logging.getLogger(__name__) class ApplicationStackTransport(object): - shutdown_msg = '__SHUTDOWN__' - - def __init_dispatcher_thread(self): - self.dispatcher_thread = threading.Thread(name=self.__class__.__name__ + ".dispatcher_thread", target=self._dispatch_messages) - #self.dispatcher_thread.daemon = True + SHUTDOWN_MSG = '__SHUTDOWN__' def __init__(self, app, stack, dispatcher=None): """ Pre-fork initialization. @@ -36,93 +26,64 @@ class ApplicationStackTransport(object): self.running = False self.dispatcher = dispatcher self.dispatcher_thread = None - self.__init_dispatcher_thread() def _dispatch_messages(self): pass def start_if_needed(self): # Don't unnecessarily start a thread that we don't need. - if self.can_run and not self.running and not self.dispatcher_thread.is_alive() and self.dispatcher and self.dispatcher.handler_count: + if self.can_run and not self.running and not self.dispatcher_thread and self.dispatcher and self.dispatcher.handler_count: self.running = True + self.dispatcher_thread = threading.Thread(name=self.__class__.__name__ + ".dispatcher_thread", target=self._dispatch_messages) self.dispatcher_thread.start() - log.debug('######## Web stack IPC message dispatcher thread started in mule %s', uwsgi.mule_id()) + log.info('%s dispatcher started', self.__class__.__name__) def stop_if_unneeded(self): - if self.can_run and self.running and self.dispatcher_thread.is_alive() and self.dispatcher and not self.dispatcher.handler_count: + if self.can_run and self.running and self.dispatcher_thread and self.dispatcher and not self.dispatcher.handler_count: self.running = False self.dispatcher_thread.join() - self.__init_dispatcher_thread() + self.dispatcher_thread = None + log.info('%s dispatcher stopped', self.__class__.__name__) def start(self): """ Post-fork initialization. """ self.can_run = True - log.debug('######## start called') self.start_if_needed() def send_message(self, msg, dest): pass def shutdown(self): - log.debug('######## TRANSPORT SHUTDOWN CALLED') self.running = False - if self.dispatcher_thread.is_alive(): - # FIXME - #self.send_message(self.shutdown_msg, 'job-handlers') + if self.dispatcher_thread: self.dispatcher_thread.join() - log.debug('######## Joined dispatcher thread') + self.dispatcher_thread = None class UWSGIFarmMessageTransport(ApplicationStackTransport): """ Communication via uWSGI Mule Farm messages. Communication is unidirectional (workers -> mules). """ - # FIXME: do you need a clear "this is a producer, this is a consumer flag? - #shutdown_msg = '__SHUTDOWN__' # Define any static lock names here, additional locks will be appended for each configured farm's message handler _locks = [] def __initialize_locks(self): num = int(uwsgi.opt.get('locks', 0)) + 1 - farms = self._farms.keys() + farms = self.stack._configured_farms.keys() need = len(farms) - #need = len(filter(lambda x: x.startswith('LOCK_'), dir(self))) if num < need: raise RuntimeError('Need %i uWSGI locks but only %i exist(s): Set `locks = %i` in uWSGI configuration' % (need, num, need - 1)) sys.exit(1) self._locks.extend(map(lambda x: 'RECV_MSG_FARM_' + x, farms)) - # This would be nice, but in my 2.0.15 uWSGI, the uwsgi module has no set_option function. And I don't know if it'd work. + # this would be nice, but in my 2.0.15 uWSGI, the uwsgi module has no set_option function, and I don't know if it'd work even if the function existed as documented #if len(self.lock_map) > 1: # uwsgi.set_option('locks', len(self.lock_map)) # log.debug('Created %s uWSGI locks' % len(self.lock_map)) def __init__(self, app, stack, dispatcher=None): super(UWSGIFarmMessageTransport, self).__init__(app, stack, dispatcher=dispatcher) - self._farms_dict = None - self._mules_list = None - self._is_mule = False - self._msg_queue = Queue() self.__initialize_locks() - @property - def _farms(self): - if self._farms_dict is None: - self._farms_dict = {} - farms = uwsgi.opt.get('farm', []) - farms = [farms] if isinstance(farms, string_types) else farms - for farm in farms: - name, mules = farm.split(':', 1) - self._farms_dict[name] = [int(m) for m in mules.split(',')] - return self._farms_dict - - @property - def _mules(self): - if self._mules_list is None: - self._mules_list = [] - mules = uwsgi.opt.get('mule', []) - self._mules_list = [mules] if isinstance(mules, string_types) or mules is True else mules - return self._mules_list - def __lock(self, name_or_id): try: uwsgi.lock(name_or_id) @@ -136,69 +97,59 @@ class UWSGIFarmMessageTransport(ApplicationStackTransport): uwsgi.unlock(self._locks.index(name_or_id)) def _farm_recv_msg_lock_num(self): - # need one lock per farm... except mules can have multiple farms... ack - return self._locks.index('RECV_MSG_FARM_' + self._farm_name) + return self._locks.index('RECV_MSG_FARM_' + self.stack._farm_name) def _dispatch_messages(self): - # We are going to do this a lot, so cache the lock number + # this could be moved to the base class if locking was abstracted and a get_message method was added + log.info('Application stack message dispatcher thread starting up') + # we are going to do this a lot, so cache the lock number lock = self._farm_recv_msg_lock_num() while self.running: msg = None self.__lock(lock) try: - log.debug('######## Mule %s acquired message receive lock, waiting for new message', uwsgi.mule_id()) + log.debug('Acquired message lock, waiting for new message') msg = uwsgi.farm_get_msg() - log.debug('######## Mule %s received message: %s', uwsgi.mule_id(), msg) - if msg == self.shutdown_msg or msg is None: - # all you need to do is pass here, self.running should already be set False by the signal handler calling the shutdown method defined in the superclass + log.debug('Received message: %s', msg) + if msg == self.SHUTDOWN_MSG: self.running = False - log.debug('######## SHUTTING DOWN %s', uwsgi.mule_id()) + else: + self.dispatcher.dispatch(msg) except: - log.exception( "Exception in mule message handling" ) + log.exception('Exception in mule message handling') finally: self.__unlock(lock) - log.debug('######## Mule %s released lock', uwsgi.mule_id()) - if msg != self.shutdown_msg and msg is not None: - self.dispatcher.dispatch(msg) - log.info('######## Mule %s message handler shutting down', uwsgi.mule_id()) + log.debug('Released lock') + log.info('Application stack message dispatcher thread exiting') + + # TODO: start_if_needed would be called on a web worker by the stack's register_message_handler function if a + # function were registered in a web handler, that should probably be prevented. def start(self): """ Post-fork initialization. + + This is mainly done here for the future possibility that we'll be able to run mules post-fork without exec()ing. In a programmed mule it could be done at __init__ time. """ - # TODO: what happens if workers > 1?? - self._is_mule = uwsgi.mule_id() > 0 - if self._is_mule: + if self.stack._is_mule: if not uwsgi.in_farm(): raise RuntimeError('Mule %s is not in a farm! Set `farm = :%s` in uWSGI configuration' % (uwsgi.mule_id(), - ','.join(map(str, range(1, len(filter(lambda x: x.endswith('galaxy/main.py'), self._mules)) + 1))))) - super(UWSGIFarmMessageTransport, self).start() - log.info('######## Mule transport started, worker id: %s, mule id: %s, farm names: %s, server name: %s', uwsgi.worker_id(), uwsgi.mule_id(), self.stack.farm_names, self.app.config.server_name) - #self._send_all_messages() - - #@property - #def _farm_name(self): - # for name, mules in self._farms.items(): - # if uwsgi.mule_id() in mules: - # return name - # return None - - def _send_all_messages(self): - # the sender doesn't have a running thread, all we are concerned with here is whether or not we've forked yet - if self.can_run and not self._is_mule: - while True: - try: - msg, dest = self._msg_queue.get_nowait() - except Empty: - break - log.debug('######## Sending message in mule %s to farm %s: %s', uwsgi.mule_id(), dest, msg) - uwsgi.farm_msg(dest, msg) - log.debug('######## Message sent') + ','.join(map(str, range(1, len(filter(lambda x: x.endswith('galaxy/main.py'), self.stack._configured_mules)) + 1))))) + elif len(self.stack._farms) > 1: + raise RuntimeError('Mule %s is in multiple farms! This configuration is not supported due to locking issues' % uwsgi.mule_id()) + # only mules receive messages so don't bother starting the dispatcher if we're not a mule (although + # currently it doesn't have any registered handlers and so wouldn't start anyway) + super(UWSGIFarmMessageTransport, self).start() + def shutdown(self): + if not self.stack._is_mule: + for farm in self.stack._configured_farms.keys(): + for mule in self.stack._configured_mules: + # this could possibly generate more than we need, but that's ok + self.send_message(self.SHUTDOWN_MSG, farm) + else: + super(UWSGIFarmMessageTransport, self).shutdown() def send_message(self, msg, dest): - #log.debug('######## Queing message in mule %s to farm %s: %s', uwsgi.mule_id(), dest, msg) - #self._msg_queue.put((msg, dest)) - #self._send_all_messages() - log.debug('######## Sending message to farm %s: %s', dest, msg) + log.debug('Sending message to farm %s: %s', dest, msg) uwsgi.farm_msg(dest, msg) diff --git a/scripts/get_uwsgi_args.py b/scripts/get_uwsgi_args.py index 0e9804ec058..cf9bdd019f2 100644 --- a/scripts/get_uwsgi_args.py +++ b/scripts/get_uwsgi_args.py @@ -12,7 +12,6 @@ from galaxy.config import ( parse_dependency_options, ) from galaxy.script import main_factory -from galaxy.util.configdict import parse_config DESCRIPTION = "Script to determine uWSGI command line arguments." @@ -20,9 +19,6 @@ COMMAND_TEMPLATE = '{virtualenv}--ini-paste {galaxy_ini} --paste-logger --die-on def _get_uwsgi_args(args, kwargs): - # FIXME: hardcoded - stack_conf = parse_config('config/stack_conf.yml', 'stack', {}) - # FIXME: these belong in stack conf handlerct = int(kwargs.get('job_handler_count', 1)) pool_name = kwargs.get('job_handler_pool_name', 'job-handlers') virtualenv = os.environ.get('VIRTUAL_ENV', None)