Best Python code snippet using hypothesis
BaseCPU.py
Source:BaseCPU.py  
1# Copyright (c) 2012-2013, 2015-2017 ARM Limited2# All rights reserved.3#4# The license below extends only to copyright in the software and shall5# not be construed as granting a license to any other intellectual6# property including but not limited to intellectual property relating7# to a hardware implementation of the functionality of the software8# licensed hereunder.  You may use the software subject to the license9# terms below provided that you ensure that this notice is replicated10# unmodified and in its entirety in all distributions of the software,11# modified or unmodified, in source code or in binary form.12#13# Copyright (c) 2005-2008 The Regents of The University of Michigan14# Copyright (c) 2011 Regents of the University of California15# All rights reserved.16#17# Redistribution and use in source and binary forms, with or without18# modification, are permitted provided that the following conditions are19# met: redistributions of source code must retain the above copyright20# notice, this list of conditions and the following disclaimer;21# redistributions in binary form must reproduce the above copyright22# notice, this list of conditions and the following disclaimer in the23# documentation and/or other materials provided with the distribution;24# neither the name of the copyright holders nor the names of its25# contributors may be used to endorse or promote products derived from26# this software without specific prior written permission.27#28# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS29# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT30# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR31# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT32# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,33# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT34# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,35# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY36# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT37# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE38# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.39#40# Authors: Nathan Binkert41#          Rick Strong42#          Andreas Hansson43#          Glenn Bergmans44from __future__ import print_function45import sys46from m5.SimObject import *47from m5.defines import buildEnv48from m5.params import *49from m5.proxy import *50from m5.util.fdthelper import *51from XBar import L2XBar52from InstTracer import InstTracer53from CPUTracers import ExeTracer54from MemObject import MemObject55from SubSystem import SubSystem56from ClockDomain import *57from Platform import Platform58default_tracer = ExeTracer()59if buildEnv['TARGET_ISA'] == 'alpha':60    from AlphaTLB import AlphaDTB as ArchDTB, AlphaITB as ArchITB61    from AlphaInterrupts import AlphaInterrupts62    from AlphaISA import AlphaISA63    default_isa_class = AlphaISA64elif buildEnv['TARGET_ISA'] == 'sparc':65    from SparcTLB import SparcTLB as ArchDTB, SparcTLB as ArchITB66    from SparcInterrupts import SparcInterrupts67    from SparcISA import SparcISA68    default_isa_class = SparcISA69elif buildEnv['TARGET_ISA'] == 'x86':70    from X86TLB import X86TLB as ArchDTB, X86TLB as ArchITB71    from X86LocalApic import X86LocalApic72    from X86ISA import X86ISA73    default_isa_class = X86ISA74elif buildEnv['TARGET_ISA'] == 'mips':75    from MipsTLB import MipsTLB as ArchDTB, MipsTLB as ArchITB76    from MipsInterrupts import MipsInterrupts77    from MipsISA import MipsISA78    default_isa_class = MipsISA79elif buildEnv['TARGET_ISA'] == 'arm':80    from ArmTLB import ArmTLB as ArchDTB, ArmTLB as ArchITB81    from ArmTLB import ArmStage2IMMU, ArmStage2DMMU82    from ArmInterrupts import ArmInterrupts83    from ArmISA import ArmISA84    default_isa_class = ArmISA85elif buildEnv['TARGET_ISA'] == 'power':86    from PowerTLB import PowerTLB as ArchDTB, PowerTLB as ArchITB87    from PowerInterrupts import PowerInterrupts88    from PowerISA import PowerISA89    default_isa_class = PowerISA90elif buildEnv['TARGET_ISA'] == 'riscv':91    from RiscvTLB import RiscvTLB as ArchDTB, RiscvTLB as ArchITB92    from RiscvInterrupts import RiscvInterrupts93    from RiscvISA import RiscvISA94    default_isa_class = RiscvISA95class BaseCPU(MemObject):96    type = 'BaseCPU'97    abstract = True98    cxx_header = "cpu/base.hh"99    cxx_exports = [100        PyBindMethod("switchOut"),101        PyBindMethod("takeOverFrom"),102        PyBindMethod("switchedOut"),103        PyBindMethod("flushTLBs"),104        PyBindMethod("totalInsts"),105        PyBindMethod("scheduleInstStop"),106        PyBindMethod("scheduleLoadStop"),107        PyBindMethod("getCurrentInstCount"),108    ]109    @classmethod110    def memory_mode(cls):111        """Which memory mode does this CPU require?"""112        return 'invalid'113    @classmethod114    def require_caches(cls):115        """Does the CPU model require caches?116        Some CPU models might make assumptions that require them to117        have caches.118        """119        return False120    @classmethod121    def support_take_over(cls):122        """Does the CPU model support CPU takeOverFrom?"""123        return False124    def takeOverFrom(self, old_cpu):125        self._ccObject.takeOverFrom(old_cpu._ccObject)126    system = Param.System(Parent.any, "system object")127    cpu_id = Param.Int(-1, "CPU identifier")128    socket_id = Param.Unsigned(0, "Physical Socket identifier")129    numThreads = Param.Unsigned(1, "number of HW thread contexts")130    pwr_gating_latency = Param.Cycles(300,131        "Latency to enter power gating state when all contexts are suspended")132    power_gating_on_idle = Param.Bool(False, "Control whether the core goes "\133        "to the OFF power state after all thread are disabled for "\134        "pwr_gating_latency cycles")135    function_trace = Param.Bool(False, "Enable function trace")136    function_trace_start = Param.Tick(0, "Tick to start function trace")137    checker = Param.BaseCPU(NULL, "checker CPU")138    syscallRetryLatency = Param.Cycles(10000, "Cycles to wait until retry")139    do_checkpoint_insts = Param.Bool(True,140        "enable checkpoint pseudo instructions")141    do_statistics_insts = Param.Bool(True,142        "enable statistics pseudo instructions")143    profile = Param.Latency('0ns', "trace the kernel stack")144    do_quiesce = Param.Bool(True, "enable quiesce instructions")145    wait_for_remote_gdb = Param.Bool(False,146        "Wait for a remote GDB connection");147    workload = VectorParam.Process([], "processes to run")148    dtb = Param.BaseTLB(ArchDTB(), "Data TLB")149    itb = Param.BaseTLB(ArchITB(), "Instruction TLB")150    if buildEnv['TARGET_ISA'] == 'sparc':151        interrupts = VectorParam.SparcInterrupts(152                [], "Interrupt Controller")153        isa = VectorParam.SparcISA([], "ISA instance")154    elif buildEnv['TARGET_ISA'] == 'alpha':155        interrupts = VectorParam.AlphaInterrupts(156                [], "Interrupt Controller")157        isa = VectorParam.AlphaISA([], "ISA instance")158    elif buildEnv['TARGET_ISA'] == 'x86':159        interrupts = VectorParam.X86LocalApic([], "Interrupt Controller")160        isa = VectorParam.X86ISA([], "ISA instance")161    elif buildEnv['TARGET_ISA'] == 'mips':162        interrupts = VectorParam.MipsInterrupts(163                [], "Interrupt Controller")164        isa = VectorParam.MipsISA([], "ISA instance")165    elif buildEnv['TARGET_ISA'] == 'arm':166        istage2_mmu = Param.ArmStage2MMU(ArmStage2IMMU(), "Stage 2 trans")167        dstage2_mmu = Param.ArmStage2MMU(ArmStage2DMMU(), "Stage 2 trans")168        interrupts = VectorParam.ArmInterrupts(169                [], "Interrupt Controller")170        isa = VectorParam.ArmISA([], "ISA instance")171    elif buildEnv['TARGET_ISA'] == 'power':172        UnifiedTLB = Param.Bool(True, "Is this a Unified TLB?")173        interrupts = VectorParam.PowerInterrupts(174                [], "Interrupt Controller")175        isa = VectorParam.PowerISA([], "ISA instance")176    elif buildEnv['TARGET_ISA'] == 'riscv':177        interrupts = VectorParam.RiscvInterrupts(178                [], "Interrupt Controller")179        isa = VectorParam.RiscvISA([], "ISA instance")180    else:181        print("Don't know what TLB to use for ISA %s" %182              buildEnv['TARGET_ISA'])183        sys.exit(1)184    max_insts_all_threads = Param.Counter(0,185        "terminate when all threads have reached this inst count")186    max_insts_any_thread = Param.Counter(0,187        "terminate when any thread reaches this inst count")188    simpoint_start_insts = VectorParam.Counter([],189        "starting instruction counts of simpoints")190    max_loads_all_threads = Param.Counter(0,191        "terminate when all threads have reached this load count")192    max_loads_any_thread = Param.Counter(0,193        "terminate when any thread reaches this load count")194    progress_interval = Param.Frequency('0Hz',195        "frequency to print out the progress message")196    switched_out = Param.Bool(False,197        "Leave the CPU switched out after startup (used when switching " \198        "between CPU models)")199    tracer = Param.InstTracer(default_tracer, "Instruction tracer")200    icache_port = MasterPort("Instruction Port")201    dcache_port = MasterPort("Data Port")202    _cached_ports = ['icache_port', 'dcache_port']203    if buildEnv['TARGET_ISA'] in ['x86', 'arm']:204        _cached_ports += ["itb.walker.port", "dtb.walker.port"]205    _uncached_slave_ports = []206    _uncached_master_ports = []207    if buildEnv['TARGET_ISA'] == 'x86':208        _uncached_slave_ports += ["interrupts[0].pio",209                                  "interrupts[0].int_slave"]210        _uncached_master_ports += ["interrupts[0].int_master"]211    def createInterruptController(self):212        if buildEnv['TARGET_ISA'] == 'sparc':213            self.interrupts = [SparcInterrupts() for i in xrange(self.numThreads)]214        elif buildEnv['TARGET_ISA'] == 'alpha':215            self.interrupts = [AlphaInterrupts() for i in xrange(self.numThreads)]216        elif buildEnv['TARGET_ISA'] == 'x86':217            self.apic_clk_domain = DerivedClockDomain(clk_domain =218                                                      Parent.clk_domain,219                                                      clk_divider = 16)220            self.interrupts = [X86LocalApic(clk_domain = self.apic_clk_domain,221                                           pio_addr=0x2000000000000000)222                               for i in xrange(self.numThreads)]223            _localApic = self.interrupts224        elif buildEnv['TARGET_ISA'] == 'mips':225            self.interrupts = [MipsInterrupts() for i in xrange(self.numThreads)]226        elif buildEnv['TARGET_ISA'] == 'arm':227            self.interrupts = [ArmInterrupts() for i in xrange(self.numThreads)]228        elif buildEnv['TARGET_ISA'] == 'power':229            self.interrupts = [PowerInterrupts() for i in xrange(self.numThreads)]230        elif buildEnv['TARGET_ISA'] == 'riscv':231            self.interrupts = \232                [RiscvInterrupts() for i in xrange(self.numThreads)]233        else:234            print("Don't know what Interrupt Controller to use for ISA %s" %235                  buildEnv['TARGET_ISA'])236            sys.exit(1)237    def connectCachedPorts(self, bus):238        for p in self._cached_ports:239            exec('self.%s = bus.slave' % p)240    def connectUncachedPorts(self, bus):241        for p in self._uncached_slave_ports:242            exec('self.%s = bus.master' % p)243        for p in self._uncached_master_ports:244            exec('self.%s = bus.slave' % p)245    def connectAllPorts(self, cached_bus, uncached_bus = None):246        self.connectCachedPorts(cached_bus)247        if not uncached_bus:248            uncached_bus = cached_bus249        self.connectUncachedPorts(uncached_bus)250    def addPrivateSplitL1Caches(self, ic, dc, iwc = None, dwc = None):251        self.icache = ic252        self.dcache = dc253        self.icache_port = ic.cpu_side254        self.dcache_port = dc.cpu_side255        self._cached_ports = ['icache.mem_side', 'dcache.mem_side']256        if buildEnv['TARGET_ISA'] in ['x86', 'arm']:257            if iwc and dwc:258                self.itb_walker_cache = iwc259                self.dtb_walker_cache = dwc260                self.itb.walker.port = iwc.cpu_side261                self.dtb.walker.port = dwc.cpu_side262                self._cached_ports += ["itb_walker_cache.mem_side", \263                                       "dtb_walker_cache.mem_side"]264            else:265                self._cached_ports += ["itb.walker.port", "dtb.walker.port"]266            # Checker doesn't need its own tlb caches because it does267            # functional accesses only268            if self.checker != NULL:269                self._cached_ports += ["checker.itb.walker.port", \270                                       "checker.dtb.walker.port"]271    def addTwoLevelCacheHierarchy(self, ic, dc, l2c, iwc=None, dwc=None,272                                  xbar=None):273        self.addPrivateSplitL1Caches(ic, dc, iwc, dwc)274        self.toL2Bus = xbar if xbar else L2XBar()275        self.connectCachedPorts(self.toL2Bus)276        self.l2cache = l2c277        self.toL2Bus.master = self.l2cache.cpu_side278        self._cached_ports = ['l2cache.mem_side']279    def createThreads(self):280        # If no ISAs have been created, assume that the user wants the281        # default ISA.282        if len(self.isa) == 0:283            self.isa = [ default_isa_class() for i in xrange(self.numThreads) ]284        else:285            if len(self.isa) != int(self.numThreads):286                raise RuntimeError("Number of ISA instances doesn't "287                                   "match thread count")288        if self.checker != NULL:289            self.checker.createThreads()290    def addCheckerCpu(self):291        pass292    def createPhandleKey(self, thread):293        # This method creates a unique key for this cpu as a function of a294        # certain thread295        return 'CPU-%d-%d-%d' % (self.socket_id, self.cpu_id, thread)296    #Generate simple CPU Device Tree structure297    def generateDeviceTree(self, state):298        """Generate cpu nodes for each thread and the corresponding part of the299        cpu-map node. Note that this implementation does not support clusters300        of clusters. Note that GEM5 is not compatible with the official way of301        numbering cores as defined in the Device Tree documentation. Where the302        cpu_id needs to reset to 0 for each cluster by specification, GEM5303        expects the cpu_id to be globally unique and incremental. This304        generated node adheres the GEM5 way of doing things."""305        if bool(self.switched_out):306            return307        cpus_node = FdtNode('cpus')308        cpus_node.append(state.CPUCellsProperty())309        #Special size override of 0310        cpus_node.append(FdtPropertyWords('#size-cells', [0]))311        # Generate cpu nodes312        for i in range(int(self.numThreads)):313            reg = (int(self.socket_id)<<8) + int(self.cpu_id) + i314            node = FdtNode("cpu@%x" % reg)315            node.append(FdtPropertyStrings("device_type", "cpu"))316            node.appendCompatible(["gem5,arm-cpu"])317            node.append(FdtPropertyWords("reg", state.CPUAddrCells(reg)))318            platform, found = self.system.unproxy(self).find_any(Platform)319            if found:320                platform.annotateCpuDeviceNode(node, state)321            else:322                warn("Platform not found for device tree generation; " \323                     "system or multiple CPUs may not start")324            freq = round(self.clk_domain.unproxy(self).clock[0].frequency)325            node.append(FdtPropertyWords("clock-frequency", freq))326            # Unique key for this CPU327            phandle_key = self.createPhandleKey(i)328            node.appendPhandle(phandle_key)329            cpus_node.append(node)...input_vectors_r2.py
Source:input_vectors_r2.py  
...43        for element in input_class:44            if re.match(re.compile(element), imp) is not None:45                return True46        return False47    def get_interrupts(self, function_list):48        interrupts = []49        for function in function_list:50            try:51                interrupts.extend(self._get_interrupts_of_function(function))52            except (TypeError, KeyError):  # function issues53                continue54        return interrupts55    def _get_interrupts_of_function(self, function):56        interrupts = []57        for instruction in self.get_function_instructions(function)['ops']:58            if 'opcode' in instruction and instruction['type'] == 'swi':59                for trap in ['syscall', 'swi', 'int 0x80']:60                    if trap in instruction['opcode']:61                        interrupts.append(instruction['offset'])62        return interrupts63    def find_input_vectors(self):64        input_vectors = []65        function_list = self.api.cmdj('aflj')66        if not function_list:67            return input_vectors68        for function in function_list:69            if self._is_imported_function(function):70                input_vectors.extend(self.find_input_vectors_of_function(function))71        interrupts = self.get_interrupts(function_list)72        if interrupts:73            input_vectors.append({74                'class': 'kernel',75                'name': 'interrupts',76                'count': len(interrupts),77                'xrefs': interrupts78            })79        return input_vectors80    def find_input_vectors_of_function(self, function):81        input_vectors = []82        clean_import = function['name'].replace(self.config['import_prefix'], '')83        for input_class in self.config['input_classes']:84            if self.matches_import(clean_import.lower(), self.config['input_classes'][input_class]):85                input_vectors.append({...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
