From 1b0bfd1044dc00fcba9ea859cd72259b85e1b11a Mon Sep 17 00:00:00 2001 From: Vo Minh Thu Date: Wed, 28 Mar 2012 11:03:20 +0200 Subject: [PATCH 1/8] [IMP] cron: alternative implementation: - The previous implementation was optimized to make few queries to the database but needed to keep some internal state. - That state was updated whenever the ir_cron table was modified by the ORM (this works only when the cron and web processes/threads are inside a single OpenERP server instance). - The new implementation is instead polling the database. - This is deemed acceptable in `normal` situation (i.e. not a SaaS with thousand of databases). - This makes it possible to avoid sharing state or the use of IPC. - This makes it possible to add/remove additional worker processes, possibly on different machines. - The code of the older implementation is removed in this commit but will be added back in a later commit: this is the 6.1 stable branch and we don't want to change the existing installation, but simply provide a solution for those running OpenERP with Gunicorn (which uses processes for which no cron state were shared). bzr revid: vmt@openerp.com-20120328090320-vshsfv3gt1ck34s1 --- openerp-cron-worker | 22 +++++ openerp-server | 3 - openerp/addons/base/ir/ir_cron.py | 98 ++++++------------ openerp/cron.py | 158 ------------------------------ openerp/modules/registry.py | 15 --- openerp/service/__init__.py | 9 +- 6 files changed, 52 insertions(+), 253 deletions(-) create mode 100755 openerp-cron-worker diff --git a/openerp-cron-worker b/openerp-cron-worker new file mode 100755 index 00000000000..4f86da928a5 --- /dev/null +++ b/openerp-cron-worker @@ -0,0 +1,22 @@ +#!/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. +""" + +import openerp + +if __name__ == '__main__': + openerp.modules.module.initialize_sys_path() + openerp.modules.loading.open_openerp_namespace() + openerp.tools.config['log_handler'] = ['openerp.addons.base.ir.ir_cron:DEBUG'] + openerp.netsvc.init_logger() + import openerp.addons.base + openerp.addons.base.ir.ir_cron.ir_cron._run(['xx']) + +# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: diff --git a/openerp-server b/openerp-server index f574ebb354f..19c60397a08 100755 --- a/openerp-server +++ b/openerp-server @@ -93,9 +93,6 @@ def preload_registry(dbname): """ Preload a registry, and start the cron.""" try: db, registry = openerp.pooler.get_db_and_pool(dbname, update_module=config['init'] or config['update'], pooljobs=False) - - # jobs will start to be processed later, when openerp.cron.start_master_thread() is called by openerp.service.start_services() - registry.schedule_cron_jobs() except Exception: _logger.exception('Failed to initialize database `%s`.', dbname) diff --git a/openerp/addons/base/ir/ir_cron.py b/openerp/addons/base/ir/ir_cron.py index 3c437c93282..2514116cf19 100644 --- a/openerp/addons/base/ir/ir_cron.py +++ b/openerp/addons/base/ir/ir_cron.py @@ -31,7 +31,6 @@ import netsvc import openerp import pooler import tools -from openerp.cron import WAKE_UP_NOW from osv import fields, osv from tools import DEFAULT_SERVER_DATETIME_FORMAT from tools.safe_eval import safe_eval as eval @@ -142,17 +141,15 @@ class ir_cron(osv.osv): except Exception, e: self._handle_callback_exception(cr, uid, model_name, method_name, args, job_id, e) - def _run_job(self, cr, job, now): + def _run_job(self, cr, job): """ Run a given job taking care of the repetition. - The cursor has a lock on the job (aquired by _run_jobs_multithread()) and this - method is run in a worker thread (spawned by _run_jobs_multithread())). + The cursor has a lock on the job (aquired by _acquire_job()). :param job: job to be run (as a dictionary). - :param now: timestamp (result of datetime.now(), no need to call it multiple time). - """ try: + now = datetime.utcnow() nextcall = datetime.strptime(job['nextcall'], DEFAULT_SERVER_DATETIME_FORMAT) numbercall = job['numbercall'] @@ -171,45 +168,29 @@ class ir_cron(osv.osv): cr.execute("UPDATE ir_cron SET nextcall=%s, numbercall=%s"+addsql+" WHERE id=%s", (nextcall.strftime(DEFAULT_SERVER_DATETIME_FORMAT), numbercall, job['id'])) - if numbercall: - # Reschedule our own main cron thread if necessary. - # This is really needed if this job runs longer than its rescheduling period. - nextcall = calendar.timegm(nextcall.timetuple()) - openerp.cron.schedule_wakeup(nextcall, cr.dbname) finally: cr.commit() cr.close() - openerp.cron.release_thread_slot() - def _run_jobs_multithread(self): + @classmethod + def _acquire_job(cls, db_name): # TODO remove 'check' argument from addons/base_action_rule/base_action_rule.py - """ Process the cron jobs by spawning worker threads. + """ 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, spawns a thread to run + 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). - - The cursor used to lock the job in database is given to the worker - thread (which has to close it itself). - + locked to be taken care of by another thread) and stop. """ - db = self.pool.db + db = openerp.sql_db.db_connect(db_name) cr = db.cursor() - db_name = db.dbname try: - jobs = {} # mapping job ids to jobs for all jobs being processed. - now = datetime.now() # 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(): - if not openerp.cron.get_thread_slots(): - break - jobs[job['id']] = job - task_cr = db.cursor() try: # Try to grab an exclusive lock on the job row from within the task transaction @@ -233,31 +214,11 @@ class ir_cron(osv.osv): # we're exiting due to an exception while acquiring the lot task_cr.close() - # Got the lock on the job row, now spawn a thread to execute it in the transaction with the lock - task_thread = threading.Thread(target=self._run_job, name=job['name'], args=(task_cr, job, now)) - # force non-daemon task threads (the runner thread must be daemon, and this property is inherited by default) - task_thread.setDaemon(False) - openerp.cron.take_thread_slot() - task_thread.start() - _logger.debug('Cron execution thread for job `%s` spawned', job['name']) - - # Find next earliest job ignoring currently processed jobs (by this and other cron threads) - find_next_time_query = """SELECT min(nextcall) AS min_next_call - FROM ir_cron WHERE numbercall != 0 AND active""" - if jobs: - cr.execute(find_next_time_query + " AND id NOT IN %s", (tuple(jobs.keys()),)) - else: - cr.execute(find_next_time_query) - next_call = cr.dictfetchone()['min_next_call'] - - if next_call: - next_call = calendar.timegm(time.strptime(next_call, DEFAULT_SERVER_DATETIME_FORMAT)) - else: - # no matching cron job found in database, re-schedule arbitrarily in 1 day, - # this delay will likely be modified when running jobs complete their tasks - next_call = time.time() + (24*3600) - - openerp.cron.schedule_wakeup(next_call, db_name) + # Got the lock on the job row, run its code + _logger.debug('Starting job `%s`.', job['name']) + registry = openerp.pooler.get_pool(db_name) + registry[cls._name]._run_job(task_cr, job) + return True except Exception, ex: _logger.warning('Exception in cron:', exc_info=True) @@ -266,18 +227,21 @@ class ir_cron(osv.osv): cr.commit() cr.close() - 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 - # immediately our changes and restart the cron agent in order to apply the change - # immediately. The commit() is needed because as soon as the cron is (re)started it - # will query the database with its own cursor, possibly before the end of the - # current transaction. - # This commit() is not an issue in most cases, but we must absolutely avoid it - # when the server is only starting or loading modules (hence the test on pool._init). - if not self.pool._init: - cr.commit() - openerp.cron.schedule_wakeup(WAKE_UP_NOW, self.pool.db.dbname) + return False + + @classmethod + def _run(cls, db_names): + while True: + t1 = time.time() + for db_name in db_names: + while(cls._acquire_job(db_name)): + pass + t2 = time.time() + t = t2 - t1 + if t > 60: + _logger.warning('Cron worker: processing all jobs took more than 1 minute to complete (%ss.).', int(t)) + else: + time.sleep(60 - t) def _try_lock(self, cr, uid, ids, context=None): """Try to grab a dummy exclusive write-lock to the rows with the given ids, @@ -294,20 +258,16 @@ class ir_cron(osv.osv): def create(self, cr, uid, vals, context=None): res = super(ir_cron, self).create(cr, uid, vals, context=context) - self.update_running_cron(cr) return res def write(self, cr, uid, ids, vals, context=None): self._try_lock(cr, uid, ids, context) res = super(ir_cron, self).write(cr, uid, ids, vals, context=context) - self.update_running_cron(cr) return res def unlink(self, cr, uid, ids, context=None): self._try_lock(cr, uid, ids, context) res = super(ir_cron, self).unlink(cr, uid, ids, context=context) - self.update_running_cron(cr) return res -ir_cron() # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: diff --git a/openerp/cron.py b/openerp/cron.py index 8551ed7dadc..5c6b860865b 100644 --- a/openerp/cron.py +++ b/openerp/cron.py @@ -49,167 +49,9 @@ import tools _logger = logging.getLogger(__name__) -# 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 -# cause it to be loaded when the schedule time is reached, even if it was -# unloaded in the mean time. Normally a database's wake-up is cancelled by -# the RegistryManager when the database is unloaded - so this should not -# cause it to be reloaded. -# # TODO: perhaps in the future we could consider a flag on ir.cron jobs # that would cause database wake-up even if the database has not been # loaded yet or was already unloaded (e.g. 'force_db_wakeup' or something) -# -# Each element is a triple (timestamp, database-name, boolean). The boolean -# specifies if the wake-up is canceled (so a wake-up can be canceled without -# relying on the heapq implementation detail; no need to remove the job from -# the heapq). -_wakeups = [] -# Mapping of database names to the wake-up defined in the heapq, -# so that we can cancel the wake-up without messing with the heapq -# invariant: lookup the wake-up by database-name, then set -# its third element to True. -_wakeup_by_db = {} - -# Re-entrant lock to protect the above _wakeups and _wakeup_by_db variables. -# We could use a simple (non-reentrant) lock if the runner function below -# was more fine-grained, but we are fine with the loop owning the lock -# while spawning a few threads. -_wakeups_lock = threading.RLock() - -# Maximum number of threads allowed to process cron jobs concurrently. This -# variable is set by start_master_thread using openerp.conf.max_cron_threads. -_thread_slots = None - -# A (non re-entrant) lock to protect the above _thread_slots variable. -_thread_slots_lock = threading.Lock() - -# Sleep duration limits - must not loop too quickly, but can't sleep too long -# either, because a new job might be inserted in ir_cron with a much sooner -# execution date than current known ones. We won't see it until we wake! -MAX_SLEEP = 60 # 1 min -MIN_SLEEP = 1 # 1 sec - -# Dummy wake-up timestamp that can be used to force a database wake-up asap -WAKE_UP_NOW = 1 - -def get_thread_slots(): - """ Return the number of available thread slots """ - return _thread_slots - - -def release_thread_slot(): - """ Increment the number of available thread slots """ - global _thread_slots - with _thread_slots_lock: - _thread_slots += 1 - - -def take_thread_slot(): - """ Decrement the number of available thread slots """ - global _thread_slots - with _thread_slots_lock: - _thread_slots -= 1 - - -def cancel(db_name): - """ Cancel the next wake-up of a given database, if any. - - :param db_name: database name for which the wake-up is canceled. - - """ - _logger.debug("Cancel next wake-up for database '%s'.", db_name) - with _wakeups_lock: - if db_name in _wakeup_by_db: - _wakeup_by_db[db_name][2] = True - - -def cancel_all(): - """ Cancel all database wake-ups. """ - _logger.debug("Cancel all database wake-ups") - global _wakeups - global _wakeup_by_db - with _wakeups_lock: - _wakeups = [] - _wakeup_by_db = {} - - -def schedule_wakeup(timestamp, db_name): - """ Schedule a new wake-up for a database. - - If an earlier wake-up is already defined, the new wake-up is discarded. - If another wake-up is defined, that wake-up is discarded and the new one - is scheduled. - - :param db_name: database name for which a new wake-up is scheduled. - :param timestamp: when the wake-up is scheduled. - - """ - if not timestamp: - return - with _wakeups_lock: - if db_name in _wakeup_by_db: - task = _wakeup_by_db[db_name] - if not task[2] and timestamp > task[0]: - # existing wakeup is valid and occurs earlier than new one - return - task[2] = True # cancel existing task - task = [timestamp, db_name, False] - heapq.heappush(_wakeups, task) - _wakeup_by_db[db_name] = task - _logger.debug("Wake-up scheduled for database '%s' @ %s", db_name, - 'NOW' if timestamp == WAKE_UP_NOW else timestamp) - -def runner(): - """Neverending function (intended to be run in a dedicated thread) that - checks every 60 seconds the next database wake-up. TODO: make configurable - """ - while True: - runner_body() - -def runner_body(): - with _wakeups_lock: - while _wakeups and _wakeups[0][0] < time.time() and get_thread_slots(): - task = heapq.heappop(_wakeups) - timestamp, db_name, canceled = task - if canceled: - continue - del _wakeup_by_db[db_name] - registry = openerp.pooler.get_pool(db_name) - if not registry._init: - _logger.debug("Database '%s' wake-up! Firing multi-threaded cron job processing", db_name) - registry['ir.cron']._run_jobs_multithread() - amount = MAX_SLEEP - with _wakeups_lock: - # Sleep less than MAX_SLEEP if the next known wake-up will happen before that. - if _wakeups and get_thread_slots(): - amount = min(MAX_SLEEP, max(MIN_SLEEP, _wakeups[0][0] - time.time())) - _logger.debug("Going to sleep for %ss", amount) - time.sleep(amount) - -def start_master_thread(): - """ Start the above runner function in a daemon thread. - - The thread is a typical daemon thread: it will never quit and must be - terminated when the main process exits - with no consequence (the processing - threads it spawns are not marked daemon). - - """ - global _thread_slots - _thread_slots = openerp.conf.max_cron_threads - db_maxconn = tools.config['db_maxconn'] - if _thread_slots >= tools.config.get('db_maxconn', 64): - _logger.warning("Connection pool size (%s) is set lower than max number of cron threads (%s), " - "this may cause trouble if you reach that number of parallel cron tasks.", - db_maxconn, _thread_slots) - if _thread_slots: - t = threading.Thread(target=runner, name="openerp.cron.master_thread") - t.setDaemon(True) - t.start() - _logger.debug("Master cron daemon started!") - else: - _logger.info("No master cron daemon (0 workers needed).") # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: diff --git a/openerp/modules/registry.py b/openerp/modules/registry.py index 9ec3e6f456a..5ed8ae3dbdf 100644 --- a/openerp/modules/registry.py +++ b/openerp/modules/registry.py @@ -27,7 +27,6 @@ import threading import openerp.sql_db import openerp.osv.orm -import openerp.cron import openerp.tools import openerp.modules.db import openerp.tools.config @@ -110,15 +109,6 @@ class Registry(object): return res - def schedule_cron_jobs(self): - """ Make the cron thread care about this registry/database jobs. - This will initiate the cron thread to check for any pending jobs for - this registry/database as soon as possible. Then it will continuously - monitor the ir.cron model for future jobs. See openerp.cron for - details. - """ - openerp.cron.schedule_wakeup(openerp.cron.WAKE_UP_NOW, self.db.dbname) - def clear_caches(self): """ Clear the caches This clears the caches associated to methods decorated with @@ -213,9 +203,6 @@ class RegistryManager(object): finally: cr.close() - if pooljobs: - registry.schedule_cron_jobs() - return registry @classmethod @@ -233,8 +220,6 @@ class RegistryManager(object): if db_name in cls.registries: cls.registries[db_name].clear_caches() del cls.registries[db_name] - openerp.cron.cancel(db_name) - @classmethod def delete_all(cls): diff --git a/openerp/service/__init__.py b/openerp/service/__init__.py index c387d272e26..41e7200de85 100644 --- a/openerp/service/__init__.py +++ b/openerp/service/__init__.py @@ -28,7 +28,6 @@ import netrpc_server import web_services import websrv_lib -import openerp.cron import openerp.modules import openerp.netsvc import openerp.osv @@ -51,7 +50,7 @@ _logger = logging.getLogger(__name__) def start_services(): """ Start all services. - Services include the different servers and cron threads. + Services include the different servers. """ # Instantiate local services (this is a legacy design). @@ -64,9 +63,6 @@ def start_services(): #http_server.init_static_http() netrpc_server.init_servers() - # Start the main cron thread. - openerp.cron.start_master_thread() - # Start the top-level servers threads (normally HTTP, HTTPS, and NETRPC). openerp.netsvc.Server.startAll() @@ -76,9 +72,6 @@ def start_services(): def stop_services(): """ Stop all services. """ - # stop scheduling new jobs; we will have to wait for the jobs to complete below - openerp.cron.cancel_all() - openerp.netsvc.Server.quitAll() openerp.wsgi.core.stop_server() config = openerp.tools.config From 18de40b351a008dbf8af04778636bd62ef7bfd01 Mon Sep 17 00:00:00 2001 From: Vo Minh Thu Date: Wed, 28 Mar 2012 13:51:09 +0200 Subject: [PATCH 2/8] [IMP] cron: added signal handler for Ctrl-C. bzr revid: vmt@openerp.com-20120328115109-9offjmjqzq6w9k6q --- openerp-cron-worker | 44 +++++++++++++++++++++++++++++++ openerp/addons/base/ir/ir_cron.py | 17 ++++++++++-- 2 files changed, 59 insertions(+), 2 deletions(-) diff --git a/openerp-cron-worker b/openerp-cron-worker index 4f86da928a5..12d58f3fec9 100755 --- a/openerp-cron-worker +++ b/openerp-cron-worker @@ -9,14 +9,58 @@ OpenERP server but depending on deployment needs, independent worker processes can be used. """ +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) + if __name__ == '__main__': + setup_signal_handlers() openerp.modules.module.initialize_sys_path() openerp.modules.loading.open_openerp_namespace() openerp.tools.config['log_handler'] = ['openerp.addons.base.ir.ir_cron:DEBUG'] openerp.netsvc.init_logger() import openerp.addons.base + print "OpenERP cron jobs worker. Hit Ctrl-C to exit." openerp.addons.base.ir.ir_cron.ir_cron._run(['xx']) # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: diff --git a/openerp/addons/base/ir/ir_cron.py b/openerp/addons/base/ir/ir_cron.py index 2514116cf19..0147e6b0bdc 100644 --- a/openerp/addons/base/ir/ir_cron.py +++ b/openerp/addons/base/ir/ir_cron.py @@ -38,6 +38,14 @@ from tools.translate import _ _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 '')) @@ -231,17 +239,22 @@ class ir_cron(osv.osv): @classmethod def _run(cls, db_names): - while True: + global quit_signal_received + while not quit_signal_received: t1 = time.time() for db_name in db_names: while(cls._acquire_job(db_name)): - pass + 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 _try_lock(self, cr, uid, ids, context=None): """Try to grab a dummy exclusive write-lock to the rows with the given ids, From 006159c38b1313833bd4d2c959b5e80c8598a933 Mon Sep 17 00:00:00 2001 From: Vo Minh Thu Date: Wed, 28 Mar 2012 15:13:12 +0200 Subject: [PATCH 3/8] [IMP] cron: added comments. bzr revid: vmt@openerp.com-20120328131312-520xm4f4hd3q887e --- openerp/addons/base/ir/ir_cron.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/openerp/addons/base/ir/ir_cron.py b/openerp/addons/base/ir/ir_cron.py index 0147e6b0bdc..357debb919c 100644 --- a/openerp/addons/base/ir/ir_cron.py +++ b/openerp/addons/base/ir/ir_cron.py @@ -186,9 +186,11 @@ class ir_cron(osv.osv): """ 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 stop. + 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() @@ -239,6 +241,10 @@ class ir_cron(osv.osv): @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. + """ global quit_signal_received while not quit_signal_received: t1 = time.time() From 7a901e8216b37c1f791e66c62288a58f0d9f4c1d Mon Sep 17 00:00:00 2001 From: Vo Minh Thu Date: Thu, 29 Mar 2012 11:18:46 +0200 Subject: [PATCH 4/8] [IMP] cron: the db_names to monitor can be a callable. bzr revid: vmt@openerp.com-20120329091846-6t1ofh57d6ldlwky --- openerp-cron-worker | 26 +++++++++++++++++-- openerp/addons/base/ir/ir_cron.py | 43 +++++++++++++++++++++---------- 2 files changed, 54 insertions(+), 15 deletions(-) diff --git a/openerp-cron-worker b/openerp-cron-worker index 12d58f3fec9..5db5c319da4 100755 --- a/openerp-cron-worker +++ b/openerp-cron-worker @@ -53,14 +53,36 @@ def setup_signal_handlers(): 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__': + 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') + 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." setup_signal_handlers() openerp.modules.module.initialize_sys_path() openerp.modules.loading.open_openerp_namespace() - openerp.tools.config['log_handler'] = ['openerp.addons.base.ir.ir_cron:DEBUG'] openerp.netsvc.init_logger() import openerp.addons.base print "OpenERP cron jobs worker. Hit Ctrl-C to exit." - openerp.addons.base.ir.ir_cron.ir_cron._run(['xx']) + openerp.addons.base.ir.ir_cron.ir_cron._run(db_names) # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: diff --git a/openerp/addons/base/ir/ir_cron.py b/openerp/addons/base/ir/ir_cron.py index 357debb919c..a2cee2944b7 100644 --- a/openerp/addons/base/ir/ir_cron.py +++ b/openerp/addons/base/ir/ir_cron.py @@ -230,6 +230,13 @@ class ir_cron(osv.osv): registry[cls._name]._run_job(task_cr, job) 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) @@ -244,23 +251,33 @@ class ir_cron(osv.osv): """ 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: - t1 = time.time() - for db_name in db_names: - while(cls._acquire_job(db_name)): - 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)) + if callable(db_names): + names = db_names() else: - job_in_progress = False - time.sleep(60 - t) - job_in_progress = True + names = db_names + for x in xrange(5): + if quit_signal_received: + return + t1 = time.time() + for db_name in names: + while(cls._acquire_job(db_name)): + 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 _try_lock(self, cr, uid, ids, context=None): """Try to grab a dummy exclusive write-lock to the rows with the given ids, From c3503add6eec8b2e2ae8a93a655376aece1e19f0 Mon Sep 17 00:00:00 2001 From: Vo Minh Thu Date: Thu, 29 Mar 2012 13:55:31 +0200 Subject: [PATCH 5/8] [IMP] cron: re-added the threaded cron: - the new process workers implementation is present along the old one - new code impacts minimally the old one: just one flag is added to schedule_wakeup(). bzr revid: vmt@openerp.com-20120329115531-9mbh4mtvktyyi9bv --- openerp-cron-worker | 14 +-- openerp/__init__.py | 1 + openerp/addons/base/ir/ir_cron.py | 148 +++++++++++++++++++++++++- openerp/cron.py | 167 ++++++++++++++++++++++++++++++ openerp/modules/registry.py | 15 +++ openerp/service/__init__.py | 9 +- 6 files changed, 344 insertions(+), 10 deletions(-) diff --git a/openerp-cron-worker b/openerp-cron-worker index 5db5c319da4..a3b2ab0ac55 100755 --- a/openerp-cron-worker +++ b/openerp-cron-worker @@ -66,23 +66,25 @@ def list_databases(): 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 + import openerp.addons.base + print "OpenERP cron jobs worker. Hit Ctrl-C to exit." 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." - setup_signal_handlers() - openerp.modules.module.initialize_sys_path() - openerp.modules.loading.open_openerp_namespace() - openerp.netsvc.init_logger() - import openerp.addons.base - print "OpenERP cron jobs worker. Hit Ctrl-C to exit." openerp.addons.base.ir.ir_cron.ir_cron._run(db_names) # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: diff --git a/openerp/__init__.py b/openerp/__init__.py index 521a9ef2dd8..cc355664abc 100644 --- a/openerp/__init__.py +++ b/openerp/__init__.py @@ -27,6 +27,7 @@ SUPERUSER_ID = 1 import addons import conf +import cron import loglevels import modules import netsvc diff --git a/openerp/addons/base/ir/ir_cron.py b/openerp/addons/base/ir/ir_cron.py index a2cee2944b7..cf54a904844 100644 --- a/openerp/addons/base/ir/ir_cron.py +++ b/openerp/addons/base/ir/ir_cron.py @@ -31,6 +31,7 @@ import netsvc import openerp import pooler import tools +from openerp.cron import WAKE_UP_NOW from osv import fields, osv from tools import DEFAULT_SERVER_DATETIME_FORMAT from tools.safe_eval import safe_eval as eval @@ -149,7 +150,131 @@ class ir_cron(osv.osv): except Exception, e: self._handle_callback_exception(cr, uid, model_name, method_name, args, job_id, e) - def _run_job(self, cr, job): + def _run_job(self, cr, job, now): + """ Run a given job taking care of the repetition. + + The cursor has a lock on the job (aquired by _run_jobs_multithread()) and this + method is run in a worker thread (spawned by _run_jobs_multithread())). + + :param job: job to be run (as a dictionary). + :param now: timestamp (result of datetime.now(), no need to call it multiple time). + + """ + try: + 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'])) + + if numbercall: + # Reschedule our own main cron thread if necessary. + # This is really needed if this job runs longer than its rescheduling period. + nextcall = calendar.timegm(nextcall.timetuple()) + openerp.cron.schedule_wakeup(nextcall, cr.dbname) + finally: + cr.commit() + cr.close() + openerp.cron.release_thread_slot() + + def _run_jobs_multithread(self): + # TODO remove 'check' argument from addons/base_action_rule/base_action_rule.py + """ Process the cron jobs by spawning worker threads. + + This selects in database all the jobs that should be processed. It then + tries to lock each of them and, if it succeeds, spawns a thread to run + the cron job (if it doesn't succeed, it means the job was already + locked to be taken care of by another thread). + + The cursor used to lock the job in database is given to the worker + thread (which has to close it itself). + + """ + db = self.pool.db + cr = db.cursor() + db_name = db.dbname + try: + jobs = {} # mapping job ids to jobs for all jobs being processed. + now = datetime.now() + # 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(): + if not openerp.cron.get_thread_slots(): + break + jobs[job['id']] = job + + 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, now spawn a thread to execute it in the transaction with the lock + task_thread = threading.Thread(target=self._run_job, name=job['name'], args=(task_cr, job, now)) + # force non-daemon task threads (the runner thread must be daemon, and this property is inherited by default) + task_thread.setDaemon(False) + openerp.cron.take_thread_slot() + task_thread.start() + _logger.debug('Cron execution thread for job `%s` spawned', job['name']) + + # Find next earliest job ignoring currently processed jobs (by this and other cron threads) + find_next_time_query = """SELECT min(nextcall) AS min_next_call + FROM ir_cron WHERE numbercall != 0 AND active""" + if jobs: + cr.execute(find_next_time_query + " AND id NOT IN %s", (tuple(jobs.keys()),)) + else: + cr.execute(find_next_time_query) + next_call = cr.dictfetchone()['min_next_call'] + + if next_call: + next_call = calendar.timegm(time.strptime(next_call, DEFAULT_SERVER_DATETIME_FORMAT)) + else: + # no matching cron job found in database, re-schedule arbitrarily in 1 day, + # this delay will likely be modified when running jobs complete their tasks + next_call = time.time() + (24*3600) + + openerp.cron.schedule_wakeup(next_call, db_name) + + except Exception, ex: + _logger.warning('Exception in cron:', exc_info=True) + + finally: + 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()). @@ -157,7 +282,7 @@ class ir_cron(osv.osv): :param job: job to be run (as a dictionary). """ try: - now = datetime.utcnow() + now = datetime.now() nextcall = datetime.strptime(job['nextcall'], DEFAULT_SERVER_DATETIME_FORMAT) numbercall = job['numbercall'] @@ -227,7 +352,7 @@ class ir_cron(osv.osv): # Got the lock on the job row, run its code _logger.debug('Starting job `%s`.', job['name']) registry = openerp.pooler.get_pool(db_name) - registry[cls._name]._run_job(task_cr, job) + registry[cls._name]._process_job(task_cr, job) return True except psycopg2.ProgrammingError, e: @@ -279,6 +404,19 @@ class ir_cron(osv.osv): 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 + # immediately our changes and restart the cron agent in order to apply the change + # immediately. The commit() is needed because as soon as the cron is (re)started it + # will query the database with its own cursor, possibly before the end of the + # current transaction. + # This commit() is not an issue in most cases, but we must absolutely avoid it + # when the server is only starting or loading modules (hence the test on pool._init). + if not self.pool._init: + cr.commit() + openerp.cron.schedule_wakeup(WAKE_UP_NOW, self.pool.db.dbname) + def _try_lock(self, cr, uid, ids, context=None): """Try to grab a dummy exclusive write-lock to the rows with the given ids, to make sure a following write() or unlink() will not block due @@ -294,16 +432,20 @@ class ir_cron(osv.osv): def create(self, cr, uid, vals, context=None): res = super(ir_cron, self).create(cr, uid, vals, context=context) + self.update_running_cron(cr) return res def write(self, cr, uid, ids, vals, context=None): self._try_lock(cr, uid, ids, context) res = super(ir_cron, self).write(cr, uid, ids, vals, context=context) + self.update_running_cron(cr) return res def unlink(self, cr, uid, ids, context=None): self._try_lock(cr, uid, ids, context) res = super(ir_cron, self).unlink(cr, uid, ids, context=context) + self.update_running_cron(cr) return res +ir_cron() # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: diff --git a/openerp/cron.py b/openerp/cron.py index 5c6b860865b..317d69522a2 100644 --- a/openerp/cron.py +++ b/openerp/cron.py @@ -49,9 +49,176 @@ import tools _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 +# cause it to be loaded when the schedule time is reached, even if it was +# unloaded in the mean time. Normally a database's wake-up is cancelled by +# the RegistryManager when the database is unloaded - so this should not +# cause it to be reloaded. +# # TODO: perhaps in the future we could consider a flag on ir.cron jobs # that would cause database wake-up even if the database has not been # loaded yet or was already unloaded (e.g. 'force_db_wakeup' or something) +# +# Each element is a triple (timestamp, database-name, boolean). The boolean +# specifies if the wake-up is canceled (so a wake-up can be canceled without +# relying on the heapq implementation detail; no need to remove the job from +# the heapq). +_wakeups = [] +# Mapping of database names to the wake-up defined in the heapq, +# so that we can cancel the wake-up without messing with the heapq +# invariant: lookup the wake-up by database-name, then set +# its third element to True. +_wakeup_by_db = {} + +# Re-entrant lock to protect the above _wakeups and _wakeup_by_db variables. +# We could use a simple (non-reentrant) lock if the runner function below +# was more fine-grained, but we are fine with the loop owning the lock +# while spawning a few threads. +_wakeups_lock = threading.RLock() + +# Maximum number of threads allowed to process cron jobs concurrently. This +# variable is set by start_master_thread using openerp.conf.max_cron_threads. +_thread_slots = None + +# A (non re-entrant) lock to protect the above _thread_slots variable. +_thread_slots_lock = threading.Lock() + +# Sleep duration limits - must not loop too quickly, but can't sleep too long +# either, because a new job might be inserted in ir_cron with a much sooner +# execution date than current known ones. We won't see it until we wake! +MAX_SLEEP = 60 # 1 min +MIN_SLEEP = 1 # 1 sec + +# Dummy wake-up timestamp that can be used to force a database wake-up asap +WAKE_UP_NOW = 1 + +def get_thread_slots(): + """ Return the number of available thread slots """ + return _thread_slots + + +def release_thread_slot(): + """ Increment the number of available thread slots """ + global _thread_slots + with _thread_slots_lock: + _thread_slots += 1 + + +def take_thread_slot(): + """ Decrement the number of available thread slots """ + global _thread_slots + with _thread_slots_lock: + _thread_slots -= 1 + + +def cancel(db_name): + """ Cancel the next wake-up of a given database, if any. + + :param db_name: database name for which the wake-up is canceled. + + """ + _logger.debug("Cancel next wake-up for database '%s'.", db_name) + with _wakeups_lock: + if db_name in _wakeup_by_db: + _wakeup_by_db[db_name][2] = True + + +def cancel_all(): + """ Cancel all database wake-ups. """ + _logger.debug("Cancel all database wake-ups") + global _wakeups + global _wakeup_by_db + with _wakeups_lock: + _wakeups = [] + _wakeup_by_db = {} + +def schedule_wakeup(timestamp, db_name): + """ Schedule a new wake-up for a database. + + If an earlier wake-up is already defined, the new wake-up is discarded. + If another wake-up is defined, that wake-up is discarded and the new one + is scheduled. + + :param db_name: database name for which a new wake-up is scheduled. + :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: + if db_name in _wakeup_by_db: + task = _wakeup_by_db[db_name] + if not task[2] and timestamp > task[0]: + # existing wakeup is valid and occurs earlier than new one + return + task[2] = True # cancel existing task + task = [timestamp, db_name, False] + heapq.heappush(_wakeups, task) + _wakeup_by_db[db_name] = task + _logger.debug("Wake-up scheduled for database '%s' @ %s", db_name, + 'NOW' if timestamp == WAKE_UP_NOW else timestamp) + +def runner(): + """Neverending function (intended to be run in a dedicated thread) that + checks every 60 seconds the next database wake-up. TODO: make configurable + """ + while True: + runner_body() + +def runner_body(): + with _wakeups_lock: + while _wakeups and _wakeups[0][0] < time.time() and get_thread_slots(): + task = heapq.heappop(_wakeups) + timestamp, db_name, canceled = task + if canceled: + continue + del _wakeup_by_db[db_name] + registry = openerp.pooler.get_pool(db_name) + if not registry._init: + _logger.debug("Database '%s' wake-up! Firing multi-threaded cron job processing", db_name) + registry['ir.cron']._run_jobs_multithread() + amount = MAX_SLEEP + with _wakeups_lock: + # Sleep less than MAX_SLEEP if the next known wake-up will happen before that. + if _wakeups and get_thread_slots(): + amount = min(MAX_SLEEP, max(MIN_SLEEP, _wakeups[0][0] - time.time())) + _logger.debug("Going to sleep for %ss", amount) + time.sleep(amount) + +def start_master_thread(): + """ Start the above runner function in a daemon thread. + + The thread is a typical daemon thread: it will never quit and must be + terminated when the main process exits - with no consequence (the processing + threads it spawns are not marked daemon). + + """ + global _thread_slots + _thread_slots = openerp.conf.max_cron_threads + db_maxconn = tools.config['db_maxconn'] + if _thread_slots >= tools.config.get('db_maxconn', 64): + _logger.warning("Connection pool size (%s) is set lower than max number of cron threads (%s), " + "this may cause trouble if you reach that number of parallel cron tasks.", + db_maxconn, _thread_slots) + if _thread_slots: + t = threading.Thread(target=runner, name="openerp.cron.master_thread") + t.setDaemon(True) + t.start() + _logger.debug("Master cron daemon started!") + else: + _logger.info("No master cron daemon (0 workers needed).") # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: diff --git a/openerp/modules/registry.py b/openerp/modules/registry.py index 5ed8ae3dbdf..9ec3e6f456a 100644 --- a/openerp/modules/registry.py +++ b/openerp/modules/registry.py @@ -27,6 +27,7 @@ import threading import openerp.sql_db import openerp.osv.orm +import openerp.cron import openerp.tools import openerp.modules.db import openerp.tools.config @@ -109,6 +110,15 @@ class Registry(object): return res + def schedule_cron_jobs(self): + """ Make the cron thread care about this registry/database jobs. + This will initiate the cron thread to check for any pending jobs for + this registry/database as soon as possible. Then it will continuously + monitor the ir.cron model for future jobs. See openerp.cron for + details. + """ + openerp.cron.schedule_wakeup(openerp.cron.WAKE_UP_NOW, self.db.dbname) + def clear_caches(self): """ Clear the caches This clears the caches associated to methods decorated with @@ -203,6 +213,9 @@ class RegistryManager(object): finally: cr.close() + if pooljobs: + registry.schedule_cron_jobs() + return registry @classmethod @@ -220,6 +233,8 @@ class RegistryManager(object): if db_name in cls.registries: cls.registries[db_name].clear_caches() del cls.registries[db_name] + openerp.cron.cancel(db_name) + @classmethod def delete_all(cls): diff --git a/openerp/service/__init__.py b/openerp/service/__init__.py index 41e7200de85..c387d272e26 100644 --- a/openerp/service/__init__.py +++ b/openerp/service/__init__.py @@ -28,6 +28,7 @@ import netrpc_server import web_services import websrv_lib +import openerp.cron import openerp.modules import openerp.netsvc import openerp.osv @@ -50,7 +51,7 @@ _logger = logging.getLogger(__name__) def start_services(): """ Start all services. - Services include the different servers. + Services include the different servers and cron threads. """ # Instantiate local services (this is a legacy design). @@ -63,6 +64,9 @@ def start_services(): #http_server.init_static_http() netrpc_server.init_servers() + # Start the main cron thread. + openerp.cron.start_master_thread() + # Start the top-level servers threads (normally HTTP, HTTPS, and NETRPC). openerp.netsvc.Server.startAll() @@ -72,6 +76,9 @@ def start_services(): def stop_services(): """ Stop all services. """ + # stop scheduling new jobs; we will have to wait for the jobs to complete below + openerp.cron.cancel_all() + openerp.netsvc.Server.quitAll() openerp.wsgi.core.stop_server() config = openerp.tools.config From 0498572623614a282cd60fc9c5b0baa4c4a3864e Mon Sep 17 00:00:00 2001 From: Vo Minh Thu Date: Thu, 29 Mar 2012 15:20:28 +0200 Subject: [PATCH 6/8] [IMP] cron: re-enabled thread cron (was unnecessarilly removed in a previous commit). bzr revid: vmt@openerp.com-20120329132028-3cjq7mt0yqbk2g00 --- openerp-server | 3 +++ 1 file changed, 3 insertions(+) diff --git a/openerp-server b/openerp-server index 19c60397a08..f574ebb354f 100755 --- a/openerp-server +++ b/openerp-server @@ -93,6 +93,9 @@ def preload_registry(dbname): """ Preload a registry, and start the cron.""" try: db, registry = openerp.pooler.get_db_and_pool(dbname, update_module=config['init'] or config['update'], pooljobs=False) + + # jobs will start to be processed later, when openerp.cron.start_master_thread() is called by openerp.service.start_services() + registry.schedule_cron_jobs() except Exception: _logger.exception('Failed to initialize database `%s`.', dbname) From ea6184d3d57631df61d52f3ccbe01f62b01fa40f Mon Sep 17 00:00:00 2001 From: Vo Minh Thu Date: Fri, 30 Mar 2012 12:34:59 +0200 Subject: [PATCH 7/8] [IMP] cron: re-use the rlimits (used with Gunicorn) with the cron processes. bzr revid: vmt@openerp.com-20120330103459-mj7it2zucpiwlc8d --- openerp-cron-worker | 22 +++++++++++++++++++++- openerp/addons/base/ir/ir_cron.py | 17 ++++++++++++++++- openerp/wsgi/core.py | 2 +- 3 files changed, 38 insertions(+), 3 deletions(-) diff --git a/openerp-cron-worker b/openerp-cron-worker index a3b2ab0ac55..dd80870dba0 100755 --- a/openerp-cron-worker +++ b/openerp-cron-worker @@ -6,7 +6,26 @@ 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. +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 @@ -79,6 +98,7 @@ if __name__ == '__main__': openerp.cron.enable_schedule_wakeup = False 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) diff --git a/openerp/addons/base/ir/ir_cron.py b/openerp/addons/base/ir/ir_cron.py index cf54a904844..54a2ceab65b 100644 --- a/openerp/addons/base/ir/ir_cron.py +++ b/openerp/addons/base/ir/ir_cron.py @@ -391,7 +391,22 @@ class ir_cron(osv.osv): return t1 = time.time() for db_name in names: - while(cls._acquire_job(db_name)): + 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() diff --git a/openerp/wsgi/core.py b/openerp/wsgi/core.py index 028bb47b349..4fd108ca9ec 100644 --- a/openerp/wsgi/core.py +++ b/openerp/wsgi/core.py @@ -506,7 +506,7 @@ def post_request(worker, req, environ): 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. From d94b776f0cebf92467e276cad798aef436a74289 Mon Sep 17 00:00:00 2001 From: Vo Minh Thu Date: Mon, 2 Apr 2012 14:01:58 +0200 Subject: [PATCH 8/8] [IMP] cron: use multi-process signaling. bzr revid: vmt@openerp.com-20120402120158-m3m4s67pjueemane --- openerp-cron-worker | 1 + openerp/addons/base/ir/ir_cron.py | 2 ++ 2 files changed, 3 insertions(+) diff --git a/openerp-cron-worker b/openerp-cron-worker index dd80870dba0..6deb5db01ef 100755 --- a/openerp-cron-worker +++ b/openerp-cron-worker @@ -96,6 +96,7 @@ if __name__ == '__main__': 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." diff --git a/openerp/addons/base/ir/ir_cron.py b/openerp/addons/base/ir/ir_cron.py index 54a2ceab65b..a46e4f32446 100644 --- a/openerp/addons/base/ir/ir_cron.py +++ b/openerp/addons/base/ir/ir_cron.py @@ -351,8 +351,10 @@ class ir_cron(osv.osv): # 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: