How to use _distribute_users method in locust

Best Python code snippet using locust

dispatch.py

Source:dispatch.py

...148        When a rebalance is required because of added and/or removed workers, we compute the desired state as if149        we started from 0 user. So, if we were currently running 500 users, then the `_distribute_users` will150        perform a fake ramp-up without any waiting and return the final distribution.151        """152        users_on_workers, user_gen, worker_gen, active_users = self._distribute_users(self._current_user_count)153        self._users_on_workers = users_on_workers154        self._active_users = active_users155        # It's important to reset the generators by using the ones from `_distribute_users`156        # so that the next iterations are smooth and continuous.157        self._user_generator = user_gen158        self._worker_node_generator = worker_gen159        self._rebalance = True160    @contextlib.contextmanager161    def _wait_between_dispatch_iteration_context(self) -> None:162        t0_rel = time.perf_counter()163        # We don't use `try: ... finally: ...` because we don't want to sleep164        # if there's an exception within the context.165        yield166        delta = time.perf_counter() - t0_rel167        self._dispatch_iteration_durations.append(delta)168        # print("Dispatch cycle took {:.3f}ms".format(delta * 1000))169        if self._current_user_count == self._target_user_count:170            # No sleep when this is the last dispatch iteration171            return172        sleep_duration = max(0.0, self._wait_between_dispatch - delta)173        gevent.sleep(sleep_duration)174    def _add_users_on_workers(self) -> Dict[str, Dict[str, int]]:175        """Add users on the workers until the target number of users is reached for the current dispatch iteration176        :return: The users that we want to run on the workers177        """178        current_user_count_target = min(179            self._current_user_count + self._user_count_per_dispatch_iteration, self._target_user_count180        )181        for user in self._user_generator:182            worker_node = next(self._worker_node_generator)183            self._users_on_workers[worker_node.id][user] += 1184            self._current_user_count += 1185            self._active_users.append((worker_node, user))186            if self._current_user_count >= current_user_count_target:187                return self._users_on_workers188    def _remove_users_from_workers(self) -> Dict[str, Dict[str, int]]:189        """Remove users from the workers until the target number of users is reached for the current dispatch iteration190        :return: The users that we want to run on the workers191        """192        current_user_count_target = max(193            self._current_user_count - self._user_count_per_dispatch_iteration, self._target_user_count194        )195        while True:196            try:197                worker_node, user = self._active_users.pop()198            except IndexError:199                return self._users_on_workers200            self._users_on_workers[worker_node.id][user] -= 1201            self._current_user_count -= 1202            if self._current_user_count == 0 or self._current_user_count <= current_user_count_target:203                return self._users_on_workers204    def _distribute_users(205        self, target_user_count: int206    ) -> Tuple[dict, Generator[str, None, None], typing.Iterator["WorkerNode"], List[Tuple["WorkerNode", str]]]:207        """208        This function might take some time to complete if the `target_user_count` is a big number. A big number209        is typically > 50 000. However, this function is only called if a worker is added or removed while a test210        is running. Such a situation should be quite rare.211        """212        user_gen = self._user_gen()213        worker_gen = itertools.cycle(self._worker_nodes)214        users_on_workers = {215            worker_node.id: {user_class.__name__: 0 for user_class in self._user_classes}216            for worker_node in self._worker_nodes217        }218        active_users = []...

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.