<s> import time <EOL> from pypy . module . thread import gil <EOL> from pypy . module . thread . test import test_ll_thread <EOL> from pypy . rpython . lltypesystem import rffi <EOL> from pypy . module . thread import ll_thread as thread <EOL> from pypy . rlib . objectmodel import we_are_translated <EOL> class FakeEC ( object ) : <EOL> pass <EOL> class FakeActionFlag ( object ) : <EOL> def register_action ( self , action ) : <EOL> pass <EOL> def get ( self ) : <EOL> return <NUM_LIT:0> <EOL> def set ( self , x ) : <EOL> pass <EOL> class FakeSpace ( object ) : <EOL> def __init__ ( self ) : <EOL> self . actionflag = FakeActionFlag ( ) <EOL> def _freeze_ ( self ) : <EOL> return True <EOL> def getexecutioncontext ( self ) : <EOL> return FakeEC ( ) <EOL> def getbuiltinmodule ( self , name ) : <EOL> raise NotImplementedError <EOL> class GILTests ( test_ll_thread . AbstractGCTestClass ) : <EOL> use_threads = True <EOL> bigtest = False <EOL> def test_one_thread ( self ) : <EOL> if self . bigtest : <EOL> N = <NUM_LIT> <EOL> else : <EOL> N = <NUM_LIT:100> <EOL> space = FakeSpace ( ) <EOL> class State : <EOL> pass <EOL> state = State ( ) <EOL> def runme ( ) : <EOL> for i in range ( N ) : <EOL> state . data . append ( ( thread . get_ident ( ) , i ) ) <EOL> state . threadlocals . yield_thread ( ) <EOL> def bootstrap ( ) : <EOL> try : <EOL> runme ( ) <EOL> finally : <EOL> thread . gc_thread_die ( ) <EOL> def f ( ) : <EOL> state . data = [ ] <EOL> state . threadlocals = gil . GILThreadLocals ( ) <EOL> state . threadlocals . setup_threads ( space ) <EOL> thread . gc_thread_prepare ( ) <EOL> subident = thread . start_new_thread ( bootstrap , ( ) ) <EOL> mainident = thread . get_ident ( ) <EOL> runme ( ) <EOL> still_waiting = <NUM_LIT> <EOL> while len ( state . data ) < <NUM_LIT:2> * N : <EOL> if not still_waiting : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> still_waiting -= <NUM_LIT:1> <EOL> if not we_are_translated ( ) : gil . before_external_call ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> if not we_are_translated ( ) : gil . after_external_call ( ) <EOL> i1 = i2 = <NUM_LIT:0> <EOL> for tid , i in state . data : <EOL> if tid == mainident : <EOL> assert i == i1 ; i1 += <NUM_LIT:1> <EOL> elif tid == subident : <EOL> assert i == i2 ; i2 += <NUM_LIT:1> <EOL> else : <EOL> assert <NUM_LIT:0> <EOL> assert i1 == N <EOL> assert i2 == N <EOL> return len ( state . data ) <EOL> fn = self . getcompiled ( f , [ ] ) <EOL> res = fn ( ) <EOL> assert res == <NUM_LIT:2> * N <EOL> class TestRunDirectly ( GILTests ) : <EOL> def getcompiled ( self , f , argtypes ) : <EOL> return f <EOL> class TestUsingFramework ( GILTests ) : <EOL> gcpolicy = '<STR_LIT>' <EOL> bigtest = True </s>
