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.

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.

Part of #229 and extended work on #566.

Author: thiell

lib/ClusterShell/Propagation.py

@@ -424,3 +424,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)
@@ -1384,8 +1386,22 @@ def _pchannel_release(self, gateway, metaworker):
                 logger.debug("pchannel_release: destroying channel %s",
                             chanworker.eh)
                 chanworker.abort()
-                # delete gateway reference
-                del self.gateways[gwstr]
+
+    def _pchannel_close(self, gateway, chanworker):
+        """Close propagation channel.
+
+        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