pool: shutdown xrootd TPC event-loop group asynchronously

Motivation:

Netty has two interesting features: first, shutting down is done
"gracefully", where it waits at least one second for activity to calm
down and can wait up to a maximum of three seconds; second, it supports
asynchronous shutdown, where the various event-loops gracefully shutdown
concurrently.

Currently the Xrootd TPC event-loop does not take advantage of the
asynchronous shutdown.  It waits until the main activity has finished
before starting its own graceful shutdown.

Modification:

Update Netty transfer service to support subclasses adding additional
thread-groups that should be shutdown both gracefully and
asynchronously.

Update Xrootd netty transfer service to use this feature to shutdown the
TPC event-loop asynchronously.

Result:

No user obserable changes.

Pools will shut down slightly faster.

Target: master
Requires-notes: no
Requires-book: no
Patch: https://rb.dcache.org/r/12413/
Acked-by: Tigran Mkrtchyan

modules/dcache-xrootd/src/main/java/org/dcache/xrootd/pool/XrootdTransferService.java

@@ -354,15 +354,10 @@ protected void initChannel(Channel ch) throws Exception
         pipeline.addLast("transfer", handler);
     }
 
-    @PreDestroy
     @Override
-    public synchronized void shutdown() {
-        super.shutdown();
-        thirdPartyClientGroup.shutdownGracefully(1, 3, TimeUnit.SECONDS);
-        try {
-            thirdPartyClientGroup.terminationFuture().sync();
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-        }
+    protected void initialiseShutdown()
+    {
+        super.initialiseShutdown();
+        shutdownGracefully(thirdPartyClientGroup);
     }
 }

modules/dcache/src/main/java/org/dcache/pool/movers/NettyTransferService.java

@@ -43,6 +43,8 @@
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.nio.channels.CompletionHandler;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executors;
@@ -137,6 +139,8 @@ public abstract class NettyTransferService<P extends ProtocolInfo>
 
     private CellAddressCore address;
 
+    private final List<io.netty.util.concurrent.Future<?>> shutdownFutures = new ArrayList<>();
+
     public NettyTransferService(String name)
     {
         this.name = name;
@@ -296,17 +300,35 @@ public synchronized void init()
 
     @PreDestroy
     public synchronized void shutdown()
+    {
+        LOGGER.debug("NettyTransferService#shutdown started");
+        initialiseShutdown();
+        awaitShutdownCompletion();
+        LOGGER.debug("NettyTransferService#shutdown completed");
+    }
+
+    protected void initialiseShutdown()
     {
         stopServer();
         timeoutScheduler.shutdown();
 
-        acceptGroup.shutdownGracefully(1, 3, TimeUnit.SECONDS);
-        socketGroup.shutdownGracefully(1, 3, TimeUnit.SECONDS);
+        shutdownGracefully(acceptGroup);
+        shutdownGracefully(socketGroup);
+    }
+
+    protected void shutdownGracefully(NioEventLoopGroup group)
+    {
+        io.netty.util.concurrent.Future<?> terminationFuture = group.shutdownGracefully(1, 3, TimeUnit.SECONDS);
+        shutdownFutures.add(terminationFuture);
+    }
 
+    private void awaitShutdownCompletion()
+    {
         try {
             if (timeoutScheduler.awaitTermination(3, TimeUnit.SECONDS)) {
-                acceptGroup.terminationFuture().sync();
-                socketGroup.terminationFuture().sync();
+                for (io.netty.util.concurrent.Future<?> f : shutdownFutures) {
+                    f.sync();
+                }
             }
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();