diff --git a/proteus/proteus/config.py b/proteus/proteus/config.py index 9b4f82a738..072c99f5f2 100644 --- a/proteus/proteus/config.py +++ b/proteus/proteus/config.py @@ -183,7 +183,7 @@ class _TrytondMethod(object): def __call__(self, *args, **kwargs): from trytond.rpc import RPC from trytond.tools import is_instance_method - from trytond.transaction import Transaction + from trytond.transaction import Transaction, TransactionError from trytond.worker import run_task if self._name in self._object.__rpc__: @@ -193,23 +193,35 @@ class _TrytondMethod(object): else: raise TypeError('%s is not callable' % self._name) - with Transaction().start(self._config.database_name, - self._config.user, readonly=rpc.readonly) as transaction: - args, kwargs, transaction.context, transaction.timestamp = \ - rpc.convert(self._object, *args, **kwargs) - meth = getattr(self._object, self._name) - if (rpc.instantiate is None - or not is_instance_method(self._object, self._name)): - result = rpc.result(meth(*args, **kwargs)) - else: - assert rpc.instantiate == 0 - inst = args.pop(0) - if hasattr(inst, self._name): - result = rpc.result(meth(inst, *args, **kwargs)) - else: - result = [rpc.result(meth(i, *args, **kwargs)) - for i in inst] - transaction.commit() + extras = {} + while True: + with Transaction().start(self._config.database_name, + self._config.user, readonly=rpc.readonly, + **extras) as transaction: + try: + (c_args, c_kwargs, + transaction.context, transaction.timestamp) \ + = rpc.convert(self._object, *args, **kwargs) + meth = getattr(self._object, self._name) + if (rpc.instantiate is None + or not is_instance_method( + self._object, self._name)): + result = rpc.result(meth(*c_args, **c_kwargs)) + else: + assert rpc.instantiate == 0 + inst = c_args.pop(0) + if hasattr(inst, self._name): + result = rpc.result( + meth(inst, *c_args, **c_kwargs)) + else: + result = [rpc.result(meth(i, *c_args, **c_kwargs)) + for i in inst] + except TransactionError as e: + transaction.rollback() + e.fix(extras) + continue + transaction.commit() + break while transaction.tasks: task_id = transaction.tasks.pop() run_task(self._config.database_name, task_id) diff --git a/trytond/CHANGELOG b/trytond/CHANGELOG index f537843802..c463afe338 100644 --- a/trytond/CHANGELOG +++ b/trytond/CHANGELOG @@ -1,3 +1,4 @@ +* Lock table on transaction start * Use EXISTS for search on O2M with many records * Replace target_search by use of count() * Manage domain on id in single value list as unique diff --git a/trytond/doc/ref/transaction.rst b/trytond/doc/ref/transaction.rst index 291bfa4b7a..6f2a62681e 100644 --- a/trytond/doc/ref/transaction.rst +++ b/trytond/doc/ref/transaction.rst @@ -4,6 +4,15 @@ Transaction =========== +.. exception:: TransactionError + + The base class for transaction error that need to retry the transaction. + +.. method:: TransactionError.fix(extras) + + Update the extras argument of :meth:`~Transaction.start` to restart the + transaction without the error. + .. class:: Transaction Represents a Tryton transaction that contains thread-local parameters of a @@ -57,7 +66,7 @@ Transaction Return a monotonic time used to populate :attr:~Transaction.started_at. -.. method:: Transaction.start(database_name, user[, readonly[, context[, close[, autocommit]]]]) +.. method:: Transaction.start(database_name, user[, readonly[, context[, close[, autocommit, \**extras]]]]) Start a new transaction and return a `context manager`_. The non-readonly transaction will be committed when exiting the ``with`` @@ -89,13 +98,18 @@ Transaction record rules. The user will be restored when exiting the ``with`` statement. +.. method:: Transaction.lock_table(table) + + Raise a :exc:`TransactionError` to retry the transaction if the table has + not been locked at the start. + .. method:: Transaction.set_current_transaction(transaction) Add a specific ``transaction`` on the top of the transaction stack. A transaction is commited or rollbacked only when its last reference is popped from the stack. -.. method:: Transaction.new_transaction([autocommit[, readonly]]) +.. method:: Transaction.new_transaction([autocommit[, readonly, \**extras]]) Create a new transaction with the same database, user and context as the original transaction and adds it to the stack of transactions. diff --git a/trytond/trytond/commandline.py b/trytond/trytond/commandline.py index 3cfc8823c7..6b3848e820 100644 --- a/trytond/trytond/commandline.py +++ b/trytond/trytond/commandline.py @@ -148,6 +148,9 @@ def get_parser_console(): help="The number of commands to remember in the command history") parser.add_argument("--readonly", dest="readonly", action='store_true', help="Start a readonly transaction") + parser.add_argument( + "--lock-table", dest="lock_tables", nargs='+', default=[], + metavar='TABLE', help="Lock tables") parser.epilog = "To store changes, `transaction.commit()` must be called." return parser diff --git a/trytond/trytond/console.py b/trytond/trytond/console.py index 9654adc116..8243f40644 100644 --- a/trytond/trytond/console.py +++ b/trytond/trytond/console.py @@ -46,7 +46,8 @@ def run(options): pool.init() with Transaction().start( - db_name, 0, readonly=options.readonly) as transaction: + db_name, 0, readonly=options.readonly, + _lock_tables=options.lock_tables) as transaction: local = { 'pool': pool, 'transaction': transaction, diff --git a/trytond/trytond/ir/cron.py b/trytond/trytond/ir/cron.py index 200c364b7b..9e78798f1f 100644 --- a/trytond/trytond/ir/cron.py +++ b/trytond/trytond/ir/cron.py @@ -15,7 +15,7 @@ from trytond.pool import Pool from trytond.pyson import Eval from trytond.status import processing from trytond.tools import timezone as tz -from trytond.transaction import Transaction +from trytond.transaction import Transaction, TransactionError from trytond.worker import run_task logger = logging.getLogger(__name__) @@ -171,18 +171,27 @@ class Cron(DeactivableMixin, ModelSQL, ModelView): return (time.monotonic() - started) * 1000 started = time.monotonic() name = '' % (cron.id, db_name, cron.method) - for count in range(retry, -1, -1): - if count != retry: + transaction_extras = {} + count = 0 + while True: + if count: time.sleep(0.02 * (retry - count)) try: with processing(name), \ - transaction.new_transaction() as cron_trans: + transaction.new_transaction( + **transaction_extras) as cron_trans: cron.run_once() cron_trans.commit() except Exception as e: + if isinstance(e, TransactionError): + cron_trans.rollback() + e.fix(transaction_extras) + continue if (isinstance(e, backend.DatabaseOperationalError) - and count): - logger.debug("Retry: %i", retry - count + 1) + and count < retry): + cron_trans.rollback() + count += 1 + logger.debug("Retry: %i", count) continue if isinstance(e, (UserError, UserWarning)): Error.log(cron, e) diff --git a/trytond/trytond/model/modelsql.py b/trytond/trytond/model/modelsql.py index 465bb2bdf6..5ca0d316bb 100644 --- a/trytond/trytond/model/modelsql.py +++ b/trytond/trytond/model/modelsql.py @@ -2034,7 +2034,7 @@ class ModelSQL(ModelStorage): with connection.cursor() as cursor: cursor.execute(*query) else: - database.lock(connection, cls._table) + transaction.lock_table(cls._table) def convert_from(table, tables): diff --git a/trytond/trytond/protocols/dispatcher.py b/trytond/trytond/protocols/dispatcher.py index 77e3756ebc..c431a83c11 100644 --- a/trytond/trytond/protocols/dispatcher.py +++ b/trytond/trytond/protocols/dispatcher.py @@ -13,7 +13,7 @@ from trytond.exceptions import ( ConcurrencyException, LoginException, RateLimitException, UserError, UserWarning) from trytond.tools import is_instance_method -from trytond.transaction import Transaction +from trytond.transaction import Transaction, TransactionError from trytond.worker import run_task from trytond.wsgi import app @@ -185,11 +185,14 @@ def _dispatch(request, pool, *args, **kwargs): started = time.monotonic() retry = config.getint('database', 'retry') - for count in range(retry, -1, -1): - if count != retry: + count = 0 + transaction_extras = {} + while True: + if count: time.sleep(0.02 * (retry - count)) - with Transaction().start(pool.database_name, user, - readonly=rpc.readonly) as transaction: + with Transaction().start( + pool.database_name, user, readonly=rpc.readonly, + **transaction_extras) as transaction: try: c_args, c_kwargs, transaction.context, transaction.timestamp \ = rpc.convert(obj, *args, **kwargs) @@ -206,10 +209,15 @@ def _dispatch(request, pool, *args, **kwargs): else: result = [rpc.result(meth(i, *c_args, **c_kwargs)) for i in inst] + except TransactionError as e: + transaction.rollback() + e.fix(transaction_extras) + continue except backend.DatabaseOperationalError: - if count and not rpc.readonly: + if count < retry and not rpc.readonly: transaction.rollback() - logger.debug("Retry: %i", retry - count + 1) + count += 1 + logger.debug("Retry: %i", count) continue logger.exception(log_message, *log_args, duration()) raise diff --git a/trytond/trytond/tests/test_modelsql.py b/trytond/trytond/tests/test_modelsql.py index c920ed5b94..ef438a7594 100644 --- a/trytond/trytond/tests/test_modelsql.py +++ b/trytond/trytond/tests/test_modelsql.py @@ -14,7 +14,7 @@ from trytond.model.exceptions import ( from trytond.model.modelsql import split_subquery_domain from trytond.pool import Pool from trytond.tests.test_tryton import activate_module, with_transaction -from trytond.transaction import Transaction +from trytond.transaction import Transaction, TransactionError class ModelSQLTestCase(unittest.TestCase): @@ -671,11 +671,26 @@ class ModelSQLTestCase(unittest.TestCase): Model = pool.get('test.modelsql.lock') transaction = Transaction() - with transaction.new_transaction(): - Model.lock() - with transaction.new_transaction(): - with self.assertRaises(backend.DatabaseOperationalError): + args1 = {} + while True: + with transaction.new_transaction(**args1): + try: Model.lock() + except TransactionError as e: + e.fix(args1) + continue + with self.assertRaises( + backend.DatabaseOperationalError): + args2 = {} + while True: + with transaction.new_transaction(**args2): + try: + Model.lock() + except TransactionError as e: + e.fix(args2) + continue + break + break @with_transaction() def test_search_or_to_union(self): diff --git a/trytond/trytond/tests/test_tryton.py b/trytond/trytond/tests/test_tryton.py index 5f79d0f440..4fc5ac7cca 100644 --- a/trytond/trytond/tests/test_tryton.py +++ b/trytond/trytond/tests/test_tryton.py @@ -34,7 +34,7 @@ from trytond.model.fields import Function from trytond.pool import Pool, isregisteredby from trytond.pyson import PYSONDecoder, PYSONEncoder from trytond.tools import file_open, find_dir, is_instance_method -from trytond.transaction import Transaction +from trytond.transaction import Transaction, TransactionError from trytond.wizard import StateAction, StateView __all__ = [ @@ -214,17 +214,23 @@ def with_transaction(user=1, context=None): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): - transaction = Transaction() - with transaction.start(DB_NAME, user, context=context): - try: - result = func(*args, **kwargs) - finally: - transaction.rollback() - # Clear remaining tasks - transaction.tasks.clear() - # Drop the cache as the transaction is rollbacked - Cache.drop(DB_NAME) - return result + extras = {} + while True: + with Transaction().start( + DB_NAME, user, context=context, + **extras) as transaction: + try: + result = func(*args, **kwargs) + except TransactionError as e: + e.fix(extras) + continue + finally: + transaction.rollback() + # Clear remaining tasks + transaction.tasks.clear() + # Drop the cache as the transaction is rollbacked + Cache.drop(DB_NAME) + return result return wrapper return decorator diff --git a/trytond/trytond/transaction.py b/trytond/trytond/transaction.py index 5605838962..8681e81e87 100644 --- a/trytond/trytond/transaction.py +++ b/trytond/trytond/transaction.py @@ -16,6 +16,21 @@ _cache_record = config.getint('cache', 'record') logger = logging.getLogger(__name__) +class TransactionError(Exception): + def fix(self, extras): + pass + + +class _TransactionLockError(TransactionError): + def __init__(self, table): + super().__init__() + self._table = table + + def fix(self, extras): + super().fix(extras) + extras.setdefault('_lock_tables', []).append(self._table) + + def record_cache_size(transaction): return transaction.context.get('_record_cache_size', _cache_record) @@ -104,45 +119,49 @@ class Transaction(object): return self.cache[(self.user, keys)] def start(self, database_name, user, readonly=False, context=None, - close=False, autocommit=False): + close=False, autocommit=False, **extras): ''' Start transaction ''' - from trytond import backend - assert self.user is None - assert self.database is None - assert self.close is None - assert self.context is None - # Compute started_at before connect to ensure - # it is strictly before all transactions started after - # but it may be also before transactions started before - self.started_at = self.monotonic_time() - if not database_name: - database = backend.Database().connect() - else: - database = backend.Database(database_name).connect() - Flavor.set(backend.Database.flavor) - self.connection = database.get_connection(readonly=readonly, - autocommit=autocommit) - self.user = user - self.database = database - self.readonly = readonly - self.close = close - self.context = ImmutableDict(context or {}) - self.create_records = defaultdict(set) - self.delete_records = defaultdict(set) - self.trigger_records = defaultdict(set) - self.check_warnings = set() - self.timestamp = {} - self.counter = 0 - self._datamanagers = [] - if database_name: - from trytond.cache import Cache - try: + try: + from trytond import backend + assert self.user is None + assert self.database is None + assert self.close is None + assert self.context is None + # Compute started_at before connect to ensure + # it is strictly before all transactions started after + # but it may be also before transactions started before + self.started_at = self.monotonic_time() + if not database_name: + database = backend.Database().connect() + else: + database = backend.Database(database_name).connect() + Flavor.set(backend.Database.flavor) + self.connection = database.get_connection(readonly=readonly, + autocommit=autocommit) + self.user = user + self.database = database + self.readonly = readonly + self.close = close + self.context = ImmutableDict(context or {}) + self.create_records = defaultdict(set) + self.delete_records = defaultdict(set) + self.trigger_records = defaultdict(set) + self.check_warnings = set() + self.timestamp = {} + self.counter = 0 + self._datamanagers = [] + lock_tables = extras.get('_lock_tables', []) + for table in lock_tables: + self.database.lock(self.connection, table) + self._locked_tables = set(lock_tables) + if database_name: + from trytond.cache import Cache Cache.sync(self) - except BaseException: - self.stop(False) - raise + except BaseException: + self.stop(False) + raise return self def __enter__(self): @@ -219,15 +238,19 @@ class Transaction(object): self.user = user return manager + def lock_table(self, table): + if table not in self._locked_tables: + raise _TransactionLockError(table) + def set_current_transaction(self, transaction): self._local.transactions.append(transaction) return transaction - def new_transaction(self, autocommit=False, readonly=False): + def new_transaction(self, autocommit=False, readonly=False, **extras): transaction = Transaction(new=True) return transaction.start(self.database.name, self.user, context=self.context, close=self.close, readonly=readonly, - autocommit=autocommit) + autocommit=autocommit, **extras) def commit(self): from trytond.cache import Cache diff --git a/trytond/trytond/worker.py b/trytond/trytond/worker.py index 546e607462..d2a5f28dd0 100644 --- a/trytond/trytond/worker.py +++ b/trytond/trytond/worker.py @@ -16,7 +16,7 @@ from trytond.config import config from trytond.exceptions import UserError, UserWarning from trytond.pool import Pool from trytond.status import processing -from trytond.transaction import Transaction +from trytond.transaction import Transaction, TransactionError __all__ = ['work'] logger = logging.getLogger(__name__) @@ -126,10 +126,14 @@ def run_task(pool, task_id): name = '' % (task_id, pool.database_name) retry = config.getint('database', 'retry') try: - for count in range(retry, -1, -1): - if count != retry: + count = 0 + transaction_extras = {} + while True: + if count: time.sleep(0.02 * (retry - count)) - with Transaction().start(pool.database_name, 0) as transaction: + with Transaction().start( + pool.database_name, 0, + **transaction_extras) as transaction: try: try: task, = Queue.search([('id', '=', task_id)]) @@ -139,10 +143,15 @@ def run_task(pool, task_id): with processing(name): task.run() break + except TransactionError as e: + transaction.rollbacked() + e.fix(transaction_extras) + continue except backend.DatabaseOperationalError: - if count: + if count < retry: transaction.rollback() - logger.debug("Retry: %i", retry - count + 1) + count += 1 + logger.debug("Retry: %i", count) continue raise except (UserError, UserWarning) as e: