pool: Use ListenableFuture rather than CompletionHandler

The pool queuing and mover wrapping code relies on the
CompletionHandler interface from the NIO 2 async code.
This interface is a poor fit for our needs as it doesn't
provide means of cancelling the result (which means we
had to add a custom Cancellable interface) and has
features like attachables, that we don't need.

The Guava ListenableFuture - an extension of the regular
JDK Future concept - fits our needs much better. This
patch refactors the code to use ListenableFuture and this
allows custom cancellation code to be removed, and also
stream lines the code a bit.

The patch moves some error handling from SimpleIoScheduler
to PoolIORequest. The code was tightly coupled to
PoolIORequest anyway and fits more naturally inside that
class.

The patch is part of a series of pathes to clean up the
pool mover handling.

Target: trunk
Require-notes: no
Require-book: no
Acked-by: Paul Millar <paul.millar@desy.de>
Patch: http://rb.dcache.org/r/5308

Author: gbehrmann

modules/dcache/src/main/java/org/dcache/chimera/nfsv41/mover/NfsExcecutionService.java

@@ -1,13 +1,15 @@
 package org.dcache.chimera.nfsv41.mover;
 
+import com.google.common.util.concurrent.AbstractFuture;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
 import org.ietf.jgss.GSSException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
-import java.nio.channels.CompletionHandler;
 import java.util.List;
 
 import diskCacheV111.vehicles.PoolPassiveIoFileMessage;
@@ -17,7 +19,6 @@
 import org.dcache.cells.CellMessageSender;
 import org.dcache.chimera.ChimeraFsException;
 import org.dcache.chimera.nfs.v4.xdr.stateid4;
-import org.dcache.pool.classic.Cancelable;
 import org.dcache.pool.classic.MoverExecutorService;
 import org.dcache.pool.classic.PoolIORequest;
 import org.dcache.pool.classic.PoolIOTransfer;
@@ -62,8 +63,7 @@ public void shutdown() throws IOException {
     }
 
     @Override
-    public Cancelable execute(PoolIORequest request, final CompletionHandler completionHandler) {
-
+    public ListenableFuture<Void> execute(PoolIORequest request) {
         try {
             NFS4ProtocolInfo nfs4ProtocolInfo = (NFS4ProtocolInfo) request.getTransfer().getProtocolInfo();
             PoolIOTransfer transfer = request.getTransfer();
@@ -80,22 +80,26 @@ public Cancelable execute(PoolIORequest request, final CompletionHandler complet
             request.sendToDoor(new PoolPassiveIoFileMessage<>(
                     request.getPoolAddress().getCellName(), _localSocketAddresses, stateid));
 
-            return new Cancelable() {
+            /* An NFS mover doesn't complete until it is cancelled (the door sends a mover kill
+             * message when the file is closed).
+             */
+            return new AbstractFuture<Void>() {
                 @Override
-                public void cancel() {
+                public boolean cancel(boolean mayInterruptIfRunning)
+                {
                     _nfsIO.removeHandler(moverBridge);
                     try {
                         repositoryChannel.close();
                     } catch (IOException e) {
                         _log.error("failed to close RAF", e);
                     }
-                    completionHandler.completed(null, null);
+                    set(null);
+                    return false;
                 }
             };
-        } catch (Throwable e) {
-            completionHandler.failed(e, null);
+        } catch (Exception e) {
+            return Futures.immediateFailedFuture(e);
         }
-        return null;
     }
 
     public void setEnableGss(boolean withGss) {

modules/dcache/src/main/java/org/dcache/pool/classic/Cancelable.java

@@ -25,5 +25,5 @@ public interface IoProcessable {
      *
      * @return true if successful, false otherwise
      */
-    public boolean kill();
+    public void kill();
 }

modules/dcache/src/main/java/org/dcache/pool/classic/IoProcessable.java

@@ -1,12 +1,14 @@
 package org.dcache.pool.classic;
 
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.nio.channels.CompletionHandler;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Callable;
 import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
 
 import dmg.cells.nucleus.CDC;
 
@@ -21,94 +23,50 @@ public class LegacyMoverExecutorService implements MoverExecutorService
     private final static String _name =
         LegacyMoverExecutorService.class.getSimpleName();
 
-    private final ExecutorService _executor =
-        Executors.newCachedThreadPool(
-            new ThreadFactory()
-            {
-                private int _counter;
-
-                private ThreadFactory _factory =
-                    Executors.defaultThreadFactory();
-
-                @Override
-                public Thread newThread(Runnable r) {
-                    Thread t = _factory.newThread(r);
-                    t.setName(_name + "-worker-" + ++_counter);
-                    return t;
-                }
-            });
+    private final ListeningExecutorService _executor =
+            MoreExecutors.listeningDecorator(Executors.newCachedThreadPool(
+                    new ThreadFactoryBuilder().setNameFormat(_name + "-worker-%d").build()));
 
     @Override
-    public Cancelable execute(final PoolIORequest request,
-                          final CompletionHandler completionHandler)
+    public ListenableFuture<Void> execute(PoolIORequest request)
     {
-        final MoverTask moverTask = new MoverTask(request, completionHandler);
-        _executor.execute(moverTask);
-        return moverTask;
+        return _executor.submit(new MoverTask(request));
     }
 
     public void shutdown()
     {
         _executor.shutdown();
     }
 
-    private static class MoverTask implements Runnable, Cancelable {
-
+    private static class MoverTask implements Callable<Void>
+    {
         private final PoolIORequest _request;
         private final CDC _cdc = new CDC();
-        private final CompletionHandler _completionHandler;
 
-        private Thread _thread;
-        private boolean _needInterruption;
-
-        public MoverTask(PoolIORequest request, CompletionHandler completionHandler) {
+        public MoverTask(PoolIORequest request) {
             _request = request;
-            _completionHandler = completionHandler;
         }
 
         @Override
-        public void run() {
+        public Void call() throws Exception {
             try {
-                setThread();
                 _cdc.restore();
-                try {
-                    _request.getTransfer().transfer();
-                } catch (Throwable e) {
-                    _completionHandler.failed(e, null);
-                    throw e;
-                }
-                _completionHandler.completed(null, null);
+                _request.getTransfer().transfer();
+                return null;
+            } catch (RuntimeException e) {
+                _log.error("Transfer failed due to a bug: {}", e);
+                throw e;
             } catch (Exception e) {
                 _log.error("Transfer failed: {}", e.toString());
+                throw e;
             } catch (Throwable e) {
                 _log.error("Transfer failed:", e);
                 Thread t = Thread.currentThread();
                 t.getUncaughtExceptionHandler().uncaughtException(t, e);
+                throw e;
             } finally {
-                cleanThread();
                 CDC.clear();
             }
         }
-
-        private synchronized void setThread() throws InterruptedException {
-            if(_needInterruption) {
-                throw new InterruptedException("Thread interrupted before excecution");
-            }
-
-            _thread = Thread.currentThread();
-        }
-
-        private synchronized void cleanThread() {
-            _thread = null;
-        }
-
-        @Override
-        public synchronized void cancel() {
-            if (_thread != null) {
-                _thread.interrupt();
-            } else {
-                _needInterruption = true;
-            }
-        }
     }
 }

modules/dcache/src/main/java/org/dcache/pool/classic/LegacyMoverExecutorService.java

@@ -1,12 +1,11 @@
 package org.dcache.pool.classic;
 
-import java.nio.channels.CompletionHandler;
+import com.google.common.util.concurrent.ListenableFuture;
 
 /**
  *
  * @since 1.9.11
  */
 public interface MoverExecutorService {
-
-    Cancelable execute(PoolIORequest transfer, CompletionHandler completionHandler);
+    ListenableFuture<Void> execute(PoolIORequest transfer);
 }

modules/dcache/src/main/java/org/dcache/pool/classic/MoverExecutorService.java

@@ -1,11 +1,15 @@
 package org.dcache.pool.classic;
 
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.io.Serializable;
-import java.nio.channels.CompletionHandler;
+import java.util.concurrent.CancellationException;
 
 import diskCacheV111.util.CacheException;
 import diskCacheV111.util.PnfsId;
@@ -17,6 +21,8 @@
 import dmg.cells.nucleus.NoRouteToCellException;
 
 import org.dcache.cells.CellStub;
+import org.dcache.pool.FaultAction;
+import org.dcache.pool.FaultEvent;
 import org.dcache.pool.FaultListener;
 import org.dcache.vehicles.FileAttributes;
 
@@ -42,7 +48,7 @@ public class PoolIORequest implements IoProcessable {
     private final static Logger _log = LoggerFactory.getLogger(PoolIORequest.class);
     private final FaultListener _faultListener;
 
-    private Cancelable _mover;
+    private ListenableFuture<Void> _mover;
     /**
      * Request creation time.
      */
@@ -61,7 +67,6 @@ public class PoolIORequest implements IoProcessable {
     /** transfer status error message */
     private volatile String _errorMessage = "";
 
-    private boolean _canceled;
     private final CellStub _billing;
     private final CellStub _door;
     private final CellAddressCore _pool;
@@ -176,26 +181,54 @@ public long getClientId() {
     }
 
     @Override
-    public synchronized boolean kill() {
+    public synchronized void kill() {
         _state = CANCELED;
-        _canceled = true;
-
-        if (_mover == null) {
-            return false;
+        if (_mover != null) {
+            _mover.cancel(true);
         }
-
-        _mover.cancel();
-        return true;
     }
 
-    synchronized void transfer(MoverExecutorService moverExecutorService, CompletionHandler<Object,Object> completionHandler) {
-        _startTime = System.currentTimeMillis();
-        if(_canceled) {
-            completionHandler.failed( new InterruptedException("Mover canceled"), null);
-        } else {
-            _state = RUNNING;
-            _mover = moverExecutorService.execute(this, completionHandler);
+    synchronized ListenableFuture<Void> transfer(MoverExecutorService moverExecutorService) {
+        if (_state != QUEUED) {
+            return Futures.immediateFailedFuture(new InterruptedException("Mover canceled"));
         }
+
+        _state = RUNNING;
+        _startTime = System.currentTimeMillis();
+        _mover = moverExecutorService.execute(this);
+        final SettableFuture<Void> future = SettableFuture.create();
+        Futures.addCallback(_mover,
+                new FutureCallback<Void>()
+                {
+                    @Override
+                    public void onSuccess(Void result)
+                    {
+                        future.set(null);
+                    }
+
+                    @Override
+                    public void onFailure(Throwable e)
+                    {
+                        int rc;
+                        String msg;
+                        if (e instanceof CancellationException) {
+                            rc = CacheException.DEFAULT_ERROR_CODE;
+                            msg = "Transfer was killed";
+                        } else if (e instanceof CacheException) {
+                            rc = ((CacheException) e).getRc();
+                            msg = e.getMessage();
+                            if (rc == CacheException.ERROR_IO_DISK) {
+                                getFaultListener().faultOccurred(new FaultEvent("repository", FaultAction.DISABLED, msg, e));
+                            }
+                        } else {
+                            rc = CacheException.UNEXPECTED_SYSTEM_EXCEPTION;
+                            msg = "Transfer failed due to unexpected exception: " + e;
+                        }
+                        setTransferStatus(rc, msg);
+                        future.set(null);
+                    }
+                });
+        return future;
     }
 
     void close()