test that threads waiting for connections are handled when the pool is replaced.
(self)
| 1300 | @testing.requires.threading_with_mock |
| 1301 | @testing.requires.timing_intensive |
| 1302 | def test_waiters_handled(self): |
| 1303 | """test that threads waiting for connections are |
| 1304 | handled when the pool is replaced. |
| 1305 | |
| 1306 | """ |
| 1307 | mutex = threading.Lock() |
| 1308 | dbapi = MockDBAPI() |
| 1309 | |
| 1310 | def creator(): |
| 1311 | with mutex: |
| 1312 | return dbapi.connect() |
| 1313 | |
| 1314 | success = [] |
| 1315 | for timeout in (None, 30): |
| 1316 | for max_overflow in (0, -1, 3): |
| 1317 | p = pool.QueuePool( |
| 1318 | creator=creator, |
| 1319 | pool_size=2, |
| 1320 | timeout=timeout, |
| 1321 | max_overflow=max_overflow, |
| 1322 | ) |
| 1323 | |
| 1324 | def waiter(p, timeout, max_overflow): |
| 1325 | success_key = (timeout, max_overflow) |
| 1326 | conn = p.connect() |
| 1327 | success.append(success_key) |
| 1328 | time.sleep(0.1) |
| 1329 | conn.close() |
| 1330 | |
| 1331 | c1 = p.connect() # noqa |
| 1332 | c2 = p.connect() |
| 1333 | |
| 1334 | threads = [] |
| 1335 | for i in range(2): |
| 1336 | t = threading.Thread( |
| 1337 | target=waiter, args=(p, timeout, max_overflow) |
| 1338 | ) |
| 1339 | t.daemon = True |
| 1340 | t.start() |
| 1341 | threads.append(t) |
| 1342 | |
| 1343 | # this sleep makes sure that the |
| 1344 | # two waiter threads hit upon wait() |
| 1345 | # inside the queue, before we invalidate the other |
| 1346 | # two conns |
| 1347 | time.sleep(0.2) |
| 1348 | p._invalidate(c2) |
| 1349 | |
| 1350 | for t in threads: |
| 1351 | t.join(join_timeout) |
| 1352 | |
| 1353 | eq_(len(success), 12, "successes: %s" % success) |
| 1354 | |
| 1355 | def test_connrec_invalidated_within_checkout_no_race(self): |
| 1356 | """Test that a concurrent ConnectionRecord.invalidate() which |