[pypy-svn] r45934 - pypy/branch/pypy-more-rtti-inprogress/module/thread/test

[arigo at codespeak.net]

Author: arigo
Date: Thu Aug 23 15:42:58 2007
New Revision: 45934

Modified:
   pypy/branch/pypy-more-rtti-inprogress/module/thread/test/test_ll_thread.py
Log:
Write test_start_new_thread in the same style as os_thread.py.
Check that the references held to other objects don't leak.


Modified: pypy/branch/pypy-more-rtti-inprogress/module/thread/test/test_ll_thread.py
==============================================================================
--- pypy/branch/pypy-more-rtti-inprogress/module/thread/test/test_ll_thread.py	(original)
+++ pypy/branch/pypy-more-rtti-inprogress/module/thread/test/test_ll_thread.py	Thu Aug 23 15:42:58 2007
@@ -2,6 +2,8 @@
 from pypy.module.thread.ll_thread import *
 from pypy.translator.c.test.test_genc import compile
 from pypy.rpython.lltypesystem import lltype
+from pypy.rpython.lltypesystem.lloperation import llop
+from pypy.rlib.objectmodel import free_non_gc_object
 import py
 
 def test_lock():
@@ -24,27 +26,56 @@
 
 def test_start_new_thread():
     import time
+    
+
+    class State:
+        pass
+    state = State()
+    
+    class Z:
+        def __init__(self, value):
+            self.value = value
+        def __del__(self):
+            state.freed_counter += 1
+            
     class Y:
         _alloc_flavor_ = 'raw'
-        
-        def __init__(self):
-            self.top = []
 
         def bootstrap(self):
-            self.top.append(get_ident())
+            state.my_thread_ident = get_ident()
+            assert state.my_thread_ident == get_ident()
+            state.seen_value = self.z.value
+            self.z = None
+            free_non_gc_object(self)
+            state.done = 1
 
-    def f():
+    def g(i):
         y = Y()
+        y.z = Z(i)
         start_new_thread(Y.bootstrap, (y,))
-        time.sleep(.3)
-        res = len(y.top)
-        lltype.free(y, flavor='raw')
-        return 1 == get_ident()
+    g.dont_inline = True
+
+    def f():
+        main_ident = get_ident()
+        assert main_ident == get_ident()
+        state.freed_counter = 0
+        for i in range(50):
+            state.done = 0
+            state.seen_value = 0
+            g(i)
+            while not state.done:
+                time.sleep(0.01)
+            assert state.my_thread_ident != main_ident
+            assert state.seen_value == i
+        # try to force Boehm to do some freeing
+        for i in range(3):
+            llop.gc__collect(lltype.Void)
+        return state.freed_counter
 
-    # for some reason, refcounting does not handle _alloc_flavor_ = 'raw'
-    # XXX is this testable on top of llinterp at all?
     fn = compile(f, [], gcpolicy='boehm')
-    assert fn() == False
+    freed_counter = fn()
+    print freed_counter
+    assert freed_counter > 0
 
 def test_prebuilt_lock():
     py.test.skip("Does not work (prebuilt opaque object)")