(self)
| 1919 | |
| 1920 | @threading_helper.requires_working_threading() |
| 1921 | def test_lru_cache_threaded(self): |
| 1922 | n, m = 5, 11 |
| 1923 | def orig(x, y): |
| 1924 | return 3 * x + y |
| 1925 | f = self.module.lru_cache(maxsize=n*m)(orig) |
| 1926 | hits, misses, maxsize, currsize = f.cache_info() |
| 1927 | self.assertEqual(currsize, 0) |
| 1928 | |
| 1929 | start = threading.Event() |
| 1930 | def full(k): |
| 1931 | start.wait(10) |
| 1932 | for _ in range(m): |
| 1933 | self.assertEqual(f(k, 0), orig(k, 0)) |
| 1934 | |
| 1935 | def clear(): |
| 1936 | start.wait(10) |
| 1937 | for _ in range(2*m): |
| 1938 | f.cache_clear() |
| 1939 | |
| 1940 | orig_si = sys.getswitchinterval() |
| 1941 | support.setswitchinterval(1e-6) |
| 1942 | try: |
| 1943 | # create n threads in order to fill cache |
| 1944 | threads = [threading.Thread(target=full, args=[k]) |
| 1945 | for k in range(n)] |
| 1946 | with threading_helper.start_threads(threads): |
| 1947 | start.set() |
| 1948 | |
| 1949 | hits, misses, maxsize, currsize = f.cache_info() |
| 1950 | if self.module is py_functools: |
| 1951 | # XXX: Why can be not equal? |
| 1952 | self.assertLessEqual(misses, n) |
| 1953 | self.assertLessEqual(hits, m*n - misses) |
| 1954 | else: |
| 1955 | self.assertEqual(misses, n) |
| 1956 | self.assertEqual(hits, m*n - misses) |
| 1957 | self.assertEqual(currsize, n) |
| 1958 | |
| 1959 | # create n threads in order to fill cache and 1 to clear it |
| 1960 | threads = [threading.Thread(target=clear)] |
| 1961 | threads += [threading.Thread(target=full, args=[k]) |
| 1962 | for k in range(n)] |
| 1963 | start.clear() |
| 1964 | with threading_helper.start_threads(threads): |
| 1965 | start.set() |
| 1966 | finally: |
| 1967 | sys.setswitchinterval(orig_si) |
| 1968 | |
| 1969 | @threading_helper.requires_working_threading() |
| 1970 | def test_lru_cache_threaded2(self): |
nothing calls this directly
no test coverage detected