From fb32f722ab3cc2d3ea8235fc3a2e411813e39bd0 Mon Sep 17 00:00:00 2001 From: Nico Andrade Date: Fri, 2 Oct 2020 17:24:27 -0400 Subject: [PATCH 01/11] Add timeout for MongoDB backend --- cachier/core.py | 9 ++++++++- cachier/mongo_core.py | 15 ++++++++++++--- 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/cachier/core.py b/cachier/core.py index 1e106d4..d0d811b 100644 --- a/cachier/core.py +++ b/cachier/core.py @@ -80,6 +80,7 @@ def cachier( mongetter=None, cache_dir=None, hash_params=None, + wait_for_calc_timeout=0 ): """A persistent, stale-free memoization decorator. @@ -118,6 +119,12 @@ def cachier( and returns a hash key for them. This parameter can be used to enable the use of cachier with functions that get arguments that are not automatically hashable by Python. + wait_for_calc_timeout: int, optional, for MongoDB only + The maximum time to wait for an ongoing calculation. When a + process started to calculate the value setting being_calculated to + True, any process trying to read the same entry will wait a maximum of + seconds specified in this parameter. 0 means wait forever. + Once the timeout expires the calculation will be triggered. """ # print('Inside the wrapper maker') # print('mongetter={}'.format(mongetter)) @@ -125,7 +132,7 @@ def cachier( # print('next_time={}'.format(next_time)) if mongetter: - core = _MongoCore(mongetter, stale_after, next_time) + core = _MongoCore(mongetter, stale_after, next_time, wait_for_calc_timeout) else: core = _PickleCore( # pylint: disable=R0204 stale_after=stale_after, diff --git a/cachier/mongo_core.py b/cachier/mongo_core.py index e05b6da..933c084 100644 --- a/cachier/mongo_core.py +++ b/cachier/mongo_core.py @@ -37,7 +37,7 @@ class _MongoCore(_BaseCore): _INDEX_NAME = 'func_1_key_1' - def __init__(self, mongetter, stale_after, next_time): + def __init__(self, mongetter, stale_after, next_time, wait_for_calc_timeout): if 'pymongo' not in sys.modules: warnings.warn(( "Cachier warning: pymongo was not found. " @@ -45,6 +45,7 @@ class _MongoCore(_BaseCore): _BaseCore.__init__(self, stale_after, next_time) self.mongetter = mongetter self.mongo_collection = self.mongetter() + self.wait_for_calc_timeout = wait_for_calc_timeout index_inf = self.mongo_collection.index_information() if _MongoCore._INDEX_NAME not in index_inf: func1key1 = IndexModel( @@ -131,13 +132,21 @@ class _MongoCore(_BaseCore): pass # don't care in this case def wait_on_entry_calc(self, key): + time_spent = 0 while True: time.sleep(MONGO_SLEEP_DURATION_IN_SEC) + time_spent += 1 key, entry = self.get_entry_by_key(key) if entry is None: raise RecalculationNeeded() - if entry is not None and not entry['being_calculated']: - return entry['value'] + + if entry is not None: + if not entry['being_calculated']: + return entry['value'] + + if self.wait_for_calc_timeout > 0 and time_spent >= self.wait_for_calc_timeout: + raise RecalculationNeeded() + def clear_cache(self): self.mongo_collection.delete_many( From 466f7669b490b400cf94874ec66edaf26dc37584 Mon Sep 17 00:00:00 2001 From: Nico Andrade Date: Fri, 2 Oct 2020 18:15:51 -0400 Subject: [PATCH 02/11] Add calculation timeout in MongoDB handler to avoid deadlocks --- cachier/mongo_core.py | 2 +- tests/test_mongo_core.py | 50 ++++++++++++++++++++++++++++++++++++++-- 2 files changed, 49 insertions(+), 3 deletions(-) diff --git a/cachier/mongo_core.py b/cachier/mongo_core.py index 933c084..b501be4 100644 --- a/cachier/mongo_core.py +++ b/cachier/mongo_core.py @@ -135,7 +135,7 @@ class _MongoCore(_BaseCore): time_spent = 0 while True: time.sleep(MONGO_SLEEP_DURATION_IN_SEC) - time_spent += 1 + time_spent += MONGO_SLEEP_DURATION_IN_SEC key, entry = self.get_entry_by_key(key) if entry is None: raise RecalculationNeeded() diff --git a/tests/test_mongo_core.py b/tests/test_mongo_core.py index b6fd845..9f0906f 100644 --- a/tests/test_mongo_core.py +++ b/tests/test_mongo_core.py @@ -99,7 +99,6 @@ def _stale_after_mongo(arg_1, arg_2): """Some function.""" return random() + arg_1 + arg_2 - def test_mongo_stale_after(): """Testing MongoDB core stale_after functionality.""" _stale_after_mongo.clear_cache() @@ -142,6 +141,53 @@ def test_mongo_being_calculated(): assert res1 == res2 +@cachier(mongetter=_test_mongetter, stale_after=MONGO_DELTA, next_time=False, wait_for_calc_timeout=2) +def _wait_for_calc_timeout_mongo_fast(arg_1, arg_2): + """Some function.""" + sleep(1) + return random() + arg_1 + arg_2 + + +def test_mongo_wait_for_calc_timeout_ok(): + """Testing MongoDB core handling of waiting for results.""" + _wait_for_calc_timeout_mongo_fast.clear_cache() + val1 = _wait_for_calc_timeout_mongo_fast(1, 2) + val2 = _wait_for_calc_timeout_mongo_fast(1, 2) + assert val1 == val2 + + +@cachier(mongetter=_test_mongetter, stale_after=MONGO_DELTA, next_time=False, wait_for_calc_timeout=2) +def _wait_for_calc_timeout_mongo_slow(arg_1, arg_2): + """Some function.""" + sleep(3) + return random() + arg_1 + arg_2 + + +def _calls_wait_for_calc_timeout_mongo_slow(res_queue): + res = _wait_for_calc_timeout_mongo_slow(1, 2) + res_queue.put(res) + + +def test_mongo_wait_for_calc_timeout_slow(): + """Testing MongoDB core handling of waiting for results.""" + _wait_for_calc_timeout_mongo_slow.clear_cache() + res_queue = queue.Queue() + thread1 = threading.Thread( + target=_calls_wait_for_calc_timeout_mongo_slow, kwargs={'res_queue': res_queue}) + thread2 = threading.Thread( + target=_calls_wait_for_calc_timeout_mongo_slow, kwargs={'res_queue': res_queue}) + + thread1.start() + thread2.start() + sleep(3) + thread1.join() + thread2.join() + assert res_queue.qsize() == 2 + res1 = res_queue.get() + res2 = res_queue.get() + assert res1 != res2 + + class _BadMongoCollection: def __init__(self, mongetter): @@ -188,7 +234,7 @@ def test_stalled_mongo_db_cache(): @cachier(mongetter=_test_mongetter) def _stalled_func(): return 1 - core = _MongoCore(_test_mongetter, None, False) + core = _MongoCore(_test_mongetter, None, False, 0) core.set_func(_stalled_func) core.clear_cache() with pytest.raises(RecalculationNeeded): From a676ad422d6e34d71860969f86d4bb674411d451 Mon Sep 17 00:00:00 2001 From: Nico Andrade Date: Fri, 2 Oct 2020 18:54:50 -0400 Subject: [PATCH 03/11] Add calculation timeout in MongoDB handler to avoid deadlocks --- tests/test_mongo_core.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/tests/test_mongo_core.py b/tests/test_mongo_core.py index 9f0906f..660626f 100644 --- a/tests/test_mongo_core.py +++ b/tests/test_mongo_core.py @@ -92,7 +92,7 @@ def test_mongo_core(): MONGO_DELTA = timedelta(seconds=3) - +MONGO_DELTA_LONG = timedelta(seconds=10) @cachier(mongetter=_test_mongetter, stale_after=MONGO_DELTA, next_time=False) def _stale_after_mongo(arg_1, arg_2): @@ -156,7 +156,8 @@ def test_mongo_wait_for_calc_timeout_ok(): assert val1 == val2 -@cachier(mongetter=_test_mongetter, stale_after=MONGO_DELTA, next_time=False, wait_for_calc_timeout=2) + +@cachier(mongetter=_test_mongetter, stale_after=MONGO_DELTA_LONG, next_time=False, wait_for_calc_timeout=2) def _wait_for_calc_timeout_mongo_slow(arg_1, arg_2): """Some function.""" sleep(3) @@ -185,7 +186,9 @@ def test_mongo_wait_for_calc_timeout_slow(): assert res_queue.qsize() == 2 res1 = res_queue.get() res2 = res_queue.get() - assert res1 != res2 + assert res1 != res2 # Timeout kicked in, hence two calls were done + res3 = _wait_for_calc_timeout_mongo_slow(1, 2) + assert res2 == res3 # The cached value is returned class _BadMongoCollection: From 3ab1a8d2b43174dca3083e444cc32c78bff28d98 Mon Sep 17 00:00:00 2001 From: Nico Andrade Date: Fri, 2 Oct 2020 19:10:30 -0400 Subject: [PATCH 04/11] Add calculation timeout in MongoDB handler to avoid deadlocks --- tests/test_mongo_core.py | 23 ++++++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/tests/test_mongo_core.py b/tests/test_mongo_core.py index 660626f..bfff4e5 100644 --- a/tests/test_mongo_core.py +++ b/tests/test_mongo_core.py @@ -147,15 +147,32 @@ def _wait_for_calc_timeout_mongo_fast(arg_1, arg_2): sleep(1) return random() + arg_1 + arg_2 +def _calls_wait_for_calc_timeout_mongo_fast(res_queue): + res = _wait_for_calc_timeout_mongo_fast(1, 2) + res_queue.put(res) + def test_mongo_wait_for_calc_timeout_ok(): - """Testing MongoDB core handling of waiting for results.""" _wait_for_calc_timeout_mongo_fast.clear_cache() val1 = _wait_for_calc_timeout_mongo_fast(1, 2) val2 = _wait_for_calc_timeout_mongo_fast(1, 2) assert val1 == val2 + res_queue = queue.Queue() + thread1 = threading.Thread( + target=_calls_wait_for_calc_timeout_mongo_fast, kwargs={'res_queue': res_queue}) + thread2 = threading.Thread( + target=_calls_wait_for_calc_timeout_mongo_fast, kwargs={'res_queue': res_queue}) + thread1.start() + thread2.start() + sleep(2) + thread1.join() + thread2.join() + assert res_queue.qsize() == 2 + res1 = res_queue.get() + res2 = res_queue.get() + assert res1 == res2 # Timeout did not kick in, a single call was done @cachier(mongetter=_test_mongetter, stale_after=MONGO_DELTA_LONG, next_time=False, wait_for_calc_timeout=2) def _wait_for_calc_timeout_mongo_slow(arg_1, arg_2): @@ -170,7 +187,7 @@ def _calls_wait_for_calc_timeout_mongo_slow(res_queue): def test_mongo_wait_for_calc_timeout_slow(): - """Testing MongoDB core handling of waiting for results.""" + """Testing for calls that time out are performed again.""" _wait_for_calc_timeout_mongo_slow.clear_cache() res_queue = queue.Queue() thread1 = threading.Thread( @@ -186,7 +203,7 @@ def test_mongo_wait_for_calc_timeout_slow(): assert res_queue.qsize() == 2 res1 = res_queue.get() res2 = res_queue.get() - assert res1 != res2 # Timeout kicked in, hence two calls were done + assert res1 != res2 # Timeout kicked in. Two calls were done res3 = _wait_for_calc_timeout_mongo_slow(1, 2) assert res2 == res3 # The cached value is returned From cbd515c4301decd78dbd0ce6656e3ee7ad7df7f2 Mon Sep 17 00:00:00 2001 From: Nico Andrade Date: Fri, 2 Oct 2020 19:12:31 -0400 Subject: [PATCH 05/11] Update test documentation --- tests/test_mongo_core.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_mongo_core.py b/tests/test_mongo_core.py index bfff4e5..8cb99ca 100644 --- a/tests/test_mongo_core.py +++ b/tests/test_mongo_core.py @@ -153,6 +153,7 @@ def _calls_wait_for_calc_timeout_mongo_fast(res_queue): def test_mongo_wait_for_calc_timeout_ok(): + """ Testing calls that avoid timeouts store the values in cache. """ _wait_for_calc_timeout_mongo_fast.clear_cache() val1 = _wait_for_calc_timeout_mongo_fast(1, 2) val2 = _wait_for_calc_timeout_mongo_fast(1, 2) @@ -187,7 +188,7 @@ def _calls_wait_for_calc_timeout_mongo_slow(res_queue): def test_mongo_wait_for_calc_timeout_slow(): - """Testing for calls that time out are performed again.""" + """Testing for calls timing out to be performed twice when needed.""" _wait_for_calc_timeout_mongo_slow.clear_cache() res_queue = queue.Queue() thread1 = threading.Thread( From 7f2b0dc35cca705ea4b78483a84d5d1318937fbe Mon Sep 17 00:00:00 2001 From: Nico Andrade Date: Fri, 2 Oct 2020 21:51:21 -0400 Subject: [PATCH 06/11] Update test for slow timeout --- tests/test_mongo_core.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_mongo_core.py b/tests/test_mongo_core.py index 8cb99ca..dbd3052 100644 --- a/tests/test_mongo_core.py +++ b/tests/test_mongo_core.py @@ -206,7 +206,7 @@ def test_mongo_wait_for_calc_timeout_slow(): res2 = res_queue.get() assert res1 != res2 # Timeout kicked in. Two calls were done res3 = _wait_for_calc_timeout_mongo_slow(1, 2) - assert res2 == res3 # The cached value is returned + assert res2 == res3 or res1 == res3 # One of the cached values is returned class _BadMongoCollection: From 101fbbf0464a7402805f611568af3a77ab6aa630 Mon Sep 17 00:00:00 2001 From: Nico Andrade Date: Mon, 5 Oct 2020 12:44:05 -0400 Subject: [PATCH 07/11] Add prints to follow the flow in CI --- cachier/mongo_core.py | 11 ++++++----- tests/test_mongo_core.py | 5 ++++- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/cachier/mongo_core.py b/cachier/mongo_core.py index b501be4..f481823 100644 --- a/cachier/mongo_core.py +++ b/cachier/mongo_core.py @@ -134,18 +134,19 @@ class _MongoCore(_BaseCore): def wait_on_entry_calc(self, key): time_spent = 0 while True: + print("Waiting for Mongo cache...") time.sleep(MONGO_SLEEP_DURATION_IN_SEC) time_spent += MONGO_SLEEP_DURATION_IN_SEC key, entry = self.get_entry_by_key(key) if entry is None: raise RecalculationNeeded() - if entry is not None: - if not entry['being_calculated']: - return entry['value'] + if not entry['being_calculated']: + return entry['value'] - if self.wait_for_calc_timeout > 0 and time_spent >= self.wait_for_calc_timeout: - raise RecalculationNeeded() + if self.wait_for_calc_timeout > 0 and time_spent >= self.wait_for_calc_timeout: + print("Got an entry. Is not valid. Force recomputation.", time_spent, self.wait_for_calc_timeout, time_spent >= self.wait_for_calc_timeout) + raise RecalculationNeeded() def clear_cache(self): diff --git a/tests/test_mongo_core.py b/tests/test_mongo_core.py index dbd3052..726c61e 100644 --- a/tests/test_mongo_core.py +++ b/tests/test_mongo_core.py @@ -177,12 +177,14 @@ def test_mongo_wait_for_calc_timeout_ok(): @cachier(mongetter=_test_mongetter, stale_after=MONGO_DELTA_LONG, next_time=False, wait_for_calc_timeout=2) def _wait_for_calc_timeout_mongo_slow(arg_1, arg_2): + print("_wait_for_calc_timeout_mongo_slow") """Some function.""" sleep(3) return random() + arg_1 + arg_2 def _calls_wait_for_calc_timeout_mongo_slow(res_queue): + print("_calls_wait_for_calc_timeout_mongo_slow") res = _wait_for_calc_timeout_mongo_slow(1, 2) res_queue.put(res) @@ -190,6 +192,7 @@ def _calls_wait_for_calc_timeout_mongo_slow(res_queue): def test_mongo_wait_for_calc_timeout_slow(): """Testing for calls timing out to be performed twice when needed.""" _wait_for_calc_timeout_mongo_slow.clear_cache() + print("Cache cleared") res_queue = queue.Queue() thread1 = threading.Thread( target=_calls_wait_for_calc_timeout_mongo_slow, kwargs={'res_queue': res_queue}) @@ -198,7 +201,7 @@ def test_mongo_wait_for_calc_timeout_slow(): thread1.start() thread2.start() - sleep(3) + sleep(4) thread1.join() thread2.join() assert res_queue.qsize() == 2 From 3c339cb9278b1cd1e79f0e3d936ec730df2dc970 Mon Sep 17 00:00:00 2001 From: Nico Andrade Date: Mon, 5 Oct 2020 13:09:29 -0400 Subject: [PATCH 08/11] Add prints to follow the flow in CI --- cachier/mongo_core.py | 3 ++- tests/test_mongo_core.py | 8 +++++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/cachier/mongo_core.py b/cachier/mongo_core.py index f481823..b1c986f 100644 --- a/cachier/mongo_core.py +++ b/cachier/mongo_core.py @@ -144,8 +144,9 @@ class _MongoCore(_BaseCore): if not entry['being_calculated']: return entry['value'] + print("Got an entry. It is being calculated. Do we wait?", time_spent, self.wait_for_calc_timeout, time_spent >= self.wait_for_calc_timeout) if self.wait_for_calc_timeout > 0 and time_spent >= self.wait_for_calc_timeout: - print("Got an entry. Is not valid. Force recomputation.", time_spent, self.wait_for_calc_timeout, time_spent >= self.wait_for_calc_timeout) + print("Tired of waiting. RecalculationNeeded.", time_spent, self.wait_for_calc_timeout, time_spent >= self.wait_for_calc_timeout) raise RecalculationNeeded() diff --git a/tests/test_mongo_core.py b/tests/test_mongo_core.py index 726c61e..7895daa 100644 --- a/tests/test_mongo_core.py +++ b/tests/test_mongo_core.py @@ -201,15 +201,17 @@ def test_mongo_wait_for_calc_timeout_slow(): thread1.start() thread2.start() + sleep(1) + res3 = _wait_for_calc_timeout_mongo_slow(1, 2) sleep(4) thread1.join() - thread2.join() + thread2.join() assert res_queue.qsize() == 2 res1 = res_queue.get() res2 = res_queue.get() assert res1 != res2 # Timeout kicked in. Two calls were done - res3 = _wait_for_calc_timeout_mongo_slow(1, 2) - assert res2 == res3 or res1 == res3 # One of the cached values is returned + res4 = _wait_for_calc_timeout_mongo_slow(1, 2) + assert res1 == res4 or res2 == res4 or res3 == res4 # One of the cached values is returned class _BadMongoCollection: From 4cb2afa4388b20f0083261d7c504e3edf970a3bb Mon Sep 17 00:00:00 2001 From: Nico Andrade Date: Mon, 5 Oct 2020 13:27:07 -0400 Subject: [PATCH 09/11] Add prints to follow the flow in CI --- cachier/mongo_core.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/cachier/mongo_core.py b/cachier/mongo_core.py index b1c986f..fe506ed 100644 --- a/cachier/mongo_core.py +++ b/cachier/mongo_core.py @@ -134,7 +134,6 @@ class _MongoCore(_BaseCore): def wait_on_entry_calc(self, key): time_spent = 0 while True: - print("Waiting for Mongo cache...") time.sleep(MONGO_SLEEP_DURATION_IN_SEC) time_spent += MONGO_SLEEP_DURATION_IN_SEC key, entry = self.get_entry_by_key(key) @@ -144,9 +143,7 @@ class _MongoCore(_BaseCore): if not entry['being_calculated']: return entry['value'] - print("Got an entry. It is being calculated. Do we wait?", time_spent, self.wait_for_calc_timeout, time_spent >= self.wait_for_calc_timeout) if self.wait_for_calc_timeout > 0 and time_spent >= self.wait_for_calc_timeout: - print("Tired of waiting. RecalculationNeeded.", time_spent, self.wait_for_calc_timeout, time_spent >= self.wait_for_calc_timeout) raise RecalculationNeeded() From adc9143f3a9b2a98699f2285fddcf9f67c96efbb Mon Sep 17 00:00:00 2001 From: Nico Andrade <33872298+non-senses@users.noreply.github.com> Date: Thu, 8 Oct 2020 16:00:14 -0400 Subject: [PATCH 10/11] Update test_mongo_core.py --- tests/test_mongo_core.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_mongo_core.py b/tests/test_mongo_core.py index 7895daa..862c6b1 100644 --- a/tests/test_mongo_core.py +++ b/tests/test_mongo_core.py @@ -178,7 +178,7 @@ def test_mongo_wait_for_calc_timeout_ok(): @cachier(mongetter=_test_mongetter, stale_after=MONGO_DELTA_LONG, next_time=False, wait_for_calc_timeout=2) def _wait_for_calc_timeout_mongo_slow(arg_1, arg_2): print("_wait_for_calc_timeout_mongo_slow") - """Some function.""" + """Some slow function.""" sleep(3) return random() + arg_1 + arg_2 From 5afa69098c66256431db5db15b66c4852e37c545 Mon Sep 17 00:00:00 2001 From: Nico Andrade <33872298+non-senses@users.noreply.github.com> Date: Thu, 8 Oct 2020 16:02:07 -0400 Subject: [PATCH 11/11] Remove prints and a comment --- tests/test_mongo_core.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/tests/test_mongo_core.py b/tests/test_mongo_core.py index 862c6b1..f34208a 100644 --- a/tests/test_mongo_core.py +++ b/tests/test_mongo_core.py @@ -177,14 +177,11 @@ def test_mongo_wait_for_calc_timeout_ok(): @cachier(mongetter=_test_mongetter, stale_after=MONGO_DELTA_LONG, next_time=False, wait_for_calc_timeout=2) def _wait_for_calc_timeout_mongo_slow(arg_1, arg_2): - print("_wait_for_calc_timeout_mongo_slow") - """Some slow function.""" sleep(3) return random() + arg_1 + arg_2 def _calls_wait_for_calc_timeout_mongo_slow(res_queue): - print("_calls_wait_for_calc_timeout_mongo_slow") res = _wait_for_calc_timeout_mongo_slow(1, 2) res_queue.put(res) @@ -192,7 +189,6 @@ def _calls_wait_for_calc_timeout_mongo_slow(res_queue): def test_mongo_wait_for_calc_timeout_slow(): """Testing for calls timing out to be performed twice when needed.""" _wait_for_calc_timeout_mongo_slow.clear_cache() - print("Cache cleared") res_queue = queue.Queue() thread1 = threading.Thread( target=_calls_wait_for_calc_timeout_mongo_slow, kwargs={'res_queue': res_queue})