dcache: make admin pin command asynchronous

Motivation:
When pinning a file via admin interface and that file needs to be staged, the call blocks until it finishes.

Modification:
Make the admin-triggered pin request asynchronous.

Result:
A pin request via admin interface does no longer wait for the pin to be established before returning.

Target: master
Requires-notes: no
Requires-book: no
Patch: https://rb.dcache.org/r/14152/
Acked-by: Dmitry Litvintsev

modules/dcache/src/main/java/org/dcache/pinmanager/PinManagerCLI.java

@@ -96,22 +96,22 @@ public void setDao(PinDao dao) {
     }
 
     private Future<PinManagerPinMessage>
-    pin(PnfsId pnfsId, String requestId, long lifetime)
+    pin(PnfsId pnfsId, String requestId, long lifetime, boolean replyWhenStarted)
           throws CacheException {
         DCapProtocolInfo protocolInfo =
               new DCapProtocolInfo("DCap", 3, 0, new InetSocketAddress("localhost", 0));
         PinManagerPinMessage message = new PinManagerPinMessage(FileAttributes.ofPnfsId(pnfsId),
               protocolInfo, requestId, lifetime);
+        message.setReplyWhenStarted(replyWhenStarted);
         return _pinProcessor.messageArrived(message);
     }
 
     @Command(name = "pin", hint = "pin a file to disk",
           description = "Pins a file to disk for some time. A file may be pinned forever by " +
                 "specifying a lifetime of -1. Pinning a file may involve staging it " +
                 "or copying it from one pool to another. For that reason pinning may " +
-                "take awhile and the pin command may time out. The pin request will " +
-                "however stay active and progress may be tracked by listing the pins " +
-                "on the file.")
+                "take awhile. The pin command will return immediately, but the pin request " +
+                "will stay active and progress may be tracked by listing the pins on the file.")
     public class PinCommand implements Callable<String> {
 
         @Argument(index = 0)
@@ -124,13 +124,8 @@ public class PinCommand implements Callable<String> {
         public String call()
               throws CacheException, ExecutionException, InterruptedException {
             long millis = (lifetime == -1) ? -1 : TimeUnit.SECONDS.toMillis(lifetime);
-            PinManagerPinMessage message = pin(pnfsId, null, millis).get();
-            if (message.getExpirationTime() == null) {
-                return String.format("[%d] %s pinned", message.getPinId(), pnfsId);
-            } else {
-                return String.format("[%d] %s pinned until %tc",
-                      message.getPinId(), pnfsId,
-                      message.getExpirationTime());
+            PinManagerPinMessage message = pin(pnfsId, null, millis, true).get();
+            return String.format("A pin request for %s has been sent", pnfsId);
             }
         }
     }
@@ -526,7 +521,7 @@ public void run() {
             List<PnfsId> list = new ArrayList(_tasks.keySet());
             for (PnfsId pnfsId : list) {
                 try {
-                    _tasks.put(pnfsId, pin(pnfsId, _requestId, _lifetime));
+                    _tasks.put(pnfsId, pin(pnfsId, _requestId, _lifetime, false));
                 } catch (CacheException e) {
                     _tasks.remove(pnfsId);
                     _errors.append("    ").append(pnfsId).