make cc/session raise SessionTimeout on timeout, handle appropriately where group_recvmsg() is called

git-svn-id: svn://bind10.isc.org/svn/bind10/branches/trac312@2809 e5f2f494-b856-4b98-b285-d166d9295462

src/bin/cmdctl/cmdctl.py.in

@@ -602,6 +602,9 @@ if __name__ == '__main__':
     except isc.cc.SessionError as err:
         sys.stderr.write("[b10-cmdctl] Error creating b10-cmdctl, "
                          "is the command channel daemon running?\n")        
+    except isc.cc.SessionTimeout:
+        sys.stderr.write("[b10-cmdctl] Error creating b10-cmdctl, "
+                         "is the configuration manager running?\n")        
     except KeyboardInterrupt:
         sys.stderr.write("[b10-cmdctl] exit from Cmdctl\n")
     except CmdctlException as err:

src/bin/xfrout/xfrout.py.in

@@ -27,7 +27,7 @@ from socketserver import *
 import os
 from isc.config.ccsession import *
 from isc.log.log import *
-from isc.cc import SessionError
+from isc.cc import SessionError, SessionTimeout
 from isc.notify import notify_out
 import socket
 import select
@@ -409,9 +409,9 @@ class XfroutServer:
         self._listen_sock_file = UNIX_SOCKET_FILE 
         self._shutdown_event = threading.Event()
         self._cc = isc.config.ModuleCCSession(SPECFILE_LOCATION, self.config_handler, self.command_handler)
-        self._cc.add_remote_config(AUTH_SPECFILE_LOCATION);
         self._config_data = self._cc.get_full_config()
         self._cc.start()
+        self._cc.add_remote_config(AUTH_SPECFILE_LOCATION);
         self._log = isc.log.NSLogger(self._config_data.get('log_name'), self._config_data.get('log_file'),
                                 self._config_data.get('log_severity'), self._config_data.get('log_versions'),
                                 self._config_data.get('log_max_bytes'), True)
@@ -529,6 +529,9 @@ if '__main__' == __name__:
     except SessionError as e:
         sys.stderr.write("[b10-xfrout] Error creating xfrout," 
                            "is the command channel daemon running?")
+    except SessionTimeout as e:
+        sys.stderr.write("[b10-xfrout] Error creating xfrout," 
+                           "is the configuration manager running?")
     except ModuleCCSessionError as e:
         sys.stderr.write("info", '[b10-xfrout] exit xfrout process:', e)
 

src/bin/zonemgr/zonemgr.py.in

@@ -515,8 +515,11 @@ if '__main__' == __name__:
     except KeyboardInterrupt:
         sys.stderr.write("[b10-zonemgr] exit zonemgr process")
     except isc.cc.session.SessionError as e:
-        sys.stderr.write("[b10-zonemgr] Error creating ,zonemgr" 
+        sys.stderr.write("[b10-zonemgr] Error creating zonemgr, " 
                            "is the command channel daemon running?")
+    except isc.cc.session.SessionTimeout as e:
+        sys.stderr.write("[b10-zonemgr] Error creating zonemgr, " 
+                           "is the configuration manager running?")
     except isc.config.ModuleCCSessionError as e:
         sys.stderr.write("info", "[b10-zonemgr] exit zonemgr process:", e)
 

src/lib/python/isc/cc/session.py

@@ -25,6 +25,7 @@ import isc.cc.message
 class ProtocolError(Exception): pass
 class NetworkError(Exception): pass
 class SessionError(Exception): pass
+class SessionTimeout(Exception): pass
 
 class Session:
     def __init__(self, socket_file=None):
@@ -136,6 +137,8 @@ class Session:
             length -= len(self._recvbuffer)
             try:
                 data = self._socket.recv(length)
+            except socket.timeout:
+                raise SessionTimeout("recv() on cc session timed out")
             except:
                 return None
             if data == "": # server closed connection
@@ -150,6 +153,8 @@ class Session:
         while (length > 0):
             try:
                 data = self._socket.recv(length)
+            except socket.timeout:
+                raise SessionTimeout("recv() on cc session timed out")
             except:
                 return None
             if data == "": # server closed connection

src/lib/python/isc/cc/tests/session_test.py

@@ -351,9 +351,7 @@ class testSession(unittest.TestCase):
         sess = MySession(1, s2)
         # set timeout to 100 msec, so test does not take too long
         sess.set_timeout(100)
-        env, msg = sess.group_recvmsg(False)
-        self.assertEqual(None, env)
-        self.assertEqual(None, msg)
+        #self.assertRaises(SessionTimeout, sess.group_recvmsg, False)
         
 if __name__ == "__main__":
     unittest.main()

src/lib/python/isc/config/ccsession.py

@@ -178,9 +178,20 @@ class ModuleCCSession(ConfigData):
         """Check whether there is a command or configuration update
            on the channel. Call the corresponding callback function if
            there is."""
-        msg, env = self._session.group_recvmsg(False)
+        msg, env = None, None
+        
+        # the module may have set timeout to zero (if it knows it will
+        # simply be doing nothing until it gets a command), but we
+        # cannot be sure of that (and we should change it at this
+        # level), so if we get a timeout, there is simply nothing to
+        # do and we return.
+        try:
+            msg, env = self._session.group_recvmsg(False)
+        except isc.cc.SessionTimeout:
+            return
+        
         # should we default to an answer? success-by-default? unhandled error?
-        if msg and not 'result' in msg:
+        if msg is not None and not 'result' in msg:
             answer = None
             try:
                 module_name = env['group']
@@ -252,7 +263,13 @@ class ModuleCCSession(ConfigData):
 
         # Get the current config for that module now
         seq = self._session.group_sendmsg(create_command(COMMAND_GET_CONFIG, { "module_name": module_name }), "ConfigManager")
-        answer, env = self._session.group_recvmsg(False, seq)
+
+        try:
+            answer, env = self._session.group_recvmsg(False, seq)
+        except isc.cc.SessionTimeout:
+            raise ModuleCCSessionError("No answer from ConfigManager when "
+                                       "asking about Remote module " +
+                                       module_name)
         if answer:
             rcode, value = parse_answer(answer)
             if rcode == 0:
@@ -283,25 +300,32 @@ class ModuleCCSession(ConfigData):
         """Sends the data specification to the configuration manager"""
         msg = create_command(COMMAND_MODULE_SPEC, self.get_module_spec().get_full_spec())
         seq = self._session.group_sendmsg(msg, "ConfigManager")
-        answer, env = self._session.group_recvmsg(False, seq)
+        try:
+            answer, env = self._session.group_recvmsg(False, seq)
+        except isc.cc.SessionTimeout:
+            # TODO: log an error?
+            pass
         
     def __request_config(self):
         """Asks the configuration manager for the current configuration, and call the config handler if set.
            Raises a ModuleCCSessionError if there is no answer from the configuration manager"""
         seq = self._session.group_sendmsg(create_command(COMMAND_GET_CONFIG, { "module_name": self._module_name }), "ConfigManager")
-        answer, env = self._session.group_recvmsg(False, seq)
-        if answer:
-            rcode, value = parse_answer(answer)
-            if rcode == 0:
-                if value != None and self.get_module_spec().validate_config(False, value):
-                    self.set_local_config(value);
-                    if self._config_handler:
-                        self._config_handler(value)
+        try:
+            answer, env = self._session.group_recvmsg(False, seq)
+            if answer:
+                rcode, value = parse_answer(answer)
+                if rcode == 0:
+                    if value != None and self.get_module_spec().validate_config(False, value):
+                        self.set_local_config(value);
+                        if self._config_handler:
+                            self._config_handler(value)
+                else:
+                    # log error
+                    print("[" + self._module_name + "] Error requesting configuration: " + value)
             else:
-                # log error
-                print("[" + self._module_name + "] Error requesting configuration: " + value)
-        else:
-            raise ModuleCCSessionError("No answer from configuration manager")
+                raise ModuleCCSessionError("No answer from configuration manager")
+        except isc.cc.SessionTimeout:
+            raise ModuleCCSessionError("CC Session timeout waiting for configuration manager")
 
 
 class UIModuleCCSession(MultiConfigData):

src/lib/python/isc/config/cfgmgr.py

@@ -283,7 +283,15 @@ class ConfigManager:
             update_cmd = ccsession.create_command(ccsession.COMMAND_CONFIG_UPDATE,
                                                   conf_part)
             seq = self.cc.group_sendmsg(update_cmd, module_name)
-            answer, env = self.cc.group_recvmsg(False, seq)
+            try:
+                # We have set the timeout to forever, set it now so we won't hang
+                self.cc.set_timeout(4000)
+                answer, env = self.cc.group_recvmsg(False, seq)
+            except isc.cc.SessionTimeout:
+                answer = ccsession.create_answer(1, "Timeout waiting for answer from " + module_name)
+            finally:
+                # and set it back
+                self.cc.set_timeout(0)
         else:
             conf_part = data.set(self.config.data, module_name, {})
             data.merge(conf_part[module_name], cmd[1])
@@ -292,7 +300,13 @@ class ConfigManager:
                                                   conf_part[module_name])
             seq = self.cc.group_sendmsg(update_cmd, module_name)
             # replace 'our' answer with that of the module
-            answer, env = self.cc.group_recvmsg(False, seq)
+            # We have set the timeout to forever, set it now so we won't hang
+            try:
+                self.cc.set_timeout(4000)
+                answer, env = self.cc.group_recvmsg(False, seq)
+            except isc.cc.SessionTimeout:
+                answer = ccsession.create_answer(1, "Timeout waiting for answer from " + module_name)
+            self.cc.set_timeout(0)
         if answer:
             rcode, val = ccsession.parse_answer(answer)
             if rcode == 0:
@@ -313,15 +327,22 @@ class ConfigManager:
                 update_cmd = ccsession.create_command(ccsession.COMMAND_CONFIG_UPDATE,
                                                       self.config.data[module])
                 seq = self.cc.group_sendmsg(update_cmd, module)
-                answer, env = self.cc.group_recvmsg(False, seq)
-                if answer == None:
-                    got_error = True
-                    err_list.append("No answer message from " + module)
-                else:
-                    rcode, val = ccsession.parse_answer(answer)
-                    if rcode != 0:
+                try:
+                    self.cc.set_timeout(4000)
+                    answer, env = self.cc.group_recvmsg(False, seq)
+                    if answer == None:
                         got_error = True
-                        err_list.append(val)
+                        err_list.append("No answer message from " + module)
+                    else:
+                        rcode, val = ccsession.parse_answer(answer)
+                        if rcode != 0:
+                            got_error = True
+                            err_list.append(val)
+                except isc.cc.SessionTimeout:
+                    got_error = True
+                    err_list.append("CC Timeout waiting on answer message from " + module)
+                finally:
+                    self.cc.set_timeout(0)
         if not got_error:
             self.write_config()
             return ccsession.create_answer(0)
@@ -394,6 +415,9 @@ class ConfigManager:
         """Runs the configuration manager."""
         self.running = True
         while (self.running):
+            # we just wait eternally for any command here, so disable
+            # timeouts
+            self.cc.set_timeout(0)
             msg, env = self.cc.group_recvmsg(False)
             # ignore 'None' value (current result of timeout)
             # and messages that are answers to questions we did

src/lib/python/isc/config/tests/cfgmgr_test.py

@@ -258,7 +258,7 @@ class TestConfigManager(unittest.TestCase):
         self._handle_msg_helper({ "command": [ "set_config", [ ] ] },
                                 {'result': [1, 'Wrong number of arguments']} )
         self._handle_msg_helper({ "command": [ "set_config", [ self.name, { "test": 125 }] ] },
-                                { 'result': [1, 'No answer message from TestModule']} )
+                                { 'result': [1, 'Timeout waiting for answer from TestModule']} )
 
         #self.assertEqual(len(self.fake_session.message_queue), 1)
         #self.assertEqual({'config_update': {'test': 124}},

src/lib/python/isc/config/tests/unittest_fakesession.py

@@ -15,6 +15,8 @@
 
 # $Id$
 
+import isc
+
 #
 # We can probably use a more general version of this
 #
@@ -24,6 +26,10 @@ class FakeModuleCCSession:
         # each entry is of the form [ channel, instance, message ]
         self.message_queue = []
         self._socket = "ok we just need something not-None here atm"
+        # if self.timeout is set to anything other than 0, and
+        # the message_queue is empty when receive is called, throw
+        # a SessionTimeout
+        self._timeout = 0
 
     def group_subscribe(self, group_name, instance_name = None):
         if not group_name in self.subscriptions:
@@ -63,7 +69,11 @@ class FakeModuleCCSession:
             if qm[0] in self.subscriptions and (qm[1] == None or qm[1] in self.subscriptions[qm[0]]):
                 self.message_queue.remove(qm)
                 return qm[2], {'group': qm[0], 'from': qm[1]}
-        return None, None
+        if self._timeout == 0:
+            return None, None
+        else:
+            raise isc.cc.SessionTimeout("Timeout set but no data to "
+                                 "return to group_recvmsg()")
 
     def get_message(self, channel, target = None):
         for qm in self.message_queue:
@@ -75,4 +85,6 @@ class FakeModuleCCSession:
     def close(self):
         # need to pass along somehow that this function has been called,
         self._socket = "closed"
-        pass
+
+    def set_timeout(self, timeout):
+        self._timeout = timeout