Skip to content

Add thread-safe Singleton implementation with comprehensive tests #94

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Harsh2806 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add thread-safe Singleton implementation with comprehensive tests #94

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 1 addition & 15 deletions WeatherRoutingTool/algorithms/genetic/patcher.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
43 changes: 43 additions & 0 deletions tests/test_singleton_threadsafe.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
41 changes: 41 additions & 0 deletions wrt_singleton.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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]