Thread safe cache

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 0.3.1
+
+- Make `ExpressionCache` thread-safe, it raised a spurious `FPMLValidationError` when an entry was evicted mid-lookup
+
 ## 0.3.0
 
 - Add configurable LRU cache of compiled FHIRPath expressions instead of parsing them on every evaluation #37 (@ruscoder)

python/README.md

@@ -169,7 +169,7 @@ Parsing FHIRPath expressions is expensive, so expressions can be compiled once a
 kept in memory, zero disables caching.
 
 Entries are keyed by the expression only, while compilation binds the model and the user-defined
-functions, so keep one long-living cache per `fp_options`.
+functions, so keep one long-living cache per `fp_options`. A cache is safe to share between threads.
 
 ```python
 from fhirpathpy.models import models

python/fpml/core/cache.py

@@ -1,3 +1,4 @@
+import threading
 from collections import OrderedDict
 from typing import Any, Callable, Optional, cast
 
@@ -21,30 +22,36 @@ class ExpressionCache:
 
     Entries are keyed by the expression only, while compilation binds the model
     and the user-defined functions, so use a separate cache per fp_options.
-    Zero max size disables caching.
+    Zero max size disables caching. Instances are safe to share between threads.
     """
 
     def __init__(self, max_size: int) -> None:
         self.max_size = max_size
         self._compiled: OrderedDict[str, CompiledExpression] = OrderedDict()
+        self._lock = threading.Lock()
 
     def compile(self, expression: str, fp_options: Optional[FPOptions]) -> CompiledExpression:
-        cached = self._compiled.get(expression)
-        if cached is not None:
-            self._compiled.move_to_end(expression)
+        with self._lock:
+            cached = self._compiled.get(expression)
+            if cached is not None:
+                self._compiled.move_to_end(expression)
 
-            return cached
+                return cached
 
+        # Compiling outside the lock, so that it does not hold up the other threads
         compiled = compile_expression(expression, fp_options)
+
         if self.max_size > 0:
-            self._compiled[expression] = compiled
-            if len(self._compiled) > self.max_size:
-                self._compiled.popitem(last=False)
+            with self._lock:
+                self._compiled[expression] = compiled
+                if len(self._compiled) > self.max_size:
+                    self._compiled.popitem(last=False)
 
         return compiled
 
     def clear(self) -> None:
-        self._compiled.clear()
+        with self._lock:
+            self._compiled.clear()
 
     @property
     def size(self) -> int:

python/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fpml"
-version = "0.3.0"
+version = "0.3.1"
 description = "The FHIRPath mapping language is a data DSL designed to convert data from QuestionnaireResponse (and not only) to any FHIR Resource."
 authors = [{ name = "Beda Software", email = "ilya@beda.software" }]
 maintainers = [

python/tests/core/test_cache.py

@@ -1,3 +1,7 @@
+import sys
+from collections.abc import Iterator
+from itertools import repeat
+from threading import Thread
 from typing import Any
 
 import pytest
@@ -41,6 +45,44 @@ def test_clear_drops_compiled_expressions() -> None:
     assert cache.compile("list.key", None) is not compiled
 
 
+def test_stays_consistent_when_shared_between_threads(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Stub the compilation, so that the cache bookkeeping is what threads contend over
+    monkeypatch.setattr(cache_module, "compile_expression", lambda expression, _: expression)
+    max_size = 2
+    cache = ExpressionCache(max_size=max_size)
+    errors: list[Exception] = []
+
+    def hammer(expressions: Iterator[str]) -> None:
+        for expression in expressions:
+            try:
+                cache.compile(expression, None)
+            except Exception as exc:
+                errors.append(exc)
+                return
+
+    threads = [
+        # Readers keep hitting one entry while churners evict it from under them
+        *[Thread(target=hammer, args=(repeat("hot", 20000),)) for _ in range(4)],
+        *[
+            Thread(target=hammer, args=((f"churn{index}-{n}" for n in range(20000)),))
+            for index in range(4)
+        ],
+    ]
+    switch_interval = sys.getswitchinterval()
+    # Preempt threads aggressively to widen the window between a lookup and its eviction
+    sys.setswitchinterval(1e-6)
+    try:
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+    finally:
+        sys.setswitchinterval(switch_interval)
+
+    assert errors == []
+    assert cache.size == max_size
+
+
 def test_resolve_template_compiles_repeated_expression_once(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:

ts/server/package.json

@@ -1,6 +1,6 @@
 {
     "name": "fpml-server",
-    "version": "0.3.0",
+    "version": "0.3.1",
     "description": "The FHIRPath mapping language is a data DSL designed to convert data from QuestionnaireResponse (and not only) to any FHIR Resource.",
     "author": "beda.software",
     "private": true,