HDFS-12140. Remove BPOfferService lock contention to get block pool id. Contributed by Daryn Sharp.

(cherry picked from commit e7d187a1b6)

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

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

@@ -70,6 +70,7 @@ class BPOfferService {
   volatile DatanodeRegistration bpRegistration;
 
   private final String nameserviceId;
+  private volatile String bpId;
   private final DataNode dn;
 
   /**
@@ -182,6 +183,11 @@ class BPOfferService {
   }
 
   String getBlockPoolId() {
+    // avoid lock contention unless the registration hasn't completed.
+    String id = bpId;
+    if (id != null) {
+      return id;
+    }
     readLock();
     try {
       if (bpNSInfo != null) {
@@ -197,7 +203,7 @@ class BPOfferService {
   }
 
   boolean hasBlockPoolId() {
-    return getNamespaceInfo() != null;
+    return getBlockPoolId() != null;
   }
 
   NamespaceInfo getNamespaceInfo() {
@@ -209,6 +215,28 @@ class BPOfferService {
     }
   }
 
+  @VisibleForTesting
+  NamespaceInfo setNamespaceInfo(NamespaceInfo nsInfo) throws IOException {
+    writeLock();
+    try {
+      NamespaceInfo old = bpNSInfo;
+      if (bpNSInfo != null && nsInfo != null) {
+        checkNSEquality(bpNSInfo.getBlockPoolID(), nsInfo.getBlockPoolID(),
+            "Blockpool ID");
+        checkNSEquality(bpNSInfo.getNamespaceID(), nsInfo.getNamespaceID(),
+            "Namespace ID");
+        checkNSEquality(bpNSInfo.getClusterID(), nsInfo.getClusterID(),
+            "Cluster ID");
+      }
+      bpNSInfo = nsInfo;
+      // cache the block pool id for lock-free access.
+      bpId = (nsInfo != null) ? nsInfo.getBlockPoolID() : null;
+      return old;
+    } finally {
+      writeUnlock();
+    }
+  }
+
   @Override
   public String toString() {
     readLock();
@@ -281,9 +309,10 @@ class BPOfferService {
   private void checkBlock(ExtendedBlock block) {
     Preconditions.checkArgument(block != null,
         "block is null");
-    Preconditions.checkArgument(block.getBlockPoolId().equals(getBlockPoolId()),
+    final String bpId = getBlockPoolId();
+    Preconditions.checkArgument(block.getBlockPoolId().equals(bpId),
         "block belongs to BP %s instead of BP %s",
-        block.getBlockPoolId(), getBlockPoolId());
+        block.getBlockPoolId(), bpId);
   }
 
   //This must be called only by blockPoolManager
@@ -329,8 +358,7 @@ class BPOfferService {
     }
 
     try {
-      if (this.bpNSInfo == null) {
-        this.bpNSInfo = nsInfo;
+      if (setNamespaceInfo(nsInfo) == null) {
         boolean success = false;
 
         // Now that we know the namespace ID, etc, we can pass this to the DN.
@@ -344,16 +372,9 @@ class BPOfferService {
             // The datanode failed to initialize the BP. We need to reset
             // the namespace info so that other BPService actors still have
             // a chance to set it, and re-initialize the datanode.
-            this.bpNSInfo = null;
+            setNamespaceInfo(null);
           }
         }
-      } else {
-        checkNSEquality(bpNSInfo.getBlockPoolID(), nsInfo.getBlockPoolID(),
-            "Blockpool ID");
-        checkNSEquality(bpNSInfo.getNamespaceID(), nsInfo.getNamespaceID(),
-            "Namespace ID");
-        checkNSEquality(bpNSInfo.getClusterID(), nsInfo.getClusterID(),
-            "Cluster ID");
       }
     } finally {
       writeUnlock();

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestBPOfferService.java

@@ -221,6 +221,35 @@ public class TestBPOfferService {
     }
   }
 
+  @Test
+  public void testLocklessBlockPoolId() throws Exception {
+    BPOfferService bpos = Mockito.spy(setupBPOSForNNs(mockNN1));
+
+    // bpNSInfo is not set, should take lock to check nsInfo.
+    assertNull(bpos.getBlockPoolId());
+    Mockito.verify(bpos).readLock();
+
+    // setting the bpNSInfo should cache the bp id, thus no locking.
+    Mockito.reset(bpos);
+    NamespaceInfo nsInfo = new NamespaceInfo(1, FAKE_CLUSTERID, FAKE_BPID, 0);
+    assertNull(bpos.setNamespaceInfo(nsInfo));
+    assertEquals(FAKE_BPID, bpos.getBlockPoolId());
+    Mockito.verify(bpos, Mockito.never()).readLock();
+
+    // clearing the bpNSInfo should clear the cached bp id, thus requiring
+    // locking to check the bpNSInfo.
+    Mockito.reset(bpos);
+    assertEquals(nsInfo, bpos.setNamespaceInfo(null));
+    assertNull(bpos.getBlockPoolId());
+    Mockito.verify(bpos).readLock();
+
+    // test setting it again.
+    Mockito.reset(bpos);
+    assertNull(bpos.setNamespaceInfo(nsInfo));
+    assertEquals(FAKE_BPID, bpos.getBlockPoolId());
+    Mockito.verify(bpos, Mockito.never()).readLock();
+  }
+
   /**
    * Test that DNA_INVALIDATE commands from the standby are ignored.
    */