Merge lp:~openerp-dev/openobject-server/6.1-here-comes-the-bogeyman-vmt into lp:openobject-server/6.1

[Vo Minh Thu]

[lp:~openerp-dev/openobject-server/6.1-here-comes-the-bogeyman-vmt]

4135. By Vo Minh Thu on 2012-04-02

[IMP] cron: use multi-process signaling.

[Christophe Simonis (OpenERP) (kangol)]

1/ What is this "for x in xrange(5):" in _run() ?
2/ why not use existing function for listing the databases ? http://bazaar.launchpad.net/~openerp/openobject-server/6.1/view/head:/openerp/service/web_services.py#L317

review: Needs Fixing

[Vo Minh Thu (thu)]

1. Simply to do 5 times the same thing (which will account for 5 mnutes)
2. I didn't think of it but it wouldn't work: for instance it assumes tools.config['db_name'] is a single database name. Maybe it would be useful to refactor the code but we can't to that in stable.

=== added file 'openerp-cron-worker'
--- openerp-cron-worker	1970-01-01 00:00:00 +0000
+++ openerp-cron-worker	2012-04-02 12:05:25 +0000
@@ -0,0 +1,111 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+OpenERP cron jobs worker
+
+This script executes OpenERP cron jobs. Normally, cron jobs are handled by the
+OpenERP server but depending on deployment needs, independent worker processes
+can be used. This is especially the case when the server is run via Gunicorn.
+
+OpenERP cron jobs worker re-uses openerp-server command-line options but does
+not honor all of them.
+
+Meaningful options include:
+
+  -d, --database  comma-separated list of databases to monitor for cron jobs
+                  processing. If left empty, the worker monitors all databases
+                  (given by `psql -ls`).
+
+  --addons-path   as ususal.
+
+  --cpu-time-limte
+  --virtual-memory-limit
+  --virtual-memory-reset  Those three options have the same meaning the for
+                          the server with Gunicorn. The only catch is: To
+                          not enable rlimits by default, those options are
+                          honored only when --cpu-time-limte is different than
+                          60 (its default value).
+"""
+
+import logging
+import os
+import signal
+import sys
+
+import openerp
+
+# Also use the `openerp` logger for the main script.
+_logger = logging.getLogger('openerp')
+
+# Variable keeping track of the number of calls to the signal handler defined
+# below. This variable is monitored by ``quit_on_signals()``.
+quit_signals_received = 0
+
+# TODO copy/pasted from openerp-server
+def signal_handler(sig, frame):
+    """ Signal handler: exit ungracefully on the second handled signal.
+
+    :param sig: the signal number
+    :param frame: the interrupted stack frame or None
+    """
+    global quit_signals_received
+    quit_signals_received += 1
+    import openerp.addons.base
+    openerp.addons.base.ir.ir_cron.quit_signal_received = True
+    if quit_signals_received == 1 and openerp.addons.base.ir.ir_cron.job_in_progress:
+        _logger.info("Waiting for the current job to complete.")
+        print "Waiting for the current job to complete."
+        print "Hit Ctrl-C again to force shutdown."
+    if quit_signals_received > 1:
+        # logging.shutdown was already called at this point.
+        sys.stderr.write("Forced shutdown.\n")
+        os._exit(0)
+
+# TODO copy/pasted from openerp-server
+def setup_signal_handlers():
+    """ Register the signal handler defined above. """
+    SIGNALS = map(lambda x: getattr(signal, "SIG%s" % x), "INT TERM".split())
+    if os.name == 'posix':
+        map(lambda sig: signal.signal(sig, signal_handler), SIGNALS)
+    elif os.name == 'nt':
+        import win32api
+        win32api.SetConsoleCtrlHandler(lambda sig: signal_handler(sig, None), 1)
+
+def list_databases():
+    import subprocess
+    p1 = subprocess.Popen(["psql", "-lAt"], stdout=subprocess.PIPE)
+    p2 = subprocess.Popen(["cut", "-f", "1", "-d", "|"], stdin=p1.stdout, stdout=subprocess.PIPE)
+    p1.stdout.close()  # Allow p1 to receive a SIGPIPE if p2 exits.
+    output = p2.communicate()[0]
+    databases = output.splitlines()
+    # TODO filter out non-OpenERP databases
+    databases = [d for d in databases if d not in ['template0', 'template1', 'postgres']]
+    databases = [d for d in databases if not d.startswith('postgres')]
+    return databases
+
+if __name__ == '__main__':
+    os.environ['TZ'] = 'UTC'
+    openerp.tools.config.parse_config(sys.argv[1:])
+    config = openerp.tools.config
+    if config['log_handler'] == [':INFO']:
+        # Replace the default value, which is suitable for openerp-server.
+        config['log_handler'].append('openerp.addons.base.ir.ir_cron:DEBUG')
+    setup_signal_handlers()
+    openerp.modules.module.initialize_sys_path()
+    openerp.modules.loading.open_openerp_namespace()
+    openerp.netsvc.init_logger()
+    openerp.cron.enable_schedule_wakeup = False
+    openerp.multi_process = True # enable multi-process signaling
+    import openerp.addons.base
+    print "OpenERP cron jobs worker. Hit Ctrl-C to exit."
+    print "Documentation is available at the top of the `opener-cron-worker` file."
+    if config['db_name']:
+        db_names = config['db_name'].split(',')
+        print "Monitoring %s databases." % len(db_names)
+    else:
+        db_names = list_databases
+        print "Monitored databases are auto-discovered."
+    openerp.addons.base.ir.ir_cron.ir_cron._run(db_names)
+
+# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:
 
=== modified file 'openerp/__init__.py'
--- openerp/__init__.py	2012-02-08 14:28:34 +0000
+++ openerp/__init__.py	2012-04-02 12:05:25 +0000
@@ -27,6 +27,7 @@
 
 import addons
 import conf
+import cron
 import loglevels
 import modules
 import netsvc
 
=== modified file 'openerp/addons/base/ir/ir_cron.py'
--- openerp/addons/base/ir/ir_cron.py	2012-02-02 09:21:05 +0000
+++ openerp/addons/base/ir/ir_cron.py	2012-04-02 12:05:25 +0000
@@ -39,6 +39,14 @@
 
 _logger = logging.getLogger(__name__)
 
+# This variable can be set by a signal handler to stop the infinite loop in
+# ir_cron._run()
+quit_signal_received = False
+
+# This variable can be checked to know if ir_cron._run() is processing a job or
+# sleeping.
+job_in_progress = True
+
 def str2tuple(s):
     return eval('tuple(%s)' % (s or ''))
 
@@ -266,6 +274,153 @@
             cr.commit()
             cr.close()
 
+    def _process_job(self, cr, job):
+        """ Run a given job taking care of the repetition.
+
+        The cursor has a lock on the job (aquired by _acquire_job()).
+
+        :param job: job to be run (as a dictionary).
+        """
+        try:
+            now = datetime.now() 
+            nextcall = datetime.strptime(job['nextcall'], DEFAULT_SERVER_DATETIME_FORMAT)
+            numbercall = job['numbercall']
+
+            ok = False
+            while nextcall < now and numbercall:
+                if numbercall > 0:
+                    numbercall -= 1
+                if not ok or job['doall']:
+                    self._callback(cr, job['user_id'], job['model'], job['function'], job['args'], job['id'])
+                if numbercall:
+                    nextcall += _intervalTypes[job['interval_type']](job['interval_number'])
+                ok = True
+            addsql = ''
+            if not numbercall:
+                addsql = ', active=False'
+            cr.execute("UPDATE ir_cron SET nextcall=%s, numbercall=%s"+addsql+" WHERE id=%s",
+                       (nextcall.strftime(DEFAULT_SERVER_DATETIME_FORMAT), numbercall, job['id']))
+
+        finally:
+            cr.commit()
+            cr.close()
+
+    @classmethod
+    def _acquire_job(cls, db_name):
+        # TODO remove 'check' argument from addons/base_action_rule/base_action_rule.py
+        """ Try to process one cron job.
+
+        This selects in database all the jobs that should be processed. It then
+        tries to lock each of them and, if it succeeds, run the cron job (if it
+        doesn't succeed, it means the job was already locked to be taken care
+        of by another thread) and return.
+
+        If a job was processed, returns True, otherwise returns False.
+        """
+        db = openerp.sql_db.db_connect(db_name)
+        cr = db.cursor()
+        try:
+            # Careful to compare timestamps with 'UTC' - everything is UTC as of v6.1.
+            cr.execute("""SELECT * FROM ir_cron
+                          WHERE numbercall != 0
+                              AND active AND nextcall <= (now() at time zone 'UTC')
+                          ORDER BY priority""")
+            for job in cr.dictfetchall():
+                task_cr = db.cursor()
+                try:
+                    # Try to grab an exclusive lock on the job row from within the task transaction
+                    acquired_lock = False
+                    task_cr.execute("""SELECT *
+                                       FROM ir_cron
+                                       WHERE id=%s
+                                       FOR UPDATE NOWAIT""",
+                                   (job['id'],), log_exceptions=False)
+                    acquired_lock = True
+                except psycopg2.OperationalError, e:
+                    if e.pgcode == '55P03':
+                        # Class 55: Object not in prerequisite state; 55P03: lock_not_available
+                        _logger.debug('Another process/thread is already busy executing job `%s`, skipping it.', job['name'])
+                        continue
+                    else:
+                        # Unexpected OperationalError
+                        raise
+                finally:
+                    if not acquired_lock:
+                        # we're exiting due to an exception while acquiring the lot
+                        task_cr.close()
+
+                # Got the lock on the job row, run its code
+                _logger.debug('Starting job `%s`.', job['name'])
+                openerp.modules.registry.RegistryManager.check_registry_signaling(db_name)
+                registry = openerp.pooler.get_pool(db_name)
+                registry[cls._name]._process_job(task_cr, job)
+                openerp.modules.registry.RegistryManager.signal_caches_change(db_name)
+                return True
+
+        except psycopg2.ProgrammingError, e:
+            if e.pgcode == '42P01':
+                # Class 42 — Syntax Error or Access Rule Violation; 42P01: undefined_table
+                # The table ir_cron does not exist; this is probably not an OpenERP database.
+                _logger.warning('Tried to poll an undefined table on database %s.', db_name)
+            else:
+                raise
+        except Exception, ex:
+            _logger.warning('Exception in cron:', exc_info=True)
+
+        finally:
+            cr.commit()
+            cr.close()
+
+        return False
+
+    @classmethod
+    def _run(cls, db_names):
+        """
+        Class method intended to be run in a dedicated process to handle jobs.
+        This polls the database for jobs that can be run every 60 seconds.
+
+        :param db_names: list of database names to poll or callable to
+            generate such a list.
+        """
+        global quit_signal_received
+        while not quit_signal_received:
+            if callable(db_names):
+                names = db_names()
+            else:
+                names = db_names
+            for x in xrange(5):
+                if quit_signal_received:
+                    return
+                t1 = time.time()
+                for db_name in names:
+                    while True:
+                        # Small hack to re-use the openerp-server config:
+                        # If the cpu_time_limit has not its default value, we
+                        # truly want to establish limits.
+                        if openerp.tools.config['cpu_time_limit'] != 60:
+                            openerp.wsgi.core.pre_request('dummy', 'dummy')
+                        acquired = cls._acquire_job(db_name)
+                        if openerp.tools.config['cpu_time_limit'] != 60:
+                            class W(object):
+                                alive = True
+                            worker = W()
+                            openerp.wsgi.core.post_request(worker, 'dummy', 'dummy')
+                            if not worker.alive:
+                                return
+                        if not acquired:
+                            break
+                        if quit_signal_received:
+                            return
+                t2 = time.time()
+                t = t2 - t1
+                global job_in_progress
+                if t > 60:
+                    _logger.warning('Cron worker: processing all jobs took more than 1 minute to complete (%ss.).', int(t))
+                else:
+                    job_in_progress = False
+                    time.sleep(60 - t)
+                    job_in_progress = True
+
     def update_running_cron(self, cr):
         """ Schedule as soon as possible a wake-up for this database. """
         # Verify whether the server is already started and thus whether we need to commit
 
=== modified file 'openerp/cron.py'
--- openerp/cron.py	2012-02-13 11:53:54 +0000
+++ openerp/cron.py	2012-04-02 12:05:25 +0000
@@ -49,6 +49,13 @@
 
 _logger = logging.getLogger(__name__)
 
+# Scheduling wake-ups (see below) can be disabled when the polling process
+# workers are used instead of the managed thread workers. (I.e. wake-ups are
+# not used since polling is used. And polling is used when the cron are
+# handled by running special processes, e.g. openerp-cron-worker, instead
+# of the general openerp-server script.)
+enable_schedule_wakeup = True
+
 # Heapq of database wake-ups. Note that 'database wake-up' meaning is in
 # the context of the cron management. This is not originally about loading
 # a database, although having the database name in the queue will
@@ -135,7 +142,6 @@
         _wakeups = []
         _wakeup_by_db = {}
 
-
 def schedule_wakeup(timestamp, db_name):
     """ Schedule a new wake-up for a database.
 
@@ -147,6 +153,9 @@
     :param timestamp: when the wake-up is scheduled.
 
     """
+    global enable_schedule_wakeup
+    if not enable_schedule_wakeup:
+        return
     if not timestamp:
         return
     with _wakeups_lock:
 
=== modified file 'openerp/wsgi/core.py'
--- openerp/wsgi/core.py	2012-03-05 16:37:30 +0000
+++ openerp/wsgi/core.py	2012-04-02 12:05:25 +0000
@@ -506,7 +506,7 @@
     rss, vms = psutil.Process(os.getpid()).get_memory_info()
     if vms > config['virtual_memory_reset']:
         _logger.info('Virtual memory consumption '
-            'too high, rebooting the worker.')
+            'too high, killing the worker.')
         worker.alive = False # Commit suicide after the request.
 
 # SIGXCPU (exceeded CPU time) signal handler will raise an exception.