EngineClient: improve write handling

- generate ev_written in the right places
- handle broken pipe errors

Closes #196.

Change-Id: Ib2d742bce31b6004ca904368eb7037a57eac30b4

Author: thiell

ChangeLog

@@ -1,3 +1,7 @@
+2016-05-22 S. Thiell <[email protected]>
+
+	* EngineClient.py: handle broken pipe on write() (ticket #196).
+
 2016-04-24 S. Thiell <[email protected]>
 
 	* NodeSet.py: allow empty string as valid argument for empty NodeSet

lib/ClusterShell/Engine/Engine.py

@@ -1,6 +1,6 @@
 #
-# Copyright CEA/DAM/DIF (2007-2015)
-#  Contributor: Stephane THIELL <[email protected]>
+# Copyright CEA/DAM/DIF (2007-2016)
+#  Contributor: Stephane THIELL <[email protected]>
 #
 # This file is part of the ClusterShell library.
 #
@@ -44,13 +44,17 @@
 import time
 import traceback
 
+
+LOGGER = logging.getLogger(__name__)
+
 # Engine client fd I/O event interest bits
 E_READ = 0x1
 E_WRITE = 0x2
 
 # Define epsilon value for time float arithmetic operations
 EPSILON = 1.0e-3
 
+
 class EngineException(Exception):
     """
     Base engine exception.
@@ -219,9 +223,8 @@ def arm(self, client):
                 # Just print a debug message that could help detect issues
                 # coming from a long-running timer handler.
                 if self.fire_date < time_current:
-                    logging.getLogger(__name__).debug(
-                        "Warning: passed interval time for %r (long running "
-                        "event handler?)", self.client)
+                    LOGGER.debug("Warning: passed interval time for %r "
+                                 "(long running event handler?)", self.client)
 
         def disarm(self):
             client = self.client
@@ -421,8 +424,8 @@ def _fd2client(self, fd):
             if client._reg_epoch < self._current_loopcnt:
                 return client, stream
             else:
-                self._debug("ENGINE _fd2client: ignoring just re-used FD %d" \
-                            % stream.fd)
+                LOGGER.debug("_fd2client: ignoring just re-used FD %d",
+                             stream.fd)
         return (None, None)
 
     def add(self, client):
@@ -472,10 +475,18 @@ def remove_stream(self, client, stream):
         needed read flush as needed. If no more retainable stream
         remains for this client, this method automatically removes the
         entire client from engine.
+
+        This function does nothing if the stream is not registered.
         """
+        if stream.fd not in self.reg_clifds:
+            LOGGER.debug("remove_stream: %s not registered", stream)
+            return
+
         self.unregister_stream(client, stream)
+
         # _close_stream() will flush pending read buffers so may generate events
         client._close_stream(stream.name)
+
         # client may have been removed by previous events, if not check whether
         # some retained streams still remain
         if client in self._clients and not client.streams.retained():
@@ -596,8 +607,7 @@ def set_events(self, client, stream):
             (stream.new_events, stream.events, client, stream.name))
 
         if not client.registered:
-            logging.getLogger(__name__).debug( \
-                "set_events: client %s not registered" % self)
+            LOGGER.debug("set_events: client %s not registered", self)
             return
 
         chgbits = stream.new_events ^ stream.events
@@ -705,7 +715,7 @@ def run(self, timeout):
                     # BaseException. For now, print a backtrace in debug to
                     # help detect the problem.
                     tbexc = traceback.format_exception(exc_t, exc_val, exc_tb)
-                    logging.getLogger(__name__).debug(''.join(tbexc))
+                    LOGGER.debug(''.join(tbexc))
                     raise
                 raise
         finally:
@@ -747,6 +757,6 @@ def exited(self):
         return not self.running and self._exited
 
     def _debug(self, s):
-        """library engine debugging hook"""
-        #logging.getLogger(__name__).debug(s)
+        """library engine verbose debugging hook"""
+        #LOGGER.debug(s)
         pass

lib/ClusterShell/Worker/EngineClient.py

@@ -1,6 +1,6 @@
 #
-# Copyright CEA/DAM/DIF (2009-2014)
-#  Contributor: Stephane THIELL <[email protected]>
+# Copyright CEA/DAM/DIF (2009-2016)
+#  Contributor: Stephane THIELL <[email protected]>
 #
 # This file is part of the ClusterShell library.
 #
@@ -41,6 +41,7 @@
 """
 
 import errno
+import logging
 import os
 import Queue
 import thread
@@ -51,6 +52,9 @@
 from ClusterShell.Engine.Engine import EngineBaseTimer, E_READ, E_WRITE
 
 
+LOGGER = logging.getLogger(__name__)
+
+
 class EngineClientException(Exception):
     """Generic EngineClient exception."""
 
@@ -319,20 +323,27 @@ def _handle_write(self, sname):
             try:
                 wcnt = os.write(wfile.fd, wfile.wbuf)
             except OSError, exc:
-                if (exc.errno == errno.EAGAIN):
+                if exc.errno == errno.EAGAIN:
+                    # _handle_write() is not only called by the engine but also
+                    # by _write(), so this is legit: we just try again later
                     self._set_writing(sname)
                     return
+                if exc.errno == errno.EPIPE:
+                    # broken pipe: log warning message and do NOT retry
+                    LOGGER.warning('%s: %s', self, exc)
+                    return
                 raise
             if wcnt > 0:
-                self.worker._on_written(self.key, wcnt, sname)
                 # dequeue written buffer
                 wfile.wbuf = wfile.wbuf[wcnt:]
                 # check for possible ending
                 if wfile.eof and not wfile.wbuf:
+                    self.worker._on_written(self.key, wcnt, sname)
                     # remove stream from engine (not directly)
                     self._engine.remove_stream(self, wfile)
                 else:
                     self._set_writing(sname)
+                    self.worker._on_written(self.key, wcnt, sname)
 
     def _exec_nonblock(self, commandlist, shell=False, env=None):
         """

tests/StreamWorkerTest.py

@@ -200,3 +200,100 @@ def ev_timeout(self, worker):
         self.assertRaises(OSError, os.close, rfd1)
         os.close(wfd1)
 
+    def test_006_worker_abort_on_written(self):
+        """test StreamWorker abort on ev_written"""
+
+        # This test creates a writable StreamWorker that will abort after the
+        # first write, to check whether ev_written is generated in the right
+        # place.
+
+        class TestH(EventHandler):
+            def __init__(self, testcase, rfd):
+                self.testcase = testcase
+                self.rfd = rfd
+                self.check_written = 0
+
+            def ev_written(self, worker, node, sname, size):
+                self.check_written += 1
+                self.testcase.assertEqual(os.read(self.rfd, 1024), "initial")
+                worker.abort()
+
+        rfd, wfd = os.pipe()
+
+        hdlr = TestH(self, rfd)
+        worker = StreamWorker(handler=hdlr)
+
+        worker.set_writer("test", wfd) # closefd=True
+        worker.write("initial", "test")
+
+        self.run_worker(worker)
+        self.assertEqual(hdlr.check_written, 1)
+        os.close(rfd)
+
+    def test_007_worker_abort_on_written_eof(self):
+        """test StreamWorker abort on ev_written (with EOF)"""
+
+        # This test is similar to previous test test_006 but does
+        # write() + set_write_eof().
+
+        class TestH(EventHandler):
+            def __init__(self, testcase, rfd):
+                self.testcase = testcase
+                self.rfd = rfd
+                self.check_written = 0
+
+            def ev_written(self, worker, node, sname, size):
+                self.check_written += 1
+                self.testcase.assertEqual(os.read(self.rfd, 1024), "initial")
+                worker.abort()
+
+        rfd, wfd = os.pipe()
+
+        hdlr = TestH(self, rfd)
+        worker = StreamWorker(handler=hdlr)
+
+        worker.set_writer("test", wfd) # closefd=True
+        worker.write("initial", "test")
+        worker.set_write_eof()
+
+        self.run_worker(worker)
+        self.assertEqual(hdlr.check_written, 1)
+        os.close(rfd)
+
+    def test_008_broken_pipe_on_write(self):
+        """test StreamWorker with broken pipe on write()"""
+
+        # This test creates a writable StreamWorker that will close the read
+        # side of the pipe just after the first write to generate a broken
+        # pipe error.
+
+        class TestH(EventHandler):
+            def __init__(self, testcase, rfd):
+                self.testcase = testcase
+                self.rfd = rfd
+                self.check_hup = 0
+                self.check_written = 0
+
+            def ev_hup(self, worker):
+                self.check_hup += 1
+
+            def ev_written(self, worker, node, sname, size):
+                self.check_written += 1
+                self.testcase.assertEqual(os.read(self.rfd, 1024), "initial")
+                # close reader, that will stop the StreamWorker
+                os.close(self.rfd)
+                # The following write call used to raise broken pipe before
+                # version 1.7.2.
+                worker.write("final")
+
+        rfd, wfd = os.pipe()
+
+        hdlr = TestH(self, rfd)
+        worker = StreamWorker(handler=hdlr)
+
+        worker.set_writer("test", wfd) # closefd=True
+        worker.write("initial", "test")
+
+        self.run_worker(worker)
+        self.assertEqual(hdlr.check_hup, 1)
+        self.assertEqual(hdlr.check_written, 1)