| 246 | self.env.assertTrue(cached_result.cached_execution) |
| 247 | |
| 248 | def test_14_cache_eviction(self): |
| 249 | # this tests spawns a new graph env` with a query-cache with just |
| 250 | # a single slot, then multiple clients are issuing a similar query |
| 251 | # only with a small variation to cause a cache miss which implies |
| 252 | # cache eviction of the only solt |
| 253 | # we want to make sure the execution of a recently evicted query |
| 254 | # runs to completion successfuly |
| 255 | |
| 256 | # skip if 'to_thread' is missing or if test under valgrind |
| 257 | if VALGRIND or "to_thread" not in dir(asyncio): |
| 258 | self.env.skip() |
| 259 | |
| 260 | # stop previous env |
| 261 | self.env.flush() |
| 262 | self.env.stop() |
| 263 | |
| 264 | self.env = Env(decodeResponses=True, moduleArgs='THREAD_COUNT 8 CACHE_SIZE 1') |
| 265 | |
| 266 | # eviction |
| 267 | con = self.env.getConnection() |
| 268 | graph = Graph(con, 'cache_eviction') |
| 269 | |
| 270 | # populate graph |
| 271 | graph.query("UNWIND range(0, 10000) as x CREATE ({v:'/'})") |
| 272 | |
| 273 | # _run_query is expected to be issued by multiple threads |
| 274 | def _run_query(i): |
| 275 | #random param name |
| 276 | param_name = 'p_' + str(i) |
| 277 | q = f"MATCH (n) WHERE n.v = ${param_name} RETURN count(n)" |
| 278 | params = {param_name : '/'} |
| 279 | g = Graph(self.env.getConnection(), 'cache_eviction') |
| 280 | count = g.query(q, params).result_set[0][0] |
| 281 | self.env.assertEqual(count, 10001) |
| 282 | |
| 283 | tasks = [] |
| 284 | loop = asyncio.get_event_loop() |
| 285 | for i in range(1, 50): |
| 286 | tasks.append(loop.create_task(asyncio.to_thread(_run_query, i))) |
| 287 | |
| 288 | loop.run_until_complete(asyncio.wait(tasks)) |
| 289 | |