Tree: implement per-gateway abort

This commit adds the functionality to abort a specific gateway channel from
the initiator. Until now, this was not properly handled. This also fixes
gateway failover.

Changes:

* Implement TreeWorker._gateway_abort() that can be used to abort/cancel
  all tasks being done by the TreeWorker via the specified gateway. In
  case of such abort (likely due to some gateway failure), a special return
  code 76 (os.EX_PROTOCOL) is used for closing all running remote commands via
  this gateway. This return code is sometimes used to specify a "Remote
  protocol error" / "An error occurred in a remote communication protocol"
  which seems appropriate here.

* Implement a new Task._pchannel_closing() method that is called on
  PropagationChannel.ev_close(), so deterministically every time a gateway
  channel is closing (self-initiated or not). This method performs necessary
  cleanup actions, but most notably calls TreeWorker._gateway_abort(gateway)
  on each worker currently using the gateway channel.

* Update Task._pchannel_release() so that it now calls
  PropagationChannel._close() instead of Worker.abort() to properly reset
  the channel's opened/setup flags.

* Updated TreeWorkerTest with tests to better cover the above and gateway
  failover.

Part of #229 and extended work on #566.

Author: thiell

lib/ClusterShell/Communication.py

@@ -203,10 +203,9 @@ def _open(self):
         xmlgen = XMLGenerator(self.worker, encoding=ENCODING)
         xmlgen.startElement('channel', {'version': __version__})
 
-    def _close(self):
+    def _close(self, abort=False):
         """close an already opened channel"""
-        send_endtag = self.opened
-        if send_endtag:
+        if self.opened and not abort:
             XMLGenerator(self.worker, encoding=ENCODING).endElement('channel')
         self.worker.abort()
         self.opened = self.setup = False

lib/ClusterShell/Propagation.py

@@ -253,8 +253,7 @@ def recv(self, msg):
         if msg.type == EndMessage.ident:
             #??#self.ptree.notify_close()
             self.logger.debug("got EndMessage; closing")
-            # abort worker (now working)
-            self.worker.abort()
+            self._close()
         elif msg.type == StdErrMessage.ident and msg.srcid == 0:
             # Handle error messages when channel is not established yet
             # or if messages are non-routed (eg. gateway-related)
@@ -423,3 +422,6 @@ def ev_close(self, worker, timedout):
                 self.logger.debug("channel was not set up: redistributing...")
                 for mw in set(self.task.gateways[gateway][1]):
                     mw._relaunch(gateway)
+
+        # update Task that we are closing
+        worker.task._pchannel_close(gateway, worker)

lib/ClusterShell/Task.py

@@ -1365,8 +1365,10 @@ def _pchannel(self, gateway, metaworker):
     def _pchannel_release(self, gateway, metaworker):
         """Release propagation channel associated to gateway.
 
-        Lookup by gateway, decref associated metaworker set and release
-        channel worker if needed.
+        Lookup by gateway, decref associated metaworker set and abort channel
+        worker if not used anymore.
+
+        Called by TreeWorker._check_fini()
         """
         logger = logging.getLogger(__name__)
         logger.debug("pchannel_release %s %s", gateway, metaworker)
@@ -1381,11 +1383,29 @@ def _pchannel_release(self, gateway, metaworker):
             chanworker, metaworkers = self.gateways[gwstr]
             metaworkers.remove(metaworker)
             if len(metaworkers) == 0:
-                logger.debug("pchannel_release: destroying channel %s",
+                logger.debug("pchannel_release: closing channel %s",
                             chanworker.eh)
-                chanworker.abort()
-                # delete gateway reference
-                del self.gateways[gwstr]
+                # Call PropagationChannel._close() that will close the channel
+                # properly, update the opened/setup flags and abort the worker.
+                # We might be in an event handler and we want to make sure we
+                # ignore any pending messages from this gateway from now on.
+                chanworker.eh._close(abort=True)
+
+    def _pchannel_close(self, gateway, chanworker):
+        """A propagation channel is closing.
+
+        Perform necessary cleanup actions when a gateway channel is closing.
+
+        Called by PropagationChannel.ev_close().
+        """
+        logger = logging.getLogger(__name__)
+        logger.debug("pchannel_closing: %s", gateway)
+        chwrk, metaworkers = self.gateways[gateway]
+        assert chwrk is chanworker, (chwrk, chanworker)
+        metaworkers_copy = list(metaworkers)
+        for mw in metaworkers_copy:
+            mw._gateway_abort(gateway)
+        del self.gateways[gateway]
 
 
 def task_self(defaults=None):

lib/ClusterShell/Worker/Tree.py

@@ -564,6 +564,18 @@ def set_write_eof(self):
 
         self._set_write_eof_remote()
 
+    def _gateway_abort(self, gateway):
+        """Abort on gateway failure"""
+        if gateway not in self.gwtargets:
+            self.logger.warning("TreeWorker._gateway_abort %s not found",
+                                gateway)
+            return
+        targets = self.gwtargets[gateway]
+        self.logger.debug("TreeWorker._gateway_abort %s found: targets=%s",
+                          gateway, targets)
+        for target in NodeSet.fromlist(targets): # targets is a mutable list
+            self._on_remote_node_close(target, os.EX_PROTOCOL, gateway)
+
     def abort(self):
         """Abort processing any action by this worker."""
         # Not yet supported by TreeWorker