diff --git a/CHANGELOG b/CHANGELOG index e72a778170..d23802d550 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -1,3 +1,4 @@ +* Allow to use channel to synchronize Cache * Display ids and rule names on access error * Move field definition from Model to Field * Use number of verbose flag as log level diff --git a/doc/topics/configuration.rst b/doc/topics/configuration.rst index 61f0665f68..cc40841d9b 100644 --- a/doc/topics/configuration.rst +++ b/doc/topics/configuration.rst @@ -192,6 +192,8 @@ clean_timeout ~~~~~~~~~~~~~ The minimum number of seconds between two cleanings of the cache. +If the value is 0, the notification between processes will be done using +channels if the back-end supports them. Default: `300` diff --git a/trytond/cache.py b/trytond/cache.py index e73ff96d88..5583d38a3d 100644 --- a/trytond/cache.py +++ b/trytond/cache.py @@ -1,5 +1,9 @@ # 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 json +import logging +import select +import threading from collections import OrderedDict, defaultdict from datetime import datetime from weakref import WeakKeyDictionary @@ -15,6 +19,7 @@ from trytond.tools import resolve __all__ = ['BaseCache', 'Cache', 'LRUDict'] _clear_timeout = config.getint('cache', 'clean_timeout', default=5 * 60) +logger = logging.getLogger(__name__) def _cast(column): @@ -84,6 +89,10 @@ class MemoryCache(BaseCache): _reset = WeakKeyDictionary() _clean_last = datetime.now() _default_lower = Transaction.monotonic_time() + _listener = {} + _listener_lock = threading.Lock() + _table = 'ir_cache' + _channel = _table def __init__(self, name, size_limit=1024, context=True): super(MemoryCache, self).__init__(name, size_limit, context) @@ -101,8 +110,9 @@ class MemoryCache(BaseCache): try: return self._transaction_cache[transaction] except KeyError: - self._transaction_cache[transaction] = LRUDict(self.size_limit) - return self._transaction_cache[transaction] + cache = self._database_cache.default_factory() + self._transaction_cache[transaction] = cache + return cache else: return self._database_cache[dbname] @@ -129,13 +139,28 @@ class MemoryCache(BaseCache): self._reset.setdefault(transaction, set()).add(self._name) self._transaction_cache.pop(transaction, None) + def _clear(self, dbname, timestamp=None): + logger.debug("clearing cache '%s' of '%s'", self._name, dbname) + self._timestamp[dbname] = timestamp + self._database_cache[dbname] = self._database_cache.default_factory() + self._transaction_lower[dbname] = max( + Transaction.monotonic_time(), + self._transaction_lower.get(dbname, self._default_lower)) + @classmethod def sync(cls, transaction): + dbname = transaction.database.name + if not _clear_timeout and transaction.database.has_channel(): + with cls._listener_lock: + if dbname not in cls._listener: + cls._listener[dbname] = listener = threading.Thread( + target=cls._listen, args=(dbname,), daemon=True) + listener.start() + return if (datetime.now() - cls._clean_last).total_seconds() < _clear_timeout: return - dbname = transaction.database.name with transaction.connection.cursor() as cursor: - table = Table('ir_cache') + table = Table(cls._table) cursor.execute(*table.select(_cast(table.timestamp), table.name)) timestamps = {} for timestamp, name in cursor.fetchall(): @@ -147,34 +172,40 @@ class MemoryCache(BaseCache): continue inst_timestamp = inst._timestamp.get(dbname) if not inst_timestamp or timestamp > inst_timestamp: - inst._timestamp[dbname] = timestamp - inst._database_cache[dbname] = LRUDict(inst.size_limit) - inst._transaction_lower[dbname] = max( - transaction.monotonic_time(), - inst._transaction_lower.get(dbname, cls._default_lower)) + inst._clear(dbname, timestamp) cls._clean_last = datetime.now() @classmethod def commit(cls, transaction): - table = Table('ir_cache') + table = Table(cls._table) reset = cls._reset.setdefault(transaction, set()) if not reset: return dbname = transaction.database.name with transaction.connection.cursor() as cursor: - 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()], + if not _clear_timeout and transaction.database.has_channel(): + 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()], + 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( _cast(Max(table.timestamp)), @@ -182,11 +213,8 @@ class MemoryCache(BaseCache): timestamp, = cursor.fetchone() inst = cls._instances[name] - inst._timestamp[dbname] = timestamp - inst._database_cache[dbname] = LRUDict(inst.size_limit) - inst._transaction_lower[dbname] = max( - transaction.monotonic_time(), - inst._transaction_lower.get(dbname, cls._default_lower)) + inst._clear(dbname, timestamp) + reset.clear() @classmethod def rollback(cls, transaction): @@ -197,11 +225,61 @@ class MemoryCache(BaseCache): @classmethod def drop(cls, dbname): + with cls._listener_lock: + listener = cls._listener.pop(dbname, None) + if listener: + Database = backend.get('Database') + database = Database(dbname) + conn = database.get_connection() + try: + cursor = conn.cursor() + cursor.execute('NOTIFY "%s"' % cls._channel) + conn.commit() + finally: + database.put_connection(conn) + listener.join() for inst in cls._instances.values(): inst._timestamp.pop(dbname, None) inst._database_cache.pop(dbname, None) inst._transaction_lower.pop(dbname, None) + @classmethod + def _listen(cls, dbname): + Database = backend.get('Database') + database = Database(dbname) + if not database.has_channel(): + raise NotImplementedError + + logger.info("listening on channel '%s' of '%s'", cls._channel, dbname) + conn = database.get_connection() + try: + cursor = conn.cursor() + cursor.execute('LISTEN "%s"' % cls._channel) + conn.commit() + + while cls._listener.get(dbname) == threading.current_thread(): + readable, _, _ = select.select([conn], [], []) + if not readable: + continue + + conn.poll() + while conn.notifies: + notification = conn.notifies.pop() + if notification.payload: + reset = json.loads(notification.payload) + for name in reset: + inst = cls._instances[name] + inst._clear(dbname) + except Exception: + logger.error( + "cache listener on '%s' crashed", dbname, exc_info=True) + raise + finally: + database.put_connection(conn) + with cls._listener_lock: + if cls._listener.get(dbname) == threading.current_thread(): + del cls._listener[dbname] + if config.get('cache', 'class'): Cache = resolve(config.get('cache', 'class')) diff --git a/trytond/tests/test_cache.py b/trytond/tests/test_cache.py index 537c204a42..76bcd81bf1 100644 --- a/trytond/tests/test_cache.py +++ b/trytond/tests/test_cache.py @@ -4,7 +4,7 @@ import time import unittest -from trytond import cache as cache_mod +from trytond import backend, cache as cache_mod from trytond.cache import freeze, MemoryCache from trytond.tests.test_tryton import with_transaction, activate_module from trytond.tests.test_tryton import DB_NAME, USER @@ -17,20 +17,6 @@ cache = MemoryCache('test.cache') class CacheTestCase(unittest.TestCase): "Test Cache" - @classmethod - def setUpClass(cls): - activate_module('tests') - - def setUp(self): - super().setUp() - clear_timeout = cache_mod._clear_timeout - cache_mod._clear_timeout = 1 - self.addCleanup( - setattr, cache_mod, '_clear_timeout', clear_timeout) - - def tearDown(self): - MemoryCache.drop(DB_NAME) - def testFreeze(self): "Test freeze" self.assertEqual(freeze([1, 2, 3]), (1, 2, 3)) @@ -53,6 +39,27 @@ class CacheTestCase(unittest.TestCase): ('string', 'test'), ]))]))])) + +class MemoryCacheTestCase(unittest.TestCase): + "Test Cache" + + @classmethod + def setUpClass(cls): + activate_module('tests') + + def setUp(self): + super().setUp() + clear_timeout = cache_mod._clear_timeout + cache_mod._clear_timeout = 1 + self.addCleanup( + setattr, cache_mod, '_clear_timeout', clear_timeout) + + def tearDown(self): + MemoryCache.drop(DB_NAME) + + def wait_cache_sync(self): + pass + @with_transaction() def test_memory_cache_set_get(self): "Test MemoryCache set/get" @@ -90,6 +97,7 @@ class CacheTestCase(unittest.TestCase): self.assertEqual(cache.get('foo'), 'bar') transaction2.commit() + self.wait_cache_sync() self.assertEqual(cache.get('foo'), None) def test_memory_cache_sync(self): @@ -124,9 +132,31 @@ class CacheTestCase(unittest.TestCase): self.assertEqual(cache.get('foo'), None) +@unittest.skipIf(backend.name() == 'sqlite', "SQLite has not channel") +class MemoryCacheChannelTestCase(MemoryCacheTestCase): + "Test Cache with channel" + + def setUp(self): + super().setUp() + clear_timeout = cache_mod._clear_timeout + cache_mod._clear_timeout = 0 + self.addCleanup( + setattr, cache_mod, '_clear_timeout', clear_timeout) + + def wait_cache_sync(self): + time.sleep(1) + + @unittest.skip("No cache sync on transaction start with channel") + def test_memory_cache_sync(self): + super().test_memory_cache_sync() + + def suite(): func = unittest.TestLoader().loadTestsFromTestCase suite = unittest.TestSuite() - for testcase in (CacheTestCase,): + for testcase in [ + CacheTestCase, + MemoryCacheTestCase, + MemoryCacheChannelTestCase]: suite.addTests(func(testcase)) return suite