Add transactional queue and workers

issue7406
review45461002

CHANGELOG

@@ -1,3 +1,4 @@
+* Add transactional queue and workers
 * Add __table_handler__ to ModelSQL
 * Support partial index
 * Allow to use SQL expression for index action

bin/trytond-worker

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+# This file is part of Tryton.  The COPYRIGHT file at the top level of
+# this repository contains the full copyright notices and license terms.
+import sys
+import os
+
+DIR = os.path.abspath(os.path.normpath(os.path.join(__file__,
+    '..', '..', 'trytond')))
+if os.path.isdir(DIR):
+    sys.path.insert(0, os.path.dirname(DIR))
+
+import trytond.commandline as commandline
+from trytond.config import config
+
+parser = commandline.get_parser_worker()
+options = parser.parse_args()
+config.update_etc(options.configfile)
+commandline.config_log(options)
+
+# Import after application is configured
+import trytond.worker as worker
+
+with commandline.pidfile(options):
+    worker.work(options)

doc/ref/models/models.rst

@@ -41,6 +41,11 @@ Class attributes are:
 
     The definition of the field ``id`` of records.
 
+.. attribute:: Model.__queue__
+
+    It returns a queue caller for the model. The called method will be pushed
+    into the queue.
+
 Class methods:
 
 .. classmethod:: Model.__setup__()

doc/topics/configuration.rst

@@ -179,6 +179,17 @@ The minimum number of seconds between two cleanings of the cache.
 
 Default: `300`
 
+queue
+-----
+
+worker
+~~~~~~
+
+Activate asynchronous processing of the tasks. Otherwise they are performed at
+the end of the requests.
+
+Default: `False`
+
 table
 -----
 

doc/topics/index.rst

@@ -28,6 +28,7 @@ Introduction to all the key parts of trytond:
     wizard
     reports/index
     rpc
+    task_queue
     user_application
     modules/index
     translation

doc/topics/start_server.rst

@@ -42,6 +42,18 @@ with this command line::
 The server will wake up every minutes and preform the scheduled actions defined
 in the `database`.
 
+Worker service
+==============
+
+If you want to use a pool of workers to run :ref:`asynchronously some tasks
+<topics-task-queue>`, you must activate the worker in the `queue` section of
+the :ref:`configuration <topics-configuration>` and run the worker manager with
+this command line::
+
+    trytond-worker -c <config file> -d <database>
+
+The manager will dispatch tasks from the queue to a pool of worker processes.
+
 Services options
 ================
 

doc/topics/task_queue.rst

@@ -0,0 +1,57 @@
+.. _topics-task-queue
+
+==========
+Task Queue
+==========
+
+Tryton provides a way to run asynchronously some tasks. You must activate the
+worker in the `queue` section of the :ref:`configuration
+<topics-configuration>` and :ref:`run the worker manager
+<topics-start-server>` otherwise the tasks will be run at the end of the
+transaction.
+
+A task is the parameters that defines how to call a method from a
+:class:`trytond.model.Model`. This include the
+:attr:`trytond.transaction.Transaction.context`, the
+:attr:`trytond.transaction.Transaction.user` and the arguments.
+The first argument of the method must be an instance or a list of instances of
+:class:`trytond.model.Model`. This other arguments must be JSON-ifiable.
+
+A task is pushed into the `queue` by calling the desired method on the
+:attr:`trytond.model.Model.__queue__`. This stores in the queue all the current
+parameters of the call and it will be execute by a worker or at the end of the
+transaction if no worker is configured.
+The following :attr:`trytond.transaction.Transaction.context` keys are used as
+parameters for the queue:
+
+    - `queue_name`: The name of the queue. Default value is `default`.
+
+    - `queue_scheduled_at`: A `datetime.timedelta` to add to current time to
+      define when the task should be started. Default value is `None` which
+      means directly.
+
+    - `queue_expected_at`: A `datetime.timedelta` to add to current time to
+      define when the task should be finished. Default value is `None` which
+      means as soon as possible.
+
+.. warning::
+
+    There is no access right verification during the execution of the task.
+
+Example:
+
+.. highlight:: python
+
+    from trytond.model import Model
+
+    class MyModel(Model):
+        "My Model"
+        __name__ = 'my_model'
+
+        @classmethod
+        def launch(cls, records):
+            for record in records:
+                cls.__queue__.process(record, 42)
+
+        def process(self, value):
+            self.value = value

setup.py

@@ -60,7 +60,12 @@ setup(name=name,
             'tryton.cfg', '*.xml', '*.html', 'view/*.xml', 'locale/*.po'],
         'trytond.tests': ['tryton.cfg', '*.xml', 'forbidden.txt'],
         },
-    scripts=['bin/trytond', 'bin/trytond-admin', 'bin/trytond-cron'],
+    scripts=[
+        'bin/trytond',
+        'bin/trytond-admin',
+        'bin/trytond-cron',
+        'bin/trytond-worker',
+        ],
     classifiers=[
         'Development Status :: 5 - Production/Stable',
         'Environment :: No Input/Output (Daemon)',

trytond/backend/database.py

@@ -135,6 +135,10 @@ class DatabaseInterface(object):
         '''
         raise NotImplementedError
 
+    def lock_id(self, id, timeout=None):
+        """Return SQL function to lock resource"""
+        raise NotImplementedError
+
     def has_constraint(self, constraint):
         '''
         Return True if database handle constraint.
@@ -215,6 +219,10 @@ class DatabaseInterface(object):
             return
         raise NotImplementedError
 
+    def has_channel(self):
+        "Return True if database supports LISTEN/NOTIFY channel"
+        return False
+
     def sql_type(self, type_):
         'Return the SQLType tuple corresponding to the SQL type'
         pass

trytond/backend/postgresql/database.py

@@ -70,6 +70,14 @@ class Unaccent(Function):
     _function = 'unaccent'
 
 
+class AdvisoryLock(Function):
+    _function = 'pg_advisory_xact_lock'
+
+
+class TryAdvisoryLock(Function):
+    _function = 'pg_try_advisory_xact_lock'
+
+
 class Database(DatabaseInterface):
 
     _lock = RLock()
@@ -288,6 +296,12 @@ class Database(DatabaseInterface):
         cursor = connection.cursor()
         cursor.execute('LOCK "%s" IN EXCLUSIVE MODE NOWAIT' % table)
 
+    def lock_id(self, id, timeout=None):
+        if not timeout:
+            return TryAdvisoryLock(id)
+        else:
+            return AdvisoryLock(id)
+
     def has_constraint(self, constraint):
         return True
 
@@ -459,6 +473,10 @@ class Database(DatabaseInterface):
                 'FROM "%s"' % name)
         return cursor.fetchone()[0]
 
+    def has_channel(self):
+        return True
+
+
 register_type(UNICODE)
 if PYDATE:
     register_type(PYDATE)

trytond/backend/sqlite/database.py

@@ -18,7 +18,7 @@ except ImportError:
     import sqlite3 as sqlite
     from sqlite3 import IntegrityError as DatabaseIntegrityError
     from sqlite3 import OperationalError as DatabaseOperationalError
-from sql import Flavor, Table, Query, Expression
+from sql import Flavor, Table, Query, Expression, Literal
 from sql.functions import (Function, Extract, Position, Substring,
     Overlay, CharLength, CurrentTimestamp, Trim)
 
@@ -419,6 +419,9 @@ class Database(DatabaseInterface):
     def lock(self, connection, table):
         pass
 
+    def lock_id(self, id, timeout=None):
+        return Literal(True)
+
     def has_constraint(self, constraint):
         return False
 

trytond/commandline.py

@@ -40,6 +40,19 @@ def get_parser_daemon():
     return parser
 
 
+def get_parser_worker():
+    parser = get_parser_daemon()
+    parser.add_argument("--name", dest='name',
+        help="work only on the named queue")
+    parser.add_argument("-n", dest='processes', type=int,
+        help="number of processes to use")
+    parser.add_argument("--max", dest='maxtasksperchild', type=int,
+        help="number of tasks a worker process before being replaced")
+    parser.add_argument("-t", "--timeout", dest='timeout', default=60,
+        type=int, help="maximum timeout when waiting notification")
+    return parser
+
+
 def get_parser_admin():
     parser = get_parser()
 

trytond/config.py

@@ -60,6 +60,8 @@ class TrytonConfigParser(configparser.RawConfigParser):
         self.set('cache', 'model', 200)
         self.set('cache', 'record', 2000)
         self.set('cache', 'field', 100)
+        self.add_section('queue')
+        self.set('queue', 'worker', False)
         self.add_section('ssl')
         self.add_section('email')
         self.set('email', 'uri', 'smtp://localhost:25')

trytond/ir/__init__.py

@@ -20,6 +20,7 @@ from .cache import *
 from .date import *
 from .trigger import *
 from .session import *
+from . import queue
 
 
 def register():
@@ -85,6 +86,7 @@ def register():
         TriggerLog,
         Session,
         SessionWizard,
+        queue.Queue,
         module='ir', type_='model')
     Pool.register(
         TranslationSet,

trytond/ir/queue.py

@@ -0,0 +1,222 @@
+# This file is part of Tryton.  The COPYRIGHT file at the top level of
+# this repository contains the full copyright notices and license terms.
+import datetime
+
+from sql import With, Literal, Null
+from sql.aggregate import Min
+from sql.functions import CurrentTimestamp, Extract
+
+from trytond.config import config
+from trytond.model import ModelSQL, fields
+from trytond.pool import Pool
+from trytond.tools import grouped_slice
+from trytond.transaction import Transaction
+
+has_worker = config.getboolean('queue', 'worker', default=False)
+
+
+class Queue(ModelSQL):
+    "Queue"
+    __name__ = 'ir.queue'
+    name = fields.Char("Name", required=True)
+
+    data = fields.Dict(None, "Data")
+
+    enqueued_at = fields.Timestamp("Enqueued at", required=True)
+    dequeued_at = fields.Timestamp("Dequeued at")
+    finished_at = fields.Timestamp("Finished at")
+
+    scheduled_at = fields.Timestamp("Scheduled at",
+        help="When the task can start.")
+    expected_at = fields.Timestamp("Expected at",
+        help="When the task should be done.")
+
+    @classmethod
+    def __register__(cls, module_name):
+        queue = cls.__table__()
+        super().__register__(module_name)
+        table_h = cls.__table_handler__(module_name)
+
+        # Add index for candidates
+        table_h.index_action([
+                queue.scheduled_at.nulls_first,
+                queue.expected_at.nulls_first,
+                queue.dequeued_at,
+                queue.name,
+                ], action='add')
+
+    @classmethod
+    def default_enqueued_at(cls):
+        return datetime.datetime.now()
+
+    @classmethod
+    def copy(cls, records, default=None):
+        if default is None:
+            default = {}
+        else:
+            default = default.copy()
+        default.setdefault('enqueued_at')
+        default.setdefault('dequeued_at')
+        default.setdefault('finished_at')
+        return super(Queue, cls).copy(records, default=default)
+
+    @classmethod
+    def push(cls, name, data, scheduled_at=None, expected_at=None):
+        transaction = Transaction()
+        database = transaction.database
+        cursor = transaction.connection.cursor()
+        with transaction.set_user(0):
+            record, = cls.create([{
+                        'name': name,
+                        'data': data,
+                        'scheduled_at': scheduled_at,
+                        'expected_at': expected_at,
+                        }])
+        if database.has_channel():
+            cursor.execute('NOTIFY "%s"', (cls.__name__,))
+        if not has_worker:
+            transaction.tasks.append(record.id)
+        return record.id
+
+    @classmethod
+    def pull(cls, database, connection, name=None):
+        cursor = connection.cursor()
+        queue = cls.__table__()
+
+        candidates = With('id', 'scheduled_at', 'expected_at',
+            query=queue.select(
+                queue.id,
+                queue.scheduled_at,
+                queue.expected_at,
+                where=((queue.name == name) if name else Literal(True))
+                & (queue.dequeued_at == Null),
+                order_by=[
+                    queue.scheduled_at.nulls_first,
+                    queue.expected_at.nulls_first]))
+        selected = With('id', query=candidates.select(
+                candidates.id,
+                where=((candidates.scheduled_at <= CurrentTimestamp())
+                    | (candidates.scheduled_at == Null))
+                & database.lock_id(candidates.id),
+                order_by=[
+                    candidates.scheduled_at.nulls_first,
+                    candidates.expected_at.nulls_first],
+                limit=1))
+        next_timeout = With('seconds', query=candidates.select(
+                Min(Extract('second',
+                        candidates.scheduled_at - CurrentTimestamp())
+                    ),
+                where=candidates.scheduled_at >= CurrentTimestamp()))
+
+        task_id, seconds = None, None
+        if database.has_returning():
+            query = queue.update([queue.dequeued_at], [CurrentTimestamp()],
+                where=queue.id == selected.select(selected.id),
+                with_=[candidates, selected, next_timeout],
+                returning=[
+                    queue.id, next_timeout.select(next_timeout.seconds)])
+            cursor.execute(*query)
+            row = cursor.fetchone()
+            if row:
+                task_id, seconds = row
+        else:
+            query = queue.select(queue.id,
+                where=queue.id == selected.select(selected.id),
+                with_=[candidates, selected])
+            cursor.execute(*query)
+            row = cursor.fetchone()
+            if row:
+                task_id, = row
+                query = queue.update([queue.dequeued_at], [CurrentTimestamp()],
+                    where=queue.id == task_id)
+                cursor.execute(*query)
+            query = next_timeout.select(next_timeout.seconds)
+            cursor.execute(*query)
+            row = cursor.fetchone()
+            if row:
+                seconds, = row
+
+        if not task_id and database.has_channel():
+            cursor.execute('LISTEN "%s"', (cls.__name__,))
+        return task_id, seconds
+
+    def run(self):
+        transaction = Transaction()
+        Model = Pool().get(self.data['model'])
+        with transaction.set_user(self.data['user']), \
+                transaction.set_context(self.data['context']):
+            instances = self.data['instances']
+            # Ensure record ids still exist
+            if isinstance(instances, int):
+                with transaction.set_context(active_test=False):
+                    if Model.search([('id', '=', instances)]):
+                        instances = Model(instances)
+                    else:
+                        instances = None
+            else:
+                ids = set()
+                with transaction.set_context(active_test=False):
+                    for sub_ids in grouped_slice(instances):
+                        records = Model.search([('id', 'in', list(sub_ids))])
+                        ids.update(map(int, records))
+                if ids:
+                    instances = Model.browse(
+                        [i for i in instances if i in ids])
+                else:
+                    instances = None
+            if instances is not None:
+                getattr(Model, self.data['method'])(
+                    instances, *self.data['args'], **self.data['kwargs'])
+        if not self.dequeued_at:
+            self.dequeued_at = datetime.datetime.now()
+        self.finished_at = datetime.datetime.now()
+        self.save()
+
+    @classmethod
+    def caller(cls, model):
+        return _Model(cls, model)
+
+
+class _Model(object):
+    def __init__(self, queue, model):
+        self.__queue = queue
+        self.__model = model
+
+    def __getattr__(self, name):
+        return _Method(self.__queue, self.__model, name)
+
+
+class _Method(object):
+    def __init__(self, queue, model, name):
+        self.__queue = queue
+        self.__model = model
+        self.__name = name
+
+    def __call__(self, instances, *args, **kwargs):
+        transaction = Transaction()
+        context = transaction.context
+        name = context.pop('queue_name', 'default')
+        now = datetime.datetime.now()
+        scheduled_at = context.pop('queue_scheduled_at', None)
+        if scheduled_at is not None:
+            scheduled_at = now + scheduled_at
+        expected_at = context.pop('queue_expected_at', None)
+        context.pop('_check_access', None)
+        if expected_at is not None:
+            expected_at = now + expected_at
+        try:
+            instances = list(map(int, instances))
+        except TypeError:
+            instances = int(instances)
+        data = {
+            'model': self.__model.__name__,
+            'method': self.__name,
+            'user': transaction.user,
+            'context': context,
+            'instances': instances,
+            'args': args,
+            'kwargs': kwargs,
+            }
+        return self.__queue.push(
+            name, data,
+            scheduled_at=scheduled_at, expected_at=expected_at)

trytond/model/model.py

@@ -7,7 +7,7 @@ from functools import total_ordering
 
 from trytond.model import fields
 from trytond.error import WarningErrorMixin
-from trytond.pool import Pool, PoolBase
+from trytond.pool import Pool, PoolBase, PoolMeta
 from trytond.pyson import PYSONEncoder
 from trytond.transaction import Transaction
 from trytond.url import URLMixin
@@ -16,8 +16,16 @@ from trytond.rpc import RPC
 __all__ = ['Model']
 
 
+class ModelMeta(PoolMeta):
+    @property
+    def __queue__(self):
+        pool = Pool()
+        Queue = pool.get('ir.queue')
+        return Queue.caller(self)
+
+
 @total_ordering
-class Model(WarningErrorMixin, URLMixin, PoolBase):
+class Model(WarningErrorMixin, URLMixin, PoolBase, metaclass=ModelMeta):
     """
     Define a model in Tryton.
     """

trytond/protocols/dispatcher.py

@@ -21,6 +21,7 @@ from trytond.exceptions import (
     RateLimitException)
 from trytond.tools import is_instance_method
 from trytond.wsgi import app
+from trytond.worker import run_task
 from .wrappers import with_pool
 
 logger = logging.getLogger(__name__)
@@ -186,6 +187,9 @@ def _dispatch(request, pool, *args, **kwargs):
                 raise
             # Need to commit to unlock SQLite database
             transaction.commit()
+        while transaction.tasks:
+            task_id = transaction.tasks.pop()
+            run_task(pool, task_id)
         if request.authorization.type == 'session':
             context = {'_request': request.context}
             try:

trytond/res/ir.xml

@@ -311,6 +311,21 @@ this repository contains the full copyright notices and license terms. -->
             <field name="perm_create" eval="True"/>
             <field name="perm_delete" eval="True"/>
         </record>
+        <record model="ir.model.access" id="access_ir_queue">
+            <field name="model" search="[('model', '=', 'ir.queue')]"/>
+            <field name="perm_read" eval="False"/>
+            <field name="perm_write" eval="False"/>
+            <field name="perm_create" eval="False"/>
+            <field name="perm_delete" eval="False"/>
+        </record>
+        <record model="ir.model.access" id="access_ir_queue_admin">
+            <field name="model" search="[('model', '=', 'ir.queue')]"/>
+            <field name="group" ref="group_admin"/>
+            <field name="perm_read" eval="True"/>
+            <field name="perm_write" eval="False"/>
+            <field name="perm_create" eval="False"/>
+            <field name="perm_delete" eval="False"/>
+        </record>
         <record model="ir.model.access" id="access_ir_lang">
             <field name="model" search="[('model', '=', 'ir.lang')]"/>
             <field name="perm_read" eval="True"/>

trytond/transaction.py

@@ -31,6 +31,7 @@ class _Local(local):
     def __init__(self):
         # Transaction stack control
         self.transactions = []
+        self.tasks = []
 
 
 class Transaction(object):
@@ -64,6 +65,10 @@ class Transaction(object):
             instance = transactions[-1]
         return instance
 
+    @property
+    def tasks(self):
+        return self._local.tasks
+
     def get_cache(self):
         from trytond.cache import LRUDict
         keys = tuple(((key, self.context[key])

trytond/worker.py

@@ -0,0 +1,115 @@
+# This file is part of Tryton.  The COPYRIGHT file at the top level of
+# this repository contains the full copyright notices and license terms.
+import logging
+import select
+import signal
+import time
+from multiprocessing import Pool as MPool, cpu_count
+
+from sql import Flavor
+
+from trytond import backend
+from trytond.config import config
+from trytond.pool import Pool
+from trytond.transaction import Transaction
+
+__all__ = ['work']
+logger = logging.getLogger(__name__)
+Database = backend.get('Database')
+DatabaseOperationalError = backend.get('DatabaseOperationalError')
+
+
+class Queue(object):
+    def __init__(self, pool, mpool):
+        self.database = Database(pool.database_name).connect()
+        self.connection = self.database.get_connection(autocommit=True)
+        self.pool = pool
+        self.mpool = mpool
+
+    def pull(self, name=None):
+        Queue = self.pool.get('ir.queue')
+        return Queue.pull(self.database, self.connection, name=name)
+
+    def run(self, task_id):
+        return self.mpool.apply_async(
+            run_task, (self.pool.database_name, task_id))
+
+
+class TaskList(list):
+    def filter(self):
+        for t in list(self):
+            if t.ready():
+                self.remove(t)
+        return self
+
+
+def work(options):
+    Flavor.set(Database.flavor)
+    if not config.getboolean('queue', 'worker', default=False):
+        return
+    try:
+        processes = options.processes or cpu_count()
+    except NotImplementedError:
+        processes = 1
+    logger.info("start %d workers", processes)
+    mpool = MPool(
+        processes, initializer, (options,), options.maxtasksperchild)
+    queues = [Queue(pool, mpool) for pool in initializer(options, False)]
+
+    tasks = TaskList()
+    timeout = options.timeout
+    try:
+        while True:
+            while len(tasks.filter()) >= processes:
+                time.sleep(0.1)
+            for queue in queues:
+                task_id, next_ = queue.pull(options.name)
+                timeout = min(
+                    next_ or options.timeout, timeout, options.timeout)
+                if task_id:
+                    tasks.append(queue.run(task_id))
+                    break
+            else:
+                connections = [q.connection for q in queues]
+                connections, _, _ = select.select(connections, [], [], timeout)
+                for connection in connections:
+                    connection.poll()
+                    while connection.notifies:
+                        connection.notifies.pop(0)
+    except KeyboardInterrupt:
+        mpool.close()
+
+
+def initializer(options, worker=True):
+    if worker:
+        signal.signal(signal.SIGINT, signal.SIG_IGN)
+    pools = []
+    database_list = Pool.database_list()
+    for database_name in options.database_names:
+        pool = Pool(database_name)
+        if database_name not in database_list:
+            with Transaction().start(database_name, 0, readonly=True):
+                pool.init()
+        pools.append(pool)
+    return pools
+
+
+def run_task(pool, task_id):
+    if not isinstance(pool, Pool):
+        pool = Pool(pool)
+    Queue = pool.get('ir.queue')
+    logger.info('task "%d" started', task_id)
+    try:
+        for count in range(config.getint('database', 'retry'), -1, -1):
+            with Transaction().start(pool.database_name, 0) as transaction:
+                try:
+                    Queue(task_id).run()
+                    break
+                except DatabaseOperationalError:
+                    if count:
+                        transaction.rollback()
+                        continue
+                    raise
+        logger.info('task "%d" done', task_id)
+    except Exception:
+        logger.critical('task "%d" failed', task_id, exc_info=True)