pool enable on flight checksum calculation for nfs

Acked-by: Paul Millar Target: master Require-notes: yes Require-book: no
dCache · Feb 18, 2015 · 8f8cbca · 8f8cbca
1 parent 559a5c6
commit 8f8cbca
Show file tree

Hide file tree

Showing 3 changed files with 63 additions and 5 deletions.
diff --git a/modules/dcache-nfs/src/main/java/org/dcache/chimera/nfsv41/mover/NfsMover.java b/modules/dcache-nfs/src/main/java/org/dcache/chimera/nfsv41/mover/NfsMover.java
@@ -20,12 +20,16 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.base.Optional;
+import com.google.common.base.Throwables;
+
 import java.io.IOException;
 import java.nio.channels.CompletionHandler;
 import java.util.Collections;
 import java.util.Set;
 
 import diskCacheV111.util.CacheException;
+import diskCacheV111.util.ChecksumFactory;
 import diskCacheV111.util.DiskErrorCacheException;
 import diskCacheV111.util.PnfsHandler;
 import diskCacheV111.vehicles.PoolIoFileMessage;
@@ -37,12 +41,18 @@
 import org.dcache.nfs.v4.xdr.stateid4;
 import org.dcache.nfs.v4.xdr.verifier4;
 import org.dcache.pool.classic.Cancellable;
+import org.dcache.pool.movers.ChecksumChannel;
+import org.dcache.pool.movers.IoMode;
+
 import org.dcache.pool.movers.MoverChannel;
 import org.dcache.pool.movers.MoverChannelMover;
 import org.dcache.pool.repository.ReplicaDescriptor;
+import org.dcache.pool.repository.RepositoryChannel;
 import org.dcache.util.Checksum;
 import org.dcache.vehicles.FileAttributes;
 
+import static com.google.common.base.Preconditions.checkState;
+
 public class NfsMover extends MoverChannelMover<NFS4ProtocolInfo, NfsMover> {
 
     private static final Logger _log = LoggerFactory.getLogger(NfsTransferService.class);
@@ -53,25 +63,54 @@ public class NfsMover extends MoverChannelMover<NFS4ProtocolInfo, NfsMover> {
     private volatile CompletionHandler<Void, Void> _completionHandler;
     private final verifier4 _bootVerifier;
 
+    private final ChecksumFactory _checksumFactory;
+    private ChecksumChannel _checksumChannel;
+
     public NfsMover(ReplicaDescriptor handle, PoolIoFileMessage message, CellPath pathToDoor,
-            NfsTransferService nfsTransferService, PnfsHandler pnfsHandler) {
+            NfsTransferService nfsTransferService, PnfsHandler pnfsHandler,
+            ChecksumFactory checksumFactory) {
         super(handle, message, pathToDoor, nfsTransferService, MoverChannel.AllocatorMode.SOFT);
         _nfsIO = nfsTransferService.getNfsMoverHandler();
         _state = new MoverState();
         _namespace = pnfsHandler;
         _bootVerifier = nfsTransferService.getBootVerifier();
+        _checksumFactory = checksumFactory;
     }
 
     @Override
     public Set<Checksum> getActualChecksums() {
-        return Collections.emptySet();
+        return (_checksumChannel == null)
+                ? Collections.<Checksum>emptySet()
+                : Optional.fromNullable(_checksumChannel.getChecksum()).asSet();
     }
 
     @Override
     public Set<Checksum> getExpectedChecksums() {
         return Collections.emptySet();
     }
 
+    @Override
+    public synchronized RepositoryChannel openChannel() throws DiskErrorCacheException {
+        checkState(_checksumChannel == null);
+        RepositoryChannel channel = super.openChannel();
+        try {
+            if (getIoMode() == IoMode.WRITE && _checksumFactory != null) {
+                channel = _checksumChannel = new ChecksumChannel(channel, _checksumFactory);
+            }
+        } catch (Throwable t) {
+            /* This should only happen in case of JVM Errors or if the checksum digest cannot be
+             * instantiated (which, barring bugs, should never happen).
+             */
+            try {
+                channel.close();
+            } catch (IOException e) {
+                t.addSuppressed(e);
+            }
+            Throwables.propagate(t);
+        }
+        return channel;
+    }
+
     public stateid4 getStateId() {
         org.dcache.chimera.nfs.v4.xdr.stateid4 legacyStateid =  getProtocolInfo().stateId();
         return new stateid4(legacyStateid.other, legacyStateid.seqid.value);

diff --git a/modules/dcache-nfs/src/main/java/org/dcache/chimera/nfsv41/mover/NfsTransferService.java b/modules/dcache-nfs/src/main/java/org/dcache/chimera/nfsv41/mover/NfsTransferService.java
@@ -34,6 +34,7 @@
 import org.dcache.pool.FaultEvent;
 import org.dcache.pool.FaultListener;
 import org.dcache.pool.classic.Cancellable;
+import org.dcache.pool.classic.ChecksumModule;
 import org.dcache.pool.classic.PostTransferService;
 import org.dcache.pool.classic.TransferService;
 import org.dcache.pool.movers.Mover;
@@ -45,8 +46,9 @@
 import org.dcache.utils.Bytes;
 import org.dcache.xdr.OncRpcException;
 
-import static com.google.common.collect.Iterables.toArray;
-import static com.google.common.collect.Iterables.transform;
+import static com.google.common.collect.Iterables.*;
+import diskCacheV111.util.ChecksumFactory;
+import java.security.NoSuchAlgorithmException;
 
 /**
  * Factory and transfer service for NFS movers.
@@ -67,6 +69,7 @@ public class NfsTransferService extends AbstractCellComponent
     private final verifier4 _bootVerifierBytes = toVerifier(_bootVerifier);
     private boolean _sortMultipathList;
     private PnfsHandler _pnfsHandler;
+    private ChecksumModule _checksumModule;
 
     public void init() throws IOException, GSSException, OncRpcException {
 
@@ -114,6 +117,11 @@ public void setDoorStub(CellStub cellStub) {
         _door = cellStub;
     }
 
+    @Required
+    public void setChecksumModule(ChecksumModule checksumModule) {
+        _checksumModule = checksumModule;
+    }
+
     public void shutdown() throws IOException {
         _nfsIO.shutdown();
         _nfsIO.getNFSServer().getStateHandler().shutdown();
@@ -122,7 +130,17 @@ public void shutdown() throws IOException {
     @Override
     public Mover<?> createMover(ReplicaDescriptor handle, PoolIoFileMessage message, CellPath pathToDoor) throws CacheException
     {
-        return new NfsMover(handle, message, pathToDoor, this, _pnfsHandler);
+        ChecksumFactory checksumFactory;
+        if (_checksumModule.hasPolicy(ChecksumModule.PolicyFlag.ON_TRANSFER)) {
+            try {
+                checksumFactory = _checksumModule.getPreferredChecksumFactory(handle);
+            } catch (NoSuchAlgorithmException e) {
+                throw new CacheException("Failed to instantiate NFS mover due to unsupported checksum type: " + e.getMessage(), e);
+            }
+        } else {
+            checksumFactory = null;
+        }
+        return new NfsMover(handle, message, pathToDoor, this, _pnfsHandler, checksumFactory);
     }
 
     @Override

diff --git a/modules/dcache/src/main/resources/org/dcache/pool/classic/pool.xml b/modules/dcache/src/main/resources/org/dcache/pool/classic/pool.xml
@@ -338,6 +338,7 @@
       <property name="pnfsHandler" ref="pnfs"/>
       <property name="faultListener" ref="pool"/>
       <property name="doorStub" ref="doorStub"/>
+      <property name="checksumModule" ref="csm"/>
   </bean>
 
   <bean id="xrootd-transfer-service" class="org.dcache.xrootd.pool.XrootdTransferService"