Don't remove a closed dispatcher, just make it un read/write-able.

src/supervisor/dispatchers.py

@@ -53,7 +53,6 @@ class PDispatcher:
         if not self.closed:
             self.process.config.options.logger.debug(
                 'fd %s closed, stopped monitoring %s' % (self.fd, self))
-            self.process.remove_dispatcher(self.fd)
             self.closed = True
 
 class POutputDispatcher(PDispatcher):
@@ -200,6 +199,8 @@ class POutputDispatcher(PDispatcher):
         return False
     
     def readable(self):
+        if self.closed:
+            return False
         return True
 
     def handle_read_event(self):
@@ -262,6 +263,8 @@ class PEventListenerDispatcher(PDispatcher):
     
     def readable(self):
         self.handle_listener_state_change()
+        if self.closed:
+            return False
         return True
 
     def handle_read_event(self):
@@ -378,7 +381,7 @@ class PInputDispatcher(PDispatcher):
         self.input_buffer = ''
 
     def writable(self):
-        if self.input_buffer:
+        if self.input_buffer and not self.closed:
             return True
         return False
 

src/supervisor/process.py

@@ -455,12 +455,6 @@ class Subprocess:
                 self._assertInState(ProcessStates.STARTING)
                 self.change_state(ProcessStates.RUNNING)
 
-    def remove_dispatcher(self, fd):
-        # called by dispatcher itself to remove itself when it detects
-        # that the child end of the pipe has been closed
-        del self.dispatchers[fd]
-
-        
 class ProcessGroupBase:
     def __init__(self, config):
         self.config = config

src/supervisor/tests/base.py

@@ -321,7 +321,6 @@ class DummyProcess:
         self.output_fd_drained = None
         self.transitioned = False
         self.write_error = None
-        self.dispatchers_removed = []
 
     def reopenlogs(self):
         self.logs_reopened = True
@@ -393,9 +392,6 @@ class DummyProcess:
     def transition(self):
         self.transitioned = True
 
-    def remove_dispatcher(self, fd):
-        self.dispatchers_removed.append(fd)
-
 class DummyPConfig:
     def __init__(self, options, name, command, priority=999, autostart=True,
                  autorestart=True, startsecs=10, startretries=999,

src/supervisor/tests/test_dispatchers.py

@@ -32,13 +32,22 @@ class POutputDispatcherTests(unittest.TestCase):
         dispatcher = self._makeOne(process)
         self.assertEqual(dispatcher.writable(), False)
         
-    def test_readable(self):
+    def test_readable_open(self):
         options = DummyOptions()
         config = DummyPConfig(options, 'process1', '/bin/process1')
         process = DummyProcess(config)
         dispatcher = self._makeOne(process)
+        dispatcher.closed = False
         self.assertEqual(dispatcher.readable(), True)
 
+    def test_readable_closed(self):
+        options = DummyOptions()
+        config = DummyPConfig(options, 'process1', '/bin/process1')
+        process = DummyProcess(config)
+        dispatcher = self._makeOne(process)
+        dispatcher.closed = True
+        self.assertEqual(dispatcher.readable(), False)
+
     def test_handle_write_event(self):
         options = DummyOptions()
         config = DummyPConfig(options, 'process1', '/bin/process1')
@@ -336,10 +345,8 @@ class POutputDispatcherTests(unittest.TestCase):
         process = DummyProcess(config)
         dispatcher = self._makeOne(process)
         dispatcher.close()
-        self.assertEqual(process.dispatchers_removed, [0])
         self.assertEqual(dispatcher.closed, True)
         dispatcher.close() # make sure we don't error if we try to close twice
-        self.assertEqual(process.dispatchers_removed, [0])
         self.assertEqual(dispatcher.closed, True)
         
                         
@@ -352,16 +359,32 @@ class PInputDispatcherTests(unittest.TestCase):
         channel = 'stdin'
         return self._getTargetClass()(process, channel, 0)
 
-    def test_writable_nodata(self):
+    def test_writable_open_nodata(self):
         process = DummyProcess(None)
         dispatcher = self._makeOne(process)
         dispatcher.input_buffer = 'a'
+        dispatcher.closed = False
         self.assertEqual(dispatcher.writable(), True)
 
-    def test_writable_withdata(self):
+    def test_writable_open_withdata(self):
+        process = DummyProcess(None)
+        dispatcher = self._makeOne(process)
+        dispatcher.input_buffer = ''
+        dispatcher.closed = False
+        self.assertEqual(dispatcher.writable(), False)
+
+    def test_writable_closed_nodata(self):
+        process = DummyProcess(None)
+        dispatcher = self._makeOne(process)
+        dispatcher.input_buffer = 'a'
+        dispatcher.closed = True
+        self.assertEqual(dispatcher.writable(), False)
+
+    def test_writable_closed_withdata(self):
         process = DummyProcess(None)
         dispatcher = self._makeOne(process)
         dispatcher.input_buffer = ''
+        dispatcher.closed = True
         self.assertEqual(dispatcher.writable(), False)
 
     def test_readable(self):
@@ -459,10 +482,8 @@ class PInputDispatcherTests(unittest.TestCase):
         process = DummyProcess(config)
         dispatcher = self._makeOne(process)
         dispatcher.close()
-        self.assertEqual(process.dispatchers_removed, [0])
         self.assertEqual(dispatcher.closed, True)
         dispatcher.close() # make sure we don't error if we try to close twice
-        self.assertEqual(process.dispatchers_removed, [0])
         self.assertEqual(dispatcher.closed, True)
 
 class PEventListenerDispatcherTests(unittest.TestCase):
@@ -481,13 +502,22 @@ class PEventListenerDispatcherTests(unittest.TestCase):
         dispatcher = self._makeOne(process)
         self.assertEqual(dispatcher.writable(), False)
         
-    def test_readable(self):
+    def test_readable_open(self):
         options = DummyOptions()
         config = DummyPConfig(options, 'process1', '/bin/process1')
         process = DummyProcess(config)
         dispatcher = self._makeOne(process)
+        dispatcher.closed = False
         self.assertEqual(dispatcher.readable(), True)
 
+    def test_readable_closed(self):
+        options = DummyOptions()
+        config = DummyPConfig(options, 'process1', '/bin/process1')
+        process = DummyProcess(config)
+        dispatcher = self._makeOne(process)
+        dispatcher.closed = True
+        self.assertEqual(dispatcher.readable(), False)
+
     def test_handle_write_event(self):
         options = DummyOptions()
         config = DummyPConfig(options, 'process1', '/bin/process1')
@@ -798,10 +828,8 @@ class PEventListenerDispatcherTests(unittest.TestCase):
         process = DummyProcess(config)
         dispatcher = self._makeOne(process)
         dispatcher.close()
-        self.assertEqual(process.dispatchers_removed, [0])
         self.assertEqual(dispatcher.closed, True)
         dispatcher.close() # make sure we don't error if we try to close twice
-        self.assertEqual(process.dispatchers_removed, [0])
         self.assertEqual(dispatcher.closed, True)
 
 

src/supervisor/tests/test_process.py

@@ -786,15 +786,6 @@ class SubprocessTests(unittest.TestCase):
         instance.state = 10
         self.assertEqual(instance.change_state(10), False)
 
-    def test_remove_dispatcher(self):
-        options = DummyOptions()
-        config = DummyPConfig(options, 'test', '/test')
-        instance = self._makeOne(config)
-        instance.dispatchers = {0:True}
-        instance.remove_dispatcher(0)
-        self.assertEqual(instance.dispatchers, {})
-        
-
 class ProcessGroupBaseTests(unittest.TestCase):
     def _getTargetClass(self):
         from supervisor.process import ProcessGroupBase