def _PyString_Resize(space, ref, newsize):
    """A way to resize a string object even though it is "immutable". Only use this to
    build up a brand new string object; don't use this if the string may already be
    known in other parts of the code.  It is an error to call this function if the
    refcount on the input string object is not one. Pass the address of an existing
    string object as an lvalue (it may be written into), and the new size desired.
    On success, *string holds the resized string object and 0 is returned;
    the address in *string may differ from its input value.  If the reallocation
    fails, the original string object at *string is deallocated, *string is
    set to NULL, a memory exception is set, and -1 is returned.
    """
    # XXX always create a new string so far
    py_str = rffi.cast(PyStringObject, ref[0])
    if not py_str.c_buffer:
        raise OperationError(space.w_SystemError, space.wrap(
            "_PyString_Resize called on already created string"))
    try:
        py_newstr = new_empty_str(space, newsize)
    except MemoryError:
        Py_DecRef(space, ref[0])
        ref[0] = lltype.nullptr(PyObject.TO)
        raise
    to_cp = newsize
    oldsize = py_str.c_size
    if oldsize < newsize:
        to_cp = oldsize
    for i in range(to_cp):
        py_newstr.c_buffer[i] = py_str.c_buffer[i]
    Py_DecRef(space, ref[0])
    ref[0] = rffi.cast(PyObject, py_newstr)
    return 0

 def test_poll(self):
     if not self.is_interactive:
         py.test.skip("interactive test only")
     import time, sys
     RSDL.EnableUNICODE(1)
     print
     print "Keys pressed in the Pygame window give a dot."
     print "    Wait 3 seconds to quit."
     timeout = time.time() + 3
     event = lltype.malloc(RSDL.Event, flavor='raw')
     try:
         while True:
             # busy polling
             ok = RSDL.PollEvent(event)
             ok = rffi.cast(lltype.Signed, ok)
             assert ok >= 0
             if ok > 0:
                 c_type = rffi.getintfield(event, 'c_type')
                 if c_type == RSDL.KEYDOWN:
                     sys.stderr.write('.')
                     p = rffi.cast(RSDL.KeyboardEventPtr, event)
                     if rffi.getintfield(p.c_keysym, 'c_sym') == RSDL.K_ESCAPE:
                         print 'Escape key'
                         break
                     timeout = time.time() + 3
             else:
                 if time.time() > timeout:
                     break
             time.sleep(0.05)
     finally:
         lltype.free(event, flavor='raw')

 def _base_do_getfield_f(self, struct, fielddescr):
     ofs = self.unpack_fielddescr(fielddescr)
     # --- start of GC unsafe code (no GC operation!) ---
     fieldptr = rffi.ptradd(rffi.cast(rffi.CCHARP, struct), ofs)
     fval = rffi.cast(rffi.CArrayPtr(longlong.FLOATSTORAGE), fieldptr)[0]
     # --- end of GC unsafe code ---
     return fval

 def test_mousemove(self):
     if not self.is_interactive:
         py.test.skip("interactive test only")
     print
     print "Move the Mouse up and down:"
     print "    Use Escape to quit."
     event = lltype.malloc(RSDL.Event, flavor="raw")
     directions = [False]*4
     try:
         while True:
             ok = RSDL.WaitEvent(event)
             assert rffi.cast(lltype.Signed, ok) == 1
             c_type = rffi.getintfield(event, "c_type")
             if c_type == RSDL.MOUSEMOTION:
                 m = rffi.cast(RSDL.MouseMotionEventPtr, event)
                 assert rffi.getintfield(m, "c_x") >= 0
                 assert rffi.getintfield(m, "c_y") >= 0
                 print rffi.getintfield(m, "c_xrel")
                 directions[0] |= rffi.getintfield(m, "c_xrel")>0
                 directions[1] |= rffi.getintfield(m, "c_xrel")<0
                 directions[2] |= rffi.getintfield(m, "c_yrel")>0
                 directions[3] |= rffi.getintfield(m, "c_yrel")<0
                 if False not in directions:
                     break
             elif c_type == RSDL.KEYUP:
                 p = rffi.cast(RSDL.KeyboardEventPtr, event)
                 if rffi.getintfield(p.c_keysym, 'c_sym') == RSDL.K_ESCAPE:
                     print "    test manually aborted"
                     py.test.fail(" mousemovement test aborted")
                     break  
     finally:
         lltype.free(event, flavor='raw')

    def test_forced_ptr_cast(self):
        import array
        A = lltype.Array(lltype.Signed, hints={'nolength': True})
        B = lltype.Array(lltype.Char, hints={'nolength': True})
        a = lltype.malloc(A, 10, flavor='raw')
        for i in range(10):
            a[i] = i*i

        b = rffi.cast(lltype.Ptr(B), a)

        checker = array.array('l')
        for i in range(10):
            checker.append(i*i)
        expected = checker.tostring()

        for i in range(len(expected)):
            assert b[i] == expected[i]

        c = rffi.cast(rffi.VOIDP, a)
        addr = lltype2ctypes(c)
        #assert addr == ctypes.addressof(a._obj._ctypes_storage)
        d = ctypes2lltype(rffi.VOIDP, addr)
        assert lltype.typeOf(d) == rffi.VOIDP
        assert c == d
        e = rffi.cast(lltype.Ptr(A), d)
        for i in range(10):
            assert e[i] == i*i

        c = lltype.nullptr(rffi.VOIDP.TO)
        addr = rffi.cast(lltype.Signed, c)
        assert addr == 0

        lltype.free(a, flavor='raw')
        assert not ALLOCATED     # detects memory leaks in the test

 def freeing_block(self, start, stop):
     # if [start:stop] is a raw block of assembler, then look up the
     # corresponding gcroot markers, and mark them as freed now in
     # self._gcmap by setting the 2nd address of every entry to NULL.
     gcmapstart = self.gcmapstart()
     gcmapend   = self.gcmapend()
     if gcmapstart == gcmapend:
         return
     if not self.gcmarksorted():
         asmgcroot.sort_gcmap(gcmapstart, gcmapend)
     # A note about gcmarksorted(): the deletion we do here keeps the
     # array sorted.  This avoids needing too many sort_gcmap()s.
     # Indeed, freeing_block() is typically called many times in a row,
     # so it will call sort_gcmap() at most the first time.
     startaddr = rffi.cast(llmemory.Address, start)
     stopaddr  = rffi.cast(llmemory.Address, stop)
     item = asmgcroot.binary_search(gcmapstart, gcmapend, startaddr)
     # 'item' points to one of the entries.  Because the whole array
     # is sorted, we know that it points either to the first entry we
     # want to kill, or to the previous entry.
     if item.address[0] < startaddr:
         item += asmgcroot.arrayitemsize    # go forward one entry
         assert item == gcmapend or item.address[0] >= startaddr
     while item != gcmapend and item.address[0] < stopaddr:
         item.address[1] = llmemory.NULL
         self._gcmap_deadentries += 1
         item += asmgcroot.arrayitemsize

def frame_attach(space, py_obj, w_obj):
    "Fills a newly allocated PyFrameObject with a frame object"
    frame = space.interp_w(PyFrame, w_obj)
    py_frame = rffi.cast(PyFrameObject, py_obj)
    py_frame.c_f_code = rffi.cast(PyCodeObject, make_ref(space, frame.pycode))
    py_frame.c_f_globals = make_ref(space, frame.w_globals)
    rffi.setintfield(py_frame, 'c_f_lineno', frame.f_lineno)

def frame_dealloc(space, py_obj):
    py_frame = rffi.cast(PyFrameObject, py_obj)
    py_code = rffi.cast(PyObject, py_frame.c_f_code)
    Py_DecRef(space, py_code)
    Py_DecRef(space, py_frame.c_f_globals)
    from pypy.module.cpyext.object import PyObject_dealloc
    PyObject_dealloc(space, py_obj)

def test_call_stubs():
    c0 = GcCache(False)
    ARGS = [lltype.Char, lltype.Signed]
    RES = lltype.Char
    descr1 = get_call_descr(c0, ARGS, RES)
    def f(a, b):
        return 'c'

    call_stub = descr1.call_stub
    fnptr = llhelper(lltype.Ptr(lltype.FuncType(ARGS, RES)), f)

    res = call_stub(rffi.cast(lltype.Signed, fnptr), [1, 2], None, None)
    assert res == ord('c')

    ARRAY = lltype.GcArray(lltype.Signed)
    ARGS = [lltype.Float, lltype.Ptr(ARRAY)]
    RES = lltype.Float

    def f(a, b):
        return float(b[0]) + a

    fnptr = llhelper(lltype.Ptr(lltype.FuncType(ARGS, RES)), f)
    descr2 = get_call_descr(c0, ARGS, RES)
    a = lltype.malloc(ARRAY, 3)
    opaquea = lltype.cast_opaque_ptr(llmemory.GCREF, a)
    a[0] = 1
    res = descr2.call_stub(rffi.cast(lltype.Signed, fnptr),
                           [], [opaquea], [longlong.getfloatstorage(3.5)])
    assert longlong.getrealfloat(res) == 4.5

def gethost_common(hostname, hostent, addr=None):
    if not hostent:
        raise HSocketError(hostname)
    family = rffi.getintfield(hostent, 'c_h_addrtype')
    if addr is not None and addr.family != family:
        raise CSocketError(_c.EAFNOSUPPORT)

    h_aliases = hostent.c_h_aliases
    if h_aliases:   # h_aliases can be NULL, according to SF #1511317
        aliases = rffi.charpp2liststr(h_aliases)
    else:
        aliases = []

    address_list = []
    h_addr_list = hostent.c_h_addr_list
    i = 0
    paddr = h_addr_list[0]
    while paddr:
        if family == AF_INET:
            p = rffi.cast(lltype.Ptr(_c.in_addr), paddr)
            addr = INETAddress.from_in_addr(p)
        elif AF_INET6 is not None and family == AF_INET6:
            p = rffi.cast(lltype.Ptr(_c.in6_addr), paddr)
            addr = INET6Address.from_in6_addr(p)
        else:
            raise RSocketError("unknown address family")
        address_list.append(addr)
        i += 1
        paddr = h_addr_list[i]
    return (rffi.charp2str(hostent.c_h_name), aliases, address_list)

def buffer_attach(space, py_obj, w_obj):
    """
    Fills a newly allocated PyBufferObject with the given (str) buffer object.
    """
    py_buf = rffi.cast(PyBufferObject, py_obj)
    py_buf.c_b_offset = 0
    rffi.setintfield(py_buf, 'c_b_readonly', 1)
    rffi.setintfield(py_buf, 'c_b_hash', -1)

    if isinstance(w_obj, SubBuffer):
        py_buf.c_b_offset = w_obj.offset
        w_obj = w_obj.buffer

    # If w_obj already allocated a fixed buffer, use it, and keep a
    # reference to w_obj.
    # Otherwise, b_base stays NULL, and we own the b_ptr.

    if isinstance(w_obj, StringBuffer):
        py_buf.c_b_base = lltype.nullptr(PyObject.TO)
        py_buf.c_b_ptr = rffi.cast(rffi.VOIDP, rffi.str2charp(w_obj.value))
        py_buf.c_b_size = w_obj.getlength()
    elif isinstance(w_obj, ArrayBuffer):
        w_base = w_obj.array
        py_buf.c_b_base = make_ref(space, w_base)
        py_buf.c_b_ptr = rffi.cast(rffi.VOIDP, w_obj.array._charbuf_start())
        py_buf.c_b_size = w_obj.getlength()
    else:
        raise OperationError(space.w_NotImplementedError, space.wrap(
            "buffer flavor not supported"))

        def test(encoded, endian, realendian=None):
            encoded_charp = rffi.str2charp(encoded)
            strict_charp = rffi.str2charp("strict")
            if endian is not None:
                if endian < 0:
                    value = -1
                elif endian > 0:
                    value = 1
                else:
                    value = 0
                pendian = lltype.malloc(rffi.INTP.TO, 1, flavor='raw')
                pendian[0] = rffi.cast(rffi.INT, value)
            else:
                pendian = None

            w_ustr = api.PyUnicode_DecodeUTF16(encoded_charp, len(encoded), strict_charp, pendian)
            assert space.eq_w(space.call_method(w_ustr, 'encode', space.wrap('ascii')),
                              space.wrap("abcd"))

            rffi.free_charp(encoded_charp)
            rffi.free_charp(strict_charp)
            if pendian:
                if realendian is not None:
                    assert rffi.cast(rffi.INT, realendian) == pendian[0]
                lltype.free(pendian, flavor='raw')

 def test_struct_fields(self):
     longsize = 4 if IS_32_BIT else 8
     POINT = lltype.Struct('POINT',
                           ('x', rffi.LONG),
                           ('y', rffi.SHORT),
                           ('z', rffi.VOIDP),
                           )
     y_ofs = longsize
     z_ofs = longsize*2
     p = lltype.malloc(POINT, flavor='raw')
     p.x = 42
     p.y = rffi.cast(rffi.SHORT, -1)
     p.z = rffi.cast(rffi.VOIDP, 0x1234)
     addr = rffi.cast(rffi.VOIDP, p)
     assert struct_getfield_int(types.slong, addr, 0) == 42
     assert struct_getfield_int(types.sshort, addr, y_ofs) == -1
     assert struct_getfield_int(types.pointer, addr, z_ofs) == 0x1234
     #
     struct_setfield_int(types.slong, addr, 0, 43)
     struct_setfield_int(types.sshort, addr, y_ofs, 0x1234FFFE) # 0x1234 is masked out
     struct_setfield_int(types.pointer, addr, z_ofs, 0x4321)
     assert p.x == 43
     assert p.y == -2
     assert rffi.cast(rffi.LONG, p.z) == 0x4321
     #
     lltype.free(p, flavor='raw')

 def test_simple_cast(self):
     assert rffi.cast(rffi.SIGNEDCHAR, 0x123456) == 0x56
     assert rffi.cast(rffi.SIGNEDCHAR, 0x123481) == -127
     assert rffi.cast(rffi.CHAR, 0x123456) == '\x56'
     assert rffi.cast(rffi.CHAR, 0x123481) == '\x81'
     assert rffi.cast(rffi.UCHAR, 0x123481) == 0x81
     assert not ALLOCATED     # detects memory leaks in the test

    def getattr(self, space, attr):
        try:
            attribute = self.objectType.attributesByName[attr]
        except KeyError:
            msg = "ExternalObject has no attribute '%s'" %(attr,)
            raise OperationError(space.w_AttributeError, space.wrap(msg))

        environment = self.objectType.environment

        scalarvalueindicatorptr = lltype.malloc(rffi.CArrayPtr(roci.OCIInd).TO,
                                                1, flavor='raw')
        valueindicatorptr = lltype.malloc(rffi.CArrayPtr(roci.dvoidp).TO,
                                          1, flavor='raw')
        valueptr = lltype.malloc(rffi.CArrayPtr(roci.dvoidp).TO,
                                 1, flavor='raw')
        tdoptr = lltype.malloc(rffi.CArrayPtr(roci.OCIType).TO,
                               1, flavor='raw')
        nameptr = lltype.malloc(rffi.CArrayPtr(roci.oratext).TO,
                               1, flavor='raw')
        nameptr[0] = rffi.str2charp(attr)
        namelenptr = lltype.malloc(rffi.CArrayPtr(roci.ub4).TO,
                                   1, flavor='raw')
        namelenptr[0] = rffi.cast(roci.ub4, len(attr))

        try:
            status = roci.OCIObjectGetAttr(
                environment.handle,
                environment.errorHandle,
                self.instance,
                self.indicator,
                self.objectType.tdo,
                nameptr, namelenptr, 1,
                lltype.nullptr(roci.Ptr(roci.ub4).TO), 0,
                scalarvalueindicatorptr,
                valueindicatorptr,
                valueptr,
                tdoptr)
            environment.checkForError(
                status, "ExternalObject_GetAttributeValue(): getting value")

            # determine the proper null indicator
            valueIndicator = valueindicatorptr[0]
            if not valueIndicator:
                valueIndicator = rffi.cast(roci.dvoidp,
                                           scalarvalueindicatorptr)
            value = valueptr[0]

            return convertObject(
                space, environment,
                attribute.typeCode,
                value, valueIndicator,
                self, attribute.subType)
        finally:
            lltype.free(scalarvalueindicatorptr, flavor='raw')
            lltype.free(valueindicatorptr, flavor='raw')
            lltype.free(valueptr, flavor='raw')
            lltype.free(tdoptr, flavor='raw')
            rffi.free_charp(nameptr[0])
            lltype.free(nameptr, flavor='raw')
            lltype.free(namelenptr, flavor='raw')

    def test_qsort(self):
        CMPFUNC = lltype.FuncType([rffi.VOIDP, rffi.VOIDP], rffi.INT)
        qsort = rffi.llexternal('qsort', [rffi.VOIDP,
                                          rffi.SIZE_T,
                                          rffi.SIZE_T,
                                          lltype.Ptr(CMPFUNC)],
                                lltype.Void)

        lst = [23, 43, 24, 324, 242, 34, 78, 5, 3, 10]
        A = lltype.Array(lltype.Signed, hints={'nolength': True})
        a = lltype.malloc(A, 10, flavor='raw')
        for i in range(10):
            a[i] = lst[i]

        SIGNEDPTR = lltype.Ptr(lltype.FixedSizeArray(lltype.Signed, 1))

        def my_compar(p1, p2):
            p1 = rffi.cast(SIGNEDPTR, p1)
            p2 = rffi.cast(SIGNEDPTR, p2)
            print 'my_compar:', p1[0], p2[0]
            return rffi.cast(rffi.INT, cmp(p1[0], p2[0]))

        qsort(rffi.cast(rffi.VOIDP, a),
              rffi.cast(rffi.SIZE_T, 10),
              rffi.cast(rffi.SIZE_T, llmemory.sizeof(lltype.Signed)),
              llhelper(lltype.Ptr(CMPFUNC), my_compar))

        for i in range(10):
            print a[i],
        print
        lst.sort()
        for i in range(10):
            assert a[i] == lst[i]
        lltype.free(a, flavor='raw')
        assert not ALLOCATED     # detects memory leaks in the test

def wrap_descr_delete(space, w_self, w_args, func):
    func_target = rffi.cast(descrsetfunc, func)
    check_num_args(space, w_args, 1)
    w_obj, = space.fixedview(w_args)
    res = generic_cpy_call(space, func_target, w_self, w_obj, None)
    if rffi.cast(lltype.Signed, res) == -1:
        space.fromcache(State).check_and_raise_exception(always=True)

    def llimpl_FormatError(code):
        "Return a message corresponding to the given Windows error code."
        buf = lltype.malloc(rffi.CCHARPP.TO, 1, flavor='raw')

        try:
            msglen = FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER |
                                   FORMAT_MESSAGE_FROM_SYSTEM,
                                   None,
                                   rffi.cast(DWORD, code),
                                   DEFAULT_LANGUAGE,
                                   rffi.cast(rffi.CCHARP, buf),
                                   0, None)

            if msglen <= 2:   # includes the case msglen < 0
                return fake_FormatError(code)

            # FormatMessage always appends \r\n.
            buflen = intmask(msglen - 2)
            assert buflen > 0

            result = rffi.charpsize2str(buf[0], buflen)
            LocalFree(rffi.cast(rffi.VOIDP, buf[0]))
        finally:
            lltype.free(buf, flavor='raw')

        return result

def _PyLong_FromByteArray(space, bytes, n, little_endian, signed):
    little_endian = rffi.cast(lltype.Signed, little_endian)
    signed = rffi.cast(lltype.Signed, signed)

    result = rbigint()
    negative = False

    for i in range(0, n):
        if little_endian:
            c = intmask(bytes[i])
        else:
            c = intmask(bytes[n - i - 1])
        if i == 0 and signed and c & 0x80:
            negative = True
        if negative:
            c = c ^ 0xFF
        digit = rbigint.fromint(c)

        result = result.lshift(8)
        result = result.add(digit)

    if negative:
        result = result.neg()

    return space.newlong_from_rbigint(result)

def _operate(stream, data, flush, max_length, cfunc, while_doing):
    """Common code for compress() and decompress().
    """
    # Prepare the input buffer for the stream
    with lltype.scoped_alloc(rffi.CCHARP.TO, len(data)) as inbuf:
        for i in xrange(len(data)):
            inbuf[i] = data[i]
        stream.c_next_in = rffi.cast(Bytefp, inbuf)
        rffi.setintfield(stream, 'c_avail_in', len(data))

        # Prepare the output buffer
        with lltype.scoped_alloc(rffi.CCHARP.TO, OUTPUT_BUFFER_SIZE) as outbuf:
            # Strategy: we call deflate() to get as much output data as fits in
            # the buffer, then accumulate all output into a StringBuffer
            # 'result'.
            result = StringBuilder()

            while True:
                stream.c_next_out = rffi.cast(Bytefp, outbuf)
                bufsize = OUTPUT_BUFFER_SIZE
                if max_length < bufsize:
                    if max_length <= 0:
                        err = Z_OK
                        break
                    bufsize = max_length
                max_length -= bufsize
                rffi.setintfield(stream, 'c_avail_out', bufsize)
                err = cfunc(stream, flush)
                if err == Z_OK or err == Z_STREAM_END:
                    # accumulate data into 'result'
                    avail_out = rffi.cast(lltype.Signed, stream.c_avail_out)
                    result.append_charpsize(outbuf, bufsize - avail_out)
                    # if the output buffer is full, there might be more data
                    # so we need to try again.  Otherwise, we're done.
                    if avail_out > 0:
                        break
                    # We're also done if we got a Z_STREAM_END (which should
                    # only occur when flush == Z_FINISH).
                    if err == Z_STREAM_END:
                        break
                    else:
                        continue
                elif err == Z_BUF_ERROR:
                    avail_out = rffi.cast(lltype.Signed, stream.c_avail_out)
                    # When compressing, we will only get Z_BUF_ERROR if
                    # the output buffer was full but there wasn't more
                    # output when we tried again, so it is not an error
                    # condition.
                    if avail_out == bufsize:
                        break

                # fallback case: report this error
                raise RZlibError.fromstream(stream, err, while_doing)

    # When decompressing, if the compressed stream of data was truncated,
    # then the zlib simply returns Z_OK and waits for more.  If it is
    # complete it returns Z_STREAM_END.
    return (result.build(),
            err,
            rffi.cast(lltype.Signed, stream.c_avail_in))