def test_max_ventilation_size(self):
        """Tests that we dont surpass a max ventilation size in each pool type
        (since it relies on accurate ventilation size reporting)"""
        max_ventilation_size = 10

        for pool in [DummyPool(), ProcessPool(10), ThreadPool(10)]:
            ventilator = ConcurrentVentilator(ventilate_fn=pool.ventilate,
                                              items_to_ventilate=[{'item': i} for i in range(100)],
                                              max_ventilation_queue_size=max_ventilation_size)
            pool.start(IdentityWorker, ventilator=ventilator)

            # Give time for the thread to fill the ventilation queue
            while ventilator._ventilated_items_count - ventilator._processed_items_count < max_ventilation_size:
                time.sleep(.1)

            # After stopping the ventilator queue, we should only get 10 results
            ventilator.stop()
            for _ in range(max_ventilation_size):
                pool.get_results()

            with self.assertRaises(EmptyResultError):
                pool.get_results()

            pool.stop()
            pool.join()
Tests that we dont surpass a max ventilation size in each pool type
        (since it relies on accurate ventilation size reporting)
petastorm/workers_pool/tests/test_ventilator.py