ENH: freeze the shared entry-point loader mapping

dascore/utils/plugins.py

@@ -7,9 +7,23 @@
 from importlib.metadata import entry_points
 from typing import Any
 
+from dascore.utils.mapping import FrozenDict
+
+# These caches are deliberately unsynchronized, unlike the rest of the
+# concurrency work in this subsystem. Concurrent misses on the same key can run
+# the wrapped function more than once, which is harmless here. The only side
+# effect either has is importing the plugin, and CPython runs a module body
+# exactly once behind its per-module import lock; DASCore plugins are
+# module-level classes, so once that import succeeds every racing caller
+# resolves the same already-registered object. An import that raises is dropped
+# from `sys.modules` and simply retried by the next caller, which is the
+# behavior we want anyway. Electing a single loader would only save a duplicate
+# `entry_points()` scan and, for duplicate names, a repeated warning below --
+# not worth the lock choreography it takes to arrange.
+
 
 @functools.cache
-def get_entry_point_loaders(entry_point_group: str) -> dict[str, Any]:
+def get_entry_point_loaders(entry_point_group: str) -> FrozenDict[str, Any]:
     """Return cached entry-point loaders keyed by entry-point name."""
     out: dict[str, Any] = {}
     duplicate_names: set[str] = set()
@@ -24,7 +38,8 @@ def get_entry_point_loaders(entry_point_group: str) -> dict[str, Any]:
             "Using the last registered entry point for each name."
         )
         warnings.warn(msg, UserWarning, stacklevel=2)
-    return out
+    # Every caller shares this one mapping, so it must not be mutable.
+    return FrozenDict(out)
 
 
 @functools.cache

tests/test_utils/test_plugins.py

@@ -2,6 +2,9 @@
 
 from __future__ import annotations
 
+import sys
+import threading
+from importlib.metadata import EntryPoint
 from types import SimpleNamespace
 
 import pytest
@@ -63,3 +66,79 @@ def second_loader():
 
         assert out["dup"] is second_loader
         assert out["unique"]() == "unique"
+
+    def test_loaders_mapping_is_immutable(self, monkeypatch):
+        """The cached loader mapping is shared, so callers cannot mutate it."""
+        entry_point_list = [SimpleNamespace(name="unique", load=lambda: "unique")]
+        plugin_mod.get_entry_point_loaders.cache_clear()
+        monkeypatch.setattr(
+            plugin_mod, "entry_points", lambda *, group: entry_point_list
+        )
+
+        out = plugin_mod.get_entry_point_loaders("test.group")
+
+        with pytest.raises(TypeError):
+            out["another"] = None
+
+
+class TestConcurrentEntryPointLoading:
+    """Tests for concurrent misses on the unsynchronized plugin caches."""
+
+    module_name = "dascore_slow_test_plugin"
+
+    @pytest.fixture
+    def execution_log(self, tmp_path):
+        """Return the path the test plugin appends to each time it executes."""
+        return tmp_path / "executions.txt"
+
+    @pytest.fixture
+    def slow_entry_point(self, tmp_path, execution_log, monkeypatch):
+        """Return an entry point for a plugin with a deliberately slow import."""
+        # The module records executions in a file rather than in its own
+        # namespace: a second execution builds a new module object, so a
+        # counter kept in module globals could not see the first one.
+        source = (
+            "import time\n"
+            f"open({str(execution_log)!r}, 'a').write('x')\n"
+            # Hold the import lock long enough for the other threads to reach
+            # it while this one is still running the module body.
+            "time.sleep(0.2)\n"
+            "class Thing:\n"
+            "    pass\n"
+        )
+        (tmp_path / f"{self.module_name}.py").write_text(source)
+        monkeypatch.syspath_prepend(str(tmp_path))
+        value = f"{self.module_name}:Thing"
+        yield EntryPoint(name="thing", value=value, group="test.group")
+        sys.modules.pop(self.module_name, None)
+
+    @pytest.mark.concurrency
+    def test_racing_loads_import_the_plugin_once(
+        self, slow_entry_point, execution_log, monkeypatch, run_in_threads
+    ):
+        """Racing callers share one plugin object although the cache has no lock."""
+        thread_count = 4
+        # Every thread must be inside the loader before any may proceed, so a
+        # run whose calls did not really overlap fails instead of quietly
+        # asserting something weaker.
+        entered = threading.Barrier(thread_count, timeout=30)
+
+        def load():
+            entered.wait()
+            return slow_entry_point.load()
+
+        plugin_mod.maybe_load_entry_point.cache_clear()
+        monkeypatch.setattr(
+            plugin_mod, "get_entry_point_loaders", lambda _: {"thing": load}
+        )
+
+        results = run_in_threads(
+            lambda _: plugin_mod.maybe_load_entry_point("test.group", "thing"),
+            count=thread_count,
+        )
+
+        # The cache misses on every thread, but the import system runs the
+        # module body once, so nothing observes a duplicated or half-built
+        # plugin. This is why the caches need no lock of their own.
+        assert len({id(x) for x in results}) == 1
+        assert execution_log.read_text() == "x"