async-lru/tests/benchmarks/test_bench_async_lru.py at f849a05ef739ac64de43aed3761b651ce9c53b28 · aio-libs/async-lru · GitHub

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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
import asyncio
import pytest
from async_lru import alru_cache

pytestmark = pytest.mark.codspeed

# Bounded cache (LRU)
@alru_cache(maxsize=128)
async def cached_func(x):
    return x

@alru_cache(maxsize=16, ttl=0.01)
async def cached_func_ttl(x):
    return x

# Unbounded cache (no maxsize)
@alru_cache()
async def cached_func_unbounded(x):
    return x

@alru_cache(ttl=0.01)
async def cached_func_unbounded_ttl(x):
    return x

async def uncached_func(x):
    return x

# Bounded cache benchmarks
@pytest.mark.asyncio
async def test_cache_hit_benchmark(benchmark):
    await cached_func(42)
    async def hit():
        await cached_func(42)
    await benchmark.asyncio(hit)

@pytest.mark.asyncio
async def test_cache_miss_benchmark(benchmark):
    async def miss():
        await cached_func(object())
    await benchmark.asyncio(miss)

@pytest.mark.asyncio
async def test_cache_fill_eviction_benchmark(benchmark):
    keys = list(range(256))
    async def fill():
        for k in keys:
            await cached_func(k)
    await benchmark.asyncio(fill)

@pytest.mark.asyncio
async def test_cache_clear_benchmark(benchmark):
    await cached_func(1)
    async def clear():
        await cached_func.cache_clear()
    await benchmark.asyncio(clear)

@pytest.mark.asyncio
async def test_cache_ttl_expiry_benchmark(benchmark):
    await cached_func_ttl(99)
    await asyncio.sleep(0.02)
    async def ttl_expire():
        await cached_func_ttl(99)
    await benchmark.asyncio(ttl_expire)

@pytest.mark.asyncio
async def test_cache_invalidate_benchmark(benchmark):
    await cached_func(123)
    async def invalidate():
        await cached_func.cache_invalidate(123)
    await benchmark.asyncio(invalidate)

@pytest.mark.asyncio
async def test_cache_info_benchmark(benchmark):
    await cached_func(1)
    async def info():
        cached_func.cache_info()
    await benchmark.asyncio(info)

@pytest.mark.asyncio
async def test_uncached_func_benchmark(benchmark):
    async def raw():
        await uncached_func(42)
    await benchmark.asyncio(raw)

@pytest.mark.asyncio
async def test_concurrent_cache_hit_benchmark(benchmark):
    await cached_func(77)
    async def concurrent_hit():
        await asyncio.gather(*(cached_func(77) for _ in range(10)))
    await benchmark.asyncio(concurrent_hit)

# Unbounded cache benchmarks
@pytest.mark.asyncio
async def test_cache_hit_unbounded_benchmark(benchmark):
    await cached_func_unbounded(42)
    async def hit():
        await cached_func_unbounded(42)
    await benchmark.asyncio(hit)

@pytest.mark.asyncio
async def test_cache_miss_unbounded_benchmark(benchmark):
    async def miss():
        await cached_func_unbounded(object())
    await benchmark.asyncio(miss)

@pytest.mark.asyncio
async def test_cache_clear_unbounded_benchmark(benchmark):
    await cached_func_unbounded(1)
    async def clear():
        await cached_func_unbounded.cache_clear()
    await benchmark.asyncio(clear)

@pytest.mark.asyncio
async def test_cache_ttl_expiry_unbounded_benchmark(benchmark):
    await cached_func_unbounded_ttl(99)
    await asyncio.sleep(0.02)
    async def ttl_expire():
        await cached_func_unbounded_ttl(99)
    await benchmark.asyncio(ttl_expire)

@pytest.mark.asyncio
async def test_cache_invalidate_unbounded_benchmark(benchmark):
    await cached_func_unbounded(123)
    async def invalidate():
        await cached_func_unbounded.cache_invalidate(123)
    await benchmark.asyncio(invalidate)

@pytest.mark.asyncio
async def test_cache_info_unbounded_benchmark(benchmark):
    await cached_func_unbounded(1)
    async def info():
        cached_func_unbounded.cache_info()
    await benchmark.asyncio(info)

@pytest.mark.asyncio
async def test_concurrent_cache_hit_unbounded_benchmark(benchmark):
    await cached_func_unbounded(77)
    async def concurrent_hit():
        await asyncio.gather(*(cached_func_unbounded(77) for _ in range(10)))
    await benchmark.asyncio(concurrent_hit)