1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
import weakref
from rpython.rtyper.lltypesystem import lltype, llmemory
# this is global because a header cannot be a header of more than one GcObj
header2obj = weakref.WeakKeyDictionary()
class GCHeaderBuilder(object):
def __init__(self, HDR):
"""NOT_RPYTHON"""
self.HDR = HDR
#
# The following used to be a weakref.WeakKeyDictionary(), but
# the problem is that if you have a gcobj which has already a
# weakref cached on it and the hash already cached in that
# weakref, and later the hash of the gcobj changes (because it
# is ll2ctypes-ified), then that gcobj cannot be used as a key
# in a WeakKeyDictionary any more: from this point on,
# 'ref(gcobj)' and 'ref(gcobj, callback)' return two objects
# with different hashes... and so e.g. the sequence of
# operations 'obj2header[x]=y; assert x in obj2header' fails.
#
# Instead, just use a regular dictionary and hope that not too
# many objects would be reclaimed in a given GCHeaderBuilder
# instance.
self.obj2header = {}
self.size_gc_header = llmemory.GCHeaderOffset(self)
def header_of_object(self, gcptr):
# XXX hackhackhack
gcptr = gcptr._as_obj(check=False)
if isinstance(gcptr, llmemory._gctransformed_wref):
return self.obj2header[gcptr._ptr._as_obj(check=False)]
return self.obj2header[gcptr]
def object_from_header(headerptr):
return header2obj[headerptr._as_obj(check=False)]
object_from_header = staticmethod(object_from_header)
def get_header(self, gcptr):
return self.obj2header.get(gcptr._as_obj(check=False), None)
def attach_header(self, gcptr, headerptr):
gcobj = gcptr._as_obj()
assert gcobj not in self.obj2header
# sanity checks
assert gcobj._TYPE._gckind == 'gc'
assert not isinstance(gcobj._TYPE, lltype.GcOpaqueType)
assert not gcobj._parentstructure()
self.obj2header[gcobj] = headerptr
header2obj[headerptr._obj] = gcptr._as_ptr()
def new_header(self, gcptr):
headerptr = lltype.malloc(self.HDR, immortal=True)
self.attach_header(gcptr, headerptr)
return headerptr
def _freeze_(self):
return True # for reads of size_gc_header
|
