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

trytond/cache.py

@@ -6,6 +6,7 @@ import logging
 import os
 import select
 import threading
+import time
 from collections import OrderedDict, defaultdict
 from datetime import datetime
 from weakref import WeakKeyDictionary
@@ -220,6 +221,9 @@ class MemoryCache(BaseCache):
                     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
@@ -355,9 +359,10 @@ class MemoryCache(BaseCache):
         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
 

trytond/tests/test_cache.py

@@ -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
 
@@ -81,7 +82,7 @@ class MemoryCacheTestCase(unittest.TestCase):
     def tearDown(self):
         MemoryCache.drop(DB_NAME)
 
-    def wait_cache_sync(self):
+    def wait_cache_sync(self, after=None):
         pass
 
     @with_transaction()
@@ -119,11 +120,9 @@ class MemoryCacheTestCase(unittest.TestCase):
         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):
@@ -158,8 +157,9 @@ class MemoryCacheTestCase(unittest.TestCase):
         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)
@@ -191,8 +191,11 @@ class MemoryCacheChannelTestCase(MemoryCacheTestCase):
         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):