Add mitogen.core.now() and use it everywhere; closes #614.

Author: dw

docs/api.rst

@@ -603,6 +603,14 @@ Fork Safety
 Utility Functions
 =================
 
+.. currentmodule:: mitogen.core
+.. function:: now
+
+   A reference to :func:`time.time` on Python 2, or :func:`time.monotonic` on
+   Python >3.3. We prefer :func:`time.monotonic` when available to ensure
+   timers are not impacted by system clock changes.
+
+
 .. module:: mitogen.utils
 
 A random assortment of utility functions useful on masters and children.

mitogen/core.py

@@ -362,6 +362,10 @@ def to_text(o):
     return UnicodeType(o)
 
 
+# Documented in api.rst to work around Sphinx limitation.
+now = getattr(time, 'monotonic', time.time)
+
+
 # Python 2.4
 try:
     any
@@ -636,7 +640,7 @@ def _real_profile_hook(name, func, *args):
         return func(*args)
     finally:
         path = _profile_fmt % {
-            'now': int(1e6 * time.time()),
+            'now': int(1e6 * now()),
             'identity': name,
             'pid': os.getpid(),
             'ext': '%s'
@@ -3482,9 +3486,9 @@ def _broker_shutdown(self):
         for _, (side, _) in self.poller.readers + self.poller.writers:
             self._call(side.stream, side.stream.on_shutdown)
 
-        deadline = time.time() + self.shutdown_timeout
-        while self.keep_alive() and time.time() < deadline:
-            self._loop_once(max(0, deadline - time.time()))
+        deadline = now() + self.shutdown_timeout
+        while self.keep_alive() and now() < deadline:
+            self._loop_once(max(0, deadline - now()))
 
         if self.keep_alive():
             LOG.error('%r: pending work still existed %d seconds after '

mitogen/master.py

@@ -910,9 +910,9 @@ def _build_tuple(self, fullname):
 
         if self.minify_safe_re.search(source):
             # If the module contains a magic marker, it's safe to minify.
-            t0 = time.time()
+            t0 = mitogen.core.now()
             source = mitogen.minify.minimize_source(source).encode('utf-8')
-            self.minify_secs += time.time() - t0
+            self.minify_secs += mitogen.core.now() - t0
 
         if is_pkg:
             pkg_present = get_child_modules(path)
@@ -1001,11 +1001,11 @@ def _on_get_module(self, msg):
             LOG.warning('_on_get_module(): dup request for %r from %r',
                         fullname, stream)
 
-        t0 = time.time()
+        t0 = mitogen.core.now()
         try:
             self._send_module_and_related(stream, fullname)
         finally:
-            self.get_module_secs += time.time() - t0
+            self.get_module_secs += mitogen.core.now() - t0
 
     def _send_forward_module(self, stream, context, fullname):
         if stream.protocol.remote_id != context.context_id:

mitogen/parent.py

@@ -633,7 +633,7 @@ class TimerList(object):
     :meth:`expire`. The main user interface to :class:`TimerList` is
     :meth:`schedule`.
     """
-    _now = time.time
+    _now = mitogen.core.now
 
     def __init__(self):
         self._lst = []
@@ -1124,7 +1124,7 @@ def __init__(self, **kwargs):
 
     def on_line_received(self, line):
         self.logged_partial = None
-        self.logged_lines.append((time.time(), line))
+        self.logged_lines.append((mitogen.core.now(), line))
         self.logged_lines[:] = self.logged_lines[-100:]
         return super(LineLoggingProtocolMixin, self).on_line_received(line)
 
@@ -1134,7 +1134,7 @@ def on_partial_line_received(self, line):
 
     def on_disconnect(self, broker):
         if self.logged_partial:
-            self.logged_lines.append((time.time(), self.logged_partial))
+            self.logged_lines.append((mitogen.core.now(), self.logged_partial))
             self.logged_partial = None
         super(LineLoggingProtocolMixin, self).on_disconnect(broker)
 
@@ -1324,7 +1324,7 @@ def __init__(self, max_message_size, name=None, remote_name=None,
         self.profiling = profiling
         self.unidirectional = unidirectional
         self.max_message_size = max_message_size
-        self.connect_deadline = time.time() + self.connect_timeout
+        self.connect_deadline = mitogen.core.now() + self.connect_timeout
 
 
 class Connection(object):
@@ -1819,7 +1819,7 @@ def make_chain_id(cls):
             socket.gethostname(),
             os.getpid(),
             thread.get_ident(),
-            int(1e6 * time.time()),
+            int(1e6 * mitogen.core.now()),
         )
 
     def __repr__(self):
@@ -2569,7 +2569,7 @@ def _on_broker_shutdown(self):
     def _install_timer(self, delay):
         new = self._timer is None
         self._timer = self.broker.timers.schedule(
-            when=time.time() + delay,
+            when=mitogen.core.now() + delay,
             func=self.reap,
         )
         if new:

mitogen/service.py

@@ -1109,7 +1109,7 @@ def get(cls, context, path, out_fp):
             :meth:`fetch`.
         """
         LOG.debug('get_file(): fetching %r from %r', path, context)
-        t0 = time.time()
+        t0 = mitogen.core.now()
         recv = mitogen.core.Receiver(router=context.router)
         metadata = context.call_service(
             service_name=cls.name(),
@@ -1143,5 +1143,6 @@ def get(cls, context, path, out_fp):
                       path, metadata['size'], received_bytes)
 
         LOG.debug('target.get_file(): fetched %d bytes of %r from %r in %dms',
-                  metadata['size'], path, context, 1000 * (time.time() - t0))
+                  metadata['size'], path, context,
+                  1000 * (mitogen.core.now() - t0))
         return ok, metadata

tests/bench/fork.py

@@ -3,13 +3,13 @@
 """
 
 import mitogen
-import time
+import mitogen.core
 
 @mitogen.main()
 def main(router):
-    t0 = time.time()
+    t0 = mitogen.core.now()
     for x in xrange(200):
-        t = time.time()
+        t = mitogen.core.now()
         ctx = router.fork()
         ctx.shutdown(wait=True)
-    print '++', 1000 * ((time.time() - t0) / (1.0+x))
+    print '++', 1000 * ((mitogen.core.now() - t0) / (1.0+x))

tests/bench/large_messages.py

@@ -4,7 +4,9 @@
 import subprocess
 import time
 import socket
+
 import mitogen
+import mitogen.core
 
 
 @mitogen.main()
@@ -15,12 +17,12 @@ def main(router):
     s = ' ' * n
     print('bytes in %.2fMiB string...' % (n/1048576.0),)
 
-    t0 = time.time()
+    t0 = mitogen.core.now()
     for x in range(10):
-        tt0 = time.time()
+        tt0 = mitogen.core.now()
         assert n == c.call(len, s)
-        print('took %dms' % (1000 * (time.time() - tt0),))
-    t1 = time.time()
+        print('took %dms' % (1000 * (mitogen.core.now() - tt0),))
+    t1 = mitogen.core.now()
     print('total %dms / %dms avg / %.2fMiB/sec' % (
         1000 * (t1 - t0),
         (1000 * (t1 - t0)) / (x + 1),

tests/bench/latch_roundtrip.py

@@ -6,6 +6,7 @@
 import time
 
 import mitogen
+import mitogen.core
 import mitogen.utils
 import ansible_mitogen.affinity
 
@@ -33,8 +34,8 @@ def flip_flop(ready, inp, out):
 ready.get()
 ready.get()
 
-t0 = time.time()
+t0 = mitogen.core.now()
 l1.put(None)
 t1.join()
 t2.join()
-print('++', int(1e6 * ((time.time() - t0) / (1.0+X))), 'usec')
+print('++', int(1e6 * ((mitogen.core.now() - t0) / (1.0+X))), 'usec')

tests/bench/local.py

@@ -5,6 +5,7 @@
 import time
 
 import mitogen
+import mitogen.core
 import mitogen.utils
 import ansible_mitogen.affinity
 
@@ -15,10 +16,10 @@
 
 @mitogen.main()
 def main(router):
-    t0=time.time()
+    t0 = mitogen.core.now()
     for x in range(100):
-        t = time.time()
+        t = mitogen.core.now()
         f = router.local()# debug=True)
-        tt = time.time()
+        tt = mitogen.core.now()
         print(x, 1000 * (tt - t))
-    print('%.03f ms' % (1000 * (time.time() - t0) / (1.0 + x)))
+    print('%.03f ms' % (1000 * (mitogen.core.now() - t0) / (1.0 + x)))

tests/bench/megatime.py

@@ -4,12 +4,14 @@
 import os
 import time
 
+import mitogen.core
+
 
 times = []
 for x in range(5):
-    t0 = time.time()
+    t0 = mitogen.core.now()
     os.spawnvp(os.P_WAIT, sys.argv[1], sys.argv[1:])
-    t = time.time() - t0
+    t = mitogen.core.now() - t0
     times.append(t)
     print('+++', t)