Reset multiproc values when the pid changes. (#169)

This puts us at ~2000ns for an operation,
which is ~300ns than previously but still slightly
better than we were before we switched to a single lock.

This is twice as slow as without multiproc mode.

Author: brian-brazil

README.md

@@ -416,7 +416,6 @@ This comes with a number of limitations:
 - Custom collectors do not work (e.g. cpu and memory metrics)
 - The pushgateway cannot be used
 - Gauges cannot use the `pid` label
-- Gunicorn's `preload_app` feature and equivalents are not supported
 
 There's several steps to getting this working:
 

prometheus_client/core.py

@@ -392,9 +392,9 @@ def close(self):
             self._f = None
 
 
-def _MultiProcessValue(__pid=os.getpid()):
-    pid = __pid
+def _MultiProcessValue(_pidFunc=os.getpid):
     files = {}
+    values = []
     # Use a single global lock when in multi-processing mode
     # as we presume this means there is no threading going on.
     # This avoids the need to also have mutexes in __MmapDict.
@@ -406,31 +406,51 @@ class _MmapedValue(object):
         _multiprocess = True
 
         def __init__(self, typ, metric_name, name, labelnames, labelvalues, multiprocess_mode='', **kwargs):
+            self._params = typ, metric_name, name, labelnames, labelvalues, multiprocess_mode
+            with lock:
+                self.__reset()
+                values.append(self)
+
+
+        def __reset(self):
+            self._pid = _pidFunc()
+            typ, metric_name, name, labelnames, labelvalues, multiprocess_mode = self._params
             if typ == 'gauge':
                 file_prefix = typ + '_' +  multiprocess_mode
             else:
                 file_prefix = typ
-            with lock:
-                if file_prefix not in files:
-                    filename = os.path.join(
-                            os.environ['prometheus_multiproc_dir'], '{0}_{1}.db'.format(file_prefix, pid))
-                    files[file_prefix] = _MmapedDict(filename)
+            if file_prefix not in files:
+                filename = os.path.join(
+                        os.environ['prometheus_multiproc_dir'], '{0}_{1}.db'.format(file_prefix, self._pid))
+                files[file_prefix] = _MmapedDict(filename)
             self._file = files[file_prefix]
             self._key = json.dumps((metric_name, name, labelnames, labelvalues))
             self._value = self._file.read_value(self._key)
 
+        def __check_for_pid_change(self):
+            if self._pid != _pidFunc():
+                # There has been a fork(), reset all the values.
+                for f in files.values():
+                    f.close()
+                files.clear()
+                for value in values:
+                    value.__reset()
+
         def inc(self, amount):
             with lock:
+                self.__check_for_pid_change()
                 self._value += amount
                 self._file.write_value(self._key, self._value)
 
         def set(self, value):
             with lock:
+                self.__check_for_pid_change()
                 self._value = value
                 self._file.write_value(self._key, self._value)
 
         def get(self):
             with lock:
+                self.__check_for_pid_change()
                 return self._value
 
     return _MmapedValue

tests/test_multiprocess.py

@@ -13,7 +13,7 @@ class TestMultiProcess(unittest.TestCase):
     def setUp(self):
         self.tempdir = tempfile.mkdtemp()
         os.environ['prometheus_multiproc_dir'] = self.tempdir
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(123)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 123)
         self.registry = CollectorRegistry()
         MultiProcessCollector(self.registry, self.tempdir)
 
@@ -24,7 +24,7 @@ def tearDown(self):
 
     def test_counter_adds(self):
         c1 = Counter('c', 'help', registry=None)
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         c2 = Counter('c', 'help', registry=None)
         self.assertEqual(0, self.registry.get_sample_value('c'))
         c1.inc(1)
@@ -33,7 +33,7 @@ def test_counter_adds(self):
 
     def test_summary_adds(self):
         s1 = Summary('s', 'help', registry=None)
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         s2 = Summary('s', 'help', registry=None)
         self.assertEqual(0, self.registry.get_sample_value('s_count'))
         self.assertEqual(0, self.registry.get_sample_value('s_sum'))
@@ -44,7 +44,7 @@ def test_summary_adds(self):
 
     def test_histogram_adds(self):
         h1 = Histogram('h', 'help', registry=None)
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         h2 = Histogram('h', 'help', registry=None)
         self.assertEqual(0, self.registry.get_sample_value('h_count'))
         self.assertEqual(0, self.registry.get_sample_value('h_sum'))
@@ -57,7 +57,7 @@ def test_histogram_adds(self):
 
     def test_gauge_all(self):
         g1 = Gauge('g', 'help', registry=None)
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         g2 = Gauge('g', 'help', registry=None)
         self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '123'}))
         self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '456'}))
@@ -69,7 +69,7 @@ def test_gauge_all(self):
 
     def test_gauge_liveall(self):
         g1 = Gauge('g', 'help', registry=None, multiprocess_mode='liveall')
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         g2 = Gauge('g', 'help', registry=None, multiprocess_mode='liveall')
         self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '123'}))
         self.assertEqual(0, self.registry.get_sample_value('g', {'pid': '456'}))
@@ -83,7 +83,7 @@ def test_gauge_liveall(self):
 
     def test_gauge_min(self):
         g1 = Gauge('g', 'help', registry=None, multiprocess_mode='min')
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         g2 = Gauge('g', 'help', registry=None, multiprocess_mode='min')
         self.assertEqual(0, self.registry.get_sample_value('g'))
         g1.set(1)
@@ -92,7 +92,7 @@ def test_gauge_min(self):
 
     def test_gauge_max(self):
         g1 = Gauge('g', 'help', registry=None, multiprocess_mode='max')
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         g2 = Gauge('g', 'help', registry=None, multiprocess_mode='max')
         self.assertEqual(0, self.registry.get_sample_value('g'))
         g1.set(1)
@@ -101,7 +101,7 @@ def test_gauge_max(self):
 
     def test_gauge_livesum(self):
         g1 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum')
-        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(456)
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(lambda: 456)
         g2 = Gauge('g', 'help', registry=None, multiprocess_mode='livesum')
         self.assertEqual(0, self.registry.get_sample_value('g'))
         g1.set(1)
@@ -115,6 +115,20 @@ def test_namespace_subsystem(self):
          c1.inc(1)
          self.assertEqual(1, self.registry.get_sample_value('ns_ss_c'))
 
+    def test_counter_across_forks(self):
+        pid = 0
+        def get_pid():
+            return pid
+        prometheus_client.core._ValueClass = prometheus_client.core._MultiProcessValue(get_pid)
+        c1 = Counter('c', 'help', registry=None)
+        self.assertEqual(0, self.registry.get_sample_value('c'))
+        c1.inc(1)
+        c1.inc(1)
+        pid = 1
+        c1.inc(1)
+        self.assertEqual(3, self.registry.get_sample_value('c'))
+        self.assertEqual(1, c1._value.get())
+
 
 class TestMmapedDict(unittest.TestCase):
     def setUp(self):