test_leaks.py 文件源码

def test_threaded_leak(self):
            gg = []
            def worker():
                # only main greenlet present
                gg.append(weakref.ref(greenlet.getcurrent()))
            for i in range(2):
                t = threading.Thread(target=worker)
                t.start()
                t.join()
                del t
            greenlet.getcurrent() # update ts_current
            self.recycle_threads()
            greenlet.getcurrent() # update ts_current
            gc.collect()
            greenlet.getcurrent() # update ts_current
            for g in gg:
                self.assertTrue(g() is None)