Use separate connection to query cache table
The table is a hot spot for contention so we must minimize the duration lock are kept on it. As at the start of a transaction, the cache may be sync and so it locks all the row in the table, the transaction used for this query should not be the main one that lasts a long time. Idem for the commit. issue8447 review283591002
This commit is contained in:
+50
-35
@@ -165,8 +165,9 @@ class MemoryCache(BaseCache):
|
||||
|
||||
@classmethod
|
||||
def sync(cls, transaction):
|
||||
dbname = transaction.database.name
|
||||
if not _clear_timeout and transaction.database.has_channel():
|
||||
database = transaction.database
|
||||
dbname = database.name
|
||||
if not _clear_timeout and database.has_channel():
|
||||
with cls._listener_lock:
|
||||
if dbname not in cls._listener:
|
||||
cls._listener[dbname] = listener = threading.Thread(
|
||||
@@ -175,12 +176,17 @@ class MemoryCache(BaseCache):
|
||||
return
|
||||
if (datetime.now() - cls._clean_last).total_seconds() < _clear_timeout:
|
||||
return
|
||||
with transaction.connection.cursor() as cursor:
|
||||
table = Table(cls._table)
|
||||
cursor.execute(*table.select(_cast(table.timestamp), table.name))
|
||||
timestamps = {}
|
||||
for timestamp, name in cursor.fetchall():
|
||||
timestamps[name] = timestamp
|
||||
connection = database.get_connection(readonly=True, autocommit=True)
|
||||
try:
|
||||
with connection.cursor() as cursor:
|
||||
table = Table(cls._table)
|
||||
cursor.execute(*table.select(
|
||||
_cast(table.timestamp), table.name))
|
||||
timestamps = {}
|
||||
for timestamp, name in cursor.fetchall():
|
||||
timestamps[name] = timestamp
|
||||
finally:
|
||||
database.put_connection(connection)
|
||||
for name, timestamp in timestamps.items():
|
||||
try:
|
||||
inst = cls._instances[name]
|
||||
@@ -197,39 +203,48 @@ class MemoryCache(BaseCache):
|
||||
reset = cls._reset.setdefault(transaction, set())
|
||||
if not reset:
|
||||
return
|
||||
dbname = transaction.database.name
|
||||
with transaction.connection.cursor() as cursor:
|
||||
if not _clear_timeout and transaction.database.has_channel():
|
||||
database = transaction.database
|
||||
dbname = database.name
|
||||
if not _clear_timeout and transaction.database.has_channel():
|
||||
with transaction.connection.cursor() as cursor:
|
||||
cursor.execute(
|
||||
'NOTIFY "%s", %%s' % cls._channel,
|
||||
(json.dumps(list(reset), separators=(',', ':')),))
|
||||
else:
|
||||
for name in reset:
|
||||
cursor.execute(*table.select(table.name,
|
||||
where=table.name == name,
|
||||
limit=1))
|
||||
if cursor.fetchone():
|
||||
# It would be better to insert only
|
||||
cursor.execute(*table.update([table.timestamp],
|
||||
[CurrentTimestamp()],
|
||||
else:
|
||||
connection = database.get_connection(
|
||||
readonly=False, autocommit=True)
|
||||
try:
|
||||
with connection.cursor() as cursor:
|
||||
for name in reset:
|
||||
cursor.execute(*table.select(table.name, table.id,
|
||||
table.timestamp,
|
||||
where=table.name == name,
|
||||
limit=1))
|
||||
if cursor.fetchone():
|
||||
# It would be better to insert only
|
||||
cursor.execute(*table.update([table.timestamp],
|
||||
[CurrentTimestamp()],
|
||||
where=table.name == name))
|
||||
else:
|
||||
cursor.execute(*table.insert(
|
||||
[table.timestamp, table.name],
|
||||
[[CurrentTimestamp(), name]]))
|
||||
|
||||
cursor.execute(*table.select(
|
||||
Max(table.timestamp),
|
||||
where=table.name == name))
|
||||
else:
|
||||
cursor.execute(*table.insert(
|
||||
[table.timestamp, table.name],
|
||||
[[CurrentTimestamp(), name]]))
|
||||
timestamp, = cursor.fetchone()
|
||||
|
||||
cursor.execute(*table.select(
|
||||
Max(table.timestamp),
|
||||
where=table.name == name))
|
||||
timestamp, = cursor.fetchone()
|
||||
cursor.execute(*table.select(
|
||||
_cast(Max(table.timestamp)),
|
||||
where=table.name == name))
|
||||
timestamp, = cursor.fetchone()
|
||||
|
||||
cursor.execute(*table.select(
|
||||
_cast(Max(table.timestamp)),
|
||||
where=table.name == name))
|
||||
timestamp, = cursor.fetchone()
|
||||
|
||||
inst = cls._instances[name]
|
||||
inst._clear(dbname, timestamp)
|
||||
inst = cls._instances[name]
|
||||
inst._clear(dbname, timestamp)
|
||||
connection.commit()
|
||||
finally:
|
||||
database.put_connection(connection)
|
||||
reset.clear()
|
||||
|
||||
@classmethod
|
||||
|
||||
@@ -14,6 +14,7 @@ from sql import Table
|
||||
from sql.functions import CurrentTimestamp
|
||||
|
||||
import trytond.tools as tools
|
||||
from trytond.cache import Cache
|
||||
from trytond.config import config
|
||||
from trytond.exceptions import MissingDependenciesException
|
||||
from trytond.transaction import Transaction
|
||||
@@ -266,6 +267,8 @@ def load_module_graph(graph, pool, update=None, lang=None):
|
||||
]))
|
||||
module2state[module] = 'activated'
|
||||
|
||||
# Avoid clearing cache to prevent dead lock on ir.cache table
|
||||
Cache.rollback(transaction)
|
||||
transaction.commit()
|
||||
|
||||
if not update:
|
||||
|
||||
@@ -104,6 +104,19 @@ class MemoryCacheTestCase(unittest.TestCase):
|
||||
self.wait_cache_sync()
|
||||
self.assertEqual(cache.get('foo'), None)
|
||||
|
||||
def test_memory_cache_nested_transactions(self):
|
||||
"Test MemoryCache with nested transactions"
|
||||
# Create entry in the cache table to trigger 2 updates
|
||||
with Transaction().start(DB_NAME, USER):
|
||||
cache.clear()
|
||||
# Ensure sync is performed on start
|
||||
time.sleep(cache_mod._clear_timeout)
|
||||
|
||||
with Transaction().start(DB_NAME, USER) as transaction1:
|
||||
cache.clear()
|
||||
with transaction1.new_transaction():
|
||||
cache.clear()
|
||||
|
||||
def test_memory_cache_sync(self):
|
||||
"Test MemoryCache synchronisation"
|
||||
with Transaction().start(DB_NAME, USER):
|
||||
|
||||
Reference in New Issue
Block a user