From cda36c13917fd7a0212b0e453a7237cf593c8999 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 22:55:16 +0000 Subject: [PATCH 1/4] Test that one thread at a time reads a native results instance --- .../tests/test_swig_data_threads.py | 74 +++++++++++++++++ .../tests/test_swig_data_threads.py | 83 +++++++++++++++++++ 2 files changed, 157 insertions(+) create mode 100644 device-detection/fiftyone_devicedetection_onpremise/tests/test_swig_data_threads.py create mode 100644 ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_swig_data_threads.py diff --git a/device-detection/fiftyone_devicedetection_onpremise/tests/test_swig_data_threads.py b/device-detection/fiftyone_devicedetection_onpremise/tests/test_swig_data_threads.py new file mode 100644 index 00000000..8daca00c --- /dev/null +++ b/device-detection/fiftyone_devicedetection_onpremise/tests/test_swig_data_threads.py @@ -0,0 +1,74 @@ + +""" +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 + + +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(0.02) + with self.guard: + self.active -= 1 + 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 = OverlapCountingResults() + data = SwigData(FakeElement(), results) + reader = threading.Thread(target=lambda: data.get("deviceid")) + reader.start() + time.sleep(0.005) + data.close() + reader.join() + self.assertIsNone(data.swig_results) + with self.assertRaises(Exception): + data.get("deviceid") + + +if __name__ == "__main__": + unittest.main() diff --git a/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_swig_data_threads.py b/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_swig_data_threads.py new file mode 100644 index 00000000..0b61bbfa --- /dev/null +++ b/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_swig_data_threads.py @@ -0,0 +1,83 @@ + +""" +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 + + +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(0.02) + with self.guard: + self.active -= 1 + 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 = OverlapCountingResults() + data = SwigData(FakeElement(), results) + reader = threading.Thread(target=lambda: data.get("count")) + reader.start() + time.sleep(0.005) + data.close() + reader.join() + self.assertIsNone(data.swig_results) + with self.assertRaises(Exception): + data.get("count") + + +if __name__ == "__main__": + unittest.main() From aff617ed73a24db043b59475d2ce0986549ea1b7 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 22:55:38 +0000 Subject: [PATCH 2/4] Let one thread at a time read a native results instance Reading a value writes into the native results, and the first read of a component can walk its graph, so each read, and the close that frees the results, holds a lock on the instance. --- .../swig_data.py | 21 +++++++++++++------ .../swig_data.py | 21 +++++++++++++------ 2 files changed, 30 insertions(+), 12 deletions(-) diff --git a/device-detection/fiftyone_devicedetection_onpremise/src/fiftyone_devicedetection_onpremise/swig_data.py b/device-detection/fiftyone_devicedetection_onpremise/src/fiftyone_devicedetection_onpremise/swig_data.py index 17987a59..bc4e73ef 100644 --- a/device-detection/fiftyone_devicedetection_onpremise/src/fiftyone_devicedetection_onpremise/swig_data.py +++ b/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): diff --git a/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/swig_data.py b/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/swig_data.py index 40413cf3..6c91f86c 100644 --- a/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_onpremise/swig_data.py +++ b/ip-intelligence/fiftyone_ipintelligence_onpremise/src/fiftyone_ipintelligence_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 @@ -73,6 +75,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): """! @@ -82,7 +89,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): @@ -93,12 +101,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): From 43f4f95782e78f85bba80ef357a29c93b04a1b68 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Tue, 10 Nov 2026 10:29:04 +0000 Subject: [PATCH 3/4] Hold the stand-in read open so the closing test fails unless closing waits, and name the test times --- .../tests/test_swig_data_threads.py | 52 ++++++++++++++++--- .../tests/test_swig_data_threads.py | 52 ++++++++++++++++--- 2 files changed, 90 insertions(+), 14 deletions(-) diff --git a/device-detection/fiftyone_devicedetection_onpremise/tests/test_swig_data_threads.py b/device-detection/fiftyone_devicedetection_onpremise/tests/test_swig_data_threads.py index 8daca00c..a86f20f4 100644 --- a/device-detection/fiftyone_devicedetection_onpremise/tests/test_swig_data_threads.py +++ b/device-detection/fiftyone_devicedetection_onpremise/tests/test_swig_data_threads.py @@ -11,6 +11,18 @@ 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.""" @@ -23,12 +35,26 @@ def getDeviceId(self): with self.guard: self.active += 1 self.most_active = max(self.most_active, self.active) - time.sleep(0.02) + 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 = {} @@ -58,16 +84,28 @@ def read(): self.assertEqual(1, results.most_active) def test_closing_waits_for_a_read_in_progress(self): - results = OverlapCountingResults() + results = BlockingResults() data = SwigData(FakeElement(), results) - reader = threading.Thread(target=lambda: data.get("deviceid")) + values = [] + reader = threading.Thread( + target=lambda: values.append(data.get(KEY))) reader.start() - time.sleep(0.005) - data.close() - reader.join() + 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("deviceid") + data.get(KEY) if __name__ == "__main__": diff --git a/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_swig_data_threads.py b/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_swig_data_threads.py index 0b61bbfa..065cd14b 100644 --- a/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_swig_data_threads.py +++ b/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_swig_data_threads.py @@ -20,6 +20,18 @@ 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.""" @@ -32,12 +44,26 @@ def getValueAsInteger(self, name): with self.guard: self.active += 1 self.most_active = max(self.most_active, self.active) - time.sleep(0.02) + 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"}} @@ -67,16 +93,28 @@ def read(): self.assertEqual(1, results.most_active) def test_closing_waits_for_a_read_in_progress(self): - results = OverlapCountingResults() + results = BlockingResults() data = SwigData(FakeElement(), results) - reader = threading.Thread(target=lambda: data.get("count")) + values = [] + reader = threading.Thread( + target=lambda: values.append(data.get(KEY))) reader.start() - time.sleep(0.005) - data.close() - reader.join() + 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("count") + data.get(KEY) if __name__ == "__main__": From 1149be08f886052cf158aba10c33e6423c6bf7c3 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Fri, 13 Nov 2026 23:35:24 +0000 Subject: [PATCH 4/4] Give the results thread tests the standard licence header --- .../tests/test_swig_data_threads.py | 21 +++++++++++++++++++ .../tests/test_swig_data_threads.py | 21 +++++++++++++++++++ 2 files changed, 42 insertions(+) diff --git a/device-detection/fiftyone_devicedetection_onpremise/tests/test_swig_data_threads.py b/device-detection/fiftyone_devicedetection_onpremise/tests/test_swig_data_threads.py index a86f20f4..e07f169d 100644 --- a/device-detection/fiftyone_devicedetection_onpremise/tests/test_swig_data_threads.py +++ b/device-detection/fiftyone_devicedetection_onpremise/tests/test_swig_data_threads.py @@ -1,3 +1,24 @@ +# ********************************************************************* +# 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 diff --git a/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_swig_data_threads.py b/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_swig_data_threads.py index 065cd14b..c1476c43 100644 --- a/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_swig_data_threads.py +++ b/ip-intelligence/fiftyone_ipintelligence_onpremise/tests/test_swig_data_threads.py @@ -1,3 +1,24 @@ +# ********************************************************************* +# 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