File: test_lru.py

package info (click to toggle)
mako 1.0.6%2Bds1-2
  • links: PTS, VCS
  • area: main
  • in suites: stretch
  • size: 1,556 kB
  • ctags: 1,495
  • sloc: python: 11,807; makefile: 167
file content (114 lines) | stat: -rw-r--r-- 3,227 bytes parent folder | download | duplicates (3)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
from mako.util import LRUCache
import string
import unittest
import time
import random

from mako.compat import thread

class item:
    def __init__(self, id):
        self.id = id

    def __str__(self):
        return "item id %d" % self.id

class LRUTest(unittest.TestCase):


    def testlru(self):
        l = LRUCache(10, threshold=.2)

        for id in range(1,20):
            l[id] = item(id)

        # first couple of items should be gone
        assert 1 not in l
        assert 2 not in l

        # next batch over the threshold of 10 should be present
        for id in range(11,20):
            assert id in l

        l[12]
        l[15]
        l[23] = item(23)
        l[24] = item(24)
        l[25] = item(25)
        l[26] = item(26)
        l[27] = item(27)

        assert 11 not in l
        assert 13 not in l

        for id in (25, 24, 23, 14, 12, 19, 18, 17, 16, 15):
            assert id in l

    def _disabled_test_threaded(self):
        size = 100
        threshold = .5
        all_elems = 2000
        hot_zone = list(range(30,40))
        cache = LRUCache(size, threshold)

        # element to store
        class Element(object):
            def __init__(self, id):
                self.id = id
                self.regets = 0

        # return an element.  we will favor ids in the relatively small
        # "hot zone" 25% of  the time.
        def get_elem():
            if random.randint(1,4) == 1:
                return hot_zone[random.randint(0, len(hot_zone) - 1)]
            else:
                return random.randint(1, all_elems)

        total = [0]
        # request thread.
        def request_elem():
            while True:
                total[0] += 1
                id = get_elem()
                try:
                    elem = cache[id]
                    elem.regets += 1
                except KeyError:
                    e = Element(id)
                    cache[id] = e

                time.sleep(random.random() / 1000)

        for x in range(0,20):
            _thread.start_new_thread(request_elem, ())

        # assert size doesn't grow unbounded, doesnt shrink well below size
        for x in range(0,5):
            time.sleep(1)
            print("size:", len(cache))
            assert len(cache) < size + size * threshold * 2
            assert len(cache) > size - (size * .1)

        # computs the average number of times a range of elements were "reused",
        # i.e. without being removed from the cache.
        def average_regets_in_range(start, end):
            elem = [e for e in list(cache.values()) if e.id >= start and e.id <= end]
            if len(elem) == 0:
                return 0
            avg = sum([e.regets for e in elem]) / len(elem)
            return avg

        hotzone_avg = average_regets_in_range(30, 40)
        control_avg = average_regets_in_range(450,760)
        total_avg = average_regets_in_range(0, 2000)

        # hotzone should be way above the others
        print("total fetches", total[0], "hotzone", \
                                hotzone_avg, "control", \
                                control_avg, "total", total_avg)

        assert hotzone_avg > total_avg * 5 > control_avg * 5