python类getrefcount()的实例源码

test_refcounts.py 文件源码 项目:sslstrip-hsts-openwrt 作者: adde88 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def test_1(self):
        from sys import getrefcount as grc

        f = dll._testfunc_callback_i_if
        f.restype = ctypes.c_int
        f.argtypes = [ctypes.c_int, MyCallback]

        def callback(value):
            #print "called back with", value
            return value

        self.assertEqual(grc(callback), 2)
        cb = MyCallback(callback)

        self.assertGreater(grc(callback), 2)
        result = f(-10, cb)
        self.assertEqual(result, -18)
        cb = None

        gc.collect()

        self.assertEqual(grc(callback), 2)
ngamsStatusCmd.py 文件源码 项目:ngas 作者: ICRAR 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def _getRefCounts():
    """
    Return information about all object allocated and the number of
    references pointing to each object. This can be used to check a running
    server for memory (object) leaks.

    Taken from: http://www.nightmare.com/medusa/memory-leaks.html.

    Returns:
    """
    T = TRACE()

    d = {}
    sys.modules
    # Collect all classes
    for m in sys.modules.values():
        for sym in dir(m):
            o = getattr (m, sym)
            if type(o) is types.ClassType:
                d[o] = sys.getrefcount (o)
    # sort by refcount
    pairs = map(lambda x: (x[1],x[0]), d.items())
    pairs.sort()
    pairs.reverse()
    return pairs
test_enumerate.py 文件源码 项目:web_ctp 作者: molebot 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
def test_bug1229429(self):
        # this bug was never in reversed, it was in
        # PyObject_CallMethod, and reversed_new calls that sometimes.
        if not hasattr(sys, "getrefcount"):
            return
        def f():
            pass
        r = f.__reversed__ = object()
        rc = sys.getrefcount(r)
        for i in range(10):
            try:
                reversed(f)
            except TypeError:
                pass
            else:
                self.fail("non-callable __reversed__ didn't raise!")
        self.assertEqual(rc, sys.getrefcount(r))
pixelarray_test.py 文件源码 项目:Projects 作者: it2school 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def test_ass_subscript (self):
        for bpp in (8, 16, 24, 32):
            sf = pygame.Surface ((6, 8), 0, bpp)
            sf.fill ((255, 255, 255))
            ar = pygame.PixelArray (sf)

            # Test ellipse working
            ar[...,...] = (0, 0, 0)
            self.assertEqual (ar[0,0], 0)
            self.assertEqual (ar[1,0], 0)
            self.assertEqual (ar[-1,-1], 0)
            ar[...,] = (0, 0, 255)
            self.assertEqual (ar[0,0], sf.map_rgb ((0, 0, 255)))
            self.assertEqual (ar[1,0], sf.map_rgb ((0, 0, 255)))
            self.assertEqual (ar[-1,-1], sf.map_rgb ((0, 0, 255)))
            ar[:,...] = (255, 0, 0)
            self.assertEqual (ar[0,0], sf.map_rgb ((255, 0, 0)))
            self.assertEqual (ar[1,0], sf.map_rgb ((255, 0, 0)))
            self.assertEqual (ar[-1,-1], sf.map_rgb ((255, 0, 0)))
            ar[...] = (0, 255, 0)
            self.assertEqual (ar[0,0], sf.map_rgb ((0, 255, 0)))
            self.assertEqual (ar[1,0], sf.map_rgb ((0, 255, 0)))
            self.assertEqual (ar[-1,-1], sf.map_rgb ((0, 255, 0)))

            # Ensure x and y are freed for array[x, y] = p
            # Bug fix: reference counting
            if hasattr(sys, 'getrefcount'):
                class Int(int):
                    """Unique int instances"""
                    pass

                sf = pygame.Surface ((2, 2), 0, 32)
                ar = pygame.PixelArray (sf)
                x, y = Int(0), Int(1)
                rx_before, ry_before = sys.getrefcount (x), sys.getrefcount (y)
                ar[x, y] = 0
                rx_after, ry_after = sys.getrefcount (x), sys.getrefcount (y)
                self.assertEqual (rx_after, rx_before)
                self.assertEqual (ry_after, ry_before)
base_test.py 文件源码 项目:Projects 作者: it2school 项目源码 文件源码 阅读 32 收藏 0 点赞 0 评论 0
def test_GetView_array_struct(self):
        from pygame.bufferproxy import BufferProxy

        class Exporter(self.ExporterBase):
            def __init__(self, shape, typechar, itemsize):
                super(Exporter, self).__init__(shape, typechar, itemsize)
                self.view = BufferProxy(self.__dict__)

            def get__array_struct__(self):
                return self.view.__array_struct__
            __array_struct__ = property(get__array_struct__)
            # Should not cause PgObject_GetBuffer to fail
            __array_interface__ = property(lambda self: None)

        _shape = [2, 3, 5, 7, 11]  # Some prime numbers
        for ndim in range(1, len(_shape)):
            o = Exporter(_shape[0:ndim], 'i', 2)
            v = BufferProxy(o)
            self.assertSame(v, o)
        ndim = 2
        shape = _shape[0:ndim]
        for typechar in ('i', 'u'):
            for itemsize in (1, 2, 4, 8):
                o = Exporter(shape, typechar, itemsize)
                v = BufferProxy(o)
                self.assertSame(v, o)
        for itemsize in (4, 8):
            o = Exporter(shape, 'f', itemsize)
            v = BufferProxy(o)
            self.assertSame(v, o)

        # Check returned cobject/capsule reference count
        try:
            from sys import getrefcount
        except ImportError:
            # PyPy: no reference counting
            pass
        else:
            o = Exporter(shape, typechar, itemsize)
            self.assertEqual(getrefcount(o.__array_struct__), 1)
testutils.py 文件源码 项目:psycopg2-for-aws-lambda 作者: iwitaly 项目源码 文件源码 阅读 51 收藏 0 点赞 0 评论 0
def skip_if_no_getrefcount(f):
    @wraps(f)
    def skip_if_no_getrefcount_(self):
        if not hasattr(sys, 'getrefcount'):
            return self.skipTest('skipped, no sys.getrefcount()')
        else:
            return f(self)
    return skip_if_no_getrefcount_
test_cursor.py 文件源码 项目:psycopg2-for-aws-lambda 作者: iwitaly 项目源码 文件源码 阅读 33 收藏 0 点赞 0 评论 0
def test_mogrify_leak_on_multiple_reference(self):
        # issue #81: reference leak when a parameter value is referenced
        # more than once from a dict.
        cur = self.conn.cursor()
        foo = (lambda x: x)('foo') * 10
        import sys
        nref1 = sys.getrefcount(foo)
        cur.mogrify("select %(foo)s, %(foo)s, %(foo)s", {'foo': foo})
        nref2 = sys.getrefcount(foo)
        self.assertEqual(nref1, nref2)
testutils.py 文件源码 项目:psycopg2-for-aws-lambda 作者: iwitaly 项目源码 文件源码 阅读 28 收藏 0 点赞 0 评论 0
def skip_if_no_getrefcount(f):
    @wraps(f)
    def skip_if_no_getrefcount_(self):
        if not hasattr(sys, 'getrefcount'):
            return self.skipTest('skipped, no sys.getrefcount()')
        else:
            return f(self)
    return skip_if_no_getrefcount_
test_cursor.py 文件源码 项目:psycopg2-for-aws-lambda 作者: iwitaly 项目源码 文件源码 阅读 27 收藏 0 点赞 0 评论 0
def test_mogrify_leak_on_multiple_reference(self):
        # issue #81: reference leak when a parameter value is referenced
        # more than once from a dict.
        cur = self.conn.cursor()
        foo = (lambda x: x)('foo') * 10
        import sys
        nref1 = sys.getrefcount(foo)
        cur.mogrify("select %(foo)s, %(foo)s, %(foo)s", {'foo': foo})
        nref2 = sys.getrefcount(foo)
        self.assertEqual(nref1, nref2)
test_message.py 文件源码 项目:zanph 作者: zanph 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def await_gc(obj, rc):
    """wait for refcount on an object to drop to an expected value

    Necessary because of the zero-copy gc thread,
    which can take some time to receive its DECREF message.
    """
    for i in range(50):
        # rc + 2 because of the refs in this function
        if grc(obj) <= rc + 2:
            return
        time.sleep(0.05)
test_message.py 文件源码 项目:zanph 作者: zanph 项目源码 文件源码 阅读 32 收藏 0 点赞 0 评论 0
def test_above_30(self):
        """Message above 30 bytes are never copied by 0MQ."""
        for i in range(5, 16):  # 32, 64,..., 65536
            s = (2**i)*x
            self.assertEqual(grc(s), 2)
            m = zmq.Frame(s)
            self.assertEqual(grc(s), 4)
            del m
            await_gc(s, 2)
            self.assertEqual(grc(s), 2)
            del s
test_message.py 文件源码 项目:zanph 作者: zanph 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
def test_lifecycle1(self):
        """Run through a ref counting cycle with a copy."""
        for i in range(5, 16):  # 32, 64,..., 65536
            s = (2**i)*x
            rc = 2
            self.assertEqual(grc(s), rc)
            m = zmq.Frame(s)
            rc += 2
            self.assertEqual(grc(s), rc)
            m2 = copy.copy(m)
            rc += 1
            self.assertEqual(grc(s), rc)
            buf = m2.buffer

            rc += view_rc
            self.assertEqual(grc(s), rc)

            self.assertEqual(s, b(str(m)))
            self.assertEqual(s, bytes(m2))
            self.assertEqual(s, m.bytes)
            # self.assert_(s is str(m))
            # self.assert_(s is str(m2))
            del m2
            rc -= 1
            self.assertEqual(grc(s), rc)
            rc -= view_rc
            del buf
            self.assertEqual(grc(s), rc)
            del m
            rc -= 2
            await_gc(s, rc)
            self.assertEqual(grc(s), rc)
            self.assertEqual(rc, 2)
            del s
test_message.py 文件源码 项目:zanph 作者: zanph 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
def test_lifecycle2(self):
        """Run through a different ref counting cycle with a copy."""
        for i in range(5, 16):  # 32, 64,..., 65536
            s = (2**i)*x
            rc = 2
            self.assertEqual(grc(s), rc)
            m = zmq.Frame(s)
            rc += 2
            self.assertEqual(grc(s), rc)
            m2 = copy.copy(m)
            rc += 1
            self.assertEqual(grc(s), rc)
            buf = m.buffer
            rc += view_rc
            self.assertEqual(grc(s), rc)
            self.assertEqual(s, b(str(m)))
            self.assertEqual(s, bytes(m2))
            self.assertEqual(s, m2.bytes)
            self.assertEqual(s, m.bytes)
            # self.assert_(s is str(m))
            # self.assert_(s is str(m2))
            del buf
            self.assertEqual(grc(s), rc)
            del m
            # m.buffer is kept until m is del'd
            rc -= view_rc
            rc -= 1
            self.assertEqual(grc(s), rc)
            del m2
            rc -= 2
            await_gc(s, rc)
            self.assertEqual(grc(s), rc)
            self.assertEqual(rc, 2)
            del s
explorer.py 文件源码 项目:hostapd-mana 作者: adde88 项目源码 文件源码 阅读 29 收藏 0 点赞 0 评论 0
def view_get_refcount(self, perspective):
        return sys.getrefcount(self)
test_basics.py 文件源码 项目:hostapd-mana 作者: adde88 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def test03_leak_assignment(self) :
        a = "example of private object"
        refcount = sys.getrefcount(a)
        self.obj.set_private(a)
        self.assertEqual(refcount+1, sys.getrefcount(a))
        self.obj.set_private(None)
        self.assertEqual(refcount, sys.getrefcount(a))
test_basics.py 文件源码 项目:hostapd-mana 作者: adde88 项目源码 文件源码 阅读 31 收藏 0 点赞 0 评论 0
def test04_leak_GC(self) :
        a = "example of private object"
        refcount = sys.getrefcount(a)
        self.obj.set_private(a)
        self.obj = None
        self.assertEqual(refcount, sys.getrefcount(a))
test_python_api.py 文件源码 项目:hostapd-mana 作者: adde88 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
def test_PyString_FromString(self):
        pythonapi.PyString_FromString.restype = py_object
        pythonapi.PyString_FromString.argtypes = (c_char_p,)

        s = "abc"
        refcnt = grc(s)
        pyob = pythonapi.PyString_FromString(s)
        self.assertEqual(grc(s), refcnt)
        self.assertEqual(s, pyob)
        del pyob
        self.assertEqual(grc(s), refcnt)

    # This test is unreliable, because it is possible that code in
    # unittest changes the refcount of the '42' integer.  So, it
    # is disabled by default.
test_python_api.py 文件源码 项目:hostapd-mana 作者: adde88 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
def test_PyObj_FromPtr(self):
        s = "abc def ghi jkl"
        ref = grc(s)
        # id(python-object) is the address
        pyobj = PyObj_FromPtr(id(s))
        self.assertIs(s, pyobj)

        self.assertEqual(grc(s), ref + 1)
        del pyobj
        self.assertEqual(grc(s), ref)
test_callbacks.py 文件源码 项目:hostapd-mana 作者: adde88 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def test_pyobject(self):
        o = ()
        from sys import getrefcount as grc
        for o in (), [], object():
            initial = grc(o)
            # This call leaks a reference to 'o'...
            self.check_type(py_object, o)
            before = grc(o)
            # ...but this call doesn't leak any more.  Where is the refcount?
            self.check_type(py_object, o)
            after = grc(o)
            self.assertEqual((after, o), (before, o))
test_keeprefs.py 文件源码 项目:hostapd-mana 作者: adde88 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def test_X(self):
        class X(Structure):
            _fields_ = [("p", POINTER(c_char_p))]
        x = X()
        i = c_char_p("abc def")
        from sys import getrefcount as grc
        print "2?", grc(i)
        x.p = pointer(i)
        print "3?", grc(i)
        for i in range(320):
            c_int(99)
            x.p[0]
        print x.p[0]
##        del x
##        print "2?", grc(i)
##        del i
        import gc
        gc.collect()
        for i in range(320):
            c_int(99)
            x.p[0]
        print x.p[0]
        print x.p.contents
##        print x._objects

        x.p[0] = "spam spam"
##        print x.p[0]
        print "+" * 42
        print x._objects


问题


面经


文章

微信
公众号

扫码关注公众号