<p>这个问题似乎很有趣,它给了我一个研究古比/希比的理由,为此我感谢你。</p>
<p>我试了大约两个小时让Heapy监视一个函数调用/进程,但没有用<em>zero</em>luck修改它的源代码。</p>
<p>我确实找到了使用内置Python库<a href="http://docs.python.org/library/resource.html#resource-usage" rel="noreferrer">^{<cd1>}</a>完成任务的方法。注意,文档并没有指明<code>RU_MAXRSS</code>值返回的内容。另一个用户<a href="https://stackoverflow.com/a/7669482/157744">noted</a>是以kB为单位的。运行Mac OSX 7.3并在下面的测试代码中观察我的系统资源爬升,我相信返回的值是字节而不是千字节。</p>
<p>关于我如何使用<code>resource</code>库监视库调用的10000英尺视图是在一个单独的(可监视的)线程中启动该函数,并在主线程中跟踪该进程的系统资源。下面我有两个文件,你需要运行来测试它。</p>
<p><strong>库资源监视器</strong>-<a href="https://gist.github.com/raw/b54fafd87634f017d50d/e101d619bfbc9054ad129fe5141e0c584ac35577/whatever_you_want.py" rel="noreferrer">whatever_you_want.py</a></p>
<pre><code>import resource
import time
from stoppable_thread import StoppableThread
class 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))
</code></pre>
<p><strong>可停止线程</strong>-<a href="https://gist.github.com/raw/b54fafd87634f017d50d/bed2a7c2eafed67f05ee3dce926e73ac0f8e2f9c/stoppable_thread.py" rel="noreferrer">stoppable_thread.py</a></p>
<pre><code>import threading
import time
class 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
</code></pre>