Method test_session_gc

test/test_load_balancer.py:132–159

(self)

    @client_context.require_transactions
    def test_session_gc(self):
        client = self.rs_client()
        pool = get_pool(client)
        session = client.start_session()
        session.start_transaction()
        client.test_session_gc.test.find_one({}, session=session)
        # Cleanup the transaction left open on the server
        self.addCleanup(self.client.admin.command, "killSessions", [session.session_id])
        if client_context.load_balancer:
            self.assertEqual(pool.active_sockets, 1)  # Pinned.

        task = PoolLocker(pool)
        task.start()
        self.assertTrue(task.wait(task.locked, 5), "timed out")
        # Garbage collect the session while the pool is locked to ensure we
        # don't deadlock.
        del session
        # On PyPy it can take a few rounds to collect the session.
        for _ in range(3):
            gc.collect()
        task.unlock.set()
        task.join(5)
        self.assertFalse(task.is_alive())
        self.assertIsNone(task.exc)

        wait_until(lambda: pool.active_sockets == 0, "return socket")
        # Run another operation to ensure the socket still works.
        client[self.db.name].test.delete_many({})


class PoolLocker(ExceptionCatchingTask):