KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use pid method in fMBT
Best Python code snippet using fMBT_python
event_parser.py
Source:event_parser.py 
1#!/usr/bin/env python2# coding: utf-83import sys, os4import matplotlib5matplotlib.use('Agg')6from pylab import *7from matplotlib.transforms import TransformedBbox 8class EventFactory(object):9 #CPU time pid cpu (si pid == -1 -> idle)10 cpu_event= lambda x: Event(x[0],EventFactory.Events.keys().index(x[0]),int(x[1]),int(x[2]),int(x[3]))11 #EVENT time pid 12 other_event= lambda x: Event(x[0],EventFactory.Events.keys().index(x[0]),int(x[1]),int(x[2]),-1)13 #CONTEXT CPU cpu time (se pone pid == -2) 14 context_switch_event= lambda x: Event(x[0] ,EventFactory.Events.keys().index(x[0]),int(x[3]),-2,int(x[2]))15 Events = {'LOAD': other_event,16 'CPU':cpu_event, 17 'BLOCK': other_event, 18 'UNBLOCK': other_event, 19 'EXIT': other_event,20 'CONTEXT':context_switch_event,21 'WAITING':None,22 'NOT_LOAD': None,23 'CPU_BLOCK': None}24 25 @classmethod26 def get_event(cls, event_line):27 splited_event_line= event_line.split()28 if splited_event_line[0] in EventFactory.Events.keys():29 print splited_event_line30 return EventFactory.Events[splited_event_line[0]](splited_event_line)31 else:32 return None33 34class Event(object):35 def __init__(self, event_type, event_code, time, pid, core):36 self.event_type= event_type37 self.event_code= event_code38 self.time= time39 self.pid= pid40 self.core= core41 def __str__(self):42 return 'Type: ' + self.event_type + ', Code: ' + str(self.event_code) + ', Time: ' + str(self.time) + ', Pid: ' + str(self.pid) + ', Core: ' + str(self.core)43def parseInput(fin):44 ln = 045 result = []46 cores = 047 pids= 048 settings = None49 cpus_timeline= dict()50 print fin51 for line in fin:52 ln += 153 vls = line.split()54 if line and line[0] == '#':55 if line.startswith('# SETTINGS '):56 settings = line[11:].strip()57 continue58 else:59 line= line[2:].strip() # Queda --> 'CONTEXT CPU[cpu] time60 event= EventFactory.get_event(line)61 result.append(event)62 63 if event.event_type == 'CPU':64 if (cores <=event.core):65 cores = event.core+166 if event.event_type == 'LOAD':67 if(pids <= event.pid):68 pids = event.pid +169 70 return settings, result, cores, pids71def dataGathering(data, cores, pids):72 core_resume= dict() # core: {processing_time: #, switching_time: #, idle_time: #}73 core_timeline= dict() # core: list(pids) NOTA: se supone que en cada tick hay un pid74 pid_resume= dict() # pid: {load_time: #, running_time: #, blocked: #, end_time: #}75 pids_timeline= dict() # pid: list((event_code, core))76 77 block_lapse= dict()78 for event in data:79 #core_time80 if event.core != -1:81 if event.core not in core_timeline: core_timeline[event.core]= []82 core_timeline[event.core].append(event.pid)83 84 #core_time85 if event.core != -1:86 if event.core not in core_resume: core_resume[event.core]= {'processing_time': 0, 'switching_time': 0, 'idle_time': 0}87 if event.event_type == 'CPU':88 if event.pid != -1: core_resume[event.core]['processing_time'] = core_resume[event.core]['processing_time'] + 189 else: core_resume[event.core]['idle_time'] = core_resume[event.core]['idle_time'] + 190 elif event.event_type == 'CONTEXT': core_resume[event.core]['switching_time'] = core_resume[event.core]['switching_time'] + 191 92 if event.pid != -2 and event.pid != -1:93 #task_resume94 if event.pid not in pid_resume: pid_resume[event.pid]= {'running_time': 0, 'blocked': 0}95 if event.event_type == 'LOAD': pid_resume[event.pid]['load_time']= event.time96 if event.event_type == 'CPU': pid_resume[event.pid]['running_time']= pid_resume[event.pid]['running_time'] + 197 if event.event_type == 'BLOCK': 98 if event.pid not in block_lapse: block_lapse[event.pid]= -199 if block_lapse[event.pid] == -1:100 block_lapse[event.pid]= event.time101 if event.event_type == 'UNBLOCK':102 pid_resume[event.pid]['blocked']= pid_resume[event.pid]['blocked'] + event.time - block_lapse[event.pid] + 1103 block_lapse[event.pid]= -1104 if event.event_type == 'EXIT': pid_resume[event.pid]['end_time']= event.time105 106 #task_timeline no hay interes en mostrar los context switchs107 108 if event.pid not in pids_timeline: pids_timeline[event.pid]= []109 if event.event_type == 'LOAD':110 #NOT LOADED111 for i in range(0, event.time):112 pids_timeline[event.pid].append((EventFactory.Events.keys().index('NOT_LOAD'),-1))113 elif event.event_type == 'CPU':114 prev_event_code, prev_event_core= pids_timeline[event.pid][-1]115 #POSIBLE BLOCK ANTES QUE CPU116 if len(pids_timeline[event.pid]) == event.time:117 prev_event_code, prev_event_core= pids_timeline[event.pid][-1]118 if (EventFactory.Events.keys().index('BLOCK') == prev_event_code or119 EventFactory.Events.keys().index('CPU_BLOCK') == prev_event_code):120 pid_resume[event.pid]['running_time']= pid_resume[event.pid]['running_time'] - 1 121 pids_timeline[event.pid][-1]= (EventFactory.Events.keys().index('CPU_BLOCK'),event.core)122 continue123 #WAITING EVENT124 for i in range(len(pids_timeline[event.pid])+1, event.time):125 pids_timeline[event.pid].append((EventFactory.Events.keys().index('LOAD'),-1))126 elif event.event_type == 'BLOCK':127 #ESTA BLOQUEADO Y EJECUTANDO128 if len(pids_timeline[event.pid]) == event.time:129 prev_event_code, prev_event_core= pids_timeline[event.pid][-1]130 if (EventFactory.Events.keys().index('CPU') == prev_event_code or131 EventFactory.Events.keys().index('CPU_BLOCK') == prev_event_code):132 pids_timeline[event.pid][-1]= (EventFactory.Events.keys().index('CPU_BLOCK'),prev_event_core)133 continue134 elif event.event_type == 'UNBLOCK':135 #Agrego el gap que queda de todo el tiempo bloqueado136 block_code, block_dummy_core= pids_timeline[event.pid][-1]137 if block_code != EventFactory.Events.keys().index('CPU_BLOCK'): 138 for i in range(len(pids_timeline[event.pid])+1, event.time+1):139 pids_timeline[event.pid].append((block_code,block_dummy_core))140 if event.event_type != 'UNBLOCK' and event.event_type != 'EXIT':141 pids_timeline[event.pid].append((event.event_code,event.core))142 143 return core_resume, core_timeline, pid_resume, pids_timeline144def draw_cores_resume_bars(cores_resume, filename):145 ''' pre: cores_resume = { core: {processing_time: #, switching_time: #, idle_time: #}}'''146 ''' post: horizonal bar diagram in filenaname.png '''147 148 colors={'processing_time': 'green', 'switching_time': 'yellow', 'idle_time': 'red'}149 labels={'processing_time': 'Procesando', 'switching_time': 'Cambiando contexto', 'idle_time': 'Sin uso'}150 151 fig= figure(figsize=(11.8,8.3))152 153 154 bars_lenghts= []155 for core in cores_resume:156 bars_lenghts.append(sum(cores_resume[core].values()))157 values= sorted(zip(cores_resume[core].values(), cores_resume[core].keys()), reverse= True)158 base= 0159 for value in values:160 broken_barh([(base,value[0])], (core-0.4,0.8), color=colors[value[1]], edgecolor='black')161 base= base+value[0]162 163 title('Tiempo total por tipo de tarea por core')164 yticks(cores_resume.keys())165 xlabel('Tiempo total')166 ylabel('Core')167 xlim((0,max(bars_lenghts)+1))168 ylim((-1,len(cores_resume.keys())+1))169 tight_layout()170 legend()171 savefig(filename+'.png', dpi=300, format='png')172 return None173def draw_cores_timeline_gannt(cores_timeline, filename):174 ''' pre: cores_timeline = {core: list(pid) } '''175 176 # Necesitaria tener la info algo asi como177 # core: {pid: ini_1,fin_1,ini_2,fin_2} (es decir por intervalos)178 # broken_barh necesita (inicio, longitud)179 180 colors={'pid':'#c0ffc0','switch':'#b7b7f7', 'idle':'#d0d0d0'}181 fig= figure(figsize=(11.8,8.3))182 ax = fig.add_subplot(111) 183 title('Tareas en Core por tiempo')184 yticks(cores_timeline.keys())185# ax.xaxis.set_major_locator( 186 ax.xaxis.set_major_locator( IndexLocator(2,1) )187 #xticks(range(len(cores_timeline[0])),range(0,len(cores_timeline[0]),5))188 xlabel('Tiempo')189 ylabel('Core')190 ylim((-1,len(cores_timeline.keys())))191 ax.grid(True) 192 for core in cores_timeline:193 pids_by_time= cores_timeline[core]194 intervals= dict()195 last_pid= -1196 for time in range(len(pids_by_time)):197 if last_pid != pids_by_time[time]:198 last_pid = pids_by_time[time]199 if last_pid not in intervals: intervals[last_pid]= []200 intervals[last_pid].append((time, 1))201 #intervals.push((last_pid,time,1))202 else:203 #pid, time, interval_size= intervals.pop()204 #intervals.push((pid, time, interval_size+1))205 time, interval_size= intervals[last_pid].pop()206 intervals[last_pid].append((time, interval_size+1))207 208 for pid in intervals:209 if pid >= 0:210 rect= ax.broken_barh(intervals[pid], (core-0.25, 0.5), facecolor=colors['pid'])211 for init, size in intervals[pid]:212 ax.text(init+(size/2.0),core, str(pid), ha="center", va="center", size=9, weight='bold')213 elif pid == -1:214 rect= ax.broken_barh(intervals[pid], (core-0.25, 0.5), facecolor=colors['idle'])215 elif pid == -2:216 rect= ax.broken_barh(intervals[pid], (core-0.25, 0.5), facecolor=colors['switch'])217 else:218 print 'ERROR!'219 220 tarea_dummy = Rectangle((0, 0), 1, 1, fc=colors['pid'])221 switch_dummy = Rectangle((0, 0), 1, 1, fc=colors['switch'])222 idle_dummy = Rectangle((0, 0), 1, 1, fc=colors['idle'])223 legend([tarea_dummy, switch_dummy, idle_dummy], ['Tarea','Cambio de contexto','Inactivo'])224 tight_layout()225 fig.autofmt_xdate()226 ax.legend()227 savefig(filename+'.png', dpi=300, format='png')228def draw_tasks_resume_bars(pids_resume, filename):229 ''' pre: pid: {load_time: #, running_time: #, blocked: #, end_time: #}'''230 ''' post: horizonal bar diagram in filenaname.png '''231 232 colors={'running_time': '#c0ffc0', 'blocked': '#b7b7f7', 'waiting_time': '#d0d0d0'}233 labels={'running_time': 'En ejecucion', 'blocked': 'Bloqueado', 'waiting_time': 'Esperando'}234 235 fig= figure(figsize=(11.8,8.3))236 pids_resume.pop(-1,None)237 max_time= 0238 for pid in pids_resume:239 load_time= pids_resume[pid].pop('load_time',-1)240 end_time= pids_resume[pid].pop('end_time',-1)241 max_time= max([end_time-load_time,max_time])242 pids_resume[pid]['waiting_time']= end_time - load_time - pids_resume[pid]['running_time'] - pids_resume[pid]['blocked']243 values= sorted(zip(pids_resume[pid].values(), pids_resume[pid].keys()), reverse= True)244 base= 0245 for value in values:246 broken_barh([(base,value[0])], (pid-0.25, 0.5), facecolor=colors[value[1]])247 #if pid != 0:248 # barh(pid, base + value[0], align='center', color=colors[value[1]], edgecolor='black')249 #else:250 # barh(pid, base + value[0], align='center', color=colors[value[1]], edgecolor='black', label=labels[value[1]])251 base= base + value[0]252 running_dummy = Rectangle((0, 0), 1, 1, fc=colors['running_time'])253 blocked_dummy = Rectangle((0, 0), 1, 1, fc=colors['blocked'])254 waiting_dummy = Rectangle((0, 0), 1, 1, fc=colors['waiting_time'])255 legend([running_dummy, blocked_dummy, waiting_dummy],[labels['running_time'],labels['blocked'], labels['waiting_time']],loc='best', ncol=3)256 title('Tiempo total de la tarea divido en estados')257 yticks(pids_resume.keys())258 xlabel('Tiempo total')259 ylabel('Tarea')260 ylim((-1,len(pids_resume.keys())+1))261 xlim((0,max_time+2))262 tight_layout()263 #show()264 savefig(filename+'.png', dpi=300, format='png')265 return None266def draw_pids_timeline_gannt(pids_timeline, filename):267 ''' pre: pids_timeline = { pid: list((event_code, core))}'''268 ''' post: horizonal bar diagram in filenaname.png '''269 270 # Necesitaria tener la info algo asi como271 # core: {pid: ini_1,fin_1,ini_2,fin_2} (es decir por intervalos)272 # broken_barh necesita (inicio, longitud)273 274 colors={'CPU':'#f7b7b7','BLOCK':'#b7b7f7', 'WAITING':'#e7ffe7', 'LOAD':'#e7ffe7', 'NOT_LOAD':'#d0d0d0', 'CPU_BLOCK':'#b7b7f7'}275 fig= figure(figsize=(11.8,8.3))276 ax = fig.add_subplot(111) 277 title('Estado de las tareas por tiempo')278 yticks(pids_timeline.keys(), sorted(pids_timeline.keys(),reverse=True))279# ax.xaxis.set_major_locator( 280 ax.xaxis.set_major_locator( IndexLocator(5,1) )281 #xticks(range(len(cores_timeline[0])),range(0,len(cores_timeline[0]),5))282 xlabel('Tiempo')283 ylabel('Tarea')284 ax.grid(True) 285 for pid in pids_timeline:286 pid_state_by_time= pids_timeline[pid]287 intervals= dict()288 last_state= (None,None)289 for time in range(len(pid_state_by_time)):290 if last_state != pid_state_by_time[time]:291 last_state = pid_state_by_time[time]292 if last_state not in intervals: intervals[last_state]= []293 intervals[last_state].append((time, 1))294 #intervals.push((last_pid,time,1))295 else:296 #pid, time, interval_size= intervals.pop()297 #intervals.push((pid, time, interval_size+1))298 time, interval_size= intervals[last_state].pop()299 intervals[last_state].append((time, interval_size+1))300 for event_code, core in intervals.keys():301 event_name= EventFactory.Events.keys()[event_code]302 rect= ax.broken_barh(intervals[(event_code, core)], ((len(pids_timeline)-pid-1)-0.25, 0.5), facecolor=colors[event_name])303 if core >= 0:304 for init, size in intervals[(event_code, core)]:305 ax.text(init+(size/2.0),(len(pids_timeline)-pid-1), str(core), ha="center", va="center", size=9, weight='bold')306 #else:307 #if event_code == -1:308 #elif event_code == -2:309 #rect= ax.broken_barh(intervals[(event_code, core)], (pid-0.25, 0.5), facecolor=colors['not_loaded'])310 #elif event_name == 'LOAD':311 #rect= ax.broken_barh(intervals[(event_code, core)], (pid-0.25, 0.5), facecolor=colors['load'])312 #elif event_name == 'BLOCK':313 #rect= ax.broken_barh(intervals[(event_code, core)], (pid-0.25, 0.5), facecolor=colors['blocked'])314 #else:315 # print 'ERROR!', event_code, core316 317 running_dummy = Rectangle((0, 0), 1, 1, fc=colors['CPU'])318 waiting_dummy = Rectangle((0, 0), 1, 1, fc=colors['WAITING'])319 not_loaded_dummy = Rectangle((0, 0), 1, 1, fc=colors['NOT_LOAD'])320 #load_dummy = Rectangle((0, 0), 1, 1, fc=colors['LOAD'])321 blocked_dummy = Rectangle((0, 0), 1, 1, fc=colors['BLOCK'])322 legend([not_loaded_dummy,running_dummy, waiting_dummy, blocked_dummy], ['No cargada','En ejecucion','Lista','Bloqueada'],loc='best', ncol=4)323 ylim((-1,len(pids_timeline.keys())+1))324 xlim((0,max([len(x) for x in pids_timeline.values()])+1))325 tight_layout()326 fig.autofmt_xdate()327 ax.legend()328 #show()329 savefig(filename+'.png', dpi=300, format='png')330 return None331def main(argv):332 if '-c' in argv or '--caption' in argv:333 hit = '-c' if '-c' in argv else '--caption'334 pos = argv.index(hit)335 argv.pop(pos)336 caption = argv.pop(pos)337 else:338 caption = None339 if len(argv) <= 1:340 fin = sys.stdin341 fout_cores_resume= 'out_cores_resume'342 fout_cores_timeline= 'out_cores_timeline'343 fout_pids_resume= 'out_pids_resume'344 fout_pids_timeline= 'out_pids_timeline'345 else:346 fin = open(argv[1], 'r')347 preffix= argv[1]348 fout_cores_resume= preffix + '_cores_resume' 349 fout_cores_timeline= preffix + '_cores_timeline' 350 fout_pids_resume= preffix + '_pids_resume' 351 fout_pids_timeline= preffix + '_pids_timeline' 352 353 print 'parsing input'354 settings, data, cores, pids = parseInput(fin)355 356 print 'data gathering'357 cores_resume, cores_timeline, pids_resume, pids_timeline= dataGathering(data, cores, pids)358 #todo dump de los datos359 print 'drawing cores resumen'360 draw_cores_resume_bars(cores_resume, fout_cores_resume)361 print 'drawing cores timeline'362 draw_cores_timeline_gannt(cores_timeline, fout_cores_timeline)363 print 'drawing cores timeline'364 draw_tasks_resume_bars(pids_resume, fout_pids_resume)365 print 'drawing pids timeline'366 draw_pids_timeline_gannt(pids_timeline, fout_pids_timeline)367if __name__ == "__main__":...
tasks.py
Source:tasks.py
...27 t = g = utils.container_of(g['tasks']['next'],28 task_ptr_type, "tasks")29 if t == init_task:30 return31def get_task_by_pid(pid):32 for task in task_lists():33 if int(task['pid']) == pid:34 return task35 return None36class LxTaskByPidFunc(gdb.Function):37 """Find Linux task by PID and return the task_struct variable.38$lx_task_by_pid(PID): Given PID, iterate over all tasks of the target and39return that task_struct variable which PID matches."""40 def __init__(self):41 super(LxTaskByPidFunc, self).__init__("lx_task_by_pid")42 def invoke(self, pid):43 task = get_task_by_pid(pid)44 if task:45 return task.dereference()46 else:47 raise gdb.GdbError("No task of PID " + str(pid))48LxTaskByPidFunc()49class LxPs(gdb.Command):50 """Dump Linux tasks."""51 def __init__(self):52 super(LxPs, self).__init__("lx-ps", gdb.COMMAND_DATA)53 def invoke(self, arg, from_tty):54 for task in task_lists():55 gdb.write("{address} {pid} {comm}\n".format(56 address=task,57 pid=task["pid"],58 comm=task["comm"].string()))59LxPs()60thread_info_type = utils.CachedType("struct thread_info")61ia64_task_size = None62def get_thread_info(task):63 thread_info_ptr_type = thread_info_type.get_type().pointer()64 if utils.is_target_arch("ia64"):65 global ia64_task_size66 if ia64_task_size is None:67 ia64_task_size = gdb.parse_and_eval("sizeof(struct task_struct)")68 thread_info_addr = task.address + ia64_task_size69 thread_info = thread_info_addr.cast(thread_info_ptr_type)70 else:71 thread_info = task['stack'].cast(thread_info_ptr_type)72 return thread_info.dereference()73class LxThreadInfoFunc (gdb.Function):74 """Calculate Linux thread_info from task variable.75$lx_thread_info(TASK): Given TASK, return the corresponding thread_info76variable."""77 def __init__(self):78 super(LxThreadInfoFunc, self).__init__("lx_thread_info")79 def invoke(self, task):80 return get_thread_info(task)81LxThreadInfoFunc()82class LxThreadInfoByPidFunc (gdb.Function):83 """Calculate Linux thread_info from task variable found by pid84$lx_thread_info_by_pid(PID): Given PID, return the corresponding thread_info85variable."""86 def __init__(self):87 super(LxThreadInfoByPidFunc, self).__init__("lx_thread_info_by_pid")88 def invoke(self, pid):89 task = get_task_by_pid(pid)90 if task:91 return get_thread_info(task.dereference())92 else:93 raise gdb.GdbError("No task of PID " + str(pid))...
_error.py
Source:_error.py
1#!/usr/bin/env python2# Copyright (c) 2009, Giampaolo Rodola'. All rights reserved.3# Use of this source code is governed by a BSD-style license that can be4# found in the LICENSE file.5"""psutil exception classes.6Not supposed to be used / imported directly.7Instead use psutil.NoSuchProcess, etc.8"""9class Error(Exception):10 """Base exception class. All other psutil exceptions inherit11 from this one.12 """13class NoSuchProcess(Error):14 """Exception raised when a process with a certain PID doesn't15 or no longer exists (zombie).16 """17 def __init__(self, pid, name=None, msg=None):18 Error.__init__(self)19 self.pid = pid20 self.name = name21 self.msg = msg22 if msg is None:23 if name:24 details = "(pid=%s, name=%s)" % (self.pid, repr(self.name))25 else:26 details = "(pid=%s)" % self.pid27 self.msg = "process no longer exists " + details28 def __str__(self):29 return self.msg30class AccessDenied(Error):31 """Exception raised when permission to perform an action is denied."""32 def __init__(self, pid=None, name=None, msg=None):33 Error.__init__(self)34 self.pid = pid35 self.name = name36 self.msg = msg37 if msg is None:38 if (pid is not None) and (name is not None):39 self.msg = "(pid=%s, name=%s)" % (pid, repr(name))40 elif (pid is not None):41 self.msg = "(pid=%s)" % self.pid42 else:43 self.msg = ""44 def __str__(self):45 return self.msg46class TimeoutExpired(Error):47 """Raised on Process.wait(timeout) if timeout expires and process48 is still alive.49 """50 def __init__(self, pid=None, name=None):51 Error.__init__(self)52 self.pid = pid53 self.name = name54 if (pid is not None) and (name is not None):55 self.msg = "(pid=%s, name=%s)" % (pid, repr(name))56 elif (pid is not None):57 self.msg = "(pid=%s)" % self.pid58 else:59 self.msg = ""60 def __str__(self):...
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:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run fMBT 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!!
Was this article helpful?
