From e5e92ac81acfa044221175a203326fce1f04c779 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?C=C3=A9dric=20Krier?= Date: Sat, 9 Mar 2019 17:03:38 +0100 Subject: [PATCH] Allow to use channel to synchronize Cache When the clear timeout of the cache is zero, we launch a thread to synchronize caches if the back-end support channels. The thread is started on the first call to sync for the database. The channel is also used to clear the cache of the process initiating the clearing. This allows to test the mechanism is working with a single process. So there will be a small delay even on single instance. issue8027 review68431002 --- CHANGELOG | 1 + doc/topics/configuration.rst | 2 + trytond/cache.py | 132 ++++++++++++++++++++++++++++------- trytond/tests/test_cache.py | 62 +++++++++++----- 4 files changed, 154 insertions(+), 43 deletions(-) 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