How to use infinite_cycle_gen method in locust

Best Python code snippet using locust

dispatch.py

Source:dispatch.py

...270        to keep the distribution during ramp-up and ramp-down. If we were to use a normal271        weighted round-robin algorithm, we'd get AAAAABAAAAAB which would make the distribution272        less accurate during ramp-up/down.273        """274        def infinite_cycle_gen(users: List[Tuple[Type[User], int]]) -> itertools.cycle:275            if not users:276                return itertools.cycle([None])277            # Normalize the weights so that the smallest weight will be equal to "target_min_weight".278            # The value "2" was experimentally determined because it gave a better distribution especially279            # when dealing with weights which are close to each others, e.g. 1.5, 2, 2.4, etc.280            target_min_weight = 2281            # 'Value' here means weight or fixed count282            normalized_values = [283                (284                    user.__name__,285                    round(target_min_weight * value / min(u[1] for u in users)),286                )287                for user, value in users288            ]289            generation_length_to_get_proper_distribution = sum(290                normalized_val[1] for normalized_val in normalized_values291            )292            gen = smooth(normalized_values)293            # Instead of calling `gen()` for each user, we cycle through a generator of fixed-length294            # `generation_length_to_get_proper_distribution`. Doing so greatly improves performance because295            # we only ever need to call `gen()` a relatively small number of times. The length of this generator296            # is chosen as the sum of the normalized weights. So, for users A, B, C of weights 2, 5, 6, the length is297            # 2 + 5 + 6 = 13 which would yield the distribution `CBACBCBCBCABC` that gets repeated over and over298            # until the target user count is reached.299            return itertools.cycle(gen() for _ in range(generation_length_to_get_proper_distribution))300        fixed_users = {u.__name__: u for u in self._user_classes if u.fixed_count}301        cycle_fixed_gen = infinite_cycle_gen([(u, u.fixed_count) for u in fixed_users.values()])302        cycle_weighted_gen = infinite_cycle_gen([(u, u.weight) for u in self._user_classes if not u.fixed_count])303        # Spawn users304        while True:305            if self._try_dispatch_fixed:306                self._try_dispatch_fixed = False307                current_fixed_users_count = {u: self._get_user_current_count(u) for u in fixed_users}308                spawned_classes: Set[str] = set()309                while len(spawned_classes) != len(fixed_users):310                    user_name: Optional[str] = next(cycle_fixed_gen)311                    if not user_name:312                        break313                    if current_fixed_users_count[user_name] < fixed_users[user_name].fixed_count:314                        current_fixed_users_count[user_name] += 1315                        if current_fixed_users_count[user_name] == fixed_users[user_name].fixed_count:316                            spawned_classes.add(user_name)...

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.