How to use set_cpu_state method in lisa

Best Python code snippet using lisa_python

observation.py

Source:observation.py

...67            if 'WRITE' in reg.name or 'ADDR' in reg.name:68                continue69            #for x in tqdm(range(reg.bits)):70            for x in (range(reg.bits)):71                cpu.set_cpu_state(seed_in)72                pos_val = (1<<x)73                mutate_val = seed_in[reg] ^ pos_val74                cpu.write_reg(reg, mutate_val)75                try:76                    sb, sa = cpu.execute(bytecode)77                except UcError as e:78                    continue79                except OutOfRangeException as e:80                    continue81                sbs = regs2bits(sb, state_format)82                sas = regs2bits(sa, state_format)83                if not sss.diff(sbs):84                    continue85                assert(sss.diff(sbs))86                state_list.append((sbs, sas))87        return Observation((sss, rss), state_list, bytestring, archstring, state_format)88    def _gen_seeds(self, bytestring, archstring, state_format, strategies=None):89        """Generates a set of seed states based on the state_format using the strategies defined.90        Args:91            bytestring (string): String representing the bytes of the instruction in hex without space92            archstring (string): Architecture String (X86, AMD64, ARM32, ARM64)93            state_format (list(Register)): A list of registers which defines the order of the State object94        Returns:95            A list of seed state IO tuples96        Raises:97            None98        """99        if not strategies:100            strategies = [RandomNumber(100), Bitwalk(), ZeroWalk(), BitFill(), IEEE754Extended(10)]101        seed_states = []102        # TODO: HACK to speed up, we'll ignore write and addr103        temp_state_format = [x for x in state_format if ('WRITE' not in x.name104            and 'ADDR' not in x.name)]105        for strategy in strategies:106            for seed_variation in tqdm(strategy.generator(temp_state_format)):107                seed_io = self._gen_random_seed_io(bytestring, archstring, seed_variation) 108                # check if its successful or not, if not debug print109                if seed_io:110                    seed_states.append(seed_io)111                else:112                    if self.DEBUG_LOG:113                        print("MAX_TRIES-{}-{}-{}-{}".format(bytestring, archstring, state_format, seed_variation))114                    continue115        return seed_states116    def _gen_seed_io(self, bytestring, archstring, seed_in, num_tries=255):117        """Generates a pair of in / out CPUState, seed_in, seed_out, by executing the instruction using a the provided CPUState118        Args:119            bytestring (string): String representing the bytes of the instruction in hex without space120            archstring (string): Architecture String (X86, AMD64, ARM32, ARM64)121            seed_in ( dict{Register:Integer } ): The input seed state122        Returns:123            Returns the seed input / output states as tuple(CPUState, CPUState) if successful.124            Otherwise returns None if it fails to find within num_tries125        Raises:126            Exception 127        """128        cpu = self.cpu129        bytecode = bytestring.decode('hex')130        for x in range(num_tries):131            try:132                cpu.set_cpu_state(seed_in)133                sb, sa = cpu.execute(bytecode)134                break135            except OutOfRangeException as e:136                if x == num_tries-1:137                    return None138                continue139            except UcError as e:140                if x == num_tries-1:141                    return None142                continue143        return sb,sa144    def _gen_random_seed_io(self, bytestring, archstring, seed_variation, num_tries=255):145        """Generates a pair of in / out CPUState, seed_in, seed_out, by executing the instruction using a randomly generated CPUState with the seed_variation applied.146        Args:...

hotplug_rem_ops.py

Source:hotplug_rem_ops.py

...83                _cpu_state.append(_fields[6].split(':')[1])84        85        return _cpu_state86    @trace87    def set_cpu_state(self, cpu_nr, state) :88        '''89        TBD90        '''        91        _cmd = "ssh"92        if self.priv_key :93            _cmd += " -i " + self.priv_key94        _cmd += " -o StrictHostKeyChecking=no "95        _cmd += self.login + '@' + self.host + ' '96        if state == "active" :97            _cmd += "\"/bin/echo 1 > /sys/devices/system/cpu/cpu" + str(cpu_nr)98            _cmd += "/online\""99        elif state == "inactive" :100            _cmd += "\"/bin/echo 0 > /sys/devices/system/cpu/cpu" + str(cpu_nr)101            _cmd += "/online\""102        else :103            _msg = "Invalid state (" + state + ") for cpu " + str(cpu_nr)104            _msg = " on the guest \"" + self.host + "\"."105            cberr(_msg)106            raise self.HotplugMgdConnException(_msg, 4)107        _proc_h = Popen(_cmd, bufsize=-1, shell=True, stdout=PIPE, stderr=PIPE) 108        if not _proc_h :109            _msg = "Failed to create subprocess with command " + _cmd110            cberr(_msg)111            raise self.HotplugMgdConnException(_msg, 3)112        else :113            _msg = "Successfully created subprocess with command " + _cmd114            cbdebug(_msg)115                    116        _stdout = None117        _stderr = None118        _stdout, _stderr = _proc_h.communicate()119        sleep(6)120        121        _cmd = "ssh"122        if self.priv_key :123            _cmd += " -i " + self.priv_key124        _cmd += " -o StrictHostKeyChecking=no "125        _cmd += self.login + '@' + self.host + ' '126        _cmd += "cat /sys/devices/system/cpu/cpu" + str(cpu_nr) + "/online"127        128        _proc_h = Popen(_cmd, bufsize=-1, shell=True, stdout=PIPE, stderr=PIPE) 129        if not _proc_h :130            _msg = "Failed to create subprocess with command " + _cmd131            cberr(_msg)132            raise self.HotplugMgdConnException(_msg, 3)133        else :134            _msg = "Successfully created subprocess with command " + _cmd135            cbdebug(_msg)136        137        _stdout = None138        _stderr = None139        _stdout, _stderr = _proc_h.communicate()        140        141        if state == "active" and _stdout.split('\n')[0].count('1') :142            return True143        elif state == "inactive" and _stdout.split('\n')[0].count('0') :144            return True145        else :146            _msg = "Failed to set cpu " + str(cpu_nr) + " state to " + state147            _msg += " on guest " + self.host + '.'148            cberr(_msg)149            raise self.HotplugMgdConnException(_msg, 3)150    @trace151    def set_active_cpus(self, active_cpus_tgt_nr):152        '''153        TBD154        '''        155        _cpu_state = self.get_cpus_state()156        157        _total_cpus = len(_cpu_state)158        159        _active_cpus = []160        _inactive_cpus = []161        for _cpu_number, _cpu_state in enumerate(_cpu_state) :162            if _cpu_state == '1' and _cpu_number > 0 :163                _active_cpus.append(_cpu_number)164            elif _cpu_state == '0' :165                _inactive_cpus.append(_cpu_number)166        167        active_cpus_tgt_nr = int(active_cpus_tgt_nr)168        169        if active_cpus_tgt_nr > _total_cpus :170            _msg = "Unable to activate " + str(active_cpus_tgt_nr) + " CPUs on "171            _msg += "the guest \"" + self.host + "\" (maximum number of CPUs "172            _msg += "for this guest is " + str(_total_cpus) + ")."173            cberr(_msg)174            raise self.HotplugMgdConnException(_msg, 3)175        elif active_cpus_tgt_nr < 1 :176            _msg = "At least 1 CPU must be active on the guest "177            _msg += "\"" + self.host + "\"."178            cberr(_msg)179            raise self.HotplugMgdConnException(_msg, 3)180        181        active_cpus_tgt_nr -= 1 #CPU0 is always active182        183        _active_cpus_nr = len(_active_cpus)184        if active_cpus_tgt_nr > _active_cpus_nr :185            while _active_cpus_nr < active_cpus_tgt_nr :186                _cpu_nr = _inactive_cpus.pop()187                try :188                    self.set_cpu_state(_cpu_nr, "active")189                    _active_cpus_nr += 1190                    sleep(3)191                except self.HotplugMgdConnException :192                    _msg = "Error while checking active cpus"193                    raise self.HotplugMgdConnException(_msg, 3) 194        elif active_cpus_tgt_nr < _active_cpus_nr :195            while _active_cpus_nr > active_cpus_tgt_nr :196                _cpu_nr = _active_cpus.pop()197                try :198                    self.set_cpu_state(_cpu_nr, "inactive")199                    _active_cpus_nr -= 1200                    sleep(3)201                except self.HotplugMgdConnException :202                    _msg = "Error while checking active cpus"203                    raise self.HotplugMgdConnException(_msg, 3)204        else :205            _msg = "The number of active CPUs is equal the number of "206            _msg += " targeted active CPUs. There is nothing to be done."207            cbdebug(_msg)208        ...

common.py

Source:common.py

...71) -> 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=True155            )156def set_cpu_state(node: Node, cpu: str, online: bool = False) -> bool:157    file_path = get_cpu_state_file(cpu)158    state = CPUState.OFFLINE159    if online:160        state = CPUState.ONLINE161    node.tools[Echo].write_to_file(state, node.get_pure_path(file_path), sudo=True)162    result = node.tools[Cat].read(file_path, force_run=True, sudo=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.