Memory leak tools + rewrote & tested config merger

Author: sylvinus

Makefile

@@ -10,6 +10,9 @@ test_jenkins: docker
 ssh:
 	sh -c "docker run -rm -i -t -p 27017:27017 -p 6379:6379 -p 5555:5555 -p 20020:20020 -v `pwd`:/app:rw -w /app mrq/mrq_local bash"
 
+ssh_noport:
+	sh -c "docker run -rm -i -t -v `pwd`:/app:rw -w /app mrq/mrq_local bash"
+
 lint:
 	pylint --init-hook="import sys; sys.path.append('.')" --rcfile .pylintrc mrq
 

README.md

@@ -26,12 +26,27 @@ The main features of MRQ are:
  * **Thorough testing:** Edge-cases like worker interrupts, Redis failures, ... are tested inside a Docker container.
  * **Builtin scheduler:** Schedule tasks by interval or by time of the day
  * **Greenlet tracing:** See how much time was spent in each greenlet to debug CPU-intensive jobs.
+ * **Integrated memory leak debugger:** Track down jobs leaking memory and find the leaks with objgraph.
 
 Performance
 ===========
 
 On a MacbookPro, we see 1300 jobs/second in a single worker process with very simple jobs that store results, to measure the overhead of MRQ. However what we are really measuring there is MongoDB's write performance.
 
+Hunting memory leaks
+====================
+
+Memory leaks can be a big issue with gevent workers because several tasks share the same python process.
+
+Thankfully, MRQ provides tools to track down such issues. Memory usage of each worker is graphed in the dashboard and makes it easy to see if memory leaks are happening.
+
+When a worker has a steadily growing memory usage, here are the steps to find the leak:
+
+ * Check which jobs are running on this worker and try to isolate which of them is leaking and on which queue
+ * Start a dedicated worker with ```--trace_memory --gevent 1``` on the same queue : This will start a worker doing one job at a time with memory profiling enabled. After each job you should see a report of leaked object types.
+ * Find the most unique type in the list (usually not 'list' or 'dict') and restart the worker with ```--trace_memory --gevent 1 --trace_memory_type=XXX --trace_memory_output_dir=memdbg``` (after creating the directory memdbg).
+ * There you will find a graph for each task generated by [objgraph](https://mg.pov.lt/objgraph/) which is incredibly helpful to track down the leak.
+
 Tests
 =====
 

mrq/basetasks/tests/general.py

@@ -1,6 +1,6 @@
 from time import sleep
 from mrq.task import Task
-from mrq.context import log, retry_current_job, connections
+from mrq.context import log, retry_current_job, connections, get_current_config
 import urllib2
 
 
@@ -37,6 +37,11 @@ def run(self, params):
     return len(t)
 
 
+class GetConfig(Task):
+  def run(self, params):
+    return get_current_config()
+
+
 LEAKS = []
 
 
@@ -92,5 +97,10 @@ def run(self, params):
 
     connections.mongodb_logs.tests_inserts.insert(params)
 
+    if params.get("sleep", 0) > 0:
+      sleep(params.get("sleep", 0))
+
+    return params
+
 
 MongoInsert2 = MongoInsert

mrq/config.py

@@ -3,7 +3,7 @@
 import sys
 
 
-def get_config(sources=("file", "env", "args"), env_prefix="MRQ_", defaults=None):
+def get_config(sources=("file", "env", "args"), env_prefix="MRQ_"):
 
   parser = argparse.ArgumentParser(description='Starts an RQ worker')
 
@@ -89,46 +89,53 @@ def get_config(sources=("file", "env", "args"), env_prefix="MRQ_", defaults=None
   parser.add_argument('queues', nargs='*', default=["default"],
                       help='The queues to listen on (default: \'default\')')
 
-  if "args" in sources:
-    from_args = parser.parse_args()
-  else:
-    from_args = parser.parse_args([])
-
-  # Get defaults
-  merged_config = from_args.__dict__
-  if defaults is not None:
-    merged_config.update(defaults)
-
-  # If a mrq-config.py file is in the current directory, use it!
-  default_config_file = os.path.join(os.getcwd(), "mrq-config.py")
-
-  if merged_config["config"] is None and os.path.isfile(default_config_file):
-    # print "Using config file at %s" % default_config_file
-    merged_config["config"] = default_config_file
-
-  config_module = None
-  if "file" in sources and merged_config["config"]:
-    sys.path.append(os.path.dirname(merged_config["config"]))
-    config_module = __import__(os.path.basename(merged_config["config"].replace(".py", "")))
-    sys.path.pop(-1)
-    merged_config.update({k.lower(): v for k, v in config_module.__dict__.iteritems() if k[0].isupper()})
+  default_config = parser.parse_args([]).__dict__
 
   # Keys that can't be passed from the command line
-  merged_config["tasks"] = {}
-  merged_config["scheduled_tasks"] = {}
+  default_config["tasks"] = {}
+  default_config["scheduled_tasks"] = {}
+
+  # Only keep values different from config, actually passed on the command line
+  from_args = {}
+  if "args" in sources:
+    for k, v in parser.parse_args().__dict__.iteritems():
+      if default_config[k] != v:
+        from_args[k] = v
 
+  # If we were given another config file, use it
+  if from_args.get("config"):
+    config_file = from_args.get("config")
+  # If a mrq-config.py file is in the current directory, use it!
+  elif os.path.isfile(os.path.join(os.getcwd(), "mrq-config.py")):
+    config_file = os.path.join(os.getcwd(), "mrq-config.py")
+  else:
+    config_file = None
+
+  from_file = {}
+  if config_file and "file" in sources:
+    sys.path.insert(0, os.path.dirname(config_file))
+    config_module = __import__(os.path.basename(config_file.replace(".py", "")))
+    sys.path.pop(0)
+    for k, v in config_module.__dict__.iteritems():
+
+      # We only keep variables starting with an uppercase character.
+      if k[0].isupper():
+        default_config[k.lower()] = v
+        if k.lower() not in default_config:
+          default_config[k.lower()] = v
+
+  # Merge the config in the order given by the user
+  merged_config = default_config
   for part in sources:
     for name, arg_value in merged_config.iteritems():
 
-      value = None
       if part == "env":
         value = os.environ.get(env_prefix + name.upper())
-      elif part == "args":
-        value = arg_value
-      elif part == "file":
-        value = getattr(config_module, name.upper(), None)
-
-      if value is not None:
-        merged_config[name] = value
+        if value:
+          merged_config[name] = value
+      elif part == "args" and name in from_args:
+        merged_config[name] = from_args[name]
+      elif part == "file" and name in from_file:
+        merged_config[name] = from_file[name]
 
   return merged_config

mrq/worker.py

@@ -338,6 +338,7 @@ def work_loop(self):
             # When debugging memory, intermediate psutils call like this one are
             # needed for some obscure reason. (tested in test_memoryleaks.py)
             self.get_memory()
+            gevent.sleep(0.1)
 
           free_pool_slots = self.gevent_pool.free_count()
 

requirements-base.txt

@@ -6,4 +6,5 @@ ujson==1.33
 hiredis==0.1.2
 psutil==1.2.1
 supervisor==3.0
-subprocess32==3.2.5
+subprocess32==3.2.5
+objgraph==1.8.0

tests/fixtures/config2.py

@@ -0,0 +1,4 @@
+NAME = "testworker"
+
+# There configs should be added transparently.
+ADDITIONAL_UNEXPECTED_CONFIG = "1"

tests/test_cancel.py

@@ -6,15 +6,15 @@ def test_cancel_by_path(worker):
   # Start the worker with only one greenlet so that tasks execute sequentially
   worker.start(flags="--gevent 1")
 
-  job_id1 = worker.send_task("mrq.basetasks.tests.general.Add", {"a": 41, "b": 1, "sleep": 2}, block=False)
+  job_id1 = worker.send_task("mrq.basetasks.tests.general.MongoInsert", {"a": 41, "sleep": 2}, block=False)
 
   worker.send_task("mrq.basetasks.utils.JobAction", {
     "path": "mrq.basetasks.tests.general.Add",
     "status": "queued",
     "action": "cancel"
   }, block=False)
 
-  job_id2 = worker.send_task("mrq.basetasks.tests.general.Add", {"a": 41, "b": 2}, block=False)
+  job_id2 = worker.send_task("mrq.basetasks.tests.general.MongoInsert", {"a": 43}, block=False)
 
   Job(job_id2).wait(poll_interval=0.01)
   worker.stop(deps=False)
@@ -23,7 +23,9 @@ def test_cancel_by_path(worker):
   job2 = Job(job_id2).fetch().data
 
   assert job1["status"] == "success"
-  assert job1["result"] == 42
+  assert job1["result"] == {"a": 41, "sleep": 2}
 
   assert job2["status"] == "cancel"
   assert job2.get("result") is None
+
+  assert worker.mongodb_logs.test_inserts.count() == 1

tests/test_config.py

@@ -0,0 +1,31 @@
+
+
+def test_config(worker):
+  """ Test different config passing options. """
+
+  worker.start()
+
+  cfg = worker.send_task("mrq.basetasks.tests.general.GetConfig", {}, block=True)
+
+  assert "mongodb_jobs" in cfg
+  assert cfg.get("additional_unexpected_config") is None
+
+  worker.stop()
+
+  worker.start(flags="--config tests/fixtures/config2.py")
+
+  cfg = worker.send_task("mrq.basetasks.tests.general.GetConfig", {}, block=True)
+
+  assert cfg["name"] == "testworker"
+  assert cfg.get("additional_unexpected_config") == "1"
+
+  worker.stop()
+
+  worker.start(flags="--config tests/fixtures/config2.py --name xxx")
+
+  cfg = worker.send_task("mrq.basetasks.tests.general.GetConfig", {}, block=True)
+
+  assert cfg["name"] == "xxx"
+  assert cfg.get("additional_unexpected_config") == "1"
+
+  worker.stop()

tests/test_memoryleaks.py

@@ -58,7 +58,7 @@ def test_memoryleaks_1mleak(worker):
   worker.mongodb_jobs.mrq_jobs.remove()
 
   # 1M leak!
-  diff1m = get_diff_after_jobs(worker, 10, 100000, sleep=0.05)  # sleep is needed so that psutil measurements are accurate :-/
+  diff1m = get_diff_after_jobs(worker, 10, 100000, sleep=0.2)  # sleep is needed so that psutil measurements are accurate :-/
 
   assert diff1m > 900000