KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use has_idle_threads method in localstack

Best Python code snippet using localstack_python

async_utils.py

Source:async_utils.py

...16        self.core_size = core_size or self.DEFAULT_CORE_POOL_SIZE17        super(AdaptiveThreadPool, self).__init__(max_workers=self.core_size)18    def submit(self, fn, *args, **kwargs):19        # if idle threads are available, don't spin new threads20        if self.has_idle_threads():21            return super(AdaptiveThreadPool, self).submit(fn, *args, **kwargs)22        def _run(*tmpargs):23            return fn(*args, **kwargs)24        thread = start_worker_thread(_run)25        return thread.result_future26    def has_idle_threads(self):27        if hasattr(self, "_idle_semaphore"):28            return self._idle_semaphore.acquire(timeout=0)29        num_threads = len(self._threads)30        return num_threads < self._max_workers31# Thread pool executor for running sync functions in async context.32# Note: For certain APIs like DynamoDB, we need 3x threads for each parallel request,33# as during request processing the API calls out to the DynamoDB API again (recursively).34# (TODO: This could potentially be improved if we move entirely to asyncio functions.)35THREAD_POOL = AdaptiveThreadPool()36TMP_THREADS.append(THREAD_POOL)37class AsyncThread(FuncThread):38    def __init__(self, async_func_gen=None, loop=None):39        """Pass a function that receives an event loop instance and a shutdown event,40        and returns an async function."""...

asyncio.py

Source:asyncio.py

...16        self.core_size = core_size or self.DEFAULT_CORE_POOL_SIZE17        super(AdaptiveThreadPool, self).__init__(max_workers=self.core_size)18    def submit(self, fn, *args, **kwargs):19        # if idle threads are available, don't spin new threads20        if self.has_idle_threads():21            return super(AdaptiveThreadPool, self).submit(fn, *args, **kwargs)22        def _run(*tmpargs):23            return fn(*args, **kwargs)24        thread = start_worker_thread(_run)25        return thread.result_future26    def has_idle_threads(self):27        if hasattr(self, "_idle_semaphore"):28            return self._idle_semaphore.acquire(timeout=0)29        num_threads = len(self._threads)30        return num_threads < self._max_workers31# Thread pool executor for running sync functions in async context.32# Note: For certain APIs like DynamoDB, we need 3x threads for each parallel request,33# as during request processing the API calls out to the DynamoDB API again (recursively).34# (TODO: This could potentially be improved if we move entirely to asyncio functions.)35THREAD_POOL = AdaptiveThreadPool()36TMP_THREADS.append(THREAD_POOL)37class AsyncThread(FuncThread):38    def __init__(self, async_func_gen=None, loop=None):39        """Pass a function that receives an event loop instance and a shutdown event,40        and returns an async function."""...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.