Tracking *maximum* memory usage by a Python function

It is possible to do this with memory_profiler. The function memory_usage returns a list of values, these represent the memory usage over time (by default over chunks of .1 second). If you need the maximum, just take the max of that list. Little example:

from memory_profiler import memory_usagefrom time import sleepdef f():    # a function that with growing    # memory consumption    a = [0] * 1000    sleep(.1)    b = a * 100    sleep(.1)    c = b * 100    return amem_usage = memory_usage(f)print('Memory usage (in chunks of .1 seconds): %s' % mem_usage)print('Maximum memory usage: %s' % max(mem_usage))

In my case (memory_profiler 0.25) if prints the following output:

Memory usage (in chunks of .1 seconds): [45.65625, 45.734375, 46.41015625, 53.734375]Maximum memory usage: 53.734375

This question seemed rather interesting and it gave me a reason to look into Guppy / Heapy, for that I thank you.

I tried for about 2 hours to get Heapy to do monitor a function call / process without modifying its source with zero luck.

I did find a way to accomplish your task using the built in Python library resource. Note that the documentation does not indicate what the RU_MAXRSS value returns. Another SO user noted that it was in kB. Running Mac OSX 7.3 and watching my system resources climb up during the test code below, I believe the returned values to be in Bytes, not kBytes.

A 10000ft view on how I used the resource library to monitor the library call was to launch the function in a separate (monitor-able) thread and track the system resources for that process in the main thread. Below I have the two files that you'd need to run to test it out.

Library Resource Monitor - whatever_you_want.py

import resourceimport timefrom stoppable_thread import StoppableThreadclass MyLibrarySniffingClass(StoppableThread):    def __init__(self, target_lib_call, arg1, arg2):        super(MyLibrarySniffingClass, self).__init__()        self.target_function = target_lib_call        self.arg1 = arg1        self.arg2 = arg2        self.results = None    def startup(self):        # Overload the startup function        print "Calling the Target Library Function..."    def cleanup(self):        # Overload the cleanup function        print "Library Call Complete"    def mainloop(self):        # Start the library Call        self.results = self.target_function(self.arg1, self.arg2)        # Kill the thread when complete        self.stop()def SomeLongRunningLibraryCall(arg1, arg2):    max_dict_entries = 2500    delay_per_entry = .005    some_large_dictionary = {}    dict_entry_count = 0    while(1):        time.sleep(delay_per_entry)        dict_entry_count += 1        some_large_dictionary[dict_entry_count]=range(10000)        if len(some_large_dictionary) > max_dict_entries:            break    print arg1 + " " +  arg2    return "Good Bye World"if __name__ == "__main__":    # Lib Testing Code    mythread = MyLibrarySniffingClass(SomeLongRunningLibraryCall, "Hello", "World")    mythread.start()    start_mem = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss    delta_mem = 0    max_memory = 0    memory_usage_refresh = .005 # Seconds    while(1):        time.sleep(memory_usage_refresh)        delta_mem = (resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) - start_mem        if delta_mem > max_memory:            max_memory = delta_mem        # Uncomment this line to see the memory usuage during run-time         # print "Memory Usage During Call: %d MB" % (delta_mem / 1000000.0)        # Check to see if the library call is complete        if mythread.isShutdown():            print mythread.results            break;    print "\nMAX Memory Usage in MB: " + str(round(max_memory / 1000.0, 3))

Stoppable Thread - stoppable_thread.py

import threadingimport timeclass StoppableThread(threading.Thread):    def __init__(self):        super(StoppableThread, self).__init__()        self.daemon = True        self.__monitor = threading.Event()        self.__monitor.set()        self.__has_shutdown = False    def run(self):        '''Overloads the threading.Thread.run'''        # Call the User's Startup functions        self.startup()        # Loop until the thread is stopped        while self.isRunning():            self.mainloop()        # Clean up        self.cleanup()        # Flag to the outside world that the thread has exited        # AND that the cleanup is complete        self.__has_shutdown = True    def stop(self):        self.__monitor.clear()    def isRunning(self):        return self.__monitor.isSet()    def isShutdown(self):        return self.__has_shutdown    ###############################    ### User Defined Functions ####    ###############################    def mainloop(self):        '''        Expected to be overwritten in a subclass!!        Note that Stoppable while(1) is handled in the built in "run".        '''        pass    def startup(self):        '''Expected to be overwritten in a subclass!!'''        pass    def cleanup(self):        '''Expected to be overwritten in a subclass!!'''        pass

This appears to work under Windows. Don't know about other operating systems.

In [50]: import osIn [51]: import psutilIn [52]: process = psutil.Process(os.getpid())In [53]: process.get_ext_memory_info().peak_wsetOut[53]: 41934848