from __future__ import absolute_import

import gc

import types

from rpython.rlib import jit

from rpython.rlib.objectmodel import we_are_translated, enforceargs, specialize

from rpython.rlib.objectmodel import CDefinedIntSymbolic, not_rpython

from rpython.rtyper.extregistry import ExtRegistryEntry

from rpython.rtyper.lltypesystem import lltype, llmemory

# ____________________________________________________________

# General GC features

collect = gc.collect

enable = gc.enable

disable = gc.disable

isenabled = gc.isenabled

def collect_step():

"""

If the GC is incremental, run a single gc-collect-step.

Return an integer which encodes the starting and ending GC state. Use

rgc.{old_state,new_state,is_done} to decode it.

If the GC is not incremental, do a full collection and return a value on

which rgc.is_done() return True.

"""

gc.collect()

return _encode_states(1, 0)

def _encode_states(oldstate, newstate):

return oldstate << 8 | newstate

def old_state(states):

return (states & 0xFF00) >> 8

def new_state(states):

return states & 0xFF

def is_done(states):

"""

Return True if the return value of collect_step signals the end of a major

collection

"""

old = old_state(states)

new = new_state(states)

return is_done__states(old, new)

def is_done__states(oldstate, newstate):

"Like is_done, but takes oldstate and newstate explicitly"

# a collection is considered done when it ends up in the starting state

# (which is usually represented as 0). This logic works for incminimark,

# which is currently the only gc actually used and for which collect_step

# is implemented. In case we add more GC in the future, we might want to

# delegate this logic to the GC itself, but for now it is MUCH simpler to

# just write it in plain RPython.

return oldstate != 0 and newstate == 0

def set_max_heap_size(nbytes):

"""Limit the heap size to n bytes.

"""

pass

def must_split_gc_address_space():

"""Returns True if we have a "split GC address space", i.e. if

we are translating with an option that doesn't support taking raw

addresses inside GC objects and "hacking" at them. This is

notably the case with --revdb."""

return False

# for test purposes we allow objects to be pinned and use

# the following list to keep track of the pinned objects

_pinned_objects = []

def pin(obj):

"""If 'obj' can move, then attempt to temporarily fix it. This

function returns True if and only if 'obj' could be pinned; this is

a special state in the GC. Note that can_move(obj) still returns

True even on pinned objects, because once unpinned it will indeed be

able to move again. In other words, the code that succeeded in

pinning 'obj' can assume that it won't move until the corresponding

call to unpin(obj), despite can_move(obj) still being True. (This

is important if multiple threads try to os.write() the same string:

only one of them will succeed in pinning the string.)

It is expected that the time between pinning and unpinning an object

is short. Therefore the expected use case is a single function

invoking pin(obj) and unpin(obj) only a few lines of code apart.

Note that this can return False for any reason, e.g. if the 'obj' is

already non-movable or already pinned, if the GC doesn't support

pinning, or if there are too many pinned objects.

Note further that pinning an object does not prevent it from being

collected if it is not used anymore.

"""

_pinned_objects.append(obj)

return True

class PinEntry(ExtRegistryEntry):

_about_ = pin

def compute_result_annotation(self, s_obj):

from rpython.annotator import model as annmodel

return annmodel.SomeBool()

def specialize_call(self, hop):

hop.exception_cannot_occur()

return hop.genop('gc_pin', hop.args_v, resulttype=hop.r_result)

def unpin(obj):

"""Unpin 'obj', allowing it to move again.

Must only be called after a call to pin(obj) returned True.

"""

for i in range(len(_pinned_objects)):

try:

if _pinned_objects[i] == obj:

del _pinned_objects[i]

return

except TypeError:

pass

class UnpinEntry(ExtRegistryEntry):

_about_ = unpin

def compute_result_annotation(self, s_obj):

pass

def specialize_call(self, hop):

hop.exception_cannot_occur()

hop.genop('gc_unpin', hop.args_v)

def _is_pinned(obj):

"""Method to check if 'obj' is pinned."""

for i in range(len(_pinned_objects)):

try:

if _pinned_objects[i] == obj:

return True

except TypeError:

pass

return False

class IsPinnedEntry(ExtRegistryEntry):

_about_ = _is_pinned

def compute_result_annotation(self, s_obj):

from rpython.annotator import model as annmodel

return annmodel.SomeBool()

def specialize_call(self, hop):

hop.exception_cannot_occur()

return hop.genop('gc__is_pinned', hop.args_v, resulttype=hop.r_result)

# ____________________________________________________________

# Annotation and specialization

# Support for collection.

class CollectEntry(ExtRegistryEntry):

_about_ = gc.collect

def compute_result_annotation(self, s_gen=None):

from rpython.annotator import model as annmodel

return annmodel.s_None

def specialize_call(self, hop):

hop.exception_cannot_occur()

args_v = []

if len(hop.args_s) == 1:

args_v = hop.inputargs(lltype.Signed)

return hop.genop('gc__collect', args_v, resulttype=hop.r_result)

class EnableDisableEntry(ExtRegistryEntry):

_about_ = (gc.enable, gc.disable)

def compute_result_annotation(self):

from rpython.annotator import model as annmodel

return annmodel.s_None

def specialize_call(self, hop):

hop.exception_cannot_occur()

opname = self.instance.__name__

return hop.genop('gc__%s' % opname, hop.args_v, resulttype=hop.r_result)

class IsEnabledEntry(ExtRegistryEntry):

_about_ = gc.isenabled

def compute_result_annotation(self):

from rpython.annotator import model as annmodel

return annmodel.s_Bool

def specialize_call(self, hop):

hop.exception_cannot_occur()

return hop.genop('gc__isenabled', hop.args_v, resulttype=hop.r_result)

class CollectStepEntry(ExtRegistryEntry):

_about_ = collect_step

def compute_result_annotation(self):

from rpython.annotator import model as annmodel

return annmodel.SomeInteger()

def specialize_call(self, hop):

hop.exception_cannot_occur()

return hop.genop('gc__collect_step', hop.args_v, resulttype=hop.r_result)

class SetMaxHeapSizeEntry(ExtRegistryEntry):

_about_ = set_max_heap_size

def compute_result_annotation(self, s_nbytes):

from rpython.annotator import model as annmodel

return annmodel.s_None

def specialize_call(self, hop):

[v_nbytes] = hop.inputargs(lltype.Signed)

hop.exception_cannot_occur()

return hop.genop('gc_set_max_heap_size', [v_nbytes],

resulttype=lltype.Void)

def can_move(p):

"""Check if the GC object 'p' is at an address that can move.

Must not be called with None. With non-moving GCs, it is always False.

With some moving GCs like the SemiSpace GC, it is always True.

With other moving GCs like the MiniMark GC, it can be True for some

time, then False for the same object, when we are sure that it won't

move any more.

"""

return True

class SplitAddrSpaceEntry(ExtRegistryEntry):

_about_ = must_split_gc_address_space

def compute_result_annotation(self):

config = self.bookkeeper.annotator.translator.config

result = config.translation.split_gc_address_space

return self.bookkeeper.immutablevalue(result)

def specialize_call(self, hop):

hop.exception_cannot_occur()

return hop.inputconst(lltype.Bool, hop.s_result.const)

class CanMoveEntry(ExtRegistryEntry):

_about_ = can_move

def compute_result_annotation(self, s_p):

from rpython.annotator import model as annmodel

return annmodel.SomeBool()

def specialize_call(self, hop):

hop.exception_cannot_occur()

return hop.genop('gc_can_move', hop.args_v, resulttype=hop.r_result)

def _make_sure_does_not_move(p):

"""'p' is a non-null GC object. This (tries to) make sure that the

object does not move any more, by forcing collections if needed.

Warning: should ideally only be used with the minimark GC, and only

on objects that are already a bit old, so have a chance to be

already non-movable."""

assert p

if not we_are_translated():

# for testing purpose

return not _is_pinned(p)

#

if _is_pinned(p):

# although a pinned object can't move we must return 'False'. A pinned

# object can be unpinned any time and becomes movable.

return False

i = -1

while can_move(p):

if i > 6:

raise NotImplementedError("can't make object non-movable!")

collect(i)

i += 1

return True

def needs_write_barrier(obj):

""" We need to emit write barrier if the right hand of assignment

is in nursery, used by the JIT for handling set*_gc(Const)

"""

if not obj:

return False

# XXX returning can_move() here might acidentally work for the use

# cases (see issue #2212), but this is not really safe. Now we

# just return True for any non-NULL pointer, and too bad for the

# few extra 'cond_call_gc_wb'. It could be improved e.g. to return

# False if 'obj' is a static prebuilt constant, or if we're not

# running incminimark...

return True #can_move(obj)

def _heap_stats():

raise NotImplementedError # can't be run directly

class DumpHeapEntry(ExtRegistryEntry):

_about_ = _heap_stats

def compute_result_annotation(self):

from rpython.rtyper.llannotation import SomePtr

from rpython.memory.gc.base import ARRAY_TYPEID_MAP

return SomePtr(lltype.Ptr(ARRAY_TYPEID_MAP))

def specialize_call(self, hop):

hop.exception_is_here()

return hop.genop('gc_heap_stats', [], resulttype=hop.r_result)

def copy_struct_item(source, dest, si, di):

TP = lltype.typeOf(source).TO.OF

i = 0

while i < len(TP._names):

setattr(dest[di], TP._names[i], getattr(source[si], TP._names[i]))

i += 1

class CopyStructEntry(ExtRegistryEntry):

_about_ = copy_struct_item

def compute_result_annotation(self, s_source, s_dest, si, di):

pass

def specialize_call(self, hop):

v_source, v_dest, v_si, v_di = hop.inputargs(hop.args_r[0],

hop.args_r[1],

lltype.Signed,

lltype.Signed)

hop.exception_cannot_occur()

TP = v_source.concretetype.TO.OF

for name, TP in TP._flds.iteritems():

c_name = hop.inputconst(lltype.Void, name)

v_fld = hop.genop('getinteriorfield', [v_source, v_si, c_name],

resulttype=TP)

hop.genop('setinteriorfield', [v_dest, v_di, c_name, v_fld])

@specialize.ll()

def copy_item(source, dest, si, di):

TP = lltype.typeOf(source)

if isinstance(TP.TO.OF, lltype.Struct):

copy_struct_item(source, dest, si, di)

else:

dest[di] = source[si]

@specialize.memo()

def _contains_gcptr(TP):

if not isinstance(TP, lltype.Struct):

if isinstance(TP, lltype.Ptr) and TP.TO._gckind == 'gc':

return True

return False

for TP in TP._flds.itervalues():

if _contains_gcptr(TP):

return True

return False

@jit.oopspec('list.ll_arraycopy(source, dest, source_start, dest_start, length)')

@enforceargs(None, None, int, int, int)

@specialize.ll()

def ll_arraycopy(source, dest, source_start, dest_start, length):

from rpython.rtyper.lltypesystem.lloperation import llop

from rpython.rlib.objectmodel import keepalive_until_here

# XXX: Hack to ensure that we get a proper effectinfo.write_descrs_arrays

# and also, maybe, speed up very small cases

if length <= 1:

if length == 1:

copy_item(source, dest, source_start, dest_start)

return

# supports non-overlapping copies only

if not we_are_translated():

if source == dest:

assert (source_start + length <= dest_start or

dest_start + length <= source_start)

TP = lltype.typeOf(source).TO

assert TP == lltype.typeOf(dest).TO

slowpath = False

if must_split_gc_address_space():

slowpath = True

elif _contains_gcptr(TP.OF):

# perform a write barrier that copies necessary flags from

# source to dest

if not llop.gc_writebarrier_before_copy(lltype.Bool, source, dest,

source_start, dest_start,

length):

slowpath = True

if slowpath:

# if the write barrier is not supported, or if we translate with

# the option 'split_gc_address_space', then copy by hand

i = 0

while i < length:

copy_item(source, dest, i + source_start, i + dest_start)

i += 1

return

source_addr = llmemory.cast_ptr_to_adr(source)

dest_addr = llmemory.cast_ptr_to_adr(dest)

cp_source_addr = (source_addr + llmemory.itemoffsetof(TP, 0) +

llmemory.sizeof(TP.OF) * source_start)

cp_dest_addr = (dest_addr + llmemory.itemoffsetof(TP, 0) +

llmemory.sizeof(TP.OF) * dest_start)

llmemory.raw_memcopy(cp_source_addr, cp_dest_addr,

llmemory.sizeof(TP.OF) * length)

keepalive_until_here(source)

keepalive_until_here(dest)

@jit.oopspec('rgc.ll_shrink_array(p, smallerlength)')

@enforceargs(None, int)

@specialize.ll()

def ll_shrink_array(p, smallerlength):

from rpython.rtyper.lltypesystem.lloperation import llop

from rpython.rlib.objectmodel import keepalive_until_here

if llop.shrink_array(lltype.Bool, p, smallerlength):

return p # done by the GC

# XXX we assume for now that the type of p is GcStruct containing a

# variable array, with no further pointers anywhere, and exactly one

# field in the fixed part -- like STR and UNICODE.

TP = lltype.typeOf(p).TO

newp = lltype.malloc(TP, smallerlength)

assert len(TP._names) == 2

field = getattr(p, TP._names[0])

setattr(newp, TP._names[0], field)

if must_split_gc_address_space():

# do the copying element by element

i = 0

while i < smallerlength:

newp.chars[i] = p.chars[i]

i += 1

return newp

ARRAY = getattr(TP, TP._arrayfld)

offset = (llmemory.offsetof(TP, TP._arrayfld) +

llmemory.itemoffsetof(ARRAY, 0))

source_addr = llmemory.cast_ptr_to_adr(p) + offset

dest_addr = llmemory.cast_ptr_to_adr(newp) + offset

llmemory.raw_memcopy(source_addr, dest_addr,

llmemory.sizeof(ARRAY.OF) * smallerlength)

keepalive_until_here(p)

keepalive_until_here(newp)

return newp

@jit.dont_look_inside

@specialize.ll()

def ll_arrayclear(p):

# Equivalent to memset(array, 0). Only for GcArray(primitive-type) for now.

from rpython.rlib.objectmodel import keepalive_until_here

length = len(p)

ARRAY = lltype.typeOf(p).TO

if must_split_gc_address_space():

# do the clearing element by element

from rpython.rtyper.lltypesystem import rffi

ZERO = rffi.cast(ARRAY.OF, 0)

i = 0

while i < length:

p[i] = ZERO

i += 1

else:

offset = llmemory.itemoffsetof(ARRAY, 0)

dest_addr = llmemory.cast_ptr_to_adr(p) + offset

llmemory.raw_memclear(dest_addr, llmemory.sizeof(ARRAY.OF) * length)

keepalive_until_here(p)

def no_release_gil(func):

func._dont_inline_ = True

func._no_release_gil_ = True

return func

def no_collect(func):

func._dont_inline_ = True

func._gc_no_collect_ = True

return func

def must_be_light_finalizer(func):

"""Mark a __del__ method as being a destructor, calling only a limited

set of operations. See pypy/doc/discussion/finalizer-order.rst.

If you use the same decorator on a class, this class and all its

subclasses are only allowed to have __del__ methods which are

similarly decorated (or no __del__ at all). It prevents a class

hierarchy from having destructors in some parent classes, which are

overridden in subclasses with (non-light, old-style) finalizers.

(This case is the original motivation for FinalizerQueue.)

"""

func._must_be_light_finalizer_ = True

return func

class FinalizerQueue(object):

"""A finalizer queue. See pypy/doc/discussion/finalizer-order.rst.

"""

# Must be subclassed, and the subclass needs these attributes:

#

# Class:

# the class (or base class) of finalized objects

# --or-- None to handle low-level GCREFs directly

#

# def finalizer_trigger(self):

# called to notify that new items have been put in the queue

def _freeze_(self):

return True

@specialize.arg(0)

@jit.dont_look_inside

def next_dead(self):

if we_are_translated():

from rpython.rtyper.lltypesystem.lloperation import llop

from rpython.rtyper.lltypesystem.llmemory import GCREF

from rpython.rtyper.annlowlevel import cast_gcref_to_instance

tag = FinalizerQueue._get_tag(self)

ptr = llop.gc_fq_next_dead(GCREF, tag)

if self.Class is not None:

ptr = cast_gcref_to_instance(self.Class, ptr)

return ptr

try:

return self._queue.popleft()

except (AttributeError, IndexError):

return None

@specialize.arg(0)

@jit.dont_look_inside

def register_finalizer(self, obj):

from rpython.rtyper.lltypesystem.llmemory import GCREF

if self.Class is None:

assert lltype.typeOf(obj) == GCREF

else:

assert isinstance(obj, self.Class)

if we_are_translated():

from rpython.rtyper.lltypesystem.lloperation import llop

from rpython.rtyper.annlowlevel import cast_instance_to_gcref

tag = FinalizerQueue._get_tag(self)

if self.Class is not None:

obj = cast_instance_to_gcref(obj)

llop.gc_fq_register(lltype.Void, tag, obj)

return

else:

self._untranslated_register_finalizer(obj)

@not_rpython

def _get_tag(self):

"special-cased below"

def _reset(self):

import collections

self._weakrefs = set()

self._queue = collections.deque()

def _already_registered(self, obj):

return hasattr(obj, '__enable_del_for_id')

def _untranslated_register_finalizer(self, obj):

assert not self._already_registered(obj)

if not hasattr(self, '_queue'):

self._reset()

# Fetch and check the type of 'obj'

objtyp = obj.__class__

assert isinstance(objtyp, type), (

"%r: to run register_finalizer() untranslated, "

"the object's class must be new-style" % (obj,))

assert hasattr(obj, '__dict__'), (

"%r: to run register_finalizer() untranslated, "

"the object must have a __dict__" % (obj,))

assert (not hasattr(obj, '__slots__') or

type(obj).__slots__ == () or

type(obj).__slots__ == ('__weakref__',)), (

"%r: to run register_finalizer() untranslated, "

"the object must not have __slots__" % (obj,))

# The first time, patch the method __del__ of the class, if

# any, so that we can disable it on the original 'obj' and

# enable it only on the 'newobj'

_fq_patch_class(objtyp)

# Build a new shadow object with the same class and dict

newobj = object.__new__(objtyp)

obj.__dict__ = obj.__dict__.copy() #PyPy: break the dict->obj dependency

newobj.__dict__ = obj.__dict__

# A callback that is invoked when (or after) 'obj' is deleted;

# 'newobj' is still kept alive here

def callback(wr):

self._weakrefs.discard(wr)

self._queue.append(newobj)

self.finalizer_trigger()

import weakref

wr = weakref.ref(obj, callback)

self._weakrefs.add(wr)

# Disable __del__ on the original 'obj' and enable it only on

# the 'newobj'. Use id() and not a regular reference, because

# that would make a cycle between 'newobj' and 'obj.__dict__'

# (which is 'newobj.__dict__' too).

setattr(obj, '__enable_del_for_id', id(newobj))

def _fq_patch_class(Cls):

if Cls in _fq_patched_classes:

return

if '__del__' in Cls.__dict__:

def __del__(self):

if not we_are_translated():

try:

if getattr(self, '__enable_del_for_id') != id(self):

return

except AttributeError:

pass

original_del(self)

original_del = Cls.__del__

Cls.__del__ = __del__

_fq_patched_classes.add(Cls)

for BaseCls in Cls.__bases__:

_fq_patch_class(BaseCls)

_fq_patched_classes = set()

class FqTagEntry(ExtRegistryEntry):

_about_ = FinalizerQueue._get_tag.im_func

def compute_result_annotation(self, s_fq):

assert s_fq.is_constant()

fq = s_fq.const

s_func = self.bookkeeper.immutablevalue(fq.finalizer_trigger)

self.bookkeeper.emulate_pbc_call(self.bookkeeper.position_key,

s_func, [])

if not hasattr(fq, '_fq_tag'):

fq._fq_tag = CDefinedIntSymbolic(

'0 /*FinalizerQueue TAG for %s*/' % fq.__class__.__name__,

default=fq)

return self.bookkeeper.immutablevalue(fq._fq_tag)

def specialize_call(self, hop):

from rpython.rtyper.rclass import InstanceRepr

translator = hop.rtyper.annotator.translator

fq = hop.args_s[0].const

graph = translator._graphof(fq.finalizer_trigger.im_func)

InstanceRepr.check_graph_of_del_does_not_call_too_much(hop.rtyper,

graph)

hop.exception_cannot_occur()

return hop.inputconst(lltype.Signed, hop.s_result.const)

@jit.dont_look_inside

@specialize.argtype(0)

def may_ignore_finalizer(obj):

"""Optimization hint: says that it is valid for any finalizer

for 'obj' to be ignored, depending on the GC."""

from rpython.rtyper.lltypesystem.lloperation import llop

llop.gc_ignore_finalizer(lltype.Void, obj)

@jit.dont_look_inside

def move_out_of_nursery(obj):

""" Returns another object which is a copy of obj; but at any point

(either now or in the future) the returned object might suddenly

become identical to the one returned.

NOTE: Only use for immutable objects!

NOTE: Might fail on some GCs! You have to check again

can_move() afterwards. It should always work with the default

GC. With Boehm, can_move() is always False so

move_out_of_nursery() should never be called in the first place.

"""

return obj

class MoveOutOfNurseryEntry(ExtRegistryEntry):

_about_ = move_out_of_nursery

def compute_result_annotation(self, s_obj):

return s_obj

def specialize_call(self, hop):

hop.exception_cannot_occur()

return hop.genop('gc_move_out_of_nursery', hop.args_v, resulttype=hop.r_result)

# ____________________________________________________________

@not_rpython

def get_rpy_roots():

# Return the 'roots' from the GC.

# The gc typically returns a list that ends with a few NULL_GCREFs.

return [_GcRef(x) for x in gc.get_objects()]

@not_rpython

def get_rpy_referents(gcref):

x = gcref._x

if isinstance(x, list):

d = x

elif isinstance(x, dict):

d = x.keys() + x.values()

else:

d = []

if hasattr(x, '__dict__'):

d = x.__dict__.values()

if hasattr(type(x), '__slots__'):

for slot in type(x).__slots__:

try:

d.append(getattr(x, slot))

except AttributeError:

pass

# discard objects that are too random or that are _freeze_=True

return [_GcRef(x) for x in d if _keep_object(x)]

def _keep_object(x):

if isinstance(x, type) or type(x) is types.ClassType:

return False # don't keep any type

if isinstance(x, (list, dict, str)):

return True # keep lists and dicts and strings

if hasattr(x, '_freeze_'):

return False

return type(x).__module__ != '__builtin__' # keep non-builtins

def add_memory_pressure(estimate, object=None):

"""Add memory pressure for OpaquePtrs."""

pass

class AddMemoryPressureEntry(ExtRegistryEntry):

_about_ = add_memory_pressure

def compute_result_annotation(self, s_nbytes, s_object=None):

from rpython.annotator import model as annmodel

if s_object is not None:

if not isinstance(s_object, annmodel.SomeInstance):

raise Exception("Wrong kind of object passed to "

"add memory pressure")

self.bookkeeper.memory_pressure_types.add(s_object.classdef)

return annmodel.s_None

def specialize_call(self, hop):

v_size = hop.inputarg(lltype.Signed, 0)

if len(hop.args_v) == 2:

v_obj = hop.inputarg(hop.args_r[1], 1)

args = [v_size, v_obj]

else:

args = [v_size]

hop.exception_cannot_occur()

return hop.genop('gc_add_memory_pressure', args,

resulttype=lltype.Void)

@not_rpython

def get_rpy_memory_usage(gcref):

# approximate implementation using CPython's type info

Class = type(gcref._x)

size = Class.__basicsize__

if Class.__itemsize__ > 0:

size += Class.__itemsize__ * len(gcref._x)

return size

@not_rpython

def get_rpy_type_index(gcref):

from rpython.rlib.rarithmetic import intmask

Class = gcref._x.__class__

i = intmask(id(Class))

if i < 0:

i = ~i # always return a positive number, at least

return i

def cast_gcref_to_int(gcref):

# This is meant to be used on cast_instance_to_gcref results.

# Don't use this on regular gcrefs obtained e.g. with

# lltype.cast_opaque_ptr().

if we_are_translated():

return lltype.cast_ptr_to_int(gcref)

else:

return id(gcref._x)

(TOTAL_MEMORY, TOTAL_ALLOCATED_MEMORY, TOTAL_MEMORY_PRESSURE,

PEAK_MEMORY, PEAK_ALLOCATED_MEMORY, TOTAL_ARENA_MEMORY,

TOTAL_RAWMALLOCED_MEMORY, PEAK_ARENA_MEMORY, PEAK_RAWMALLOCED_MEMORY,

NURSERY_SIZE, TOTAL_GC_TIME) = range(11)

@not_rpython

def get_stats(stat_no):

""" Long docstring goes here

"""

raise NotImplementedError

@not_rpython

def dump_rpy_heap(fd):

raise NotImplementedError

@not_rpython

def get_typeids_z():

raise NotImplementedError

@not_rpython

def get_typeids_list():

raise NotImplementedError

@not_rpython

def has_gcflag_extra():

return True

has_gcflag_extra._subopnum = 1

_gcflag_extras = set()

@not_rpython

def get_gcflag_extra(gcref):

assert gcref # not NULL!

return gcref in _gcflag_extras

get_gcflag_extra._subopnum = 2

@not_rpython

def toggle_gcflag_extra(gcref):

assert gcref # not NULL!

try:

_gcflag_extras.remove(gcref)

except KeyError:

_gcflag_extras.add(gcref)

toggle_gcflag_extra._subopnum = 3

def assert_no_more_gcflags():

if not we_are_translated():

assert not _gcflag_extras

ARRAY_OF_CHAR = lltype.Array(lltype.Char)

NULL_GCREF = lltype.nullptr(llmemory.GCREF.TO)

class _GcRef(object):

# implementation-specific: there should not be any after translation

__slots__ = ['_x', '_handle']

_TYPE = llmemory.GCREF

def __init__(self, x):

self._x = x

def __hash__(self):

return object.__hash__(self._x)

def __eq__(self, other):

if isinstance(other, lltype._ptr):

assert other == NULL_GCREF, (

"comparing a _GcRef with a non-NULL lltype ptr")

return False

assert isinstance(other, _GcRef)

return self._x is other._x

def __ne__(self, other):

return not self.__eq__(other)

def __repr__(self):

return "_GcRef(%r)" % (self._x, )

def _freeze_(self):

raise Exception("instances of rlib.rgc._GcRef cannot be translated")

def cast_instance_to_gcref(x):

# Before translation, casts an RPython instance into a _GcRef.

# After translation, it is a variant of cast_object_to_ptr(GCREF).

if we_are_translated():

from rpython.rtyper import annlowlevel

x = annlowlevel.cast_instance_to_base_ptr(x)

return lltype.cast_opaque_ptr(llmemory.GCREF, x)

else:

return _GcRef(x)

cast_instance_to_gcref._annspecialcase_ = 'specialize:argtype(0)'

def try_cast_gcref_to_instance(Class, gcref):

# Before translation, unwraps the RPython instance contained in a _GcRef.

# After translation, it is a type-check performed by the GC.

if we_are_translated():

from rpython.rtyper.rclass import OBJECTPTR, ll_isinstance

from rpython.rtyper.annlowlevel import cast_base_ptr_to_instance

if _is_rpy_instance(gcref):

objptr = lltype.cast_opaque_ptr(OBJECTPTR, gcref)

if objptr.typeptr: # may be NULL, e.g. in rdict's dummykeyobj

clsptr = _get_llcls_from_cls(Class)

if ll_isinstance(objptr, clsptr):

return cast_base_ptr_to_instance(Class, objptr)

return None

else:

if isinstance(gcref._x, Class):

return gcref._x

return None

try_cast_gcref_to_instance._annspecialcase_ = 'specialize:arg(0)'

_ffi_cache = None

def _fetch_ffi():

global _ffi_cache

if _ffi_cache is None:

try:

import _cffi_backend

_ffi_cache = _cffi_backend.FFI()

except (ImportError, AttributeError):

import py

py.test.skip("need CFFI >= 1.0")

return _ffi_cache

@jit.dont_look_inside

def hide_nonmovable_gcref(gcref):

from rpython.rtyper.lltypesystem import lltype, llmemory, rffi

if we_are_translated():

assert lltype.typeOf(gcref) == llmemory.GCREF

assert not can_move(gcref)

return rffi.cast(llmemory.Address, gcref)

else:

assert isinstance(gcref, _GcRef)

x = gcref._x

ffi = _fetch_ffi()

if not hasattr(x, '__handle'):

x.__handle = ffi.new_handle(x)

addr = int(ffi.cast("intptr_t", x.__handle))

return rffi.cast(llmemory.Address, addr)

@jit.dont_look_inside

def reveal_gcref(addr):

from rpython.rtyper.lltypesystem import lltype, llmemory, rffi

assert lltype.typeOf(addr) == llmemory.Address

if we_are_translated():

return rffi.cast(llmemory.GCREF, addr)

else:

addr = rffi.cast(lltype.Signed, addr)

if addr == 0:

return lltype.nullptr(llmemory.GCREF.TO)

ffi = _fetch_ffi()

x = ffi.from_handle(ffi.cast("void *", addr))

return _GcRef(x)

# ------------------- implementation -------------------

_cache_s_list_of_gcrefs = None

def s_list_of_gcrefs():

global _cache_s_list_of_gcrefs

if _cache_s_list_of_gcrefs is None:

from rpython.annotator import model as annmodel

from rpython.rtyper.llannotation import SomePtr

from rpython.annotator.listdef import ListDef

s_gcref = SomePtr(llmemory.GCREF)

_cache_s_list_of_gcrefs = annmodel.SomeList(

ListDef(None, s_gcref, mutated=True, resized=False))

return _cache_s_list_of_gcrefs

class Entry(ExtRegistryEntry):

_about_ = get_rpy_roots

def compute_result_annotation(self):

return s_list_of_gcrefs()

def specialize_call(self, hop):

hop.exception_cannot_occur()

return hop.genop('gc_get_rpy_roots', [], resulttype = hop.r_result)

class Entry(ExtRegistryEntry):

_about_ = get_rpy_referents

def compute_result_annotation(self, s_gcref):

from rpython.rtyper.llannotation import SomePtr

assert SomePtr(llmemory.GCREF).contains(s_gcref)

return s_list_of_gcrefs()

def specialize_call(self, hop):

vlist = hop.inputargs(hop.args_r[0])

hop.exception_cannot_occur()

return hop.genop('gc_get_rpy_referents', vlist,

resulttype=hop.r_result)

class Entry(ExtRegistryEntry):

_about_ = get_rpy_memory_usage

def compute_result_annotation(self, s_gcref):

from rpython.annotator import model as annmodel

return annmodel.SomeInteger()

def specialize_call(self, hop):

vlist = hop.inputargs(hop.args_r[0])

hop.exception_cannot_occur()

return hop.genop('gc_get_rpy_memory_usage', vlist,

resulttype = hop.r_result)

class Entry(ExtRegistryEntry):

_about_ = get_rpy_type_index

def compute_result_annotation(self, s_gcref):

from rpython.annotator import model as annmodel

return annmodel.SomeInteger()

def specialize_call(self, hop):

vlist = hop.inputargs(hop.args_r[0])

hop.exception_cannot_occur()

return hop.genop('gc_get_rpy_type_index', vlist,

resulttype = hop.r_result)

class Entry(ExtRegistryEntry):

_about_ = get_stats

def compute_result_annotation(self, s_no):

from rpython.annotator.model import SomeInteger

if not isinstance(s_no, SomeInteger):

raise Exception("expecting an integer")

return SomeInteger()

def specialize_call(self, hop):

args = hop.inputargs(lltype.Signed)

hop.exception_cannot_occur()