Author: Armin Rigo <[email protected]>
Branch: ec-keepalive
Changeset: r81545:7149d01e69d0
Date: 2016-01-04 16:46 +0100
http://bitbucket.org/pypy/pypy/changeset/7149d01e69d0/
Log: Progress, see comments
diff --git a/pypy/module/thread/__init__.py b/pypy/module/thread/__init__.py
--- a/pypy/module/thread/__init__.py
+++ b/pypy/module/thread/__init__.py
@@ -27,7 +27,7 @@
from pypy.module.thread import gil
MixedModule.__init__(self, space, *args)
prev_ec = space.threadlocals.get_ec()
- space.threadlocals = gil.GILThreadLocals()
+ space.threadlocals = gil.GILThreadLocals(space)
space.threadlocals.initialize(space)
if prev_ec is not None:
space.threadlocals._set_ec(prev_ec)
diff --git a/pypy/module/thread/test/test_gil.py
b/pypy/module/thread/test/test_gil.py
--- a/pypy/module/thread/test/test_gil.py
+++ b/pypy/module/thread/test/test_gil.py
@@ -65,7 +65,7 @@
except Exception, e:
assert 0
thread.gc_thread_die()
- my_gil_threadlocals = gil.GILThreadLocals()
+ my_gil_threadlocals = gil.GILThreadLocals(space)
def f():
state.data = []
state.datalen1 = 0
diff --git a/pypy/module/thread/threadlocals.py
b/pypy/module/thread/threadlocals.py
--- a/pypy/module/thread/threadlocals.py
+++ b/pypy/module/thread/threadlocals.py
@@ -1,5 +1,6 @@
-from rpython.rlib import rthread
+from rpython.rlib import rthread, rweaklist
from rpython.rlib.objectmodel import we_are_translated
+from rpython.rlib.rarithmetic import r_ulonglong
from pypy.module.thread.error import wrap_thread_error
from pypy.interpreter.executioncontext import ExecutionContext
@@ -13,15 +14,53 @@
a thread finishes. This works as long as the thread was started by
os_thread.bootstrap()."""
- def __init__(self):
+ _next_generation = r_ulonglong(0)
+
+ def __init__(self, space):
"NOT_RPYTHON"
- self._valuedict = {} # {thread_ident: ExecutionContext()}
+ #
+ # This object tracks code that enters and leaves threads.
+ # There are two APIs. For Python-level threads, we know when
+ # the thread starts and ends, and we call enter_thread() and
+ # leave_thread(). In a few other cases, like callbacks, we
+ # might be running in some never-seen-before thread: in this
+ # case, the callback logic needs to call try_enter_thread() at
+ # the start, and if this returns True it needs to call
+ # leave_thread() at the end.
+ #
+ # We implement an optimization for the second case (which only
+ # works if we translate with a framework GC and with
+ # rweakref). If try_enter_thread() is called in a
+ # never-seen-before thread, it still returns False and
+ # remembers the ExecutionContext with 'self._weaklist'. The
+ # next time we call try_enter_thread() again in the same
+ # thread, the ExecutionContext is reused. The optimization is
+ # not completely invisible to the user: 'thread._local()'
+ # values will remain. We can argue that it is the correct
+ # behavior to do that, and the behavior we get if the
+ # optimization is disabled is buggy (but hard to do better
+ # then).
+ #
+ # 'self._valuedict' is a dict mapping the thread idents to
+ # ExecutionContexts; it does not list the ExecutionContexts
+ # which are in 'self._weaklist'. (The latter is more precisely
+ # a list of AutoFreeECWrapper objects, defined below, which
+ # each references the ExecutionContext.)
+ #
+ self.space = space
+ self._valuedict = {}
self._cleanup_()
self.raw_thread_local = rthread.ThreadLocalReference(ExecutionContext,
loop_invariant=True)
+ def can_optimize_with_weaklist(self):
+ config = self.space.config
+ return (config.translation.rweakref and
+ rthread.ThreadLocalReference.automatic_keepalive(config))
+
def _cleanup_(self):
self._valuedict.clear()
+ self._weaklist = None
self._mainthreadident = 0
def enter_thread(self, space):
@@ -29,19 +68,36 @@
self._set_ec(space.createexecutioncontext())
def try_enter_thread(self, space):
- if rthread.get_ident() in self._valuedict:
+ # common case: the thread-local has already got a value
+ if self.raw_thread_local.get() is not None:
return False
- self.enter_thread(space)
- return True
- def _set_ec(self, ec):
+ # Else, make and attach a new ExecutionContext
+ ec = space.createexecutioncontext()
+ if not self.can_optimize_with_weaklist():
+ self._set_ec(ec)
+ return True
+
+ # If can_optimize_with_weaklist(), then 'rthread' keeps the
+ # thread-local values alive until the end of the thread. Use
+ # AutoFreeECWrapper as an object with a __del__; when this
+ # __del__ is called, it means the thread was really finished.
+ # In this case we don't want leave_thread() to be called
+ # explicitly, so we return False.
+ if self._weaklist is None:
+ self._weaklist = ListECWrappers()
+ self._weaklist.initialize()
+ self._weaklist.add_handle(AutoFreeECWrapper(self, ec))
+ self._set_ec(ec, register_in_valuedict=False)
+ return False
+
+ def _set_ec(self, ec, register_in_valuedict=True):
ident = rthread.get_ident()
if self._mainthreadident == 0 or self._mainthreadident == ident:
ec._signals_enabled = 1 # the main thread is enabled
self._mainthreadident = ident
- self._valuedict[ident] = ec
- # This logic relies on hacks and _make_sure_does_not_move().
- # It only works because we keep the 'ec' alive in '_valuedict' too.
+ if register_in_valuedict:
+ self._valuedict[ident] = ec
self.raw_thread_local.set(ec)
def leave_thread(self, space):
@@ -84,7 +140,27 @@
ec._signals_enabled = new
def getallvalues(self):
- return self._valuedict
+ if self._weaklist is None:
+ return self._valuedict
+ # This logic walks the 'self._weaklist' list and adds the
+ # ExecutionContexts to 'result'. We are careful in case there
+ # are two AutoFreeECWrappers in the list which have the same
+ # 'ident'; in this case we must keep the most recent one (the
+ # older one should be deleted soon). Moreover, entries in
+ # self._valuedict have priority because they are never
+ # outdated.
+ result = {}
+ generations = {}
+ for h in self._weaklist.get_all_handles():
+ wrapper = h()
+ if wrapper is not None:
+ key = wrapper.ident
+ prev = generations.get(key, r_ulonglong(0))
+ if wrapper.generation > prev: # implies '.generation != 0'
+ generations[key] = wrapper.generation
+ result[key] = wrapper.ec
+ result.update(self._valuedict)
+ return result
def reinit_threads(self, space):
"Called in the child process after a fork()"
@@ -98,3 +174,27 @@
self._mainthreadident = ident
self._set_ec(ec)
ec._signals_enabled = old_sig
+
+
+class AutoFreeECWrapper(object):
+
+ def __init__(self, threadlocals, ec):
+ # this makes a loop between 'self' and 'ec'. It should not prevent
+ # the __del__ method here from being called.
+ threadlocals._next_generation += 1
+ self.generation = threadlocals._next_generation
+ self.ec = ec
+ ec._threadlocals_auto_free = self
+ self.ident = rthread.get_ident()
+
+ def __del__(self):
+ from pypy.module.thread.os_local import thread_is_stopping
+ # this is always called in another thread: the thread
+ # referenced by 'self.ec' has finished at that point, and
+ # we're just after the GC which finds no more references to
+ # 'ec' (and thus to 'self').
+ self.generation = r_ulonglong(0)
+ thread_is_stopping(self.ec)
+
+class ListECWrappers(rweaklist.RWeakListMixin):
+ pass
diff --git a/rpython/rlib/rthread.py b/rpython/rlib/rthread.py
--- a/rpython/rlib/rthread.py
+++ b/rpython/rlib/rthread.py
@@ -391,7 +391,6 @@
self.get = get
self.set = set
- self.automatic_keepalive = _automatic_keepalive
def _trace_tlref(gc, obj, callback, arg):
p = llmemory.NULL
@@ -404,16 +403,13 @@
TRACETLREF = lltype.GcStruct('TRACETLREF')
_tracetlref_obj = lltype.malloc(TRACETLREF, immortal=True)
-
-def _automatic_keepalive():
- """Returns True if translated with a GC that keeps alive
- the set() value until the end of the thread. Returns False
- if you need to keep it alive yourself.
- """
- from rpython.rlib import objectmodel
- config = objectmodel.fetch_translated_config()
- return (config is not None and
- config.translation.gctransformer == "framework")
+ @staticmethod
+ def automatic_keepalive(config):
+ """Returns True if translated with a GC that keeps alive
+ the set() value until the end of the thread. Returns False
+ if you need to keep it alive yourself.
+ """
+ return config.translation.gctransformer == "framework"
tlfield_thread_ident = ThreadLocalField(lltype.Signed, "thread_ident",
diff --git a/rpython/rlib/test/test_rthread.py
b/rpython/rlib/test/test_rthread.py
--- a/rpython/rlib/test/test_rthread.py
+++ b/rpython/rlib/test/test_rthread.py
@@ -1,6 +1,7 @@
import gc, time
from rpython.rlib.rthread import *
from rpython.rlib.rarithmetic import r_longlong
+from rpython.rlib import objectmodel
from rpython.translator.c.test.test_boehm import AbstractGCTestClass
from rpython.rtyper.lltypesystem import lltype, rffi
import py
@@ -251,7 +252,6 @@
class FooBar(object):
pass
t = ThreadLocalReference(FooBar)
- assert t.automatic_keepalive() is False
def tset():
x1 = FooBar()
@@ -264,7 +264,8 @@
wr_from_thread = WrFromThread()
def f():
- assert t.automatic_keepalive() is True
+ config = objectmodel.fetch_translated_config()
+ assert t.automatic_keepalive(config) is True
wr = tset()
import gc; gc.collect() # 'x1' should not be collected
x2 = t.get()
_______________________________________________
pypy-commit mailing list
[email protected]
https://mail.python.org/mailman/listinfo/pypy-commit
