Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
refcount_before = sys.getrefcount(args)
g = greenlet.greenlet(
lambda *args: greenlet.getcurrent().parent.switch(*args))
for i in range(100):
g.switch(*args)
self.assertEqual(sys.getrefcount(args), refcount_before)
def test_kwarg_refs(self):
kwargs = {}
g = greenlet.greenlet(
lambda **kwargs: greenlet.getcurrent().parent.switch(**kwargs))
for i in range(100):
g.switch(**kwargs)
self.assertEqual(sys.getrefcount(kwargs), 2)
if greenlet.GREENLET_USE_GC:
# These only work with greenlet gc support
def recycle_threads(self):
# By introducing a thread that does sleep we allow other threads,
# that have triggered their __block condition, but did not have a
# chance to deallocate their thread state yet, to finally do so.
# The way it works is by requring a GIL switch (different thread),
# which does a GIL release (sleep), which might do a GIL switch
# to finished threads and allow them to clean up.
def worker():
time.sleep(0.001)
t = threading.Thread(target=worker)
t.start()
time.sleep(0.001)
t.join()
import gc
import sys
import unittest
import weakref
import greenlet
class GCTests(unittest.TestCase):
def test_dead_circular_ref(self):
o = weakref.ref(greenlet.greenlet(greenlet.getcurrent).switch())
gc.collect()
self.assertTrue(o() is None)
self.assertFalse(gc.garbage, gc.garbage)
if greenlet.GREENLET_USE_GC:
# These only work with greenlet gc support
def test_circular_greenlet(self):
class circular_greenlet(greenlet.greenlet):
pass
o = circular_greenlet()
o.self = o
o = weakref.ref(o)
gc.collect()
self.assertTrue(o() is None)
self.assertFalse(gc.garbage, gc.garbage)
def test_inactive_ref(self):
class inactive_greenlet(greenlet.greenlet):
def __init__(self):
greenlet.greenlet.__init__(self, run=self.run)