backup/warcprox
1
0
Fork 0
mirror of https://github.com/internetarchive/warcprox.git synced 2025-01-18 13:22:09 +01:00
Code Issues Projects Releases Activity

tweaks to memory debugging

Browse Source
This commit is contained in:
Noah Levitt 2015-10-28 21:37:21 +00:00
parent 03c506dade
commit 1b8d83203c
1 changed files with 72 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
warcprox/controller.py
View File

@ -1,5 +1,3 @@
# vim: set sw=4 et:
from __future__ import absolute_import
import logging
@ -36,26 +34,30 @@ class WarcproxController(object):
self.playback_proxy = playback_proxy
self.options = options
def run_until_shutdown(self):
"""
Start warcprox and run until shut down. Call
warcprox_controller.stop.set() to initiate graceful shutdown.
"""
proxy_thread = threading.Thread(target=self.proxy.serve_forever, name='ProxyThread')
proxy_thread.start()
self.warc_writer_thread.start()
self._last_rss = None
if self.playback_proxy is not None:
playback_proxy_thread = threading.Thread(target=self.playback_proxy.serve_forever, name='PlaybackProxyThread')
playback_proxy_thread.start()
def debug_mem(self):
self.logger.info("self.proxy.recorded_url_q.qsize()=%s", self.proxy.recorded_url_q.qsize())
if self.proxy.stats_db and hasattr(self.proxy.stats_db, "_executor"):
self.logger.info("self.proxy.stats_db._executor._work_queue.qsize()=%s",
self.proxy.stats_db._executor._work_queue.qsize())
with open("/proc/self/status") as f:
for line in f:
fields = line.split()
if len(fields) >= 2:
k, v = fields[0:2]
if k == "VmHWM:":
hwm = int(v)
elif k == "VmRSS:":
rss = int(v)
elif k == "VmData:":
data = int(v)
elif k == "VmStk:":
stk = int(v)
self.logger.info("rss=%s data=%s stack=%s hwm=%s", rss, data, stk, hwm)
self._last_rss = self._last_rss or rss # to set initial value
self.stop = threading.Event()
try:
t = time.time() - 30
while not self.stop.is_set():
time.sleep(0.5)
if time.time() - t > 60:
if rss - self._last_rss > 1024:
num_unreachable = gc.collect()
all_objects = gc.get_objects()
total_size = 0
@ -77,8 +79,15 @@ class WarcproxController(object):
biggest_objects[index] = obj
self.logger.info("%s objects totaling %s bytes", len(all_objects), total_size)
self.logger.info("=== biggest types ===")
for item in sorted(summary.items(), key=lambda item: item[1]["size"], reverse=True)[:10]:
self.logger.info("%s bytes in %s instances of %s", item[1]["size"], item[1]["count"], item[0])
self.logger.info("=== warcprox types ===")
for t in (t for t in summary if str(t).find("warcprox") >= 0):
self.logger.info("%s bytes in %s instances of %s", summary[t]["size"], summary[t]["count"], t)
for i in range(len(biggest_objects)):
obj = biggest_objects[i]
try:
@ -91,6 +100,31 @@ class WarcproxController(object):
self.logger.info("#%s (%s) (%s bytes) (%s refs) (id=%s): %s", i+1, type(obj), sys.getsizeof(obj), sys.getrefcount(obj), id(obj), value)
self.logger.info("%s unreachable objects totaling %s bytes", len(gc.garbage), sum(sys.getsizeof(x) for x in gc.garbage))
self._last_rss = rss
def run_until_shutdown(self):
"""
Start warcprox and run until shut down. Call
warcprox_controller.stop.set() to initiate graceful shutdown.
"""
proxy_thread = threading.Thread(target=self.proxy.serve_forever, name='ProxyThread')
proxy_thread.start()
self.warc_writer_thread.start()
if self.playback_proxy is not None:
playback_proxy_thread = threading.Thread(target=self.playback_proxy.serve_forever, name='PlaybackProxyThread')
playback_proxy_thread.start()
self.stop = threading.Event()
self.debug_mem()
try:
t = time.time() - 30
while not self.stop.is_set():
time.sleep(0.5)
if time.time() - t > 60:
self.debug_mem()
t = time.time()
except:
self.logger.critical("fatal exception, shutting down", exc_info=True)