#!/usr/bin/env python # -*- coding: utf-8-with-signature-unix; fill-column: 77 -*- # -*- indent-tabs-mode: nil -*- # This file is part of pyutil; see README.rst for licensing terms. from __future__ import print_function import random, unittest from pyutil.humanreadable import hr from pyutil import memutil from pyutil import odict class Bencher: def __init__(self, klass, MAXREPS=2**8, MAXTIME=5): print(klass) self.klass = klass self.MAXREPS = MAXREPS self.MAXTIME = MAXTIME self.d = {} self.lrun = None def _generic_benchmarking_init(self, n): self.d.clear() self.lrun = self.klass() for i in range(n): self.d[i] = i self.lrun[n+i] = n+i def _benchmark_init(self, n): d2 = self.klass(initialdata=self.d) assert len(d2) == len(self.d) return True def _benchmark_update(self, n): d2 = self.klass() assert len(d2) == 0 d2.update(self.d) assert len(d2) == len(self.d) return True def _benchmark_insert(self, n): d2 = self.klass() assert len(d2) == 0 for k, v, in self.d.iteritems(): d2[k] = v assert len(d2) == len(self.d) return True def _benchmark_init_and_popitem(self, n): d2 = self.klass(initialdata=self.d) assert len(d2) == len(self.d) for i in range(len(d2), 0, -1): assert len(d2) == i d2.popitem() return True def _benchmark_init_and_has_key_and_del(self, n): d2 = self.klass(initialdata=self.d) assert len(d2) == len(self.d) for k in self.d.iterkeys(): if d2.has_key(k): del d2[k] return True def _benchmark_init_and_remove(self, n): d2 = self.klass(initialdata=self.d) assert len(d2) == len(self.d) for k in self.d.iterkeys(): d2.remove(k, strictkey=False) return True def bench(self, BSIZES=(128, 250, 2048, 5000, 2**13, 2**20,)): from pyutil import benchutil funcs = ("_benchmark_insert", "_benchmark_init_and_has_key_and_del", "_benchmark_init_and_remove", "_benchmark_init_and_popitem", "_benchmark_update", "_benchmark_init",) max = 0 for func in funcs: if len(func) > max: max = len(func) for func in funcs: print(func + " " * (max + 1 - len(func))) for BSIZE in BSIZES: f = getattr(self, func) benchutil.rep_bench(f, BSIZE, self._generic_benchmarking_init, MAXREPS=self.MAXREPS, MAXTIME=self.MAXTIME) def quick_bench(): Bencher(odict.LRUCache, MAXTIME=2).bench(BSIZES=(2**7, 2**12, 2**14, 2**15, 2**16,)) Bencher(odict.LinkedListLRUCache, MAXTIME=2).bench(BSIZES=(2**7, 2**12, 2**14, 2**15,)) Bencher(odict.SmallLRUCache, MAXTIME=2).bench(BSIZES=(2**7, 2**12, 2**14, 2**15,)) def slow_bench(): Bencher(odict.LRUCache, MAXTIME=5).bench(BSIZES=[2**x for x in range(7, 21)]) Bencher(odict.LinkedListLRUCache, MAXTIME=5).bench(BSIZES=[2**x for x in range(7, 21)]) Bencher(odict.SmallLRUCache, MAXTIME=5).bench(BSIZES=[2**x for x in range(7, 17)]) MUCHADDINGSIZE=2**4 # The following parameters are for testing for memory leakage. MIN_SLOPE = 512.0 # If it leaks less than 512.0 bytes per iteration, then it's probably just some kind of noise from the interpreter or something... SAMPLES = 2**5 # MIN_SLOPE is high because samples is low, which is because taking a statistically useful numbers of samples takes too long. # For a *good* test, turn samples up as high as you can stand (maybe 2**10) and set MIN_SLOPE to about 1.0. # For a *really* good test, add a variance measure to memutil.measure_mem_leakage(), and only consider it to be leaking if the slope is > 0.1 *and* is a "pretty good" fit for the data. # MIN_SLOPE = 1.0 # SAMPLES = 2**10 class Testy(unittest.TestCase): def _test_empty_lookup(self, d) : self.assertTrue(d.get('spam') is None) def _test_key_error(self, C) : d = C() try: d['spam'] self.fail(d) except KeyError : pass def _test_insert_and_get_and_items(self, d) : d.insert("spam", "eggs") d["spam2"] = "eggs2" self.assertTrue(d.get("spam") == "eggs", str(d)) self.assertTrue(d.get("spam2") == "eggs2") self.assertTrue(d["spam"] == "eggs") self.assertTrue(d["spam2"] == "eggs2") self.assertEqual(list(d.items()), [("spam", "eggs"), ("spam2", "eggs2")], d) def _test_move_to_most_recent(self, d) : d.insert("spam", "eggs") d["spam2"] = "eggs2" self.assertTrue(d.get("spam") == "eggs", str(d)) self.assertTrue(d.get("spam2") == "eggs2") self.assertTrue(d["spam"] == "eggs") self.assertTrue(d["spam2"] == "eggs2") self.assertEqual(d.items(), [("spam", "eggs"), ("spam2", "eggs2")]) d.move_to_most_recent("spam") self.assertEqual(list(d.items()), [("spam2", "eggs2"), ("spam", "eggs")]) def _test_insert_and_remove(self, d): d.insert('spam', "eggs") self.assertTrue(d.has_key('spam')) self.assertTrue(d.get('spam') == "eggs") self.assertTrue(d['spam'] == "eggs") self.assertEqual(list(d.items()), [("spam", "eggs")]) x = d.remove('spam') self.assertTrue(x == "eggs", "x: %r" % x) self.assertTrue(not d.has_key('spam')) self.assertEqual(list(d.items()), []) d['spam'] = "eggsy" self.assertTrue(d.has_key('spam')) self.assertTrue(d.get('spam') == "eggsy") self.assertTrue(d['spam'] == "eggsy") self.assertEqual(list(d.items()), [("spam", "eggsy")]) del d['spam'] self.assertTrue(not d.has_key('spam')) self.assertEqual(list(d.items()), []) def _test_setdefault(self, d): d.setdefault('spam', "eggs") self.assertTrue(d.has_key('spam')) self.assertTrue(d.get('spam') == "eggs") self.assertTrue(d['spam'] == "eggs") self.assertEqual(list(d.items()), [("spam", "eggs")]) x = d.remove('spam') self.assertTrue(x == "eggs", "x: %r" % x) self.assertTrue(not d.has_key('spam')) self.assertEqual(list(d.items()), []) def _test_extracted_bound_method(self, d): insmeth = d.insert insmeth('spammy', "eggsy") self.assertTrue(d.get('spammy') == "eggsy") def _test_extracted_unbound_method(self, d): insumeth = d.__class__.insert insumeth(d, 'spammy', "eggsy") self.assertTrue(d.get('spammy') == "eggsy") def _test_unbound_method(self, C, d): umeth = C.insert umeth(d, 'spammy', "eggsy") self.assertTrue(d.get('spammy') == "eggsy") def _test_clear(self, d): d[11] = 11 d._assert_invariants() self.assertTrue(len(d) == 1) d.clear() d._assert_invariants() self.assertTrue(len(d) == 0) self.assertEqual(list(d.items()), []) def _test_update_from_dict(self, d): self.assertTrue(d._assert_invariants()) d['b'] = 99 self.assertTrue(d._assert_invariants()) d2={ 'a': 0, 'b': 1, 'c': 2,} d.update(d2) self.assertTrue(d._assert_invariants()) self.assertTrue(d.get('a') == 0, "d.get('a'): %s" % d.get('a')) self.assertTrue(d._assert_invariants()) self.assertTrue(d.get('b') == 1, "d.get('b'): %s" % d.get('b')) self.assertTrue(d._assert_invariants()) self.assertTrue(d.get('c') == 2) self.assertTrue(d._assert_invariants()) def _test_update_from_odict(self, d): self.assertTrue(d._assert_invariants()) d['b'] = 99 self.assertTrue(d._assert_invariants()) d2 = odict.OrderedDict() d2['a'] = 0 d2['b'] = 1 d2['c'] = 2 d.update(d2) self.assertTrue(d._assert_invariants()) self.assertTrue(d.get('a') == 0, "d.get('a'): %s" % d.get('a')) self.assertTrue(d._assert_invariants()) self.assertTrue(d.get('b') == 1, "d.get('b'): %s" % d.get('b')) self.assertTrue(d._assert_invariants()) self.assertTrue(d.get('c') == 2) self.assertTrue(d._assert_invariants()) self.assertEqual(list(d.items()), [("b", 1), ("a", 0), ("c", 2)]) def _test_popitem(self, C): c = C({"a": 1}) res = c.popitem() self.assertEqual(res, ("a", 1,)) c["a"] = 1 c["b"] = 2 res = c.popitem() self.assertEqual(res, ("b", 2,)) def _test_pop(self, C): c = C({"a": 1}) res = c.pop() self.assertEqual(res, "a") c["a"] = 1 c["b"] = 2 res = c.pop() self.assertEqual(res, "b") def _test_iterate_items(self, C): c = C({"a": 1}) c["b"] = 2 i = c.iteritems() x = i.next() self.assertEqual(x, ("a", 1,)) x = i.next() self.assertEqual(x, ("b", 2,)) try: i.next() self.fail() # Should have gotten StopIteration exception except StopIteration: pass def _test_iterate_keys(self, C): c = C({"a": 1}) c["b"] = 2 i = c.iterkeys() x = i.next() self.assertEqual(x, "a") x = i.next() self.assertEqual(x, "b") try: i.next() self.fail() # Should have gotten StopIteration exception except StopIteration: pass def _test_iterate_values(self, C): c = C({"a": 1}) c["b"] = 2 i = c.itervalues() x = i.next() self.assertTrue(x == 1) x = i.next() self.assertTrue(x == 2) try: i.next() self.fail() # Should have gotten StopIteration exception except StopIteration: pass def _test_much_adding_some_removing(self, C): c = C() for i in range(MUCHADDINGSIZE): c[i] = i if (i % 4) == 0: k = random.choice(c.keys()) del c[k] for i in range(MUCHADDINGSIZE): c[i] = i self.assertEqual(len(c), MUCHADDINGSIZE) def _test_1(self, C): c = C() c[11] = 11 c._assert_invariants() c[11] = 11 c._assert_invariants() c[11] = 1001 c._assert_invariants() c[11] = 11 c._assert_invariants() c[11] = 1001 c._assert_invariants() c[11] = 1001 c._assert_invariants() c[11] = 1001 c._assert_invariants() def _test_2(self, C): c = C() c[11] = 11 c._assert_invariants() del c[11] c._assert_invariants() c[11] = 11 c._assert_invariants() c[11] = 11 c._assert_invariants() def _test_3(self, C): c = C() c[11] = 11 c._assert_invariants() c[11] = 12 c._assert_invariants() c[11] = 13 c._assert_invariants() del c[11] c._assert_invariants() c[11] = 14 c._assert_invariants() c[11] = 15 c._assert_invariants() c[11] = 16 c._assert_invariants() def _test_has_key(self, C): c = C() c._assert_invariants() for i in range(11): c._assert_invariants() c[i] = i c._assert_invariants() del c[0] self.assertTrue(len(c) == 10) self.assertTrue(10 in c.values()) self.assertTrue(0 not in c.values()) c.has_key(1) # this touches `1' but does not make it fresher so that it will get popped next time we pop. c[1] = 1 # this touches `1' but does not make it fresher so that it will get popped. c._assert_invariants() x = c.pop() self.assertEqual(x, 10) c[99] = 99 c._assert_invariants() self.assertTrue(len(c) == 10) self.assertTrue(1 in c.values(), "C: %s, c.values(): %s" % (hr(C), hr(c.values(),),)) self.assertTrue(2 in c.values(), "C: %s, c.values(): %s" % (hr(C), hr(c.values(),),)) self.assertFalse(10 in c.values(), "C: %s, c.values(): %s" % (hr(C), hr(c.values(),),)) self.assertTrue(99 in c.values()) def _test_em(self): for klass in (odict.OrderedDict,): for testfunc in (self._test_empty_lookup, self._test_insert_and_get_and_items, self._test_insert_and_remove, self._test_extracted_bound_method, self._test_extracted_unbound_method, self._test_clear, self._test_update_from_dict, self._test_update_from_odict, self._test_setdefault,): testfunc(klass()) for testfunc in (self._test_pop, self._test_popitem, self._test_iterate_items, self._test_iterate_keys, self._test_iterate_values, self._test_key_error, ): testfunc(klass) self._test_unbound_method(klass, klass()) for klass in (odict.OrderedDict,): for testfunc in (self._test_1, self._test_2, self._test_3, self._test_has_key,): testfunc(klass) def test_em(self): self._test_em() def _mem_test_much_adding_some_removing(self): for klass in (odict.LRUCache, odict.SmallLRUCache,): return self._test_much_adding_some_removing(klass) def test_mem_leakage(self): try: self._test_mem_leakage() except memutil.NotSupportedException: print("Skipping memory leak test since measurement of current mem usage isn't implemented on this platform.") pass del test_mem_leakage # This test takes too long. def _test_mem_leakage(self): # measure one and throw it away, in order to reach a "steady state" in terms of initialization of memory state. memutil.measure_mem_leakage(self.test_em, max(2**3, SAMPLES/2**3), iterspersample=2**0) slope = memutil.measure_mem_leakage(self.test_em, max(2**3, SAMPLES/2**3), iterspersample=2**0) self.assertTrue(slope <= MIN_SLOPE, "%s leaks memory at a rate of approximately %s system bytes per invocation" % (self.test_em, "%0.3f" % slope,)) def test_mem_leakage_much_adding_some_removing(self): try: self._test_mem_leakage_much_adding_some_removing() except memutil.NotSupportedException: print("Skipping memory leak test since measurement of current mem usage isn't implemented on this platform.") pass del test_mem_leakage_much_adding_some_removing # This test takes too long. def _test_mem_leakage_much_adding_some_removing(self): # measure one and throw it away, in order to reach a "steady state" in terms of initialization of memory state. memutil.measure_mem_leakage(self._mem_test_much_adding_some_removing, SAMPLES, iterspersample=2**0) slope = memutil.measure_mem_leakage(self._mem_test_much_adding_some_removing, SAMPLES, iterspersample=2**0) self.assertTrue(slope <= MIN_SLOPE, "%s leaks memory at a rate of approximately %s system bytes per invocation" % (self._mem_test_much_adding_some_removing, "%0.3f" % slope,)) def test_obj_leakage(self): self._test_obj_leakage() del test_obj_leakage # This test takes too long. def _test_obj_leakage(self): # measure one and throw it away, in order to reach a "steady state" in terms of initialization of objects state. memutil.measure_obj_leakage(self.test_em, max(2**3, SAMPLES/2**3), iterspersample=2**0) slope = memutil.measure_obj_leakage(self.test_em, max(2**3, SAMPLES/2**3), iterspersample=2**0) self.assertTrue(slope <= MIN_SLOPE, "%s leaks objects at a rate of approximately %s system bytes per invocation" % (self.test_em, "%0.3f" % slope,)) def test_obj_leakage_much_adding_some_removing(self): self._test_obj_leakage_much_adding_some_removing() del test_obj_leakage_much_adding_some_removing # This test takes too long. def _test_obj_leakage_much_adding_some_removing(self): # measure one and throw it away, in order to reach a "steady state" in terms of initialization of objects state. memutil.measure_obj_leakage(self._mem_test_much_adding_some_removing, SAMPLES, iterspersample=2**0) slope = memutil.measure_obj_leakage(self._mem_test_much_adding_some_removing, SAMPLES, iterspersample=2**0) self.assertTrue(slope <= MIN_SLOPE, "%s leaks objects at a rate of approximately %s system bytes per invocation" % (self._mem_test_much_adding_some_removing, "%0.3f" % slope,))