def _worker_process(queue, iteration_gen, timeout, concurrency, times,
                    context, cls, method_name, args, aborted, info):

    pool = collections.deque()
    alive_threads_in_pool = 0
    finished_threads_in_pool = 0

    runner._log_worker_info(times=times, concurrency=concurrency,
                            timeout=timeout, cls=cls, method_name=method_name,
                            args=args)

    if timeout:
        timeout_queue = Queue.Queue()
        collector_thr_by_timeout = threading.Thread(
            target=utils.timeout_thread,
            args=(timeout_queue, )
        )
        collector_thr_by_timeout.start()

    iteration = next(iteration_gen)
    while iteration < times and not aborted.is_set():
        scenario_context = runner._get_scenario_context(iteration, context)
        worker_args = (queue, cls, method_name, scenario_context, args)

        thread = threading.Thread(target=runner._worker_thread,
                                  args=worker_args)

        thread.start()
        if timeout:
            timeout_queue.put((thread.ident, time.time() + timeout))
        pool.append((thread, time.time()))
        alive_threads_in_pool += 1

        while alive_threads_in_pool == concurrency:
            prev_finished_threads_in_pool = finished_threads_in_pool
            finished_threads_in_pool = 0
            for t in pool:
                if not t[0].isAlive():
                    finished_threads_in_pool += 1

            alive_threads_in_pool -= finished_threads_in_pool
            alive_threads_in_pool += prev_finished_threads_in_pool

            if alive_threads_in_pool < concurrency:
                # NOTE(boris-42): cleanup pool array. This is required because
                # in other case array length will be equal to times which
                # is unlimited big
                while pool and not pool[0][0].isAlive():
                    pool.popleft()[0].join()
                    finished_threads_in_pool -= 1
                break

            # we should wait to not create big noise with these checks
            time.sleep(0.001)
        iteration = next(iteration_gen)

    # Wait until all threads are done
    while pool:
        pool.popleft()[0].join()

    if timeout:
        timeout_queue.put((None, None,))
        collector_thr_by_timeout.join()
Start the scenario within threads.
dataset/ETHPy150Open openstack/rally/rally/plugins/common/runners/constant.py/_worker_process