tryton-do/tryton
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Lock table on transaction start

Browse Source 
When trying to lock a table, we check if it was locked at the start of the
transaction otherwise we raise a TransactionError so the all transaction is
restarted over with the table added to the table to lock.
This ensures when a table is locked that we always read the last committed
values.

Closes #10870
This commit is contained in:
Cédric Krier
2022-12-26 15:05:56 +01:00
parent 63400adeb3
commit dd222215eb
12 changed files with 196 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proteus/proteus/config.py
+30 -18
View File
@@ -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)
trytond/CHANGELOG
+1
View File
@@ -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
trytond/doc/ref/transaction.rst
+16 -2
View File
@@ -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.
trytond/trytond/commandline.py
+3
View File
@@ -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
trytond/trytond/console.py
+2 -1
View File
@@ -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,
trytond/trytond/ir/cron.py
+15 -6
View File
@@ -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 %s@%s %s>' % (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)
trytond/trytond/model/modelsql.py
+1 -1
View File
@@ -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):
trytond/trytond/protocols/dispatcher.py
+15 -7
View File
@@ -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
trytond/trytond/tests/test_modelsql.py
+20 -5
View File
@@ -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):
trytond/trytond/tests/test_tryton.py
+18 -12
View File
@@ -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
trytond/trytond/transaction.py
+60 -37
View File
@@ -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
trytond/trytond/worker.py
+15 -6
View File
@@ -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 %s@%s>' % (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: