KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use _set_tasks_max method in lisa

Best Python code snippet using lisa_python
ntttcp.py
Source:ntttcp.py
...132            sys_list = self.sys_list_udp133        for sys in sys_list:134            for variable, value in sys.items():135                sysctl.write(variable, value)136        self._set_tasks_max()137        firewall = self.node.tools[Firewall]138        firewall.stop()139    def restore_system(self, udp_mode: bool = False) -> None:140        original_settings = self._original_settings_tcp141        if udp_mode:142            original_settings = self._original_settings_udp143        sysctl = self.node.tools[Sysctl]144        for variable_list in original_settings:145            # restore back to the original value after testing146            for variable, value in variable_list.items():147                sysctl.write(variable, value)148    def help(self) -> ExecutableResult:149        return self.run("-h")150    def get_throughput(self, stdout: str) -> str:151        throughput = self.throughput_pattern.findall(stdout)152        if throughput:153            result: str = throughput[0]154        else:155            result = "cannot find throughput"156        return result157    def run_as_server_async(158        self,159        nic_name: str,160        run_time_seconds: int = 10,161        ports_count: int = 64,162        buffer_size: int = 64,163        cool_down_time_seconds: int = 1,164        warm_up_time_seconds: int = 1,165        use_epoll: bool = True,166        server_ip: str = "",167        dev_differentiator: str = "Hypervisor callback interrupts",168        run_as_daemon: bool = False,169        udp_mode: bool = False,170    ) -> Process:171        cmd = ""172        if server_ip:173            cmd += f" -r{server_ip} "174        cmd += (175            f" -P {ports_count} -t {run_time_seconds} -W {warm_up_time_seconds} "176            f"-C {cool_down_time_seconds} -b {buffer_size}k "177            f"--show-nic-packets {nic_name} "178        )179        if udp_mode:180            cmd += " -u "181        if use_epoll:182            cmd += " -e "183        if dev_differentiator:184            cmd += f" --show-dev-interrupts {dev_differentiator} "185        if run_as_daemon:186            cmd += " -D "187        process = self.node.execute_async(188            f"ulimit -n 204800 && {self.command} {cmd}", shell=True, sudo=True189        )190        # NTTTCP for Linux 1.4.0191        # ---------------------------------------------------------192        # 01:16:35 INFO: no role specified. use receiver role193        # 01:16:35 INFO: 65 threads created194        # above output means ntttcp server is ready195        process.wait_output("threads created")196        return process197    def run_as_server(198        self,199        nic_name: str,200        run_time_seconds: int = 10,201        ports_count: int = 64,202        buffer_size: int = 64,203        cool_down_time_seconds: int = 1,204        warm_up_time_seconds: int = 1,205        use_epoll: bool = True,206        server_ip: str = "",207        dev_differentiator: str = "Hypervisor callback interrupts",208        run_as_daemon: bool = False,209        udp_mode: bool = False,210    ) -> ExecutableResult:211        # -rserver_ip: run as a receiver with specified server ip address212        # -P: Number of ports listening on receiver side [default: 16] [max: 512]213        # -t: Time of test duration in seconds [default: 60]214        # -e: [receiver only] use epoll() instead of select()215        # -u: UDP mode     [default: TCP]216        # -W: Warm-up time in seconds          [default: 0]217        # -C: Cool-down time in seconds        [default: 0]218        # -b: <buffer size in n[KMG] Bytes>    [default: 65536 (receiver); 131072219        # (sender)]220        # --show-nic-packets <network interface name>: Show number of packets221        # transferred (tx and rx) through this network interface222        # --show-dev-interrupts <device differentiator>: Show number of interrupts for223        # the devices specified by the differentiator224        # Examples for differentiator: Hyper-V PCIe MSI, mlx4, Hypervisor callback225        # interrupts226        process = self.run_as_server_async(227            nic_name,228            run_time_seconds,229            ports_count,230            buffer_size,231            cool_down_time_seconds,232            warm_up_time_seconds,233            use_epoll,234            server_ip,235            dev_differentiator,236            run_as_daemon,237            udp_mode,238        )239        return self.wait_server_result(process)240    def wait_server_result(self, process: Process) -> ExecutableResult:241        return process.wait_result(242            expected_exit_code=0,243            expected_exit_code_failure_message="fail to launch ntttcp server",244        )245    def run_as_client(246        self,247        nic_name: str,248        server_ip: str,249        threads_count: int,250        run_time_seconds: int = 10,251        ports_count: int = 64,252        buffer_size: int = 64,253        cool_down_time_seconds: int = 1,254        warm_up_time_seconds: int = 1,255        dev_differentiator: str = "Hypervisor callback interrupts",256        run_as_daemon: bool = False,257        udp_mode: bool = False,258    ) -> ExecutableResult:259        # -sserver_ip: run as a sender with server ip address260        # -P: Number of ports listening on receiver side [default: 16] [max: 512]261        # -n: [sender only] number of threads per each receiver port     [default: 4]262        # [max: 25600]263        # -t: Time of test duration in seconds [default: 60]264        # -e: [receiver only] use epoll() instead of select()265        # -u: UDP mode     [default: TCP]266        # -W: Warm-up time in seconds          [default: 0]267        # -C: Cool-down time in seconds        [default: 0]268        # -b: <buffer size in n[KMG] Bytes>    [default: 65536 (receiver); 131072269        # (sender)]270        # --show-nic-packets <network interface name>: Show number of packets271        # transferred (tx and rx) through this network interface272        # --show-dev-interrupts <device differentiator>: Show number of interrupts for273        # the devices specified by the differentiator274        # Examples for differentiator: Hyper-V PCIe MSI, mlx4, Hypervisor callback275        # interrupts276        cmd = (277            f" -s{server_ip} -P {ports_count} -n {threads_count} -t {run_time_seconds} "278            f"-W {warm_up_time_seconds} -C {cool_down_time_seconds} -b {buffer_size}k "279            f"--show-nic-packets {nic_name} "280        )281        if udp_mode:282            cmd += " -u "283        if dev_differentiator:284            cmd += f" --show-dev-interrupts {dev_differentiator} "285        if run_as_daemon:286            cmd += " -D "287        result = self.node.execute(288            f"ulimit -n 204800 && {self.command} {cmd}",289            shell=True,290            sudo=True,291            expected_exit_code=0,292            expected_exit_code_failure_message=f"fail to run {self.command} {cmd}",293        )294        return result295    def create_ntttcp_result(296        self, result: ExecutableResult, role: str = "server"297    ) -> NtttcpResult:298        matched_results = self.output_pattern.match(result.stdout)299        assert matched_results, "not found matched ntttcp results."300        ntttcp_result = NtttcpResult()301        ntttcp_result.role = role302        if "Mbps" == matched_results.group("unit"):303            ntttcp_result.throughput_in_gbps = Decimal(304                Decimal(matched_results.group("throughput")) / 1000305            )306        else:307            ntttcp_result.throughput_in_gbps = Decimal(308                matched_results.group("throughput")309            )310        if matched_results.group("connections_created_time"):311            ntttcp_result.connections_created_time = Decimal(312                matched_results.group("connections_created_time")313            )314        if matched_results.group("pkts_interrupt"):315            ntttcp_result.pkts_interrupt = Decimal(316                matched_results.group("pkts_interrupt")317            )318        if matched_results.group("retrans_segs"):319            ntttcp_result.retrans_segs = Decimal(matched_results.group("retrans_segs"))320        ntttcp_result.rx_packets = Decimal(matched_results.group("rx_packets"))321        ntttcp_result.tx_packets = Decimal(matched_results.group("tx_packets"))322        ntttcp_result.cycles_per_byte = Decimal(323            matched_results.group("cycles_per_byte")324        )325        return ntttcp_result326    def create_ntttcp_tcp_performance_message(327        self,328        server_result: NtttcpResult,329        client_result: NtttcpResult,330        latency: Decimal,331        connections_num: str,332        buffer_size: int,333        test_case_name: str,334        test_result: "TestResult",335    ) -> NetworkTCPPerformanceMessage:336        other_fields: Dict[str, Any] = {}337        other_fields["tool"] = constants.NETWORK_PERFORMANCE_TOOL_NTTTCP338        other_fields["buffer_size"] = Decimal(buffer_size)339        other_fields["connections_created_time"] = int(340            client_result.connections_created_time341        )342        other_fields["connections_num"] = int(connections_num)343        other_fields["latency_us"] = latency344        other_fields["retrans_segments"] = int(client_result.retrans_segs)345        other_fields["throughput_in_gbps"] = client_result.throughput_in_gbps346        other_fields["rx_packets"] = server_result.rx_packets347        other_fields["tx_packets"] = client_result.tx_packets348        other_fields["pkts_interrupts"] = client_result.pkts_interrupt349        other_fields["sender_cycles_per_byte"] = client_result.cycles_per_byte350        other_fields["receiver_cycles_rer_byte"] = server_result.cycles_per_byte351        return create_perf_message(352            NetworkTCPPerformanceMessage,353            self.node,354            test_result,355            test_case_name,356            other_fields,357        )358    def create_ntttcp_udp_performance_message(359        self,360        server_result: NtttcpResult,361        client_result: NtttcpResult,362        connections_num: str,363        buffer_size: int,364        test_case_name: str,365        test_result: "TestResult",366    ) -> NetworkUDPPerformanceMessage:367        other_fields: Dict[str, Any] = {}368        other_fields["tool"] = constants.NETWORK_PERFORMANCE_TOOL_NTTTCP369        other_fields["protocol_type"] = TransportProtocol.Udp370        other_fields["send_buffer_size"] = Decimal(buffer_size)371        other_fields["connections_created_time"] = int(372            client_result.connections_created_time373        )374        other_fields["connections_num"] = int(connections_num)375        other_fields["tx_throughput_in_gbps"] = client_result.throughput_in_gbps376        other_fields["rx_throughput_in_gbps"] = server_result.throughput_in_gbps377        other_fields["receiver_cycles_rer_byte"] = server_result.cycles_per_byte378        other_fields["data_loss"] = (379            100380            * (client_result.throughput_in_gbps - server_result.throughput_in_gbps)381            / client_result.throughput_in_gbps382        )383        return create_perf_message(384            NetworkUDPPerformanceMessage,385            self.node,386            test_result,387            test_case_name,388            other_fields,389        )390    def _initialize(self, *args: Any, **kwargs: Any) -> None:391        firewall = self.node.tools[Firewall]392        firewall.stop()393        # save the original value for recovering394        self._original_settings_tcp: List[Dict[str, str]] = []395        self._original_settings_udp: List[Dict[str, str]] = []396        sysctl = self.node.tools[Sysctl]397        for tcp_sys in self.sys_list_tcp:398            for variable, _ in tcp_sys.items():399                self._original_settings_tcp.append({variable: sysctl.get(variable)})400        for udp_sys in self.sys_list_udp:401            for variable, _ in udp_sys.items():402                self._original_settings_udp.append({variable: sysctl.get(variable)})403    def _install(self) -> bool:404        if isinstance(self.node.os, CBLMariner):405            self.node.os.install_packages(406                [407                    "kernel-headers",408                    "binutils",409                    "glibc-devel",410                    "zlib-devel",411                ]412            )413        tool_path = self.get_tool_path()414        git = self.node.tools[Git]415        git.clone(self.repo, tool_path)416        make = self.node.tools[Make]417        code_path = tool_path.joinpath("ntttcp-for-linux/src")418        make.make_install(cwd=code_path)419        self.node.execute(420            "ln -s /usr/local/bin/ntttcp /usr/bin/ntttcp", sudo=True, cwd=code_path421        ).assert_exit_code()422        return self._check_exists()423    def _set_tasks_max(self) -> None:424        need_reboot = False425        if self.node.shell.exists(426            self.node.get_pure_path(427                "/usr/lib/systemd/system/user-.slice.d/10-defaults.conf"428            )429        ):430            self.node.tools[Sed].substitute(431                regexp="TasksMax.*",432                replacement="TasksMax=122880",433                file="/usr/lib/systemd/system/user-.slice.d/10-defaults.conf",434                sudo=True,435            )436            need_reboot = True437        elif self.node.shell.exists(...
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.