I have a custom thread subclass that calls repeatedly a method on a “bound” object. The goal is to automatically join this thread whenever the “bound” object gets GC’ed:

from threading import Thread, Event
from weakref import ref, WeakSet
from typing import Callable, Generic
from typing_extensions import TypeVar
import atexit

T = TypeVar("T", bound=object)

_workers: WeakSet = WeakSet()

@atexit.register
def stopAllWorkers():
    # copy because _workers will be mutated during enumeration.
    refs = list(_workers)
    for worker in refs:
        try:
            worker.stop()
        except:
            pass


class MaintenanceWorker(Thread, Generic[T]):

    def __init__(self, bound_to: T, interval: float, task: Callable[[T], None]):
        self._interval = interval
        self._task = task
        self._ref = ref(bound_to, lambda _: self.stop())
        self._finished = Event()
        super().__init__()

    def stop(self):
        self._finished.set()
        _workers.discard(self)

    def run(self):
        self._finished.clear()
        _workers.add(self)
        while True:
            self._finished.wait(self._interval)
            if self._finished.is_set() or (subject := self._ref()) is None:
                _workers.discard(self)
                break
            try:
                self._task(subject)
            except Exception:
                pass
            finally:
                del subject

Will the instances of the following class be able to be garbage collected at all, as one of their members is a running thread?

class Foo: 

   def __init__(self):
     self._worker = MaintenaceWorker(bound_to=self, interval=15*60.0, Foo.bar)
     self._worker.start()

   def bar(self):
     # some convoluted logic 
     ...