changeset c35bbc670aef in trytond:6.0
details: https://hg.tryton.org/trytond?cmd=changeset&node=c35bbc670aef
description:
Avoid concurrency failure in cache with channel test
We need to ensure that the listener is running on the return of the
transaction
start otherwise notification can be missed.
We must also just from time to time check that the listener is still the
current listener.
And finally we can not wait an arbitrary time for cache sync, instead
we must
rely on the _clean_last being reset with the sync happens.
issue9413
review364831002
(grafted from a32b84d9664917df3274663c06ab57fd5a7789ef)
diffstat:
trytond/cache.py | 7 ++++++-
trytond/tests/test_cache.py | 19 +++++++++++--------
2 files changed, 17 insertions(+), 9 deletions(-)
diffs (92 lines):
diff -r 37a409b66d14 -r c35bbc670aef trytond/cache.py
--- a/trytond/cache.py Tue Feb 08 18:44:55 2022 +0100
+++ b/trytond/cache.py Mon Feb 14 23:12:00 2022 +0100
@@ -6,6 +6,7 @@
import os
import select
import threading
+import time
from collections import OrderedDict, defaultdict
from datetime import datetime
from weakref import WeakKeyDictionary
@@ -216,6 +217,9 @@
cls._listener[pid, dbname] = listener = threading.Thread(
target=cls._listen, args=(dbname,), daemon=True)
listener.start()
+ while (not getattr(listener, 'listening', False)
+ and listener.is_alive()):
+ time.sleep(.01)
return
if (datetime.now() - cls._clean_last).total_seconds() < _clear_timeout:
return
@@ -354,9 +358,10 @@
try:
cursor = conn.cursor()
cursor.execute('LISTEN "%s"' % cls._channel)
+ current_thread.listening = True
while cls._listener.get((pid, dbname)) == current_thread:
- readable, _, _ = select.select([conn], [], [])
+ readable, _, _ = select.select([conn], [], [], 60)
if not readable:
continue
diff -r 37a409b66d14 -r c35bbc670aef trytond/tests/test_cache.py
--- a/trytond/tests/test_cache.py Tue Feb 08 18:44:55 2022 +0100
+++ b/trytond/tests/test_cache.py Mon Feb 14 23:12:00 2022 +0100
@@ -1,6 +1,7 @@
# 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 datetime as dt
import time
import unittest
@@ -58,7 +59,7 @@
def tearDown(self):
MemoryCache.drop(DB_NAME)
- def wait_cache_sync(self):
+ def wait_cache_sync(self, after=None):
pass
@with_transaction()
@@ -96,11 +97,9 @@
with Transaction().set_current_transaction(transaction1):
self.assertEqual(cache.get('foo'), 'bar')
+ commit_time = dt.datetime.now()
transaction2.commit()
- for n in range(10):
- if cache.get('foo') == 'baz':
- break
- self.wait_cache_sync()
+ self.wait_cache_sync(after=commit_time)
self.assertEqual(cache.get('foo'), 'baz')
def test_memory_cache_nested_transactions(self):
@@ -135,8 +134,9 @@
transaction2 = transaction1.new_transaction()
self.addCleanup(transaction2.stop)
cache.clear()
+ commit_time = dt.datetime.now()
transaction2.commit()
- self.wait_cache_sync()
+ self.wait_cache_sync(after=commit_time)
# Set value from old transaction
Transaction().set_current_transaction(transaction1)
@@ -168,8 +168,11 @@
self.addCleanup(
setattr, cache_mod, '_clear_timeout', clear_timeout)
- def wait_cache_sync(self):
- time.sleep(1)
+ def wait_cache_sync(self, after=None):
+ if after is None:
+ after = dt.datetime.now()
+ while MemoryCache._clean_last < after:
+ time.sleep(.01)
@unittest.skip("No cache sync on transaction start with channel")
def test_memory_cache_sync(self):
