Merge pull request #6363 from minrk/debug-heartbeat

add secondary `debug` flag for heartbeats

IPython/kernel/restarter.py

@@ -1,114 +1,111 b''
 """A basic kernel monitor with autorestarting.
 
 This watches a kernel's state using KernelManager.is_alive and auto
 restarts the kernel if it dies.
 
 It is an incomplete base class, and must be subclassed.
 """
 
-#-----------------------------------------------------------------------------
-#  Copyright (C) 2013  The IPython Development Team
-#
-#  Distributed under the terms of the BSD License.  The full license is in
-#  the file COPYING, distributed as part of this software.
-#-----------------------------------------------------------------------------
-
-#-----------------------------------------------------------------------------
-# Imports
-#-----------------------------------------------------------------------------
+# Copyright (c) IPython Development Team.
+# Distributed under the terms of the Modified BSD License.
 
 from IPython.config.configurable import LoggingConfigurable
 from IPython.utils.traitlets import (
     Instance, Float, Dict, Bool, Integer,
 )
 
-#-----------------------------------------------------------------------------
-# Code
-#-----------------------------------------------------------------------------
 
 class KernelRestarter(LoggingConfigurable):
     """Monitor and autorestart a kernel."""
 
     kernel_manager = Instance('IPython.kernel.KernelManager')
-
+    
+    debug = Bool(False, config=True,
+        help="""Whether to include every poll event in debugging output.
+        
+        Has to be set explicitly, because there will be *a lot* of output.
+        """
+    )
+    
     time_to_dead = Float(3.0, config=True,
         help="""Kernel heartbeat interval in seconds."""
     )
 
     restart_limit = Integer(5, config=True,
         help="""The number of consecutive autorestarts before the kernel is presumed dead."""
     )
     _restarting = Bool(False)
     _restart_count = Integer(0)
 
     callbacks = Dict()
     def _callbacks_default(self):
         return dict(restart=[], dead=[])
 
     def start(self):
         """Start the polling of the kernel."""
         raise NotImplementedError("Must be implemented in a subclass")
 
     def stop(self):
         """Stop the kernel polling."""
         raise NotImplementedError("Must be implemented in a subclass")
 
     def add_callback(self, f, event='restart'):
         """register a callback to fire on a particular event
 
         Possible values for event:
 
           'restart' (default): kernel has died, and will be restarted.
           'dead': restart has failed, kernel will be left dead.
 
         """
         self.callbacks[event].append(f)
 
     def remove_callback(self, f, event='restart'):
         """unregister a callback to fire on a particular event
 
         Possible values for event:
 
           'restart' (default): kernel has died, and will be restarted.
           'dead': restart has failed, kernel will be left dead.
 
         """
         try:
             self.callbacks[event].remove(f)
         except ValueError:
             pass
 
     def _fire_callbacks(self, event):
         """fire our callbacks for a particular event"""
         for callback in self.callbacks[event]:
             try:
                 callback()
             except Exception as e:
                 self.log.error("KernelRestarter: %s callback %r failed", event, callback, exc_info=True)
 
     def poll(self):
-        self.log.debug('Polling kernel...')
+        if self.debug:
+            self.log.debug('Polling kernel...')
         if not self.kernel_manager.is_alive():
             if self._restarting:
                 self._restart_count += 1
             else:
                 self._restart_count = 1
 
             if self._restart_count >= self.restart_limit:
                 self.log.warn("KernelRestarter: restart failed")
                 self._fire_callbacks('dead')
                 self._restarting = False
                 self._restart_count = 0
                 self.stop()
             else:
                 self.log.info('KernelRestarter: restarting kernel (%i/%i)',
                     self._restart_count,
                     self.restart_limit
                 )
                 self._fire_callbacks('restart')
                 self.kernel_manager.restart_kernel(now=True)
                 self._restarting = True
         else:
             if self._restarting:
                 self.log.debug("KernelRestarter: restart apparently succeeded")
             self._restarting = False

IPython/parallel/controller/heartmonitor.py

@@ -1,192 +1,193 b''
 #!/usr/bin/env python
 """
 A multi-heart Heartbeat system using PUB and ROUTER sockets. pings are sent out on the PUB,
 and hearts are tracked based on their DEALER identities.
-
-Authors:
-
-* Min RK
 """
-#-----------------------------------------------------------------------------
-#  Copyright (C) 2010-2011  The IPython Development Team
-#
-#  Distributed under the terms of the BSD License.  The full license is in
-#  the file COPYING, distributed as part of this software.
-#-----------------------------------------------------------------------------
+
+# Copyright (c) IPython Development Team.
+# Distributed under the terms of the Modified BSD License.
 
 from __future__ import print_function
 import time
 import uuid
 
 import zmq
 from zmq.devices import ThreadDevice, ThreadMonitoredQueue
 from zmq.eventloop import ioloop, zmqstream
 
 from IPython.config.configurable import LoggingConfigurable
 from IPython.utils.py3compat import str_to_bytes
-from IPython.utils.traitlets import Set, Instance, CFloat, Integer, Dict
+from IPython.utils.traitlets import Set, Instance, CFloat, Integer, Dict, Bool
 
 from IPython.parallel.util import log_errors
 
 class Heart(object):
     """A basic heart object for responding to a HeartMonitor.
     This is a simple wrapper with defaults for the most common
     Device model for responding to heartbeats.
 
     It simply builds a threadsafe zmq.FORWARDER Device, defaulting to using
     SUB/DEALER for in/out.
 
     You can specify the DEALER's IDENTITY via the optional heart_id argument."""
     device=None
     id=None
     def __init__(self, in_addr, out_addr, mon_addr=None, in_type=zmq.SUB, out_type=zmq.DEALER, mon_type=zmq.PUB, heart_id=None):
         if mon_addr is None:
             self.device = ThreadDevice(zmq.FORWARDER, in_type, out_type)
         else:
             self.device = ThreadMonitoredQueue(in_type, out_type, mon_type, in_prefix=b"", out_prefix=b"")
         # do not allow the device to share global Context.instance,
         # which is the default behavior in pyzmq > 2.1.10
         self.device.context_factory = zmq.Context
         
         self.device.daemon=True
         self.device.connect_in(in_addr)
         self.device.connect_out(out_addr)
         if mon_addr is not None:
             self.device.connect_mon(mon_addr)
         if in_type == zmq.SUB:
             self.device.setsockopt_in(zmq.SUBSCRIBE, b"")
         if heart_id is None:
             heart_id = uuid.uuid4().bytes
         self.device.setsockopt_out(zmq.IDENTITY, heart_id)
         self.id = heart_id
 
     def start(self):
         return self.device.start()
 
 
 class HeartMonitor(LoggingConfigurable):
     """A basic HeartMonitor class
     pingstream: a PUB stream
     pongstream: an ROUTER stream
     period: the period of the heartbeat in milliseconds"""
-
+    
+    debug = Bool(False, config=True,
+        help="""Whether to include every heartbeat in debugging output.
+        
+        Has to be set explicitly, because there will be *a lot* of output.
+        """
+    )
     period = Integer(3000, config=True,
         help='The frequency at which the Hub pings the engines for heartbeats '
         '(in ms)',
     )
     max_heartmonitor_misses = Integer(10, config=True,
         help='Allowed consecutive missed pings from controller Hub to engine before unregistering.',
     )
 
     pingstream=Instance('zmq.eventloop.zmqstream.ZMQStream')
     pongstream=Instance('zmq.eventloop.zmqstream.ZMQStream')
     loop = Instance('zmq.eventloop.ioloop.IOLoop')
     def _loop_default(self):
         return ioloop.IOLoop.instance()
 
     # not settable:
     hearts=Set()
     responses=Set()
     on_probation=Dict()
     last_ping=CFloat(0)
     _new_handlers = Set()
     _failure_handlers = Set()
     lifetime = CFloat(0)
     tic = CFloat(0)
 
     def __init__(self, **kwargs):
         super(HeartMonitor, self).__init__(**kwargs)
 
         self.pongstream.on_recv(self.handle_pong)
 
     def start(self):
         self.tic = time.time()
         self.caller = ioloop.PeriodicCallback(self.beat, self.period, self.loop)
         self.caller.start()
 
     def add_new_heart_handler(self, handler):
         """add a new handler for new hearts"""
         self.log.debug("heartbeat::new_heart_handler: %s", handler)
         self._new_handlers.add(handler)
 
     def add_heart_failure_handler(self, handler):
         """add a new handler for heart failure"""
         self.log.debug("heartbeat::new heart failure handler: %s", handler)
         self._failure_handlers.add(handler)
 
     def beat(self):
         self.pongstream.flush()
         self.last_ping = self.lifetime
 
         toc = time.time()
         self.lifetime += toc-self.tic
         self.tic = toc
-        self.log.debug("heartbeat::sending %s", self.lifetime)
+        if self.debug:
+            self.log.debug("heartbeat::sending %s", self.lifetime)
         goodhearts = self.hearts.intersection(self.responses)
         missed_beats = self.hearts.difference(goodhearts)
         newhearts = self.responses.difference(goodhearts)
         for heart in newhearts:
             self.handle_new_heart(heart)
         heartfailures, on_probation = self._check_missed(missed_beats, self.on_probation,
                                                          self.hearts)
         for failure in heartfailures:
             self.handle_heart_failure(failure)
         self.on_probation = on_probation
         self.responses = set()
         #print self.on_probation, self.hearts
         # self.log.debug("heartbeat::beat %.3f, %i beating hearts", self.lifetime, len(self.hearts))
         self.pingstream.send(str_to_bytes(str(self.lifetime)))
         # flush stream to force immediate socket send
         self.pingstream.flush()
 
     def _check_missed(self, missed_beats, on_probation, hearts):
         """Update heartbeats on probation, identifying any that have too many misses.
         """
         failures = []
         new_probation = {}
         for cur_heart in (b for b in missed_beats if b in hearts):
             miss_count = on_probation.get(cur_heart, 0) + 1
             self.log.info("heartbeat::missed %s : %s" % (cur_heart, miss_count))
             if miss_count > self.max_heartmonitor_misses:
                 failures.append(cur_heart)
             else:
                 new_probation[cur_heart] = miss_count
         return failures, new_probation
 
     def handle_new_heart(self, heart):
         if self._new_handlers:
             for handler in self._new_handlers:
                 handler(heart)
         else:
             self.log.info("heartbeat::yay, got new heart %s!", heart)
         self.hearts.add(heart)
 
     def handle_heart_failure(self, heart):
         if self._failure_handlers:
             for handler in self._failure_handlers:
                 try:
                     handler(heart)
                 except Exception as e:
                     self.log.error("heartbeat::Bad Handler! %s", handler, exc_info=True)
                     pass
         else:
             self.log.info("heartbeat::Heart %s failed :(", heart)
         self.hearts.remove(heart)
 
 
     @log_errors
     def handle_pong(self, msg):
         "a heart just beat"
         current = str_to_bytes(str(self.lifetime))
         last = str_to_bytes(str(self.last_ping))
         if msg[1] == current:
             delta = time.time()-self.tic
-            # self.log.debug("heartbeat::heart %r took %.2f ms to respond"%(msg[0], 1000*delta))
+            if self.debug:
+                self.log.debug("heartbeat::heart %r took %.2f ms to respond", msg[0], 1000*delta)
             self.responses.add(msg[0])
         elif msg[1] == last:
             delta = time.time()-self.tic + (self.lifetime-self.last_ping)
             self.log.warn("heartbeat::heart %r missed a beat, and took %.2f ms to respond", msg[0], 1000*delta)
             self.responses.add(msg[0])
         else:
             self.log.warn("heartbeat::got bad heartbeat (possibly old?): %s (current=%.3f)", msg[1], self.lifetime)