diff --git a/openerp-long-polling b/openerp-long-polling new file mode 100755 index 00000000000..304cd01c1ee --- /dev/null +++ b/openerp-long-polling @@ -0,0 +1,11 @@ +#!/usr/bin/env python + +import gevent.monkey +gevent.monkey.patch_all() +import gevent_psycopg2 +gevent_psycopg2.monkey_patch() + +import openerp + +if __name__ == "__main__": + openerp.cli.main() diff --git a/openerp/__init__.py b/openerp/__init__.py index 111677cd707..e8bb23fd12e 100644 --- a/openerp/__init__.py +++ b/openerp/__init__.py @@ -23,6 +23,14 @@ """ +import sys + +# Is the server running with gevent. +evented = False +if sys.modules.get("gevent") is not None: + evented = True + + # Make sure the OpenERP server runs in UTC. This is especially necessary # under Windows as under Linux it seems the real import of time is # sufficiently deferred so that setting the TZ environment variable @@ -61,8 +69,6 @@ wsgi.register_wsgi_handler = wsgi.wsgi_server.register_wsgi_handler # its own copy of the data structure and we don't need to care about # locks between threads. multi_process = False -# Is the server running with gevent. -evented = False def registry(database_name): """ diff --git a/openerp/cli/server.py b/openerp/cli/server.py index 2ee9e53b3ca..fc1bf61b617 100644 --- a/openerp/cli/server.py +++ b/openerp/cli/server.py @@ -221,6 +221,18 @@ def quit_on_signals(): os.unlink(config['pidfile']) sys.exit(0) +def watch_parent(beat=4): + import gevent + ppid = os.getppid() + while True: + if ppid != os.getppid(): + pid = os.getpid() + _logger.info("LongPolling (%s) Parent changed", pid) + # suicide !! + os.kill(pid, signal.SIGTERM) + return + gevent.sleep(beat) + def main(args): check_root_user() openerp.tools.config.parse_config(args) @@ -257,9 +269,15 @@ def main(args): setup_pid_file() # Some module register themselves when they are loaded so we need the # services to be running before loading any registry. - if config['workers']: - openerp.service.start_services_workers() + if not openerp.evented: + if config['workers']: + openerp.service.start_services_workers() + else: + openerp.service.start_services() else: + config['xmlrpc_port'] = config['longpolling_port'] + import gevent + gevent.spawn(watch_parent) openerp.service.start_services() rc = 0 diff --git a/openerp/service/workers.py b/openerp/service/workers.py index 9a6485b8fd3..f29bd6afb42 100644 --- a/openerp/service/workers.py +++ b/openerp/service/workers.py @@ -14,6 +14,8 @@ import signal import socket import sys import time +import subprocess +import os.path import werkzeug.serving try: @@ -35,7 +37,6 @@ class Multicorn(object): def __init__(self, app): # config self.address = (config['xmlrpc_interface'] or '0.0.0.0', config['xmlrpc_port']) - self.long_polling_address = (config['xmlrpc_interface'] or '0.0.0.0', config['longpolling_port']) self.population = config['workers'] self.timeout = config['limit_time_real'] self.limit_request = config['limit_request'] @@ -46,10 +47,10 @@ class Multicorn(object): self.socket = None self.workers_http = {} self.workers_cron = {} - self.workers_longpolling = {} self.workers = {} self.generation = 0 self.queue = [] + self.long_polling_pid = None def pipe_new(self): pipe = os.pipe() @@ -89,6 +90,14 @@ class Multicorn(object): worker.run() sys.exit(0) + def long_polling_spawn(self): + nargs = [] + sys.argv + cmd = nargs[0] + cmd = os.path.join(os.path.dirname(cmd), "openerp-long-polling") + nargs[0] = cmd + popen = subprocess.Popen(nargs) + self.long_polling_pid = popen.pid + def worker_pop(self, pid): if pid in self.workers: _logger.debug("Worker (%s) unregistered",pid) @@ -143,8 +152,8 @@ class Multicorn(object): self.worker_spawn(WorkerHTTP, self.workers_http) while len(self.workers_cron) < config['max_cron_threads']: self.worker_spawn(WorkerCron, self.workers_cron) - while len(self.workers_longpolling) < 1: - self.worker_spawn(WorkerLongPolling, self.workers_longpolling) + if not self.long_polling_pid: + self.long_polling_spawn() def sleep(self): try: @@ -183,14 +192,11 @@ class Multicorn(object): self.socket.setblocking(0) self.socket.bind(self.address) self.socket.listen(8*self.population) - # long polling socket - self.long_polling_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - self.long_polling_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - self.long_polling_socket.setblocking(0) - self.long_polling_socket.bind(self.long_polling_address) - self.long_polling_socket.listen(8) def stop(self, graceful=True): + if self.long_polling_pid is not None: + self.worker_kill(self.long_polling_pid, signal.SIGTERM) + self.long_polling_pid = None if graceful: _logger.info("Stopping gracefully") limit = time.time() + self.timeout @@ -353,49 +359,8 @@ class WorkerHTTP(Worker): def start(self): Worker.start(self) - self.multi.long_polling_socket.close() self.server = WorkerBaseWSGIServer(self.multi.app) -class WorkerLongPolling(Worker): - """ Long polling workers """ - def __init__(self, multi): - super(WorkerLongPolling, self).__init__(multi) - # Disable the watchdog feature for this kind of worker. - self.watchdog_timeout = None - - def watch_parent(self): - import gevent - while True: - if self.ppid != os.getppid(): - _logger.info("WorkerLongPolling (%s) Parent changed", self.pid) - os.kill(os.getpid(), signal.SIGTERM) - return - gevent.sleep(self.multi.beat) - - def start(self): - openerp.evented = True - _logger.info('Using gevent mode') - import gevent.monkey - gevent.monkey.patch_all() - import gevent_psycopg2 - gevent_psycopg2.monkey_patch() - from openerp.modules.registry import RegistryManager - from gevent.coros import RLock - RegistryManager.registries_lock = RLock() - - Worker.start(self) - self.multi.socket.close() - - import gevent - watcher = gevent.spawn(self.watch_parent) - - log = _logger.getChild(self.__class__.__name__) - log.write = lambda msg: log.info(msg.strip()) - - from gevent.wsgi import WSGIServer - self.server = WSGIServer(self.multi.long_polling_socket, self.multi.app, log=log) - self.server.serve_forever() - class WorkerBaseWSGIServer(werkzeug.serving.BaseWSGIServer): """ werkzeug WSGI Server patched to allow using an external listen socket """ @@ -469,7 +434,6 @@ class WorkerCron(Worker): def start(self): Worker.start(self) self.multi.socket.close() - self.multi.long_polling_socket.close() openerp.service.start_internal() # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: diff --git a/openerp/service/wsgi_server.py b/openerp/service/wsgi_server.py index df201825d31..acbfc67dd0b 100644 --- a/openerp/service/wsgi_server.py +++ b/openerp/service/wsgi_server.py @@ -449,7 +449,10 @@ def start_service(): interface = config['xmlrpc_interface'] or '0.0.0.0' port = config['xmlrpc_port'] _logger.info('HTTP service (werkzeug) running on %s:%s', interface, port) - threading.Thread(target=serve, args=(interface, port, True)).start() + if not openerp.evented: + threading.Thread(target=serve, args=(interface, port, True)).start() + else: + serve(interface, port, True) def stop_service(): """ Initiate the shutdown of the WSGI server.