Refactor stacking for reuse and overlappability

Currently, the stacking feature adds a lot of code to various parts of ShmemTrainMatcher. It would be nice if this could be cleaner, for:

• more maintainable ShmemTrainMatcher
• reusability: the DetectorAssembler also does stacking except in its own less good way
• reusability: we will at some point probably split ShmemTrainMatcher and some of the resulting matching-and-processing devices will want to stack
• improvability: the current implementation is pretty naive and though the handle_source is intended to be run in a thread pool, only one train can be processed at a time; this requires rethinking of buffer handling to change
  • bonus concern: the stacking buffers, if we start overlapping processing and decoupling this from sending, would not satisfy the safeNDArray flag in the current version

Plan for this MR:

• Move stacking functionality to "friend" class
• Get rid of global stacking buffers; idea: create, per train, a stacking context with fresh buffers
• Add some testing of stacking (not just utils, but the friend class working on tables)
• (Out of scope) Test that it works in a overlapping trains setting
• (Out of scope) Start overlapping train processing
• Refactor DetectorAssembler to use StackingFriend to stack before assembly
• Change the configuration that the manager passes to DetectorAssembler correspondingly
• Nevermind, DetectorAssembler is not a great fit actually; maybe use xarray and exploit extra-geom's partial data assembly functionality (Out of scope; see !74 (closed))
• Test everything with actual Karabo (pending node allocation)

src/tests/test_stacking.py

+import concurrent.futures
+
 from karabo.bound import Hash
 import numpy as np
 import pytest
@@ -23,7 +25,17 @@ class NotADevice:
 datas = [np.arange(i * 100, i * 100 + 8).reshape(4, 2) for i in range(3)]
 
 
-class TestSourceStacking:
+class CommonTestFixtureGuy:
+    @pytest.fixture(params=[False, True])
+    def thread_pool(self, request):
+        if request.param:
+            with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
+                yield pool
+        else:
+            yield None
+
+
+class TestSourceStacking(CommonTestFixtureGuy):
     @pytest.fixture
     def friend(self):
         source_table = [
@@ -56,8 +68,8 @@ class TestSourceStacking:
             for i, data in enumerate(datas)
         }
 
-    def test_simple(self, friend, sources):
-        friend.process(sources)
+    def test_simple(self, friend, sources, thread_pool):
+        friend.process(sources, thread_pool)
         assert "newSource" in sources
         stacked = sources["newSource"][0]["keyToStack"]
         assert stacked.shape == (4, 3, 2)
@@ -65,9 +77,9 @@ class TestSourceStacking:
             assert np.array_equal(stacked[:, i], data, equal_nan=True)
         assert not friend._device.warnings
 
-    def test_missing_source(self, friend, sources):
+    def test_missing_source(self, friend, sources, thread_pool):
         del sources["source0"]
-        friend.process(sources)
+        friend.process(sources, thread_pool)
         assert "newSource" in sources
         stacked = sources["newSource"][0]["keyToStack"]
         assert np.all(stacked[:, 0] == 0)
@@ -75,9 +87,9 @@ class TestSourceStacking:
             assert np.array_equal(stacked[:, i], data, equal_nan=True)
         assert not friend._device.warnings
 
-    def test_missing_data(self, friend, sources):
+    def test_missing_data(self, friend, sources, thread_pool):
         sources["source0"][0].erase("keyToStack")
-        friend.process(sources)
+        friend.process(sources, thread_pool)
         assert "newSource" in sources
         stacked = sources["newSource"][0]["keyToStack"]
         assert np.all(stacked[:, 0] == 0)
@@ -85,17 +97,17 @@ class TestSourceStacking:
             assert np.array_equal(stacked[:, i], data, equal_nan=True)
         assert not friend._device.warnings
 
-    def test_source_stacking_no_sources(self, friend, sources):
+    def test_source_stacking_no_sources(self, friend, sources, thread_pool):
         sources = {}
-        friend.process(sources)
+        friend.process(sources, thread_pool)
         assert friend._device.warnings
         assert not sources
 
-    def test_source_stacking_erroneous_data(self, friend, sources):
+    def test_source_stacking_erroneous_data(self, friend, sources, thread_pool):
         sources["source1"][0][
             "keyToStack"
         ] = "and now for something completely different"
-        friend.process(sources)
+        friend.process(sources, thread_pool)
         assert "newSource" in sources
         stacked = sources["newSource"][0]["keyToStack"]
         for i, data in enumerate(datas):
@@ -106,7 +118,7 @@ class TestSourceStacking:
         assert friend._device.warnings
 
 
-class TestKeyStacking:
+class TestKeyStacking(CommonTestFixtureGuy):
     @pytest.fixture
     def friend(self):
         source_table = [
@@ -138,8 +150,8 @@ class TestKeyStacking:
             h[f"key{i}"] = data
         return {"source": (h, None)}
 
-    def test_simple(self, friend, sources):
-        friend.process(sources)
+    def test_simple(self, friend, sources, thread_pool):
+        friend.process(sources, thread_pool)
         assert sources["source"][0].has("newKey")
         stacked = sources["source"][0]["newKey"]
         for i, data in enumerate(datas):