HDFS-6788. Improve synchronization in BPOfferService with read write lock. Contributed by Yongjun Zhang.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1615190 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -349,6 +349,9 @@ Release 2.6.0 - UNRELEASED
     HDFS-6802. Some tests in TestDFSClientFailover are missing @Test
     annotation. (Akira Ajisaka via wang)
 
+    HDFS-6788. Improve synchronization in BPOfferService with read write lock.
+    (Yongjun Zhang via wang)
+
   OPTIMIZATIONS
 
     HDFS-6690. Deduplicate xattr names in memory. (wang)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPOfferService.java

@@ -21,6 +21,7 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
+
 import org.apache.commons.logging.Log;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
@@ -38,6 +39,8 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 /**
  * One instance per block-pool/namespace on the DN, which handles the
@@ -91,6 +94,28 @@ class BPOfferService {
    */
   private long lastActiveClaimTxId = -1;
 
+  private final ReentrantReadWriteLock mReadWriteLock =
+      new ReentrantReadWriteLock();
+  private final Lock mReadLock  = mReadWriteLock.readLock();
+  private final Lock mWriteLock = mReadWriteLock.writeLock();
+
+  // utility methods to acquire and release read lock and write lock
+  void readLock() {
+    mReadLock.lock();
+  }
+
+  void readUnlock() {
+    mReadLock.unlock();
+  }
+
+  void writeLock() {
+    mWriteLock.lock();
+  }
+
+  void writeUnlock() {
+    mWriteLock.unlock();
+  }
+
   BPOfferService(List<InetSocketAddress> nnAddrs, DataNode dn) {
     Preconditions.checkArgument(!nnAddrs.isEmpty(),
         "Must pass at least one NN.");
@@ -136,7 +161,9 @@ class BPOfferService {
     return false;
   }
 
-  synchronized String getBlockPoolId() {
+  String getBlockPoolId() {
+    readLock();
+    try {
       if (bpNSInfo != null) {
         return bpNSInfo.getBlockPoolID();
       } else {
@@ -144,18 +171,28 @@ class BPOfferService {
             new Exception("trace"));
         return null;
       }
+    } finally {
+      readUnlock();
+    }
   }
 
   boolean hasBlockPoolId() {
     return getNamespaceInfo() != null;
   }
 
-  synchronized NamespaceInfo getNamespaceInfo() {
+  NamespaceInfo getNamespaceInfo() {
+    readLock();
+    try {
       return bpNSInfo;
+    } finally {
+      readUnlock();
+    }
   }
 
   @Override
-  public synchronized String toString() {
+  public String toString() {
+    readLock();
+    try {
       if (bpNSInfo == null) {
         // If we haven't yet connected to our NN, we don't yet know our
         // own block pool ID.
@@ -172,6 +209,9 @@ class BPOfferService {
             " (Datanode Uuid " + dn.getDatanodeUuid() +
             ")";
       }
+    } finally {
+      readUnlock();
+    }
   }
   
   void reportBadBlocks(ExtendedBlock block,
@@ -266,7 +306,9 @@ class BPOfferService {
    * verifies that this namespace matches (eg to prevent a misconfiguration
    * where a StandbyNode from a different cluster is specified)
    */
-  synchronized void verifyAndSetNamespaceInfo(NamespaceInfo nsInfo) throws IOException {
+  void verifyAndSetNamespaceInfo(NamespaceInfo nsInfo) throws IOException {
+    writeLock();
+    try {
       if (this.bpNSInfo == null) {
         this.bpNSInfo = nsInfo;
         boolean success = false;
@@ -293,6 +335,9 @@ class BPOfferService {
         checkNSEquality(bpNSInfo.getClusterID(), nsInfo.getClusterID(),
             "Cluster ID");
       }
+    } finally {
+      writeUnlock();
+    }
   }
 
   /**
@@ -300,8 +345,10 @@ class BPOfferService {
    * NN, it calls this function to verify that the NN it connected to
    * is consistent with other NNs serving the block-pool.
    */
-  synchronized void registrationSucceeded(BPServiceActor bpServiceActor,
+  void registrationSucceeded(BPServiceActor bpServiceActor,
       DatanodeRegistration reg) throws IOException {
+    writeLock();
+    try {
       if (bpRegistration != null) {
         checkNSEquality(bpRegistration.getStorageInfo().getNamespaceID(),
             reg.getStorageInfo().getNamespaceID(), "namespace ID");
@@ -317,6 +364,9 @@ class BPOfferService {
         dn.blockPoolTokenSecretManager.addKeys(getBlockPoolId(),
             reg.getExportedKeys());
       }
+    } finally {
+      writeUnlock();
+    }
   }
 
   /**
@@ -333,17 +383,24 @@ class BPOfferService {
     }
   }
 
-  synchronized DatanodeRegistration createRegistration() {
+  DatanodeRegistration createRegistration() {
+    writeLock();
+    try {
       Preconditions.checkState(bpNSInfo != null,
           "getRegistration() can only be called after initial handshake");
       return dn.createBPRegistration(bpNSInfo);
+    } finally {
+      writeUnlock();
+    }
   }
 
   /**
    * Called when an actor shuts down. If this is the last actor
    * to shut down, shuts down the whole blockpool in the DN.
    */
-  synchronized void shutdownActor(BPServiceActor actor) {
+  void shutdownActor(BPServiceActor actor) {
+    writeLock();
+    try {
       if (bpServiceToActive == actor) {
         bpServiceToActive = null;
       }
@@ -353,6 +410,9 @@ class BPOfferService {
       if (bpServices.isEmpty()) {
         dn.shutdownBlockPool(this);
       }
+    } finally {
+      writeUnlock();
+    }
   }
 
   /**
@@ -392,12 +452,17 @@ class BPOfferService {
    * @return a proxy to the active NN, or null if the BPOS has not
    * acknowledged any NN as active yet.
    */
-  synchronized DatanodeProtocolClientSideTranslatorPB getActiveNN() {
+  DatanodeProtocolClientSideTranslatorPB getActiveNN() {
+    readLock();
+    try {
       if (bpServiceToActive != null) {
         return bpServiceToActive.bpNamenode;
       } else {
         return null;
       }
+    } finally {
+      readUnlock();
+    }
   }
 
   @VisibleForTesting
@@ -424,9 +489,11 @@ class BPOfferService {
    * @param actor the actor which received the heartbeat
    * @param nnHaState the HA-related heartbeat contents
    */
-  synchronized void updateActorStatesFromHeartbeat(
+  void updateActorStatesFromHeartbeat(
       BPServiceActor actor,
       NNHAStatusHeartbeat nnHaState) {
+    writeLock();
+    try {
       final long txid = nnHaState.getTxId();
 
       final boolean nnClaimsActive =
@@ -464,6 +531,9 @@ class BPOfferService {
         assert txid >= lastActiveClaimTxId;
         lastActiveClaimTxId = txid;
       }
+    } finally {
+      writeUnlock();
+    }
   }
 
   /**
@@ -533,12 +603,15 @@ class BPOfferService {
       actor.reRegister();
       return true;
     }
-    synchronized (this) {
+    writeLock();
+    try {
       if (actor == bpServiceToActive) {
         return processCommandFromActive(cmd, actor);
       } else {
         return processCommandFromStandby(cmd, actor);
       }
+    } finally {
+      writeUnlock();
     }
   }