import operator

from rpython.annotator import model as annmodel

from rpython.flowspace.model import Constant

from rpython.rlib.rarithmetic import intmask

from rpython.rlib.unroll import unrolling_iterable

from rpython.rtyper.error import TyperError

from rpython.rtyper.lltypesystem.lltype import (

Void, Signed, Bool, Ptr, GcStruct, malloc, typeOf, nullptr)

from rpython.rtyper.lltypesystem.rstr import LLHelpers

from rpython.rtyper.rstr import AbstractStringRepr

from rpython.rtyper.rmodel import (Repr, inputconst, IteratorRepr,

externalvsinternal)

from rpython.rtyper.rint import IntegerRepr

from rpython.tool.pairtype import pairtype

class __extend__(annmodel.SomeTuple):

def rtyper_makerepr(self, rtyper):

return TupleRepr(rtyper, [rtyper.getrepr(s_item) for s_item in self.items])

def rtyper_makekey(self):

keys = [s_item.rtyper_makekey() for s_item in self.items]

return tuple([self.__class__] + keys)

_gen_eq_function_cache = {}

_gen_hash_function_cache = {}

_gen_str_function_cache = {}

def gen_eq_function(items_r):

eq_funcs = [r_item.get_ll_eq_function() or operator.eq for r_item in items_r]

key = tuple(eq_funcs)

try:

return _gen_eq_function_cache[key]

except KeyError:

autounrolling_funclist = unrolling_iterable(enumerate(eq_funcs))

def ll_eq(t1, t2):

equal_so_far = True

for i, eqfn in autounrolling_funclist:

if not equal_so_far:

return False

attrname = 'item%d' % i

item1 = getattr(t1, attrname)

item2 = getattr(t2, attrname)

equal_so_far = eqfn(item1, item2)

return equal_so_far

_gen_eq_function_cache[key] = ll_eq

return ll_eq

def gen_hash_function(items_r):

# based on CPython

hash_funcs = [r_item.get_ll_hash_function() for r_item in items_r]

key = tuple(hash_funcs)

try:

return _gen_hash_function_cache[key]

except KeyError:

autounrolling_funclist = unrolling_iterable(enumerate(hash_funcs))

def ll_hash(t):

"""Must be kept in sync with rlib.objectmodel._hash_tuple()."""

x = 0x345678

for i, hash_func in autounrolling_funclist:

attrname = 'item%d' % i

item = getattr(t, attrname)

y = hash_func(item)

x = intmask((1000003 * x) ^ y)

return x

_gen_hash_function_cache[key] = ll_hash

return ll_hash

def gen_str_function(tuplerepr):

items_r = tuplerepr.items_r

key = tuple([r_item.ll_str for r_item in items_r])

try:

return _gen_str_function_cache[key]

except KeyError:

autounrolling_funclist = unrolling_iterable(enumerate(key))

constant = LLHelpers.ll_constant

start = LLHelpers.ll_build_start

push = LLHelpers.ll_build_push

finish = LLHelpers.ll_build_finish

length = len(items_r)

def ll_str(t):

if length == 0:

return constant("()")

buf = start(2 * length + 1)

push(buf, constant("("), 0)

for i, str_func in autounrolling_funclist:

attrname = 'item%d' % i

item = getattr(t, attrname)

if i > 0:

push(buf, constant(", "), 2 * i)

push(buf, str_func(item), 2 * i + 1)

if length == 1:

push(buf, constant(",)"), 2 * length)

else:

push(buf, constant(")"), 2 * length)

return finish(buf)

_gen_str_function_cache[key] = ll_str

return ll_str

# ____________________________________________________________

#

# Concrete implementation of RPython tuples:

#

# struct tuple {

# type0 item0;

# type1 item1;

# type2 item2;

# ...

# }

def TUPLE_TYPE(field_lltypes):

if len(field_lltypes) == 0:

return Void # empty tuple

else:

fields = [('item%d' % i, TYPE) for i, TYPE in enumerate(field_lltypes)]

kwds = {'hints': {'immutable': True,

'noidentity': True}}

return Ptr(GcStruct('tuple%d' % len(field_lltypes), *fields, **kwds))

class TupleRepr(Repr):

def __init__(self, rtyper, items_r):

self.items_r = []

self.external_items_r = []

for item_r in items_r:

external_repr, internal_repr = externalvsinternal(rtyper, item_r)

self.items_r.append(internal_repr)

self.external_items_r.append(external_repr)

items_r = self.items_r

self.fieldnames = ['item%d' % i for i in range(len(items_r))]

self.lltypes = [r.lowleveltype for r in items_r]

self.tuple_cache = {}

self.lowleveltype = TUPLE_TYPE(self.lltypes)

def getitem(self, llops, v_tuple, index):

"""Generate the operations to get the index'th item of v_tuple,

in the external repr external_items_r[index]."""

v = self.getitem_internal(llops, v_tuple, index)

r_item = self.items_r[index]

r_external_item = self.external_items_r[index]

return llops.convertvar(v, r_item, r_external_item)

@classmethod

def newtuple(cls, llops, r_tuple, items_v):

# items_v should have the lowleveltype of the internal reprs

assert len(r_tuple.items_r) == len(items_v)

for r_item, v_item in zip(r_tuple.items_r, items_v):

assert r_item.lowleveltype == v_item.concretetype

#

if len(r_tuple.items_r) == 0:

return inputconst(Void, ()) # a Void empty tuple

c1 = inputconst(Void, r_tuple.lowleveltype.TO)

cflags = inputconst(Void, {'flavor': 'gc'})

v_result = llops.genop('malloc', [c1, cflags],

resulttype = r_tuple.lowleveltype)

for i in range(len(r_tuple.items_r)):

cname = inputconst(Void, r_tuple.fieldnames[i])

llops.genop('setfield', [v_result, cname, items_v[i]])

return v_result

@classmethod

def newtuple_cached(cls, hop, items_v):

r_tuple = hop.r_result

if hop.s_result.is_constant():

return inputconst(r_tuple, hop.s_result.const)

else:

return cls.newtuple(hop.llops, r_tuple, items_v)

@classmethod

def _rtype_newtuple(cls, hop):

r_tuple = hop.r_result

vlist = hop.inputargs(*r_tuple.items_r)

return cls.newtuple_cached(hop, vlist)

def convert_const(self, value):

assert isinstance(value, tuple) and len(value) == len(self.items_r)

key = tuple([Constant(item) for item in value])

try:

return self.tuple_cache[key]

except KeyError:

p = self.instantiate()

self.tuple_cache[key] = p

for obj, r, name in zip(value, self.items_r, self.fieldnames):

if r.lowleveltype is not Void:

setattr(p, name, r.convert_const(obj))

return p

def compact_repr(self):

return "TupleR %s" % ' '.join([llt._short_name() for llt in self.lltypes])

def rtype_len(self, hop):

return hop.inputconst(Signed, len(self.items_r))

def get_ll_eq_function(self):

return gen_eq_function(self.items_r)

def get_ll_hash_function(self):

return gen_hash_function(self.items_r)

# no get_ll_fasthash_function: the hash is a bit slow, better cache

# it inside dict entries

ll_str = property(gen_str_function)

def make_iterator_repr(self, variant=None):

if variant is not None:

raise TyperError("unsupported %r iterator over a tuple" %

(variant,))

if len(self.items_r) == 1:

# subclasses are supposed to set the IteratorRepr attribute

return self.IteratorRepr(self)

raise TyperError("can only iterate over tuples of length 1 for now")

def instantiate(self):

if len(self.items_r) == 0:

return dum_empty_tuple # PBC placeholder for an empty tuple

else:

return malloc(self.lowleveltype.TO)

def rtype_bltn_list(self, hop):

from rpython.rtyper.lltypesystem import rlist

nitems = len(self.items_r)

vtup = hop.inputarg(self, 0)

LIST = hop.r_result.lowleveltype.TO

cno = inputconst(Signed, nitems)

hop.exception_is_here()

vlist = hop.gendirectcall(LIST.ll_newlist, cno)

v_func = hop.inputconst(Void, rlist.dum_nocheck)

for index in range(nitems):

name = self.fieldnames[index]

ritem = self.items_r[index]

cname = hop.inputconst(Void, name)

vitem = hop.genop('getfield', [vtup, cname], resulttype = ritem)

vitem = hop.llops.convertvar(vitem, ritem, hop.r_result.item_repr)

cindex = inputconst(Signed, index)

hop.gendirectcall(rlist.ll_setitem_nonneg, v_func, vlist, cindex, vitem)

return vlist

def getitem_internal(self, llops, v_tuple, index):

"""Return the index'th item, in internal repr."""

name = self.fieldnames[index]

llresult = self.lltypes[index]

cname = inputconst(Void, name)

return llops.genop('getfield', [v_tuple, cname], resulttype = llresult)

def rtype_newtuple(hop):

return TupleRepr._rtype_newtuple(hop)

newtuple = TupleRepr.newtuple

def dum_empty_tuple(): pass

class __extend__(pairtype(TupleRepr, IntegerRepr)):

def rtype_getitem((r_tup, r_int), hop):

v_tuple, v_index = hop.inputargs(r_tup, Signed)

if not isinstance(v_index, Constant):

raise TyperError("non-constant tuple index")

if hop.has_implicit_exception(IndexError):

hop.exception_cannot_occur()

index = v_index.value

return r_tup.getitem(hop.llops, v_tuple, index)

class __extend__(TupleRepr):

def rtype_getslice(r_tup, hop):

s_start = hop.args_s[1]

s_stop = hop.args_s[2]

assert s_start.is_immutable_constant(),"tuple slicing: needs constants"

assert s_stop.is_immutable_constant(), "tuple slicing: needs constants"

start = s_start.const

stop = s_stop.const

indices = range(len(r_tup.items_r))[start:stop]

assert len(indices) == len(hop.r_result.items_r)

v_tup = hop.inputarg(r_tup, arg=0)

items_v = [r_tup.getitem_internal(hop.llops, v_tup, i)

for i in indices]

return hop.r_result.newtuple(hop.llops, hop.r_result, items_v)

class __extend__(pairtype(TupleRepr, Repr)):

def rtype_contains((r_tup, r_item), hop):

s_tup = hop.args_s[0]

if not s_tup.is_constant():

raise TyperError("contains() on non-const tuple")

t = s_tup.const

s_item = hop.args_s[1]

r_item = hop.args_r[1]

v_arg = hop.inputarg(r_item, arg=1)

ll_eq = r_item.get_ll_eq_function() or _ll_equal

v_result = None

for x in t:

s_const_item = hop.rtyper.annotator.bookkeeper.immutablevalue(x)

if not s_item.contains(s_const_item):

continue # corner case, see test_constant_tuple_contains_bug

c_tuple_item = hop.inputconst(r_item, x)

v_equal = hop.gendirectcall(ll_eq, v_arg, c_tuple_item)

if v_result is None:

v_result = v_equal

else:

v_result = hop.genop("int_or", [v_result, v_equal],

resulttype = Bool)

hop.exception_cannot_occur()

return v_result or hop.inputconst(Bool, False)

class __extend__(pairtype(TupleRepr, TupleRepr)):

def rtype_add((r_tup1, r_tup2), hop):

v_tuple1, v_tuple2 = hop.inputargs(r_tup1, r_tup2)

vlist = []

for i in range(len(r_tup1.items_r)):

vlist.append(r_tup1.getitem_internal(hop.llops, v_tuple1, i))

for i in range(len(r_tup2.items_r)):

vlist.append(r_tup2.getitem_internal(hop.llops, v_tuple2, i))

return r_tup1.newtuple_cached(hop, vlist)

rtype_inplace_add = rtype_add

def rtype_eq((r_tup1, r_tup2), hop):

s_tup = annmodel.unionof(*hop.args_s)

r_tup = hop.rtyper.getrepr(s_tup)

v_tuple1, v_tuple2 = hop.inputargs(r_tup, r_tup)

ll_eq = r_tup.get_ll_eq_function()

return hop.gendirectcall(ll_eq, v_tuple1, v_tuple2)

def rtype_ne(tup1tup2, hop):

v_res = tup1tup2.rtype_eq(hop)

return hop.genop('bool_not', [v_res], resulttype=Bool)

def convert_from_to((r_from, r_to), v, llops):

if len(r_from.items_r) == len(r_to.items_r):

if r_from.lowleveltype == r_to.lowleveltype:

return v

n = len(r_from.items_r)

items_v = []

for i in range(n):

item_v = r_from.getitem_internal(llops, v, i)

item_v = llops.convertvar(item_v,

r_from.items_r[i],

r_to.items_r[i])

items_v.append(item_v)

return r_from.newtuple(llops, r_to, items_v)

return NotImplemented

def rtype_is_((robj1, robj2), hop):

raise TyperError("cannot compare tuples with 'is'")

class __extend__(pairtype(AbstractStringRepr, TupleRepr)):

def rtype_mod((r_str, r_tuple), hop):

r_tuple = hop.args_r[1]

v_tuple = hop.args_v[1]

sourcevars = []

for i, r_arg in enumerate(r_tuple.external_items_r):

v_item = r_tuple.getitem(hop.llops, v_tuple, i)

sourcevars.append((v_item, r_arg))

return r_str.ll.do_stringformat(hop, sourcevars)

# ____________________________________________________________

#

# Iteration.

class AbstractTupleIteratorRepr(IteratorRepr):

def newiter(self, hop):

v_tuple, = hop.inputargs(self.r_tuple)

citerptr = hop.inputconst(Void, self.lowleveltype)

return hop.gendirectcall(self.ll_tupleiter, citerptr, v_tuple)

def rtype_next(self, hop):

v_iter, = hop.inputargs(self)

hop.has_implicit_exception(StopIteration) # record that we know about it

hop.exception_is_here()

v = hop.gendirectcall(self.ll_tuplenext, v_iter)

return hop.llops.convertvar(v, self.r_tuple.items_r[0], self.r_tuple.external_items_r[0])

class Length1TupleIteratorRepr(AbstractTupleIteratorRepr):

def __init__(self, r_tuple):

self.r_tuple = r_tuple

self.lowleveltype = Ptr(GcStruct('tuple1iter',

('tuple', r_tuple.lowleveltype)))

self.ll_tupleiter = ll_tupleiter

self.ll_tuplenext = ll_tuplenext

TupleRepr.IteratorRepr = Length1TupleIteratorRepr

def ll_tupleiter(ITERPTR, tuple):

iter = malloc(ITERPTR.TO)

iter.tuple = tuple

return iter

def ll_tuplenext(iter):

# for iterating over length 1 tuples only!

t = iter.tuple

if t:

iter.tuple = nullptr(typeOf(t).TO)

return t.item0

else:

raise StopIteration

def _ll_equal(x, y):

return x == y