Restore kqueue after daemonize because it's not inherited on fork()

Author: igorsobreira

supervisor/options.py

@@ -50,6 +50,7 @@
 from supervisor import loggers
 from supervisor import states
 from supervisor import xmlrpc
+from supervisor import poller
 
 mydir = os.path.abspath(os.path.dirname(__file__))
 version_txt = os.path.join(mydir, 'version.txt')
@@ -431,6 +432,7 @@ def __init__(self):
         self.process_group_configs = []
         self.parse_warnings = []
         self.signal_receiver = SignalReceiver()
+        self.poller = poller.Poller(self)
 
     def version(self, dummy):
         """Print version to stdout and exit(0).
@@ -982,6 +984,11 @@ def server_configs_from_parser(self, parser):
         return configs
 
     def daemonize(self):
+        self.poller.before_daemonize()
+        self._daemonize()
+        self.poller.after_daemonize()
+
+    def _daemonize(self):
         # To daemonize, we need to become the leader of our own session
         # (process) group.  If we do not, signals sent to our
         # parent process will also be sent to us.   This might be bad because

supervisor/poller.py

@@ -24,6 +24,13 @@ def unregister(self, fd):
     def poll(self, timeout):
         raise NotImplementedError
 
+    def before_daemonize(self):
+        pass
+
+    def after_daemonize(self):
+        pass
+
+
 class SelectPoller(BasePoller):
 
     def initialize(self):
@@ -120,13 +127,17 @@ class KQueuePoller(BasePoller):
 
     def initialize(self):
         self._kqueue = select.kqueue()
+        self.readables = set()
+        self.writables = set()
 
     def register_readable(self, fd):
+        self.readables.add(fd)
         kevent = select.kevent(fd, filter=select.KQ_FILTER_READ,
                                flags=select.KQ_EV_ADD)
         self._kqueue_control(fd, kevent)
 
     def register_writable(self, fd):
+        self.writables.add(fd)
         kevent = select.kevent(fd, filter=select.KQ_FILTER_WRITE,
                                flags=select.KQ_EV_ADD)
         self._kqueue_control(fd, kevent)
@@ -165,6 +176,16 @@ def poll(self, timeout):
 
         return readables, writables
 
+    def before_daemonize(self):
+        self._kqueue.close()
+        self._kqueue = None
+
+    def after_daemonize(self):
+        self._kqueue = select.kqueue()
+        for fd in self.readables:
+            self.register_readable(fd)
+        for fd in self.writables:
+            self.register_writable(fd)
 
 def implements_poll():
     return hasattr(select, 'poll')

supervisor/supervisord.py

@@ -43,7 +43,6 @@
 from supervisor import events
 from supervisor.states import SupervisorStates
 from supervisor.states import getProcessStateDescription
-from supervisor.poller import Poller
 
 class Supervisor:
     stopping = False # set after we detect that we are handling a stop request
@@ -53,7 +52,6 @@ class Supervisor:
 
     def __init__(self, options):
         self.options = options
-        self.poller = Poller(options)
         self.process_groups = {}
         self.ticks = {}
 
@@ -210,11 +208,11 @@ def runforever(self):
 
             for fd, dispatcher in combined_map.items():
                 if dispatcher.readable():
-                    self.poller.register_readable(fd)
+                    self.options.poller.register_readable(fd)
                 if dispatcher.writable():
-                    self.poller.register_writable(fd)
+                    self.options.poller.register_writable(fd)
 
-            r, w = self.poller.poll(timeout)
+            r, w = self.options.poller.poll(timeout)
 
             for fd in r:
                 if combined_map.has_key(fd):

supervisor/tests/base.py

@@ -67,6 +67,7 @@ def __init__(self):
         self.changed_directory = False
         self.chdir_error = None
         self.umaskset = None
+        self.poller = DummyPoller(self)
 
     def getLogger(self, *args, **kw):
         logger = DummyLogger()
@@ -1035,6 +1036,19 @@ def __init__(self, serial='abc'):
 
     def __str__(self):
         return 'dummy event'
+
+class DummyPoller:
+    def __init__(self, options):
+        self.result = [], []
+
+    def register_readable(self, fd):
+        pass
+
+    def register_writable(self, fd):
+        pass
+
+    def poll(self, timeout):
+        return self.result
 
 def dummy_handler(event, result):
     pass

supervisor/tests/test_options.py

@@ -1586,6 +1586,14 @@ def test_dropPrivileges_nonroot_different_user(self):
         msg = instance.dropPrivileges(42)
         self.assertEqual(msg, "Can't drop privilege as nonroot user")
 
+    def test_daemonize_notifies_poller_before_and_after_fork(self):
+        instance = self._makeOne()
+        instance._daemonize = lambda: None
+        instance.poller = Mock()
+        instance.daemonize()
+        instance.poller.before_daemonize.assert_called_once_with()
+        instance.poller.after_daemonize.assert_called_once_with()
+
 class TestProcessConfig(unittest.TestCase):
     def _getTargetClass(self):
         from supervisor.options import ProcessConfig

supervisor/tests/test_poller.py

@@ -2,6 +2,7 @@
 import unittest
 import errno
 import select
+from mock import Mock
 
 from supervisor.poller import SelectPoller, PollPoller, KQueuePoller
 from supervisor.poller import implements_poll, implements_kqueue
@@ -84,6 +85,7 @@ def test_register_readable(self):
         poller = self._makeOne(DummyOptions())
         poller._kqueue = kqueue
         poller.register_readable(6)
+        self.assertEqual(list(poller.readables), [6])
         self.assertEqual(len(kqueue.registered_kevents), 1)
         self.assertReadEventAdded(kqueue, kqueue.registered_kevents[0], 6)
 
@@ -92,6 +94,7 @@ def test_register_writable(self):
         poller = self._makeOne(DummyOptions())
         poller._kqueue = kqueue
         poller.register_writable(7)
+        self.assertEqual(list(poller.writables), [7])
         self.assertEqual(len(kqueue.registered_kevents), 1)
         self.assertWriteEventAdded(kqueue, kqueue.registered_kevents[0], 7)
 
@@ -144,6 +147,27 @@ def test_poll_uncaught_exception(self):
         poller.register_readable(6)
         self.assertRaises(OSError, poller.poll, 1000)
 
+    def test_before_daemonize_closes_kqueue(self):
+        mock_kqueue = Mock()
+        options = DummyOptions()
+        poller = self._makeOne(options)
+        poller._kqueue = mock_kqueue
+        poller.before_daemonize()
+        mock_kqueue.close.assert_called_once_with()
+        self.assertEqual(poller._kqueue, None)
+
+    def test_after_daemonize_restores_kqueue(self):
+        options = DummyOptions()
+        poller = self._makeOne(options)
+        poller.readables = [1]
+        poller.writables = [3]
+        poller.register_readable = Mock()
+        poller.register_writable = Mock()
+        poller.after_daemonize()
+        self.assertTrue(isinstance(poller._kqueue, select.kqueue))
+        poller.register_readable.assert_called_with(1)
+        poller.register_writable.assert_called_with(3)
+
     def assertReadEventAdded(self, kqueue, kevent, fd):
         self.assertEventAdded(kqueue, kevent, fd, select.KQ_FILTER_READ)
 

supervisor/tests/test_supervisord.py

@@ -373,8 +373,7 @@ def test_runforever_calls_tick(self):
 
     def test_runforever_poll_dispatchers(self):
         options = DummyOptions()
-        poller = DummyPoller(options)
-        poller.result = [6], [7, 8]
+        options.poller.result = [6], [7, 8]
         supervisord = self._makeOne(options)
         pconfig = DummyPConfig(options, 'foo', '/bin/foo',)
         process = DummyProcess(pconfig)
@@ -385,7 +384,6 @@ def test_runforever_poll_dispatchers(self):
         error = DummyDispatcher(writable=True, error=OSError)
         pgroup.dispatchers = {6:readable, 7:writable, 8:error}
         supervisord.process_groups = {'foo': pgroup}
-        supervisord.poller = poller
         options.test = True
         supervisord.runforever()
         self.assertEqual(pgroup.transitioned, True)
@@ -395,8 +393,7 @@ def test_runforever_poll_dispatchers(self):
 
     def test_runforever_select_dispatcher_exitnow(self):
         options = DummyOptions()
-        poller = DummyPoller(options)
-        poller.result = [6], []
+        options.poller.result = [6], []
         supervisord = self._makeOne(options)
         pconfig = DummyPConfig(options, 'foo', '/bin/foo',)
         process = DummyProcess(pconfig)
@@ -406,7 +403,6 @@ def test_runforever_select_dispatcher_exitnow(self):
         exitnow = DummyDispatcher(readable=True, error=asyncore.ExitNow)
         pgroup.dispatchers = {6:exitnow}
         supervisord.process_groups = {'foo': pgroup}
-        supervisord.poller = poller
         options.test = True
         self.assertRaises(asyncore.ExitNow, supervisord.runforever)
 
@@ -507,19 +503,6 @@ def callback(event):
         self.assertEqual(len(L), 6)
         self.assertEqual(L[-1].__class__, events.Tick3600Event)
 
-class DummyPoller:
-    def __init__(self, options):
-        self.result = []
-
-    def register_readable(self, fd):
-        pass
-
-    def register_writable(self, fd):
-        pass
-
-    def poll(self, timeout):
-        return self.result
-
 def test_suite():
     return unittest.findTestCases(sys.modules[__name__])