Always closing FCGI sockets when ref count hits zero - addresses issue with unclosed sockets on reload

src/supervisor/process.py

@@ -640,12 +640,6 @@ class ProcessGroupBase:
         for process in self.processes.values():
             process.reopenlogs()
 
-    def stop_requested(self):
-        """ Hook so that the process group gets notified by
-            it's geting stopped by an RPC interface call
-        """
-        pass
-
     def stop_all(self):
         processes = self.processes.values()
         processes.sort()
@@ -693,15 +687,6 @@ class FastCGIProcessGroup(ProcessGroup):
         except Exception, e:
             raise ValueError('Could not create FastCGI socket %s: %s' % (self.socket_manager.config(), e))
 
-    def stop_requested(self):
-        """ Overriden from ProcessGroup
-            Request close on FCGI socket (it will actually be close when all
-            the child processes are reaped)
-        """
-        self.config.options.logger.debug('Stop requested for fcgi group %s' 
-                                         % self.config.name)
-        self.socket_manager.request_close()
-
 class EventListenerPool(ProcessGroupBase):
     def __init__(self, config):
         ProcessGroupBase.__init__(self, config)

src/supervisor/rpcinterface.py

@@ -427,8 +427,6 @@ class SupervisorNamespaceRPCInterface:
         if group is None:
             raise RPCError(Faults.BAD_NAME, name)
 
-        group.stop_requested()
-
         processes = group.processes.values()
         processes.sort()
         processes = [ (group, process) for process in processes ]

src/supervisor/socket_manager.py

@@ -69,8 +69,7 @@ class SocketManager:
         self.socket = None
         self.prepared = False
         self.socket_config = socket_config
-        self.close_requested = False
-        self.ref_ctr = ReferenceCounter(on_zero=self._on_ref_ct_zero, on_non_zero=self._prepare_socket)
+        self.ref_ctr = ReferenceCounter(on_zero=self._close, on_non_zero=self._prepare_socket)
         
     def __repr__(self):
         return '<%s at %s for %s>' % (self.__class__,
@@ -91,14 +90,7 @@ class SocketManager:
     def get_socket_ref_count(self):
         self._require_prepared()
         return self.ref_ctr.get_count()
-    
-    def request_close(self):
-        if self.prepared:
-            if self.ref_ctr.get_count() == 0:
-                self._close()
-            else:
-                self.close_requested = True
-    
+        
     def _require_prepared(self):
         if not self.prepared:
             raise Exception('Socket has not been prepared')
@@ -110,12 +102,7 @@ class SocketManager:
             self.socket = self.socket_config.create_and_bind()
             self.socket.listen(socket.SOMAXCONN)
             self.prepared = True
-    
-    def _on_ref_ct_zero(self):
-        if self.close_requested:
-            self.close_requested = False
-            self._close()
-    
+
     def _close(self):
         self._require_prepared()
         if self.logger:

src/supervisor/tests/base.py

@@ -339,7 +339,6 @@ class DummySocketConfig:
 class DummySocketManager:
     def __init__(self, config, **kwargs):
         self._config = config
-        self.request_close_called = False
 
     def config(self):
         return self._config
@@ -347,9 +346,6 @@ class DummySocketManager:
     def get_socket(self):
         return DummySocket(self._config.fd)
         
-    def request_close(self):
-        self.request_close_called = True
-
 class DummyProcess:
     # Initial state; overridden by instance variables
     pid = 0 # Subprocess pid; 0 when not running
@@ -929,7 +925,6 @@ class DummyProcessGroup:
         self.all_stopped = False
         self.dispatchers = {}
         self.unstopped_processes = []
-        self.stop_was_requested = False
 
     def transition(self):
         self.transitioned = True
@@ -937,9 +932,6 @@ class DummyProcessGroup:
     def stop_all(self):
         self.all_stopped = True
         
-    def stop_requested(self):
-        self.stop_was_requested = True
-
     def get_unstopped_processes(self):
         return self.unstopped_processes
 

src/supervisor/tests/test_process.py

@@ -1418,22 +1418,7 @@ class ProcessGroupTests(ProcessGroupBaseTests):
         group.processes = {'process1': process1}
         group.transition()
         self.assertEqual(process1.transitioned, True)
-        
-class FastCGIProcessGroupTests(unittest.TestCase):
-    def _getTargetClass(self):
-        from supervisor.process import FastCGIProcessGroup
-        return FastCGIProcessGroup
-        
-    def _makeOne(self, *args, **kw):
-        return self._getTargetClass()(*args, **kw)
-        
-    def test_stop_requested_signals_socket_close(self):
-        options = DummyOptions()
-        gconfig = DummyFCGIGroupConfig(options)
-        group = self._makeOne(gconfig, socketManager=DummySocketManager)
-        group.stop_requested()
-        self.assertTrue(group.socket_manager.request_close_called)
-        
+                
 class EventListenerPoolTests(ProcessGroupBaseTests):
     def setUp(self):
         from supervisor.events import clear

src/supervisor/tests/test_rpcinterfaces.py

@@ -732,7 +732,6 @@ class SupervisorNamespaceXMLRPCInterfaceTests(TestBase):
         self.assertEqual(process1.stop_called, True)
         process2 = supervisord.process_groups['foo'].processes['process2']
         self.assertEqual(process2.stop_called, True)
-        self.assertTrue(supervisord.process_groups['foo'].stop_was_requested)
 
     def test_stopProcessGroup_nowait(self):
         options = DummyOptions()

src/supervisor/tests/test_socket_manager.py

@@ -103,7 +103,6 @@ class SocketManagerTest(unittest.TestCase):
         self.assertEqual(sock_manager.socket_config, conf)
         sock = sock_manager.get_socket()
         self.assertEqual(sock.getsockname(), ('127.0.0.1', 12345))
-        sock_manager.request_close()
 
     def test_tcp_w_ip(self):
         conf = InetStreamSocketConfig('127.0.0.1', 12345)
@@ -111,7 +110,6 @@ class SocketManagerTest(unittest.TestCase):
         self.assertEqual(sock_manager.socket_config, conf)
         sock = sock_manager.get_socket()
         self.assertEqual(sock.getsockname(), ('127.0.0.1', 12345))
-        sock_manager.request_close()
 
     def test_unix(self):
         (tf_fd, tf_name) = tempfile.mkstemp();
@@ -121,7 +119,6 @@ class SocketManagerTest(unittest.TestCase):
         sock = sock_manager.get_socket()
         self.assertEqual(sock.getsockname(), tf_name)
         sock = None
-        sock_manager.request_close()
         os.close(tf_fd)
         
     def test_socket_lifecycle(self):
@@ -139,8 +136,6 @@ class SocketManagerTest(unittest.TestCase):
         self.assertNotEqual(sock, sock2)
         #Assert that they are the same underlying socket
         self.assertEqual(sock_id, sock2_id)
-        #Request socket close
-        sock_manager.request_close()
         #Socket not actually closed yet b/c ref ct is 2
         self.assertTrue(sock_manager.is_prepared())
         self.assertFalse(sock_manager.socket.close_called)
@@ -161,10 +156,6 @@ class SocketManagerTest(unittest.TestCase):
         self.assertNotEqual(sock_id, sock3_id)
         #Drop ref ct to zero
         del sock3
-        #Assert that socket is still open until close is requested
-        self.assertTrue(sock_manager.is_prepared())
-        self.assertFalse(sock_manager.socket.close_called)
-        sock_manager.request_close()
         #Now assert that socket is closed
         self.assertFalse(sock_manager.is_prepared())
         self.assertTrue(sock_manager.socket.close_called)
@@ -178,7 +169,6 @@ class SocketManagerTest(unittest.TestCase):
         self.assertTrue(sock.listen_called)
         self.assertEqual(sock.listen_backlog, socket.SOMAXCONN)
         self.assertFalse(sock.close_called)
-        sock_manager.request_close()
     
     def test_tcp_socket_already_taken(self):
         conf = InetStreamSocketConfig('127.0.0.1', 12345)
@@ -187,7 +177,6 @@ class SocketManagerTest(unittest.TestCase):
         sock_manager2 = self._makeOne(conf)
         self.assertRaises(socket.error, sock_manager2.get_socket)
         sock = None
-        sock_manager.request_close()
         
     def test_unix_bad_sock(self):
         conf = UnixStreamSocketConfig('/notthere/foo.sock')