Let one thread at a time read a native results instance

device-detection/fiftyone_devicedetection_onpremise/src/fiftyone_devicedetection_onpremise/swig_data.py

@@ -20,6 +20,8 @@
 # such notice(s) shall fulfill the requirements of that article.
 # *********************************************************************
 
+import threading
+
 from fiftyone_pipeline_engines.aspectdata import AspectData
 from fiftyone_pipeline_core.aspectproperty_value import AspectPropertyValue
 
@@ -61,6 +63,11 @@ def __init__(self, flow_element, swig_results):
 
         self.missing_property_service = OnPremiseMissingPropertyService()
 
+        # Reading a value writes into the native results, and the first read
+        # of a component can walk its graph, so one thread at a time reads
+        # this instance.
+        self._read_lock = threading.RLock()
+
     def close(self):
 
         """!
@@ -70,7 +77,8 @@ def close(self):
         exception. Closing the FlowData closes its results.
         """
 
-        self.swig_results = None
+        with self._read_lock:
+            self.swig_results = None
 
     def get(self, key):
 
@@ -81,12 +89,13 @@ def get(self, key):
         @returns mixed
         """
 
-        # Checked here, because AspectData.get turns an exception from
-        # get_internal into a missing property message.
-        if self.swig_results is None:
-            raise Exception(RESULTS_CLOSED)
+        with self._read_lock:
+            # Checked here, because AspectData.get turns an exception from
+            # get_internal into a missing property message.
+            if self.swig_results is None:
+                raise Exception(RESULTS_CLOSED)
 
-        return super(SwigData, self).get(key)
+            return super(SwigData, self).get(key)
 
     def get_internal(self, key):
 

device-detection/fiftyone_devicedetection_onpremise/tests/test_swig_data_threads.py

@@ -0,0 +1,133 @@
+# *********************************************************************
+# This Original Work is copyright of 51 Degrees Mobile Experts Limited.
+# Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House,
+# Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU.
+#
+# This Original Work is licensed under the European Union Public Licence
+# (EUPL) v.1.2 and is subject to its terms as set out below.
+#
+# If a copy of the EUPL was not distributed with this file, You can obtain
+# one at https://opensource.org/licenses/EUPL-1.2.
+#
+# The 'Compatible Licences' set out in the Appendix to the EUPL (as may be
+# amended by the European Commission) shall be deemed incompatible for
+# the purposes of the Work and the provisions of the compatibility
+# clause in Article 5 of the EUPL shall not apply.
+#
+# If using the Work as, or as part of, a network application, by
+# including the attribution notice(s) required under Article 5 of the EUPL
+# in the end user terms of the application under an appropriate heading,
+# such notice(s) shall fulfill the requirements of that article.
+# *********************************************************************
+
+"""
+One thread at a time reads a native results instance, because reading a value
+writes into the results.
+"""
+
+import threading
+import time
+import unittest
+
+from fiftyone_devicedetection_onpremise.swig_data import SwigData
+
+
+# How long a stand-in read takes, long enough that overlapping reads are seen.
+READ_SECONDS = 0.02
+# The longest a test waits for another thread.
+WAIT_SECONDS = 10
+# How long closing is given while a read holds the results. It must still be
+# waiting at the end.
+BLOCKED_SECONDS = 0.2
+
+KEY = "deviceid"
+VALUE = "1-2-3"
+
+
+class OverlapCountingResults:
+    """Stands in for the native results and records how many reads overlap."""
+
+    def __init__(self):
+        self.guard = threading.Lock()
+        self.active = 0
+        self.most_active = 0
+
+    def getDeviceId(self):
+        with self.guard:
+            self.active += 1
+            self.most_active = max(self.most_active, self.active)
+        time.sleep(READ_SECONDS)
+        with self.guard:
+            self.active -= 1
+        return "1-2-3"
+
+
+class BlockingResults:
+    """Stands in for the native results with a read that waits to be let go,
+    so a test can act while the read is in progress."""
+
+    def __init__(self):
+        self.reading = threading.Event()
+        self.release = threading.Event()
+
+    def getDeviceId(self):
+        self.reading.set()
+        self.release.wait(WAIT_SECONDS)
+        return "1-2-3"
+
+
+class FakeElement:
+    datakey = "device"
+    properties = {}
+
+
+class SwigDataThreadTests(unittest.TestCase):
+
+    def test_reads_from_several_threads_do_not_overlap(self):
+        results = OverlapCountingResults()
+        data = SwigData(FakeElement(), results)
+        errors = []
+
+        def read():
+            try:
+                for _ in range(3):
+                    data.get("deviceid")
+            except Exception as e:
+                errors.append(e)
+
+        threads = [threading.Thread(target=read) for _ in range(6)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+
+        self.assertEqual([], errors)
+        self.assertEqual(1, results.most_active)
+
+    def test_closing_waits_for_a_read_in_progress(self):
+        results = BlockingResults()
+        data = SwigData(FakeElement(), results)
+        values = []
+        reader = threading.Thread(
+            target=lambda: values.append(data.get(KEY)))
+        reader.start()
+        self.assertTrue(results.reading.wait(WAIT_SECONDS))
+
+        closer = threading.Thread(target=data.close)
+        closer.start()
+        closer.join(BLOCKED_SECONDS)
+        # The read holds the results, so closing waits for it.
+        self.assertTrue(closer.is_alive())
+
+        results.release.set()
+        reader.join(WAIT_SECONDS)
+        closer.join(WAIT_SECONDS)
+        self.assertFalse(closer.is_alive())
+        self.assertEqual([VALUE], [value.value() for value in values])
+        self.assertIsNone(data.swig_results)
+        with self.assertRaises(Exception):
+            data.get(KEY)
+
+
+if __name__ == "__main__":
+    unittest.main()

ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/swig_data.py

@@ -21,6 +21,7 @@
 # *********************************************************************
 
 import math
+import threading
 from decimal import Decimal, ROUND_HALF_UP
 
 from fiftyone_pipeline_engines.aspectdata import AspectData
@@ -82,6 +83,11 @@ def __init__(self, flow_element, swig_results, echo_ip=None, echo_ipv6=None):
 
         self.missing_property_service = OnPremiseMissingPropertyService()
 
+        # Reading a value writes into the native results, and the first read
+        # of a component can walk its graph, so one thread at a time reads
+        # this instance.
+        self._read_lock = threading.RLock()
+
     def close(self):
 
         """!
@@ -91,7 +97,8 @@ def close(self):
         exception. Closing the FlowData closes its results.
         """
 
-        self.swig_results = None
+        with self._read_lock:
+            self.swig_results = None
 
     def get(self, key):
 
@@ -102,12 +109,13 @@ def get(self, key):
         @returns mixed
         """
 
-        # Checked here, because AspectData.get turns an exception from
-        # get_internal into a missing property message.
-        if self.swig_results is None:
-            raise Exception(RESULTS_CLOSED)
+        with self._read_lock:
+            # Checked here, because AspectData.get turns an exception from
+            # get_internal into a missing property message.
+            if self.swig_results is None:
+                raise Exception(RESULTS_CLOSED)
 
-        return super(SwigData, self).get(key)
+            return super(SwigData, self).get(key)
 
     @staticmethod
     def utf8_string_to_str(utf8_string):

ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_swig_data_threads.py

@@ -0,0 +1,142 @@
+# *********************************************************************
+# This Original Work is copyright of 51 Degrees Mobile Experts Limited.
+# Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House,
+# Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU.
+#
+# This Original Work is licensed under the European Union Public Licence
+# (EUPL) v.1.2 and is subject to its terms as set out below.
+#
+# If a copy of the EUPL was not distributed with this file, You can obtain
+# one at https://opensource.org/licenses/EUPL-1.2.
+#
+# The 'Compatible Licences' set out in the Appendix to the EUPL (as may be
+# amended by the European Commission) shall be deemed incompatible for
+# the purposes of the Work and the provisions of the compatibility
+# clause in Article 5 of the EUPL shall not apply.
+#
+# If using the Work as, or as part of, a network application, by
+# including the attribution notice(s) required under Article 5 of the EUPL
+# in the end user terms of the application under an appropriate heading,
+# such notice(s) shall fulfill the requirements of that article.
+# *********************************************************************
+
+"""
+One thread at a time reads a native results instance, because reading a value
+writes into the results.
+"""
+
+import threading
+import time
+import unittest
+
+from fiftyone_ipintelligence_onpremise.swig_data import SwigData
+
+
+class Value:
+
+    def hasValue(self):
+        return True
+
+    def getValue(self):
+        return 5
+
+
+# How long a stand-in read takes, long enough that overlapping reads are seen.
+READ_SECONDS = 0.02
+# The longest a test waits for another thread.
+WAIT_SECONDS = 10
+# How long closing is given while a read holds the results. It must still be
+# waiting at the end.
+BLOCKED_SECONDS = 0.2
+
+KEY = "count"
+VALUE = 5
+
+
+class OverlapCountingResults:
+    """Stands in for the native results and records how many reads overlap."""
+
+    def __init__(self):
+        self.guard = threading.Lock()
+        self.active = 0
+        self.most_active = 0
+
+    def getValueAsInteger(self, name):
+        with self.guard:
+            self.active += 1
+            self.most_active = max(self.most_active, self.active)
+        time.sleep(READ_SECONDS)
+        with self.guard:
+            self.active -= 1
+        return Value()
+
+
+class BlockingResults:
+    """Stands in for the native results with a read that waits to be let go,
+    so a test can act while the read is in progress."""
+
+    def __init__(self):
+        self.reading = threading.Event()
+        self.release = threading.Event()
+
+    def getValueAsInteger(self, name):
+        self.reading.set()
+        self.release.wait(WAIT_SECONDS)
+        return Value()
+
+
+class FakeElement:
+    datakey = "ip"
+    properties = {"count": {"name": "Count", "type": "int"}}
+
+
+class SwigDataThreadTests(unittest.TestCase):
+
+    def test_reads_from_several_threads_do_not_overlap(self):
+        results = OverlapCountingResults()
+        data = SwigData(FakeElement(), results)
+        errors = []
+
+        def read():
+            try:
+                for _ in range(3):
+                    data.get("count")
+            except Exception as e:
+                errors.append(e)
+
+        threads = [threading.Thread(target=read) for _ in range(6)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+
+        self.assertEqual([], errors)
+        self.assertEqual(1, results.most_active)
+
+    def test_closing_waits_for_a_read_in_progress(self):
+        results = BlockingResults()
+        data = SwigData(FakeElement(), results)
+        values = []
+        reader = threading.Thread(
+            target=lambda: values.append(data.get(KEY)))
+        reader.start()
+        self.assertTrue(results.reading.wait(WAIT_SECONDS))
+
+        closer = threading.Thread(target=data.close)
+        closer.start()
+        closer.join(BLOCKED_SECONDS)
+        # The read holds the results, so closing waits for it.
+        self.assertTrue(closer.is_alive())
+
+        results.release.set()
+        reader.join(WAIT_SECONDS)
+        closer.join(WAIT_SECONDS)
+        self.assertFalse(closer.is_alive())
+        self.assertEqual([VALUE], [value.value() for value in values])
+        self.assertIsNone(data.swig_results)
+        with self.assertRaises(Exception):
+            data.get(KEY)
+
+
+if __name__ == "__main__":
+    unittest.main()