How to use assign_interrupts method in lisa

Best Python code snippet using lisa_python

common.py

Source:common.py Github

copy

Full Screen

...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 = True133 finally:134 if not restore_state:135 restore_interrupts_assignment(file_path_list, node)136def get_cpu_state_file(cpu_id: str) -> str:137 return f"/sys/devices/system/cpu/cpu{cpu_id}/online"138def get_interrupts_assigned_cpu(device_id: str, channel_id: str) -> str:139 return f"/sys/bus/vmbus/devices/{device_id}/channels/{channel_id}/cpu"140def assign_interrupts(141 path_cpu: Dict[str, str],142 node: Node,143 target_cpu: str = "0",144) -> None:145 for path, _ in path_cpu.items():146 node.tools[Echo].write_to_file(target_cpu, node.get_pure_path(path), sudo=True)147def restore_interrupts_assignment(148 path_cpu: Dict[str, str],149 node: Node,150) -> None:151 if path_cpu:152 for path, target_cpu in path_cpu.items():153 node.tools[Echo].write_to_file(154 target_cpu, node.get_pure_path(path), sudo=True...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run lisa automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful