KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use swapcached method in autotest

Best Python code snippet using autotest_python

monitor.py

Source:monitor.py

1#!/usr/bin/env python32from time import sleep3from os import system4import os5import logger6_proc_status = '/proc/%d/status' % os.getpid()7_scale = {'kB': 1024.0, 'mB': 1024.0*1024.0,8          'KB': 1024.0, 'MB': 1024.0*1024.0}9def _VmB(VmKey):10    '''Private.11    '''12    global _proc_status, _scale13     # get pseudo file  /proc/<pid>/status14    try:15        t = open(_proc_status)16        v = t.read()17        t.close()18    except:19        return 0.0  # non-Linux?20     # get VmKey line e.g. 'VmRSS:  9999  kB\n ...'21    i = v.index(VmKey)22    v = v[i:].split(None, 3)  # whitespace23    if len(v) < 3:24        return 0.0  # invalid format?25     # convert Vm value to bytes26    return float(v[1]) * _scale[v[2]]27def memory(since=0.0):28    '''Return memory usage in bytes.29    '''30    return _VmB('VmSize:') - since31def resident(since=0.0):32    '''Return resident memory usage in bytes.33    '''34    return _VmB('VmRSS:') - since35def stacksize(since=0.0):36    '''Return stack size in bytes.37    '''38    return _VmB('VmStk:') - since39class Monitor(object):40    def __init__(self, node=None):41        self.node = node42        pid = os.getpid()43        self.tmp_stat = '/tmp/tmp_stat_%s' % pid44        self.tmp_meminfo = '/tmp/tmp_meminfo_%s' % pid45    def probe(self, cpu_reserve, mem_reserve):46        """return number of decoder instances that can run on this node."""47        result = 048        if self.node:49            status = system('ssh %s :' % self.node)50            if status != 0:51                logger.writeln('%s down' % self.node)52            else:53                cpu = self.cpu_usage()54                mem = self.mem_free()55                if logger.level >= 1:56                    logger.writeln('cpu usage: %.1f%%' % cpu)57                    logger.writeln('mem free: %s kB' % mem)58                result = int(min((100-cpu)/cpu_reserve, mem/mem_reserve))59        if logger.level >= 1:60            logger.writeln('%s decoder instances will start on %s' %61                           (result, self.node))62            logger.writeln()63        system('rm -f %s' % self.tmp_stat)64        system('rm -f %s' % self.tmp_meminfo)65        return result66    def get_stat_file(self, node=None):67        if node:68            system('ssh %s cat /proc/stat > %s' % (node, self.tmp_stat))69        else:70            system('cp /proc/stat %s' % self.tmp_stat)71    def get_meminfo_file(self, node=None):72        if node:73            system('ssh %s cat /proc/meminfo > %s' % (node, self.tmp_meminfo))74        else:75            system('cp /proc/meminfo %s' % self.tmp_meminfo)76    def cpu_usage(self, interval=1):77        times1 = self.parse_proc_stat()78        sleep(interval)79        times2 = self.parse_proc_stat()80        idle = times2[3] - times1[3]81        total = sum(t2 - t1 for t2, t1 in zip(times2, times1))82        return float(total - idle)/total*10083    def parse_proc_stat(self):84        self.get_stat_file(self.node)85        f = open(self.tmp_stat)86        l = f.readline()87        f.close()88        l = l.split()89        user_time = int(l[1])90        nice_time = int(l[2])91        sys_time = int(l[3])92        idle_time = int(l[4])93        return user_time, nice_time, sys_time, idle_time94    def mem_free(self):95        self.get_meminfo_file(self.node)96        f = open(self.tmp_meminfo)97        data = {}98        for line in f:99            line = line.split()100            name = line[0][:-1]101            value = int(line[1])102            data[name] = value103        # print('MemTotal: %s' % data['MemTotal'])104        # print('MemFree: %s' % data['MemFree'])105        # print('Buffers: %s' % data['Buffers'])106        # print('Cached: %s' % data['Cached'])107        # print('SwapCached: %s' % data['SwapCached'])108        # print('Active: %s' % data['Active'])109        # print('Inactive: %s' % data['Inactive'])110        # print('Buffers+Cached+SwapCached: %s' % (data['Buffers'] + data['Cached'] + data['SwapCached']))111        # print('Buffers+Cached+SwapCached+MemFree: %s' % (data['Buffers'] + data['Cached'] + data['SwapCached'] + data['MemFree']))112        # print('Active+Inactive: %s' % (data['Active'] + data['Inactive']))113        # print('Inactive+MemFree: %s' % (data['Inactive'] + data['MemFree']))114        free = data['MemFree'] + data['Buffers'] + data['Cached']115        return free116if __name__ == '__main__':117    m = Monitor()118    m.probe(25, 100000)119    for i in range(1, 30):120        node = 'f%s' % str(i).rjust(2, '0')121        print('probe %s' % node)122        m = Monitor(node)...

proc.py

Source:proc.py

1#!/usr/bin/env python2from ROOT import TTree, TFile3from array import array4import sys, re5if len(sys.argv) < 2:6    print 'Usage'7    print '  $ proc.py filename'8    sys.exit(-1)9ifname = sys.argv[1]10if re.search('\.[^\.]*$', ifname):11  ofname = re.sub('\.[^\.]*$', '', ifname)12else:13  ofname = ifname14ofname += '.root'15memTotal     = array('i', [0])16memFree      = array('i', [0])17buffers      = array('i', [0])18cached       = array('i', [0])19swapCached   = array('i', [0])20active       = array('i', [0])21inactive     = array('i', [0])22lowFree      = array('i', [0])23swapFree     = array('i', [0])24dirty        = array('i', [0])25writeback    = array('i', [0])26anonPages    = array('i', [0])27mapped       = array('i', [0])28slab         = array('i', [0])29pageTables   = array('i', [0])30committed_AS = array('i', [0])31vmallocUsed  = array('i', [0])32tr = TTree('tr', 'tr')33tr.Branch('memTotal', memTotal, 'memTotal/I')34tr.Branch('memFree', memFree, 'memFree/I')35tr.Branch('buffers', buffers, 'buffers/I')36tr.Branch('cached', cached, 'cached/I')37tr.Branch('swapCached', swapCached, 'swapCached/I')38tr.Branch('active', active, 'active/I')39tr.Branch('inactive', inactive, 'inactive/I')40tr.Branch('lowFree', lowFree, 'lowFree/I')41tr.Branch('swapFree', swapFree, 'swapFree/I')42tr.Branch('dirty', dirty, 'dirty/I')43tr.Branch('writeback', writeback, 'writeback/I')44tr.Branch('anonPages', anonPages, 'anonPages/I')45tr.Branch('mapped', mapped, 'mapped/I')46tr.Branch('slab', slab, 'slab/I')47tr.Branch('pageTables', pageTables, 'pageTables/I')48tr.Branch('committed_AS', committed_AS, 'committed_AS/I')49tr.Branch('vmallocUsed', vmallocUsed, 'vmallocUsed/I')50size = 051for line in open(ifname, 'r'):52    sp = line.split()53    if len(sp) != 17:54        break55    memTotal[0]     = int(sp[0])56    memFree[0]      = int(sp[1])57    buffers[0]      = int(sp[2])58    cached[0]       = int(sp[3])59    swapCached[0]   = int(sp[4])60    active[0]       = int(sp[5])61    inactive[0]     = int(sp[6])62    lowFree[0]      = int(sp[7])63    swapFree[0]     = int(sp[8])64    dirty[0]        = int(sp[9])65    writeback[0]    = int(sp[10])66    anonPages[0]    = int(sp[11])67    mapped[0]       = int(sp[12])68    slab[0]         = int(sp[13])69    pageTables[0]   = int(sp[14])70    committed_AS[0] = int(sp[15])71    vmallocUsed[0]  = int(sp[16])72    tr.Fill()73f = TFile(ofname, 'RECREATE')74tr.Write()...

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.