[MERGE] forward port branch saas-14 up to d171f282c3
This commit is contained in:
Christophe Simonis
@@ -17,6 +17,13 @@ _logger = logging.getLogger(__name__)
|
||||
BASE_VERSION = odoo.modules.load_information_from_description_file('base')['version']
|
||||
|
||||
|
||||
class BadVersion(Exception):
|
||||
pass
|
||||
|
||||
class BadModuleState(Exception):
|
||||
pass
|
||||
|
||||
|
||||
_intervalTypes = {
|
||||
'days': lambda interval: relativedelta(days=interval),
|
||||
'hours': lambda interval: relativedelta(hours=interval),
|
||||
@@ -140,34 +147,103 @@ class ir_cron(models.Model):
|
||||
cron_cr.commit()
|
||||
|
||||
@classmethod
|
||||
def _acquire_job(cls, db_name):
|
||||
# TODO remove 'check' argument from addons/base_automation/base_automation.py
|
||||
""" Try to process one cron job.
|
||||
def _process_jobs(cls, db_name):
|
||||
""" Try to process all cron jobs.
|
||||
|
||||
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.
|
||||
:raise BadVersion: if the version is different from the worker's
|
||||
:raise BadModuleState: if modules are to install/upgrade/remove
|
||||
"""
|
||||
db = odoo.sql_db.db_connect(db_name)
|
||||
threading.current_thread().dbname = db_name
|
||||
jobs = []
|
||||
try:
|
||||
with db.cursor() as cr:
|
||||
# Make sure the database we poll has the same version as the code of base
|
||||
cr.execute("SELECT 1 FROM ir_module_module WHERE name=%s AND latest_version=%s", ('base', BASE_VERSION))
|
||||
if cr.fetchone():
|
||||
# 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""")
|
||||
jobs = cr.dictfetchall()
|
||||
else:
|
||||
_logger.warning('Skipping database %s as its base version is not %s.', db_name, BASE_VERSION)
|
||||
except psycopg2.ProgrammingError, e:
|
||||
# Make sure the database has the same version as the code of
|
||||
# base and that no module must be installed/upgraded/removed
|
||||
cr.execute("SELECT latest_version FROM ir_module_module WHERE name=%s", ['base'])
|
||||
(version,) = cr.fetchone()
|
||||
cr.execute("SELECT COUNT(*) FROM ir_module_module WHERE state LIKE %s", ['to %'])
|
||||
(changes,) = cr.fetchone()
|
||||
if not version or changes:
|
||||
raise BadModuleState()
|
||||
elif version != BASE_VERSION:
|
||||
raise BadVersion()
|
||||
# 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""")
|
||||
jobs = cr.dictfetchall()
|
||||
|
||||
for job in jobs:
|
||||
lock_cr = db.cursor()
|
||||
try:
|
||||
# Try to grab an exclusive lock on the job row from within the task transaction
|
||||
# Restrict to the same conditions as for the search since the job may have already
|
||||
# been run by an other thread when cron is running in multi thread
|
||||
lock_cr.execute("""SELECT *
|
||||
FROM ir_cron
|
||||
WHERE numbercall != 0
|
||||
AND active
|
||||
AND nextcall <= (now() at time zone 'UTC')
|
||||
AND id=%s
|
||||
FOR UPDATE NOWAIT""",
|
||||
(job['id'],), log_exceptions=False)
|
||||
|
||||
locked_job = lock_cr.fetchone()
|
||||
if not locked_job:
|
||||
_logger.debug("Job `%s` already executed by another process/thread. skipping it", job['cron_name'])
|
||||
continue
|
||||
# Got the lock on the job row, run its code
|
||||
_logger.debug('Starting job `%s`.', job['cron_name'])
|
||||
job_cr = db.cursor()
|
||||
try:
|
||||
registry = odoo.registry(db_name)
|
||||
registry[cls._name]._process_job(job_cr, job, lock_cr)
|
||||
except Exception:
|
||||
_logger.exception('Unexpected exception while processing cron job %r', job)
|
||||
finally:
|
||||
job_cr.close()
|
||||
|
||||
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['cron_name'])
|
||||
continue
|
||||
else:
|
||||
# Unexpected OperationalError
|
||||
raise
|
||||
finally:
|
||||
# we're exiting due to an exception while acquiring the lock
|
||||
lock_cr.close()
|
||||
|
||||
finally:
|
||||
if hasattr(threading.current_thread(), 'dbname'):
|
||||
del threading.current_thread().dbname
|
||||
|
||||
@classmethod
|
||||
def _acquire_job(cls, db_name):
|
||||
""" Try to process all cron jobs.
|
||||
|
||||
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.
|
||||
|
||||
This method hides most exceptions related to the database's version, the
|
||||
modules' state, and such.
|
||||
"""
|
||||
try:
|
||||
cls._process_jobs(db_name)
|
||||
except BadVersion:
|
||||
_logger.warning('Skipping database %s as its base version is not %s.', db_name, BASE_VERSION)
|
||||
except BadModuleState:
|
||||
_logger.warning('Skipping database %s because of modules to install/upgrade/remove.', db_name)
|
||||
except psycopg2.ProgrammingError as 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.
|
||||
@@ -177,51 +253,6 @@ class ir_cron(models.Model):
|
||||
except Exception:
|
||||
_logger.warning('Exception in cron:', exc_info=True)
|
||||
|
||||
for job in jobs:
|
||||
lock_cr = db.cursor()
|
||||
try:
|
||||
# Try to grab an exclusive lock on the job row from within the task transaction
|
||||
# Restrict to the same conditions as for the search since the job may have already
|
||||
# been run by an other thread when cron is running in multi thread
|
||||
lock_cr.execute("""SELECT *
|
||||
FROM ir_cron
|
||||
WHERE numbercall != 0
|
||||
AND active
|
||||
AND nextcall <= (now() at time zone 'UTC')
|
||||
AND id=%s
|
||||
FOR UPDATE NOWAIT""",
|
||||
(job['id'],), log_exceptions=False)
|
||||
|
||||
locked_job = lock_cr.fetchone()
|
||||
if not locked_job:
|
||||
_logger.debug("Job `%s` already executed by another process/thread. skipping it", job['cron_name'])
|
||||
continue
|
||||
# Got the lock on the job row, run its code
|
||||
_logger.debug('Starting job `%s`.', job['cron_name'])
|
||||
job_cr = db.cursor()
|
||||
try:
|
||||
registry = odoo.registry(db_name)
|
||||
registry[cls._name]._process_job(job_cr, job, lock_cr)
|
||||
except Exception:
|
||||
_logger.exception('Unexpected exception while processing cron job %r', job)
|
||||
finally:
|
||||
job_cr.close()
|
||||
|
||||
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['cron_name'])
|
||||
continue
|
||||
else:
|
||||
# Unexpected OperationalError
|
||||
raise
|
||||
finally:
|
||||
# we're exiting due to an exception while acquiring the lock
|
||||
lock_cr.close()
|
||||
|
||||
if hasattr(threading.current_thread(), 'dbname'): # cron job could have removed it as side-effect
|
||||
del threading.current_thread().dbname
|
||||
|
||||
@api.multi
|
||||
def _try_lock(self):
|
||||
"""Try to grab a dummy exclusive write-lock to the rows with the given ids,
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
<template id="report_irmodeloverview">
|
||||
<t t-call="report.html_container">
|
||||
<t t-foreach="docs" t-as="o">
|
||||
<div class="page">
|
||||
<div class="article">
|
||||
<table class="table table-bordered mb64">
|
||||
<tr>
|
||||
<td colspan="12">
|
||||
|
||||
@@ -216,14 +216,11 @@ class ThreadedServer(CommonServer):
|
||||
registries = odoo.modules.registry.Registry.registries
|
||||
_logger.debug('cron%d polling for jobs', number)
|
||||
for db_name, registry in registries.iteritems():
|
||||
while registry.ready:
|
||||
if registry.ready:
|
||||
try:
|
||||
acquired = odoo.addons.base.ir.ir_cron.ir_cron._acquire_job(db_name)
|
||||
if not acquired:
|
||||
break
|
||||
odoo.addons.base.ir.ir_cron.ir_cron._acquire_job(db_name)
|
||||
except Exception:
|
||||
_logger.warning('cron%d encountered an Exception:', number, exc_info=True)
|
||||
break
|
||||
|
||||
def cron_spawn(self):
|
||||
""" Start the above runner function in a daemon thread.
|
||||
|
||||
Reference in New Issue
Block a user