memory usage real time profile in python

簡單的 real time mem usage 範例, 利用 guppy 來 trace memory 的 utility.

explore memory utility test case "heapy.py"

from guppy import hpy
#import pickle
import gc
import sys
import time
import re
import pprint

#--------------------------------------
ppHP = hpy()
reTOTALSIZE = re.compile("Total\s+size\s+=\s+(\d+)\s+bytes",re.M)

def parser_mem_profile(funcnm=None,memprof=None):
    """ parser mem profile
    params:
        funcnm :  func name
        memprof:  table from hpy.heap() report
    return:
        print("funcnm","mem usage")
    """
    #print ("MEMPROF",funcnm,reTOTALSIZE.findall(memprof)[0])
    # interface for driveGnuPlotStreams.pl
    print  '0:%.3f' %(float(reTOTALSIZE.findall(memprof)[0])/100000)



def parser_gc_profile(funcnm=None,gcprof=None):
    """ parser gc profile
    params:
        funcnm :  func name
        gcprof :  table from gc.collect() report
    return:
        print("funcnm","num of obj to be collected")
    """
    #print ("GCPROF",funcnm, gcprof)
    # interface for driveGnuPlotStreams.pl
    #print '1:' + str(gcprof)



def decodrator_mem_profile(): #decorator **kwargs
    """ decorator mem usage profile collect """

    def inner(func):

        def wrapper(*args, **kwargs): # func *args, **kwargs

            retval = func(*args, **kwargs)

            def _store_mem_profile():
                """ store mem profile """
                parser_mem_profile(funcnm=str(func.__name__),\
                                   memprof=str(ppHP.heap()))

            def _store_gc_profile():
                """ store garbage collect """
                parser_gc_profile(funcnm=str(func.__name__),\
                                  gcprof=str(gc.collect()))

            _store_mem_profile()
            _store_gc_profile()

        return wrapper
    return inner

#---------------------------------------------

class Pattern(object):

    def __init__(self):
        self._list = []

    def run(self):
        self._list = [ i*200 for i in range(10000) ]

    def clear(self):
        self._list = None

ppPattern = None

@decodrator_mem_profile()
def run_pattern0(wait=0.1):
    """ run test pattern0 """
    global ppPattern
    ppPattern = Pattern()
    ppPattern.run()
    time.sleep(wait)

@decodrator_mem_profile()
def free_pattern0(wait=0.1):
    """ free test pattern0 """
    global ppPattern
    ppPattern.clear()
    del ppPattern
    time.sleep(wait)


def run_test():
    """ run all tests for mem profile """

    for i in xrange(100):
        run_pattern0()
        free_pattern0()


def main():
    """ gen the PIPE test pattern to heapy_gnuplot
    ex:
        os.system((python heapy.py) | (python heapy_gnuplot))
    """

    run_test()


if __name__== "__main__":
    main()

下載 driveGnuPlotStreams.pl 當成我們的 plot func.

ps: 利用 PIPE 的方式, 把 python heay.py IO 端的 Info 存到下個 driveGnuPlotStreams.pl 的 input.

run command

(python heapy.py ; read) | perl ./driveGnuPlotStreams.pl 1 1 50 0 50 500x300+0+0 'mem' 0

results:

Downloads:
https://docs.google.com/open?id=0B35jh2lwIpeKVVJQa0dMZHhPckE
https://docs.google.com/open?id=0B35jh2lwIpeKRkpVcnVTb1VKdnM

Refs:
heapy tutorial
http://www.smira.ru/wp-content/uploads/2011/08/heapy.html
http://guppy-pe.sourceforge.net/#Heapy

real time plot(Gnuplot)
http://users.softlab.ece.ntua.gr/~ttsiod/gnuplotStreaming.html
http://www.lysium.de/blog/index.php?/archives/234-Plotting-data-with-gnuplot-in-real-time.html

real time plot(matplotlib)
http://matplotlib.org/
install
http://matplotlib.org/users/installing.html#manually-installing-pre-built-packages

garbage collection in pyhton

key words


Mark and sweep (classic mark-and-sweep implementation)


Semispace copying
(two-arena garbage collection, copying of alive objects into the other arena
happens when the active arena is full)




Generational GC (implemented as a
subclass of the Semispace copying GC, this one adds two-generation garbage
collection to distinguish between short-lived and long-living objects)

Hy-
brid GC (adding another generation to handle large objects), Mark &
Compact GC (with in-place compaction to save space, but using multiple
passes) and the Minimark GC (a combination of the previous methods,
rewritten and with a custom allocator).



Refs:
http://en.wikipedia.org/wiki/Garbage_collection_(computer_science)
http://fcamel-life.blogspot.tw/2011/12/cpython-garbage-collection.html
http://blog.ez2learn.com/2010/04/22/memory-efficient-python-with-bytearray/
http://blogs.msdn.com/b/abhinaba/archive/2009/01/25/back-to-basic-series-on-dynamic-memory-management.aspx

multiprocess + queue in python

寫了個簡單的 verilog always block simulator. 用到 multiprocessing, task queue, decorator 的技巧.

# --*-- utf8 --*--

import timeit
from collections import defaultdict
from multiprocessing import Process, Manager
import time

# global define
TASKQUEUE  = None
PROFILE    = None
STIMULATORS = defaultdict(list)

#class

class Task(object):
    """ atomic basic element for Task assign """

    def __init__(self, func, sensitives, *args, **kwargs):
        """ init
        params:
        func :     link 2 func ptr
        sensitive : (sensitive, condition)
        time : current simulation time
        ex:
        ">>>
        @decorator_block
        def add()
        ....
        Task(func, (clk,1))
        """
        self.func = func
        self.sensitives = sensitives
        self.time = 0
        self.args = args
        self.kwargs = kwargs

    def purge(self):
        self.func = None
        self.sensitives = None
        self.time = None
        self.args = None
        self.kwargs = None

    def __repr__(self):
        return "%s(Time:%s, Func:%s, Sensitive:%s, args=%s, kwargs=%s)" \
                %(self.__class__,\
                self.time,\
                self.func,\
                self.sensitives,\
                self.args,\
                self.kwargs)

    def isUpdateable(self, other):
        """  isUpdateable if other(time) > cur(time) """
        if other > self.time:
            return True
        else:
            return False


    def update(self, time=None):
        """ update """
        self.time = time


class TaskQueue(object):
    """ register each Task in TaskQueue """

    tasks = []

    def __init__(self):
        """ init """
        self.tasks = []

    def register(self, task):
        """ add task in task queue """
        assert(isinstance(task, Task))
        self.tasks.append(task)

    def dump(self):
        """ update task in task queue """
        print self.tasks

    def purge(self):
        """ purge """
        [it.purge() for it in self.tasks if it != None]

    def get(self):
        """ get tasks """
        return self.tasks


#-------------------------------------------------------

TASKQUEUE = TaskQueue()
PROFILE   = {}

def decodrator_block(**kwargs): #decorator **kwargs
    """ decorator profile collect all blocks and register it to TaskQueue """

    global TASKQUEUE

    sensitives = list(tuple(kwargs.items()))

    def inner(func):

        def wrapper(*args, **kwargs): # func *args, **kwargs

            def _store_cProfile():
                """ store cProfile """
                t = timeit.Timer()
                retval = func(*args, **kwargs)
                PROFILE[func.__name__] = str(t.timeit())

            def _store_TaskQueue():
                """ store TaskQueue """
                task = Task(func, sensitives, *args, **kwargs)
                TASKQUEUE.register(task)

            _store_cProfile()
            _store_TaskQueue()

        return wrapper
    return inner


def getValuableTasks(trigger=None):
    """ get valuable Tasks """

    global TASKQUEUE
    fit = []

    for task in TASKQUEUE.get():
        for sensitive in task.sensitives:
            if sensitive[0] == trigger:
                fit.append(task)

    return fit



def runParallelTasks(time=None,trigger=None):
    """ run each Task when the time and trigger are matched """

    valuableTasks = getValuableTasks(trigger=trigger)

    procs = []

    for valuableTask in valuableTasks:
        p = Process(target=valuableTask.func, \
                    args=valuableTask.args, \
                    kwargs=valuableTask.kwargs)
        p.start()
        procs.append(p)

    for proc in procs:
        proc.join()



#--------------------------------------------------------
def DesignUnderTest():
    """ DUT """
# as the same as verilog always block
#>>> always@(clk1.pos) begin
#>>>    a <= a+1
#>>> end

    @decodrator_block(clk1=True, clk2=True)
    def ADD0(a):
        print "@time %0.2f decodrator_block_ADD0 %d = %d + 1" %(time.time(), a+1, a)
        return a+1
#
# as the same as verilog always block
##>>> always@(clk2.pos) begin
##>>>    b <= b*3
##>>> end

    @decodrator_block(clk2=True)
    def MUX1(b):
        print "@time %0.2f decodrator_block_MUX1 %d = %d * 3" %(time.time(), b*3, b)
        return b*3

    ADD0(3)
    MUX1(3)


def preTest():
    """ pre simulation test env  """

    global STIMULATORS

    for i in range(4):
        STIMULATORS[i*2].append("clk1")
        STIMULATORS[i*4].append("clk2")

def runTest():
    """ run simulation test env """

    times = sorted(STIMULATORS.keys())
    for time in times:
        triggers = STIMULATORS[time]
        for trigger in triggers:
            runParallelTasks(time=time,trigger=trigger)


def rptTest():
    """ report simulation test env """


if __name__ == "__main__":
    DesignUnderTest()
    preTest()
    runTest()
    rptTest()

code download : https://docs.google.com/open?id=0B35jh2lwIpeKTlpEeENPNkxlNlU

Refs:
http://docs.python.org/2/library/multiprocessing.html