pypy/module/thread/test/test_gil.py::TestUsingFramework::()::test_one_thread
self = <pypy.module.thread.test.test_gil.TestUsingFramework object at 0x7fd6d13d4810>
skew = 25000
def test_one_thread(self, skew=+1):
from rpython.rlib.debug import debug_print
if self.bigtest:
N = 100000
skew *= 25000
else:
N = 100
skew *= 25
space = FakeSpace()
class State:
pass
state = State()
def runme(main=False):
j = 0
for i in range(N + [-skew, skew][main]):
state.datalen1 += 1 # try to crash if the GIL is not
state.datalen2 += 1 # correctly acquired
state.data.append((thread.get_ident(), i))
state.datalen3 += 1
state.datalen4 += 1
assert state.datalen1 == len(state.data)
assert state.datalen2 == len(state.data)
assert state.datalen3 == len(state.data)
assert state.datalen4 == len(state.data)
debug_print(main, i, state.datalen4)
rgil.yield_thread()
assert i == j
j += 1
def bootstrap():
try:
runme()
except Exception as e:
assert 0
thread.gc_thread_die()
my_gil_threadlocals = gil.GILThreadLocals(space)
def f():
state.data = []
state.datalen1 = 0
state.datalen2 = 0
state.datalen3 = 0
state.datalen4 = 0
state.threadlocals = my_gil_threadlocals
state.threadlocals.setup_threads(space)
subident = thread.start_new_thread(bootstrap, ())
mainident = thread.get_ident()
runme(True)
still_waiting = 3000
while len(state.data) < 2*N:
debug_print(len(state.data))
if not still_waiting:
llop.debug_print(lltype.Void, "timeout. progress: "
"%d of %d (= %f%%)" % \
(len(state.data), 2*N, 100*len(state.data)/(2.0*N)))
raise ValueError("time out")
still_waiting -= 1
if not we_are_translated(): rgil.release()
time.sleep(0.1)
if not we_are_translated(): rgil.acquire()
debug_print("leaving!")
i1 = i2 = 0
for tid, i in state.data:
if tid == mainident:
assert i == i1; i1 += 1
elif tid == subident:
assert i == i2; i2 += 1
else:
assert 0
assert i1 == N + skew
assert i2 == N - skew
return len(state.data)
> fn = self.getcompiled(f, [])
module/thread/test/test_gil.py:107:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../rpython/translator/c/test/test_boehm.py:40: in getcompiled
thread=self.use_threads, **extra_options)
../rpython/translator/c/test/test_genc.py:110: in compile
t.annotate()
../rpython/translator/interactive.py:83: in annotate
return self.driver.annotate()
../rpython/translator/driver.py:108: in proc
return self.proceed(backend_goal)
../rpython/translator/driver.py:568: in proceed
result = self._execute(goals, task_skip = self._maybe_skip())
../rpython/translator/tool/taskengine.py:114: in _execute
res = self._do(goal, taskcallable, *args, **kwds)
../rpython/translator/driver.py:278: in _do
res = func()
../rpython/translator/driver.py:315: in task_annotate
s = annotator.build_types(self.entry_point, self.inputtypes)
../rpython/annotator/annrpython.py:92: in build_types
return self.build_graph_types(flowgraph, inputs_s, complete_now=complete_now)
../rpython/annotator/annrpython.py:140: in build_graph_types
self.complete()
../rpython/annotator/annrpython.py:229: in complete
self.complete_pending_blocks()
../rpython/annotator/annrpython.py:224: in complete_pending_blocks
self.processblock(graph, block)
../rpython/annotator/annrpython.py:398: in processblock
self.flowin(graph, block)
../rpython/annotator/annrpython.py:501: in flowin
self.consider_op(op)
../rpython/annotator/annrpython.py:653: in consider_op
resultcell = op.consider(self)
../rpython/flowspace/operation.py:104: in consider
return spec(annotator, *self.args)
../rpython/annotator/unaryop.py:118: in simple_call_SomeObject
return s_func.call(argspec)
../rpython/annotator/unaryop.py:987: in call
return bookkeeper.pbc_call(self, args)
../rpython/annotator/bookkeeper.py:535: in pbc_call
results.append(desc.pycall(whence, args, s_previous_result, op))
../rpython/annotator/description.py:285: in pycall
graph = self.specialize(inputcells, op)
../rpython/annotator/description.py:281: in specialize
return self.specializer(self, inputcells)
../rpython/annotator/specialize.py:82: in default_specialize
graph = funcdesc.cachedgraph(key, builder=builder)
../rpython/annotator/description.py:247: in cachedgraph
graph = self.buildgraph(alt_name, builder)
../rpython/annotator/description.py:210: in buildgraph
graph = translator.buildflowgraph(self.pyobj)
../rpython/translator/translator.py:55: in buildflowgraph
graph = build_flow(func)
../rpython/flowspace/objspace.py:49: in build_flow
ctx.build_flow()
../rpython/flowspace/flowcontext.py:405: in build_flow
self.record_block(block)
../rpython/flowspace/flowcontext.py:413: in record_block
next_offset = self.handle_bytecode(next_offset)
../rpython/flowspace/flowcontext.py:492: in handle_bytecode
offset = getattr(self, methodname)(oparg)
../rpython/flowspace/flowcontext.py:999: in CALL_FUNCTION
self.call_function(oparg)
../rpython/flowspace/flowcontext.py:973: in call_function
self.pushvalue(hlop.eval(self))
../rpython/flowspace/operation.py:675: in eval
return sc(ctx, *args_w)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ = <rpython.flowspace.flowcontext.FlowContext object at 0x7fd6d138dd50>
args = ()
@register_flow_sc(locals)
def sc_locals(_, *args):
raise Exception(
> "A function calling locals() is not RPython. "
"Note that if you're translating code outside the PyPy "
"repository, a likely cause is that py.test's --assert=rewrite "
"mode is getting in the way. You should copy the file "
"pytest.ini from the root of the PyPy repository into your "
"own project.")
E Exception: A function calling locals() is not RPython. Note that if you're translating code outside the PyPy repository, a likely cause is that py.test's --assert=rewrite mode is getting in the way. You should copy the file pytest.ini from the root of the PyPy repository into your own project.
../rpython/flowspace/specialcase.py:36: Exception
builder: own-linux-x86-64 build #10810
test: pypy/module/thread/test/test_gil/py/TestUsingFramework/()/test_one_thread