<s> import gc <EOL> from pypy . module . thread . ll_thread import * <EOL> from pypy . translator . c . test . test_boehm import AbstractGCTestClass <EOL> from pypy . rpython . lltypesystem import lltype , rffi <EOL> import py <EOL> def setup_module ( mod ) : <EOL> rffi . aroundstate . _freeze_ ( ) <EOL> def test_lock ( ) : <EOL> l = allocate_lock ( ) <EOL> ok1 = l . acquire ( True ) <EOL> ok2 = l . acquire ( False ) <EOL> l . release ( ) <EOL> ok3 = l . acquire ( False ) <EOL> res = ok1 and not ok2 and ok3 <EOL> assert res == <NUM_LIT:1> <EOL> def test_thread_error ( ) : <EOL> l = allocate_lock ( ) <EOL> try : <EOL> l . release ( ) <EOL> except error : <EOL> pass <EOL> else : <EOL> py . test . fail ( "<STR_LIT>" ) <EOL> class AbstractThreadTests ( AbstractGCTestClass ) : <EOL> use_threads = True <EOL> def test_start_new_thread ( self ) : <EOL> import time <EOL> class State : <EOL> pass <EOL> state = State ( ) <EOL> class Z : <EOL> def __init__ ( self , value ) : <EOL> self . value = value <EOL> def __del__ ( self ) : <EOL> state . freed_counter += <NUM_LIT:1> <EOL> def bootstrap ( ) : <EOL> state . my_thread_ident = get_ident ( ) <EOL> assert state . my_thread_ident == get_ident ( ) <EOL> state . seen_value = state . z . value <EOL> state . z = None <EOL> state . done = <NUM_LIT:1> <EOL> def g ( i ) : <EOL> state . z = Z ( i ) <EOL> start_new_thread ( bootstrap , ( ) ) <EOL> g . _dont_inline_ = True <EOL> def f ( ) : <EOL> main_ident = get_ident ( ) <EOL> assert main_ident == get_ident ( ) <EOL> state . freed_counter = <NUM_LIT:0> <EOL> for i in range ( <NUM_LIT:50> ) : <EOL> state . done = <NUM_LIT:0> <EOL> state . seen_value = <NUM_LIT:0> <EOL> g ( i ) <EOL> gc . collect ( ) <EOL> willing_to_wait_more = <NUM_LIT:1000> <EOL> while not state . done : <EOL> willing_to_wait_more -= <NUM_LIT:1> <EOL> if not willing_to_wait_more : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> assert state . my_thread_ident != main_ident <EOL> assert state . seen_value == i <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> gc . collect ( ) <EOL> return state . freed_counter <EOL> fn = self . getcompiled ( f , [ ] ) <EOL> freed_counter = fn ( ) <EOL> print freed_counter <EOL> if self . gcpolicy == '<STR_LIT>' : <EOL> assert freed_counter > <NUM_LIT:0> <EOL> else : <EOL> assert freed_counter == <NUM_LIT:50> <EOL> def test_gc_locking ( self ) : <EOL> import time <EOL> from pypy . rlib . objectmodel import invoke_around_extcall <EOL> from pypy . rlib . objectmodel import we_are_translated <EOL> from pypy . rlib . debug import ll_assert <EOL> class State : <EOL> pass <EOL> state = State ( ) <EOL> class Z : <EOL> def __init__ ( self , i , j ) : <EOL> self . i = i <EOL> self . j = j <EOL> def run ( self ) : <EOL> j = self . j <EOL> if self . i > <NUM_LIT:1> : <EOL> g ( self . i - <NUM_LIT:1> , self . j * <NUM_LIT:2> ) <EOL> ll_assert ( j == self . j , "<STR_LIT>" ) <EOL> g ( self . i - <NUM_LIT:2> , self . j * <NUM_LIT:2> + <NUM_LIT:1> ) <EOL> else : <EOL> if len ( state . answers ) % <NUM_LIT:7> == <NUM_LIT:5> : <EOL> gc . collect ( ) <EOL> state . answers . append ( self . j ) <EOL> ll_assert ( j == self . j , "<STR_LIT>" ) <EOL> run . _dont_inline_ = True <EOL> def before_extcall ( ) : <EOL> release_NOAUTO ( state . gil ) <EOL> before_extcall . _gctransformer_hint_cannot_collect_ = True <EOL> def after_extcall ( ) : <EOL> acquire_NOAUTO ( state . gil , True ) <EOL> gc_thread_run ( ) <EOL> after_extcall . _gctransformer_hint_cannot_collect_ = True <EOL> def bootstrap ( ) : <EOL> z = state . z <EOL> state . z = None <EOL> state . bootstrapping . release ( ) <EOL> z . run ( ) <EOL> gc_thread_die ( ) <EOL> def g ( i , j ) : <EOL> state . bootstrapping . acquire ( True ) <EOL> state . z = Z ( i , j ) <EOL> gc_thread_prepare ( ) <EOL> start_new_thread ( bootstrap , ( ) ) <EOL> def f ( ) : <EOL> state . gil = allocate_ll_lock ( ) <EOL> acquire_NOAUTO ( state . gil , True ) <EOL> state . bootstrapping = allocate_lock ( ) <EOL> state . answers = [ ] <EOL> state . finished = <NUM_LIT:0> <EOL> invoke_around_extcall ( before_extcall , after_extcall ) <EOL> g ( <NUM_LIT:10> , <NUM_LIT:1> ) <EOL> done = False <EOL> willing_to_wait_more = <NUM_LIT> <EOL> while not done : <EOL> if not willing_to_wait_more : <EOL> break <EOL> willing_to_wait_more -= <NUM_LIT:1> <EOL> done = len ( state . answers ) == expected <EOL> if not we_are_translated ( ) : before_extcall ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> if not we_are_translated ( ) : after_extcall ( ) <EOL> if not we_are_translated ( ) : before_extcall ( ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> if not we_are_translated ( ) : after_extcall ( ) <EOL> return len ( state . answers ) <EOL> expected = <NUM_LIT> <EOL> try : <EOL> fn = self . getcompiled ( f , [ ] ) <EOL> finally : <EOL> rffi . aroundstate . _freeze_ ( ) <EOL> answers = fn ( ) <EOL> assert answers == expected <EOL> class TestRunDirectly ( AbstractThreadTests ) : <EOL> def getcompiled ( self , f , argtypes ) : <EOL> return f <EOL> class TestUsingBoehm ( AbstractThreadTests ) : <EOL> gcpolicy = '<STR_LIT>' <EOL> class TestUsingFramework ( AbstractThreadTests ) : <EOL> gcpolicy = '<STR_LIT>' </s>
