From 82ee9c2aa9cdca050083d0557588dc40ed874001 Mon Sep 17 00:00:00 2001
From: Harsh Suryawanshi <146956919+Harsh2806@users.noreply.github.com>
Date: Tue, 2 Dec 2025 22:41:41 +0000
Subject: [PATCH] Add thread-safe Singleton implementation with comprehensive
 tests

- Implement SingletonBase metaclass for thread-safe singleton pattern
- Add wrt_singleton.py module with reusable singleton implementation
- Add comprehensive thread-safety tests in test_singleton_threadsafe.py
- Integrate singleton pattern into patcher.py for genetic algorithm optimization
---
 .../algorithms/genetic/patcher.py             | 16 +------
 tests/test_singleton_threadsafe.py            | 43 +++++++++++++++++++
 wrt_singleton.py                              | 41 ++++++++++++++++++
 3 files changed, 85 insertions(+), 15 deletions(-)
 create mode 100644 tests/test_singleton_threadsafe.py
 create mode 100644 wrt_singleton.py

diff --git a/WeatherRoutingTool/algorithms/genetic/patcher.py b/WeatherRoutingTool/algorithms/genetic/patcher.py
index c196a01..0d732f6 100644
--- a/WeatherRoutingTool/algorithms/genetic/patcher.py
+++ b/WeatherRoutingTool/algorithms/genetic/patcher.py
@@ -4,6 +4,7 @@
 import os
 from datetime import datetime
 from pathlib import Path
+from wrt_singleton import SingletonBase
 
 import numpy as np
 from astropy import units as u
@@ -39,22 +40,7 @@ def patch(self, src: tuple, dst: tuple):
         raise NotImplementedError("This patching method is not implemented.")
 
 
-class SingletonBase(type):
-    """
-    TODO: make this thread-safe
-    Base class for Singleton implementation of patcher methods.
-
-    This is the implementation of a metaclass for those classes for which only a single instance shall be available
-    during runtime.
-    """
-    _instances = {}
-
-    def __call__(cls, *args, **kwargs):
-        if cls not in cls._instances:
-            instance = super().__call__(*args, **kwargs)
-            cls._instances[cls] = instance
 
-        return cls._instances[cls]
 
 
 # patcher variants
diff --git a/tests/test_singleton_threadsafe.py b/tests/test_singleton_threadsafe.py
new file mode 100644
index 0000000..877b09b
--- /dev/null
+++ b/tests/test_singleton_threadsafe.py
@@ -0,0 +1,43 @@
+import threading
+import time
+
+from wrt_singleton import SingletonBase
+
+
+class DummySingleton(metaclass=SingletonBase):
+    def __init__(self):
+        # small delay to increase chance of concurrent construction
+        time.sleep(0.01)
+
+
+class OtherSingleton(metaclass=SingletonBase):
+    def __init__(self):
+        time.sleep(0.005)
+
+
+def _create(instances, idx, cls):
+    instances[idx] = cls()
+
+
+def test_singleton_threadsafe_single_class():
+    n_threads = 50
+    instances = [None] * n_threads
+    threads = []
+
+    for i in range(n_threads):
+        t = threading.Thread(target=_create, args=(instances, i, DummySingleton))
+        threads.append(t)
+        t.start()
+
+    for t in threads:
+        t.join()
+
+    # all entries should be the same object
+    ids = {id(x) for x in instances}
+    assert len(ids) == 1
+
+
+def test_singleton_threadsafe_different_classes():
+    a = DummySingleton()
+    b = OtherSingleton()
+    assert a is not b
diff --git a/wrt_singleton.py b/wrt_singleton.py
new file mode 100644
index 0000000..d7c3bb9
--- /dev/null
+++ b/wrt_singleton.py
@@ -0,0 +1,41 @@
+"""Thread-safe Singleton metaclass used by the project.
+
+This helper is deliberately implemented as a top-level module so it can be
+imported for lightweight unit tests without importing the full
+`WeatherRoutingTool` package (which triggers heavy imports at package init).
+
+The metaclass guarantees that only one instance per class is created even
+when multiple threads try to instantiate concurrently.
+"""
+from __future__ import annotations
+
+import threading
+from typing import Any, Dict, Type
+
+
+class SingletonBase(type):
+    """Thread-safe Singleton metaclass.
+
+    Usage:
+        class MyClass(metaclass=SingletonBase):
+            pass
+
+        a = MyClass()
+        b = MyClass()
+        assert a is b
+    """
+
+    _instances: Dict[Type[Any], Any] = {}
+    _lock = threading.RLock()
+
+    def __call__(cls, *args, **kwargs):
+        # Fast path: return existing instance without locking
+        if cls in SingletonBase._instances:
+            return SingletonBase._instances[cls]
+
+        # Slow path: acquire lock and create instance if still missing
+        with SingletonBase._lock:
+            if cls not in SingletonBase._instances:
+                SingletonBase._instances[cls] = super().__call__(*args, **kwargs)
+
+            return SingletonBase._instances[cls]