module: summary line

work in progress...
- separating pin request handling from actual pinning
- later patch (?): pick up pins in state PINNING when HA master changes

Motivation:

Modification:

Result:

Target: master
Request: branch
Patch: rb-entry
Fixes: #github-issue
Requires-notes: no
Requires-book: no
Acked-by:

Author: lemora

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

@@ -6,6 +6,7 @@
 import static org.dcache.pinmanager.model.Pin.State.FAILED_TO_UNPIN;
 import static org.dcache.pinmanager.model.Pin.State.PINNED;
 import static org.dcache.pinmanager.model.Pin.State.PINNING;
+import static org.dcache.pinmanager.model.Pin.State.READY_TO_PIN;
 import static org.dcache.pinmanager.model.Pin.State.READY_TO_UNPIN;
 import static org.dcache.pinmanager.model.Pin.State.UNPINNING;
 
@@ -35,6 +36,7 @@ public class PinManager implements CellMessageReceiver, LeaderLatchListener, Cel
 
     private static final Logger LOGGER = LoggerFactory.getLogger(PinManager.class);
     private static final long INITIAL_EXPIRATION_DELAY = SECONDS.toMillis(15);
+    private static final long INITIAL_PIN_DELAY = SECONDS.toMillis(30);
     private static final long INITIAL_UNPIN_DELAY = SECONDS.toMillis(30);
 
     private ScheduledExecutorService executor;
@@ -44,7 +46,8 @@ public class PinManager implements CellMessageReceiver, LeaderLatchListener, Cel
 
     private long expirationPeriod;
     private TimeUnit expirationPeriodUnit;
-    // Period in which to reset all pins that failed to be unpinned from state FAILED_TO_UNPIN to READY_TO_UNPIN
+
+    /* Period in which to reset all pins that failed to be unpinned from state FAILED_TO_UNPIN to READY_TO_UNPIN */
     private Duration resetFailedUnpinsPeriod;
     private int maxUnpinsPerRun = -1;
 
@@ -94,11 +97,32 @@ public PnfsDeleteEntryNotificationMessage messageArrived(
         return message;
     }
 
+    /**
+     * Resets all pins in state PINNING to READY_TO_PIN, because after a PinManager HA master
+     * change, these are abandoned and need to be retried by the new HA master.
+     */
+    private void resetTransitivePinningStates() {
+        try {
+            // TODO: cancel these pins in PoolManager!
+            dao.update(dao.where()
+                        .state(PINNING),
+                  dao.set()
+                        .state(READY_TO_PIN)
+                        .pool(null));
+        } catch (JDOException | DataAccessException e) {
+            LOGGER.error("Database failure while trying to reset failed PINNING: {}",
+                  e.getMessage());
+        } catch (RuntimeException e) {
+            LOGGER.error("Unexpected failure while resetting PINNING pins", e);
+        }
+    }
+
     /**
      * Resets all pins in state UNPINNING and FAILED_TO_UNPIN to READY_TO_UNPIN.
      */
-    private void markAllExpiredPinsReadyToUnpin() {
+    private void resetTransitiveUnpinningStates() {
         dao.update(dao.where()
+                    .stateIsNot(READY_TO_PIN)
                     .stateIsNot(PINNING)
                     .stateIsNot(PINNED)
                     .stateIsNot(READY_TO_UNPIN),
@@ -107,18 +131,19 @@ private void markAllExpiredPinsReadyToUnpin() {
     }
 
     /**
-     * This task transitions all pins that have exceeded their lifetime and are in state PINNING or
-     * PINNED to state READY_TO_UNPIN. It removes the pool, which expires pins on its own and does
-     * not need to be contacted for regular expiries. As PoolManager is aware of the timeout for
-     * pins in state PINNING, it should also delete the request on its own if it is still ongoing.
+     * This task transitions all pins that have exceeded their lifetime and are in state
+     * READY_TO_PIN, PINNING or PINNED to state READY_TO_UNPIN. It removes the pool field, as the
+     * pool will remove these sticky bits on its own and does not need to be contacted for regular
+     * expiries. As PoolManager is aware of the timeout for PINNING requests, PoolManager should
+     * delete these ongoing request on its own as well.
      */
     private class ExpirationTask implements Runnable {
 
-        private AtomicInteger count = new AtomicInteger();
+        private final AtomicInteger count = new AtomicInteger();
 
         @Override
         public void run() {
-            NDC.push("BackgroundExpiration-" + count.incrementAndGet());
+            NDC.push("PinExpiration-" + count.incrementAndGet());
             try {
                 dao.update(dao.where()
                             .expirationTimeBefore(new Date())
@@ -145,11 +170,11 @@ public void run() {
      */
     private class ResetFailedUnpinsTask implements Runnable {
 
-        private AtomicInteger count = new AtomicInteger();
+        private final AtomicInteger count = new AtomicInteger();
 
         @Override
         public void run() {
-            NDC.push("BackgroundResetFailedUnpins-" + count.incrementAndGet());
+            NDC.push("UnpinExpiration-" + count.incrementAndGet());
             try {
                 dao.update(dao.where()
                             .state(FAILED_TO_UNPIN),
@@ -168,30 +193,40 @@ public void run() {
         }
     }
 
+    // create PinProcessor here or inject? dao issue
+    private PinProcessor pinTask;
     private UnpinProcessor unpinTask;
     private final ExpirationTask expirationTask = new ExpirationTask();
     private final ResetFailedUnpinsTask resetFailedUnpinsTask = new ResetFailedUnpinsTask();
 
+    private ScheduledFuture<?> pinFuture;
     private ScheduledFuture<?> unpinFuture;
     private ScheduledFuture<?> expirationFuture;
     private ScheduledFuture<?> resetFailedUnpinsFuture;
 
     public void init() {
         // Needs to be assigned after dao has been initialized
+        pinTask = new PinProcessor(dao, poolStub, poolMonitor, maxUnpinsPerRun);
         unpinTask = new UnpinProcessor(dao, poolStub, poolMonitor, maxUnpinsPerRun);
     }
 
     @Override
     public void isLeader() {
         LOGGER.info("Resetting existing intermediate pin states.");
-        markAllExpiredPinsReadyToUnpin();
+        resetTransitivePinningStates();
+        resetTransitiveUnpinningStates();
 
         LOGGER.info("Scheduling Expiration and Unpin tasks.");
         expirationFuture = executor.scheduleWithFixedDelay(
               new FireAndForgetTask(expirationTask),
               INITIAL_EXPIRATION_DELAY,
               expirationPeriodUnit.toMillis(expirationPeriod),
               MILLISECONDS);
+        pinFuture = executor.scheduleWithFixedDelay(
+              new FireAndForgetTask(pinTask),
+              INITIAL_PIN_DELAY,
+              expirationPeriodUnit.toMillis(expirationPeriod),
+              MILLISECONDS);
         unpinFuture = executor.scheduleWithFixedDelay(
               new FireAndForgetTask(unpinTask),
               INITIAL_UNPIN_DELAY,
@@ -208,16 +243,18 @@ public void isLeader() {
     public void notLeader() {
         LOGGER.info("Cancelling Expiration, ResetFailedUnpins and Unpin tasks.");
         expirationFuture.cancel(false);
+        pinFuture.cancel(true);
         unpinFuture.cancel(true);
         resetFailedUnpinsFuture.cancel(true);
     }
 
     @Override
     public void getInfo(PrintWriter pw) {
-        pw.printf("Expiration and unpin period:            %s %s\n", expirationPeriod,
+        pw.printf("Period for expiration and unpinning: %s %s\n", expirationPeriod,
               expirationPeriodUnit);
-        pw.printf("Reset pins that failed to unpin period: %s\n",
+        pw.printf("Period for pinning: %s %s\n", expirationPeriod, expirationPeriodUnit);
+        pw.printf("Max. unpin operations per run: %s\n", maxUnpinsPerRun);
+        pw.printf("Period for resetting pins that failed to unpin: %s\n",
               TimeUtils.describe(resetFailedUnpinsPeriod).orElse("-"));
-        pw.printf("Max unpin operations per run:           %s\n", maxUnpinsPerRun);
     }
 }

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

@@ -54,11 +54,10 @@ public class PinManagerCLI
 
     private PnfsHandler _pnfs;
     private PoolMonitor _poolMonitor;
-    private PinManager _pinManager;
     private PinDao _dao;
-    private PinRequestProcessor _pinProcessor;
-    private UnpinRequestProcessor _unpinProcessor;
-    private MovePinRequestProcessor _moveProcessor;
+    private PinRequestProcessor _pinRequestProcessor;
+    private UnpinRequestProcessor _unpinRequestProcessor;
+    private MovePinRequestProcessor _moveRequestProcessor;
 
     @Required
     public void setPnfsStub(CellStub stub) {
@@ -70,24 +69,19 @@ public void setPoolMonitor(PoolMonitor poolMonitor) {
         _poolMonitor = poolMonitor;
     }
 
-    @Required
-    public void setPinManager(PinManager pinManager) {
-        _pinManager = pinManager;
-    }
-
     @Required
     public void setPinProcessor(PinRequestProcessor processor) {
-        _pinProcessor = processor;
+        _pinRequestProcessor = processor;
     }
 
     @Required
     public void setUnpinProcessor(UnpinRequestProcessor processor) {
-        _unpinProcessor = processor;
+        _unpinRequestProcessor = processor;
     }
 
     @Required
     public void setMoveProcessor(MovePinRequestProcessor processor) {
-        _moveProcessor = processor;
+        _moveRequestProcessor = processor;
     }
 
     @Required
@@ -103,7 +97,7 @@ public void setDao(PinDao dao) {
         PinManagerPinMessage message = new PinManagerPinMessage(FileAttributes.ofPnfsId(pnfsId),
               protocolInfo, requestId, lifetime);
         message.setReplyWhenStarted(replyWhenStarted);
-        return _pinProcessor.messageArrived(message);
+        return _pinRequestProcessor.messageArrived(message);
     }
 
     @Command(name = "pin", hint = "pin a file to disk",
@@ -149,7 +143,7 @@ public String call() throws NumberFormatException, CacheException {
             if (!pin.equals("*")) {
                 message.setPinId(Long.parseLong(pin));
             }
-            _unpinProcessor.messageArrived(message);
+            _unpinRequestProcessor.messageArrived(message);
             return "The pin is now scheduled for removal";
         }
     }
@@ -179,7 +173,7 @@ public String call() throws CacheException, InterruptedException {
                   _pnfs.getFileAttributes(pnfsId, attributes);
             PinManagerExtendPinMessage message =
                   new PinManagerExtendPinMessage(fileAttributes, pin, millis);
-            message = _moveProcessor.messageArrived(message);
+            message = _moveRequestProcessor.messageArrived(message);
             if (message.getExpirationTime() == null) {
                 return String.format("[%d] %s pinned",
                       message.getPinId(), pnfsId);
@@ -400,7 +394,7 @@ public String call() throws IOException {
                 try {
                     PinManagerUnpinMessage message =
                           new PinManagerUnpinMessage(pnfsId);
-                    _unpinProcessor.messageArrived(message);
+                    _unpinRequestProcessor.messageArrived(message);
                 } catch (CacheException e) {
                     out.append(pnfsId).append(": ").append(e.getMessage()).append('\n');
                 }
@@ -544,7 +538,7 @@ public synchronized void cancel()
                     PinManagerUnpinMessage message =
                           new PinManagerUnpinMessage(pnfsId);
                     message.setRequestId(_requestId);
-                    _unpinProcessor.messageArrived(message);
+                    _unpinRequestProcessor.messageArrived(message);
                 }
                 _cancelled = true;
             }

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

@@ -7,10 +7,11 @@
 import diskCacheV111.vehicles.Message;
 import diskCacheV111.vehicles.PoolMgrSelectReadPoolMsg;
 import diskCacheV111.vehicles.ProtocolInfo;
-
 import java.io.ObjectStreamException;
 import java.util.Date;
 import java.util.EnumSet;
+import java.util.Optional;
+import jline.internal.Nullable;
 import org.dcache.auth.attributes.Restriction;
 import org.dcache.auth.attributes.Restrictions;
 import org.dcache.namespace.FileAttribute;
@@ -29,8 +30,8 @@ public class PinManagerPinMessage extends Message {
     private String _pool;
     private final String _requestId;
     private Date _expirationTime;
-    private boolean _replyWhenStarted;
     private boolean _denyStaging;
+    private boolean _pinningInProgress;
 
     public PinManagerPinMessage(FileAttributes fileAttributes,
           ProtocolInfo protocolInfo,
@@ -49,26 +50,6 @@ public PinManagerPinMessage(FileAttributes fileAttributes,
           String requestId,
           long lifetime) {
         this(fileAttributes, protocolInfo, Restrictions.none(), requestId, lifetime);
-     }
-
-    /**
-     * Choose whether to wait for the pin to be established before returning. If value is true then
-     * do not wait for file to be staged, but return straight away.  Calling getPool, getPinId and
-     * getExpirationTime will all return null, unless this request is a retry and the original
-     * request succeeded (have establishing a pin) in which case the details of that pin are
-     * available through those methods.
-     * <p>
-     * If set to false (the default) then the message returns once the pin request has been
-     * processed and the pin established, or if there was an error.
-     *
-     * @param value whether to reply after the pinning task has been started.
-     */
-    public void setReplyWhenStarted(boolean value) {
-        _replyWhenStarted = value;
-    }
-
-    public boolean isReplyWhenStarted() {
-        return _replyWhenStarted;
     }
 
     public void setDenyStaging(boolean value) {
@@ -107,8 +88,11 @@ public ProtocolInfo getProtocolInfo() {
         return _protocolInfo;
     }
 
-    public Restriction getRestriction() { return _restriction; }
+    public Restriction getRestriction() {
+        return _restriction;
+    }
 
+    @Nullable
     public String getPool() {
         return _pool;
     }
@@ -117,14 +101,22 @@ public void setPool(String pool) {
         _pool = pool;
     }
 
-    public long getPinId() {
-        return _pinId;
+    public Optional<Long> getPinId() {
+        return _pinId == 0 ? Optional.empty() : Optional.of(_pinId);
     }
 
     public void setPinId(long pinId) {
         _pinId = pinId;
     }
 
+    public void setPinningInProgress(boolean pinningInProgress) {
+        _pinningInProgress = pinningInProgress;
+    }
+
+    public boolean isPinningInProgress() {
+        return _pinningInProgress;
+    }
+
     public void setExpirationTime(Date expirationTime) {
         _expirationTime = expirationTime;
     }
@@ -134,15 +126,16 @@ public Date getExpirationTime() {
     }
 
     public void setPin(Pin pin) {
+        _pinningInProgress = false;
         setPool(pin.getPool());
         setPinId(pin.getPinId());
         setExpirationTime(pin.getExpirationTime());
     }
 
     @Override
     public String toString() {
-        return "PinManagerPinMessage[" + _fileAttributes + "," +
-              _protocolInfo + "," + _lifetime + "]";
+        return "PinManagerPinMessage[" + _fileAttributes + "," + _protocolInfo + "," + _lifetime
+              + "]";
     }
 
     public static EnumSet<FileAttribute> getRequiredAttributes() {
@@ -152,7 +145,9 @@ public static EnumSet<FileAttribute> getRequiredAttributes() {
     }
 
     private Object readResolve() throws ObjectStreamException {
-        if (_restriction == null) { _restriction = Restrictions.none(); }
+        if (_restriction == null) {
+            _restriction = Restrictions.none();
+        }
         return this;
     }