import sys import threading import weakref import gc class A: def __init__(self): self.cycle = self done = False def my_thread(): global done d = weakref.WeakValueDictionary() while not done: a = A() d[42] = a #list(d) if not d: done = True print('FAIL') # 'a' is alive at least until here, so the assert above should # always pass---but it does not. def my_thread2(): while not done: gc.collect() t1 = threading.Thread(target=my_thread) t2 = threading.Thread(target=my_thread2) sys.setswitchinterval(1e-6) t1.start() t2.start() t1.join() t2.join()