From ae06a8bca2b098a373fb86b123a5d28024d15278 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?C=C3=A9dric=20Krier?= Date: Fri, 27 Jul 2018 09:32:28 +0200 Subject: [PATCH] Add transactional queue and workers issue7406 review45461002 --- CHANGELOG | 1 + bin/trytond-worker | 24 +++ doc/ref/models/models.rst | 5 + doc/topics/configuration.rst | 11 ++ doc/topics/index.rst | 1 + doc/topics/start_server.rst | 12 ++ doc/topics/task_queue.rst | 57 +++++++ setup.py | 7 +- trytond/backend/database.py | 8 + trytond/backend/postgresql/database.py | 18 ++ trytond/backend/sqlite/database.py | 5 +- trytond/commandline.py | 13 ++ trytond/config.py | 2 + trytond/ir/__init__.py | 2 + trytond/ir/queue.py | 222 +++++++++++++++++++++++++ trytond/model/model.py | 12 +- trytond/protocols/dispatcher.py | 4 + trytond/res/ir.xml | 15 ++ trytond/transaction.py | 5 + trytond/worker.py | 115 +++++++++++++ 20 files changed, 535 insertions(+), 4 deletions(-) create mode 100755 bin/trytond-worker create mode 100644 doc/topics/task_queue.rst create mode 100644 trytond/ir/queue.py create mode 100644 trytond/worker.py diff --git a/CHANGELOG b/CHANGELOG index ab63c46a64..b197443e73 100644 --- a/CHANGELOG +++ b/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 diff --git a/bin/trytond-worker b/bin/trytond-worker new file mode 100755 index 0000000000..968d109c9d --- /dev/null +++ b/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) diff --git a/doc/ref/models/models.rst b/doc/ref/models/models.rst index c4611f0d2f..b8a3c0a3da 100644 --- a/doc/ref/models/models.rst +++ b/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__() diff --git a/doc/topics/configuration.rst b/doc/topics/configuration.rst index 17a681858d..c335b00adf 100644 --- a/doc/topics/configuration.rst +++ b/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 ----- diff --git a/doc/topics/index.rst b/doc/topics/index.rst index 853e6dac73..ed67383c5a 100644 --- a/doc/topics/index.rst +++ b/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 diff --git a/doc/topics/start_server.rst b/doc/topics/start_server.rst index ffb6148f85..a502713bde 100644 --- a/doc/topics/start_server.rst +++ b/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 +`, you must activate the worker in the `queue` section of +the :ref:`configuration ` and run the worker manager with +this command line:: + + trytond-worker -c -d + +The manager will dispatch tasks from the queue to a pool of worker processes. + Services options ================ diff --git a/doc/topics/task_queue.rst b/doc/topics/task_queue.rst new file mode 100644 index 0000000000..22921ba305 --- /dev/null +++ b/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 +` and :ref:`run the worker manager +` 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 diff --git a/setup.py b/setup.py index c8178bbb87..fc0dfe61b8 100644 --- a/setup.py +++ b/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)', diff --git a/trytond/backend/database.py b/trytond/backend/database.py index 43111fc2ff..e614be668e 100644 --- a/trytond/backend/database.py +++ b/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 diff --git a/trytond/backend/postgresql/database.py b/trytond/backend/postgresql/database.py index 1903cb220a..7051161620 100644 --- a/trytond/backend/postgresql/database.py +++ b/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) diff --git a/trytond/backend/sqlite/database.py b/trytond/backend/sqlite/database.py index 82c8c84c1a..42d0d59323 100644 --- a/trytond/backend/sqlite/database.py +++ b/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 diff --git a/trytond/commandline.py b/trytond/commandline.py index 89a0d63e8f..a39e78e0ae 100644 --- a/trytond/commandline.py +++ b/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() diff --git a/trytond/config.py b/trytond/config.py index 799190c96d..8feaf4621c 100644 --- a/trytond/config.py +++ b/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') diff --git a/trytond/ir/__init__.py b/trytond/ir/__init__.py index 1e0a5aedc1..ab5789337f 100644 --- a/trytond/ir/__init__.py +++ b/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, diff --git a/trytond/ir/queue.py b/trytond/ir/queue.py new file mode 100644 index 0000000000..2f42a1a965 --- /dev/null +++ b/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) diff --git a/trytond/model/model.py b/trytond/model/model.py index 1b36b2582d..0474296cf9 100644 --- a/trytond/model/model.py +++ b/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. """ diff --git a/trytond/protocols/dispatcher.py b/trytond/protocols/dispatcher.py index e40599fb1a..8f9f2631d8 100644 --- a/trytond/protocols/dispatcher.py +++ b/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: diff --git a/trytond/res/ir.xml b/trytond/res/ir.xml index 063791ec7d..0b1f96dbf8 100644 --- a/trytond/res/ir.xml +++ b/trytond/res/ir.xml @@ -311,6 +311,21 @@ this repository contains the full copyright notices and license terms. --> + + + + + + + + + + + + + + + diff --git a/trytond/transaction.py b/trytond/transaction.py index 349b0b2d13..6d393499e8 100644 --- a/trytond/transaction.py +++ b/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]) diff --git a/trytond/worker.py b/trytond/worker.py new file mode 100644 index 0000000000..056cf83f0a --- /dev/null +++ b/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)