Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use set_interrupts_assigned_cpu method in lisa

Best Python code snippet using lisa_python

cpusuite.py

Source:cpusuite.py

...165    def verify_cpu_offline_channel_add(self, log: Logger, node: Node) -> None:166        # skip test if kernel doesn't support cpu hotplug167        check_runnable(node)168        # set vmbus channels target cpu into 0 if kernel supports this feature.169        file_path_list = set_interrupts_assigned_cpu(log, node)170        # when kernel doesn't support above feature, we have to rely on current vm's171        # cpu usage. then collect the cpu not in used exclude cpu0.172        idle_cpus = get_idle_cpus(node)173        log.debug(f"idle cpus: {idle_cpus}")174        # save origin current channel175        origin_device_channel = (176            node.tools[Ethtool].get_device_channels_info("eth0", True)177        ).current_channels178        # set channel count into 1 to get idle cpus179        if len(idle_cpus) == 0:180            node.tools[Ethtool].change_device_channels_info("eth0", 1)181            idle_cpus = get_idle_cpus(node)182            log.debug(f"idle cpus: {idle_cpus}")183        if len(idle_cpus) == 0:...

common.py

Source:common.py

...12    if not node.tools[KernelConfig].is_built_in("CONFIG_HOTPLUG_CPU"):13        raise SkippedException(14            f"the distro {node.os.name} doesn't support cpu hotplug."15        )16def set_interrupts_assigned_cpu(17    log: Logger, node: Node, target_cpu: str = "0"18) -> Dict[str, str]:19    uname = node.tools[Uname]20    kernel_version = uname.get_linux_information().kernel_version21    dmesg = node.tools[Dmesg]22    lsvmbus = node.tools[Lsvmbus]23    vmbus_version = dmesg.get_vmbus_version()24    file_path_list: Dict[str, str] = {}25    # the vmbus interrupt channel reassignment feature is available in 5.8+ kernel and26    # vmbus version in 4.1+, the vmbus version is negotiated with the host.27    if kernel_version >= "5.8.0" and vmbus_version >= "4.1.0":28        # save the raw cpu number for each channel for restoring later.29        channels = lsvmbus.get_device_channels(force_run=True)30        for channel in channels:31            for channel_vp_map in channel.channel_vp_map:32                current_target_cpu = channel_vp_map.target_cpu33                if current_target_cpu == target_cpu:34                    continue35                file_path_list[36                    get_interrupts_assigned_cpu(37                        channel.device_id, channel_vp_map.rel_id38                    )39                ] = current_target_cpu40        # set all vmbus channel interrupts go into cpu target_cpu.41        assign_interrupts(file_path_list, node, target_cpu)42    else:43        # if current distro doesn't support this feature, the backup dict will be empty,44        # there is nothing we can restore later, the case will rely on actual cpu usage45        # on vm, if no idle cpu, then case will be skipped.46        log.debug(47            f"current distro {node.os.name}, os version {kernel_version}, "48            f"vmbus version {vmbus_version} doesn't support "49            "change channels target cpu featue."50        )51    return file_path_list52def get_idle_cpus(node: Node) -> List[str]:53    lsvmbus = node.tools[Lsvmbus]54    channels = lsvmbus.get_device_channels(force_run=True)55    # get all cpu in used from vmbus channels assignment56    cpu_in_used = set()57    for channel in channels:58        for channel_vp_map in channel.channel_vp_map:59            target_cpu = channel_vp_map.target_cpu60            if target_cpu == "0":61                continue62            cpu_in_used.add(target_cpu)63    # get all cpu exclude cpu 0, usually cpu 0 is not allowed to do hotplug64    cpu_count = node.tools[Lscpu].get_core_count()65    all_cpu = list(range(1, cpu_count))66    # get the idle cpu by excluding in used cpu from all cpu67    idle_cpu = [str(x) for x in all_cpu if str(x) not in cpu_in_used]68    return idle_cpu69def set_cpu_state_serial(70    log: Logger, node: Node, idle_cpu: List[str], state: str71) -> None:72    for target_cpu in idle_cpu:73        log.debug(f"setting cpu{target_cpu} to {state}.")74        if state == CPUState.ONLINE:75            set_state = set_cpu_state(node, target_cpu, True)76        else:77            set_state = set_cpu_state(node, target_cpu, False)78        if not set_state:79            raise BadEnvironmentStateException(80                (81                    f"Expected cpu{target_cpu} state: {state}."82                    f"The test failed leaving cpu{target_cpu} in a bad state."83                ),84            )85def set_idle_cpu_offline_online(log: Logger, node: Node, idle_cpu: List[str]) -> None:86    for target_cpu in idle_cpu:87        set_offline = set_cpu_state(node, target_cpu, False)88        log.debug(f"set cpu{target_cpu} from online to offline.")89        exception_message = (90            f"expected cpu{target_cpu} state: {CPUState.OFFLINE}(offline), "91            f"actual state: {CPUState.ONLINE}(online)."92        )93        if not set_offline:94            raise BadEnvironmentStateException(95                exception_message,96                f"the test failed leaving cpu{target_cpu} in a bad state.",97            )98        set_online = set_cpu_state(node, target_cpu, True)99        log.debug(f"set cpu{target_cpu} from offline to online.")100        exception_message = (101            f"expected cpu{target_cpu} state: {CPUState.ONLINE}(online), "102            f"actual state: {CPUState.OFFLINE}(offline)."103        )104        if not set_online:105            raise BadEnvironmentStateException(106                exception_message,107                f"the test failed leaving cpu{target_cpu} in a bad state.",108            )109def verify_cpu_hot_plug(log: Logger, node: Node, run_times: int = 1) -> None:110    check_runnable(node)111    file_path_list: Dict[str, str] = {}112    restore_state = False113    try:114        for iteration in range(1, run_times + 1):115            log.debug(f"start the {iteration} time(s) testing.")116            restore_state = False117            # set vmbus channels target cpu into 0 if kernel supports this feature.118            file_path_list = set_interrupts_assigned_cpu(log, node)119            # when kernel doesn't support above feature, we have to rely on current vm's120            # cpu usage. then collect the cpu not in used exclude cpu0.121            idle_cpu = get_idle_cpus(node)122            if 0 == len(idle_cpu):123                raise SkippedException(124                    "all of the cpu are associated vmbus channels,"125                    " no idle cpu can be used to test hotplug."126                )127            # start to take idle cpu from online to offline, then offline to online.128            set_idle_cpu_offline_online(log, node, idle_cpu)129            # when kernel doesn't support set vmbus channels target cpu feature, the130            # dict which stores original status is empty, nothing need to be restored.131            restore_interrupts_assignment(file_path_list, node)132            restore_state = True...

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.