Revert "HDFS-15202 Boost short circuit cache (rebase PR-1884) (#2016)"

This reverts commit 86e6aa8.

Author: jojochuang

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/ClientContext.java

@@ -77,7 +77,7 @@ public class ClientContext {
   /**
    * Caches short-circuit file descriptors, mmap regions.
    */
-  private final ShortCircuitCache[] shortCircuitCache;
+  private final ShortCircuitCache shortCircuitCache;
 
   /**
    * Caches TCP and UNIX domain sockets for reuse.
@@ -132,23 +132,13 @@ public class ClientContext {
    */
   private DeadNodeDetector deadNodeDetector = null;
 
-  /**
-   * ShortCircuitCache array size.
-   */
-  private final int clientShortCircuitNum;
-
   private ClientContext(String name, DfsClientConf conf,
       Configuration config) {
     final ShortCircuitConf scConf = conf.getShortCircuitConf();
 
     this.name = name;
     this.confString = scConf.confAsString();
-    this.clientShortCircuitNum = conf.getClientShortCircuitNum();
-    this.shortCircuitCache = new ShortCircuitCache[this.clientShortCircuitNum];
-    for (int i = 0; i < this.clientShortCircuitNum; i++) {
-      this.shortCircuitCache[i] = ShortCircuitCache.fromConf(scConf);
-    }
-
+    this.shortCircuitCache = ShortCircuitCache.fromConf(scConf);
     this.peerCache = new PeerCache(scConf.getSocketCacheCapacity(),
         scConf.getSocketCacheExpiry());
     this.keyProviderCache = new KeyProviderCache(
@@ -238,11 +228,7 @@ public String getConfString() {
   }
 
   public ShortCircuitCache getShortCircuitCache() {
-    return shortCircuitCache[0];
-  }
-
-  public ShortCircuitCache getShortCircuitCache(long idx) {
-    return shortCircuitCache[(int) (idx % clientShortCircuitNum)];
+    return shortCircuitCache;
   }
 
   public PeerCache getPeerCache() {

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/client/HdfsClientConfigKeys.java

@@ -144,8 +144,6 @@ public interface HdfsClientConfigKeys {
       "dfs.short.circuit.shared.memory.watcher.interrupt.check.ms";
   int     DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS_DEFAULT =
       60000;
-  String DFS_CLIENT_SHORT_CIRCUIT_NUM = "dfs.client.short.circuit.num";
-  int DFS_CLIENT_SHORT_CIRCUIT_NUM_DEFAULT = 1;
   String  DFS_CLIENT_SLOW_IO_WARNING_THRESHOLD_KEY =
       "dfs.client.slow.io.warning.threshold.ms";
   long    DFS_CLIENT_SLOW_IO_WARNING_THRESHOLD_DEFAULT = 30000;

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/client/impl/BlockReaderFactory.java

@@ -476,8 +476,7 @@ private BlockReader getBlockReaderLocal() throws IOException {
               "giving up on BlockReaderLocal.", this, pathInfo);
       return null;
     }
-    ShortCircuitCache cache =
-        clientContext.getShortCircuitCache(block.getBlockId());
+    ShortCircuitCache cache = clientContext.getShortCircuitCache();
     ExtendedBlockId key = new ExtendedBlockId(block.getBlockId(),
         block.getBlockPoolId());
     ShortCircuitReplicaInfo info = cache.fetchOrCreate(key, this);
@@ -528,8 +527,7 @@ public ShortCircuitReplicaInfo createShortCircuitReplicaInfo() {
       if (curPeer.fromCache) remainingCacheTries--;
       DomainPeer peer = (DomainPeer)curPeer.peer;
       Slot slot = null;
-      ShortCircuitCache cache =
-          clientContext.getShortCircuitCache(block.getBlockId());
+      ShortCircuitCache cache = clientContext.getShortCircuitCache();
       try {
         MutableBoolean usedPeer = new MutableBoolean(false);
         slot = cache.allocShmSlot(datanode, peer, usedPeer,
@@ -584,8 +582,7 @@ public ShortCircuitReplicaInfo createShortCircuitReplicaInfo() {
    */
   private ShortCircuitReplicaInfo requestFileDescriptors(DomainPeer peer,
           Slot slot) throws IOException {
-    ShortCircuitCache cache =
-        clientContext.getShortCircuitCache(block.getBlockId());
+    ShortCircuitCache cache = clientContext.getShortCircuitCache();
     final DataOutputStream out =
         new DataOutputStream(new BufferedOutputStream(peer.getOutputStream(), SMALL_BUFFER_SIZE));
     SlotId slotId = slot == null ? null : slot.getSlotId();

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/client/impl/DfsClientConf.java

@@ -142,7 +142,6 @@ public class DfsClientConf {
   private final long refreshReadBlockLocationsMS;
 
   private final ShortCircuitConf shortCircuitConf;
-  private final int clientShortCircuitNum;
 
   private final long hedgedReadThresholdMillis;
   private final int hedgedReadThreadpoolSize;
@@ -273,6 +272,8 @@ public DfsClientConf(Configuration conf) {
         HdfsClientConfigKeys.
             DFS_CLIENT_REFRESH_READ_BLOCK_LOCATIONS_MS_DEFAULT);
 
+    shortCircuitConf = new ShortCircuitConf(conf);
+
     hedgedReadThresholdMillis = conf.getLong(
         HedgedRead.THRESHOLD_MILLIS_KEY,
         HedgedRead.THRESHOLD_MILLIS_DEFAULT);
@@ -295,17 +296,6 @@ public DfsClientConf(Configuration conf) {
     leaseHardLimitPeriod =
         conf.getLong(HdfsClientConfigKeys.DFS_LEASE_HARDLIMIT_KEY,
             HdfsClientConfigKeys.DFS_LEASE_HARDLIMIT_DEFAULT) * 1000;
-
-    shortCircuitConf = new ShortCircuitConf(conf);
-    clientShortCircuitNum = conf.getInt(
-            HdfsClientConfigKeys.DFS_CLIENT_SHORT_CIRCUIT_NUM,
-            HdfsClientConfigKeys.DFS_CLIENT_SHORT_CIRCUIT_NUM_DEFAULT);
-    Preconditions.checkArgument(clientShortCircuitNum >= 1,
-            HdfsClientConfigKeys.DFS_CLIENT_SHORT_CIRCUIT_NUM +
-                    "can't be less then 1.");
-    Preconditions.checkArgument(clientShortCircuitNum <= 5,
-            HdfsClientConfigKeys.DFS_CLIENT_SHORT_CIRCUIT_NUM +
-                    "can't be more then 5.");
   }
 
   @SuppressWarnings("unchecked")
@@ -611,13 +601,6 @@ public long getSlowIoWarningThresholdMs() {
     return slowIoWarningThresholdMs;
   }
 
-  /*
-   * @return the clientShortCircuitNum
-   */
-  public int getClientShortCircuitNum() {
-    return clientShortCircuitNum;
-  }
-
   /**
    * @return the hedgedReadThresholdMillis
    */

hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml

@@ -4188,16 +4188,6 @@
   </description>
 </property>
 
-<property>
-  <name>dfs.client.short.circuit.num</name>
-  <value>1</value>
-  <description>
-    Number of short-circuit caches. This setting should
-    be in the range 1 - 5. Lower values will result in lower CPU consumption; higher
-    values may speed up massive parallel reading files.
-  </description>
-</property>
-
 <property>
   <name>dfs.client.read.striped.threadpool.size</name>
   <value>18</value>

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/fs/TestEnhancedByteBufferAccess.java

@@ -358,7 +358,7 @@ public void testZeroCopyMmapCache() throws Exception {
     fsIn.close();
     fsIn = fs.open(TEST_PATH);
     final ShortCircuitCache cache = ClientContext.get(
-        CONTEXT, conf).getShortCircuitCache(0);
+        CONTEXT, conf).getShortCircuitCache();
     cache.accept(new CountingVisitor(0, 5, 5, 0));
     results[0] = fsIn.read(null, BLOCK_SIZE,
         EnumSet.of(ReadOption.SKIP_CHECKSUMS));
@@ -654,12 +654,12 @@ public void testZeroCopyReadOfCachedData() throws Exception {
         BLOCK_SIZE), byteBufferToArray(result2));
     fsIn2.releaseBuffer(result2);
     fsIn2.close();
-
+    
     // check that the replica is anchored 
     final ExtendedBlock firstBlock =
         DFSTestUtil.getFirstBlock(fs, TEST_PATH);
     final ShortCircuitCache cache = ClientContext.get(
-        CONTEXT, conf).getShortCircuitCache(0);
+        CONTEXT, conf).getShortCircuitCache();
     waitForReplicaAnchorStatus(cache, firstBlock, true, true, 1);
     // Uncache the replica
     fs.removeCacheDirective(directiveId);

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/client/impl/TestBlockReaderFactory.java

@@ -389,7 +389,7 @@ private void testShortCircuitCacheUnbufferWithDisableInterval(
 
         try (FSDataInputStream in = dfs.open(testFile)) {
           Assert.assertEquals(0,
-              dfs.getClient().getClientContext().getShortCircuitCache(0)
+              dfs.getClient().getClientContext().getShortCircuitCache()
                   .getReplicaInfoMapSize());
 
           final byte[] buf = new byte[testFileLen];
@@ -398,12 +398,12 @@ private void testShortCircuitCacheUnbufferWithDisableInterval(
 
           // Set cache size to 0 so the replica marked evictable by unbuffer
           // will be purged immediately.
-          dfs.getClient().getClientContext().getShortCircuitCache(0)
+          dfs.getClient().getClientContext().getShortCircuitCache()
               .setMaxTotalSize(0);
           LOG.info("Unbuffering");
           in.unbuffer();
           Assert.assertEquals(0,
-              dfs.getClient().getClientContext().getShortCircuitCache(0)
+              dfs.getClient().getClientContext().getShortCircuitCache()
                   .getReplicaInfoMapSize());
 
           DFSTestUtil.appendFile(dfs, testFile, "append more data");
@@ -432,7 +432,7 @@ private void validateReadResult(final DistributedFileSystem dfs,
       final int expectedScrRepMapSize) {
     Assert.assertThat(expected, CoreMatchers.is(actual));
     Assert.assertEquals(expectedScrRepMapSize,
-        dfs.getClient().getClientContext().getShortCircuitCache(0)
+        dfs.getClient().getClientContext().getShortCircuitCache()
             .getReplicaInfoMapSize());
   }
 
@@ -467,7 +467,7 @@ public void testShortCircuitReadFromServerWithoutShm() throws Exception {
         calculateFileContentsFromSeed(SEED, TEST_FILE_LEN);
     Assert.assertTrue(Arrays.equals(contents, expected));
     final ShortCircuitCache cache =
-        fs.getClient().getClientContext().getShortCircuitCache(0);
+        fs.getClient().getClientContext().getShortCircuitCache();
     final DatanodeInfo datanode = new DatanodeInfoBuilder()
         .setNodeID(cluster.getDataNodes().get(0).getDatanodeId())
         .build();
@@ -516,7 +516,7 @@ public void testShortCircuitReadFromClientWithoutShm() throws Exception {
         calculateFileContentsFromSeed(SEED, TEST_FILE_LEN);
     Assert.assertTrue(Arrays.equals(contents, expected));
     final ShortCircuitCache cache =
-        fs.getClient().getClientContext().getShortCircuitCache(0);
+        fs.getClient().getClientContext().getShortCircuitCache();
     Assert.assertEquals(null, cache.getDfsClientShmManager());
     cluster.shutdown();
     sockDir.close();
@@ -548,7 +548,7 @@ public void testShortCircuitCacheShutdown() throws Exception {
         calculateFileContentsFromSeed(SEED, TEST_FILE_LEN);
     Assert.assertTrue(Arrays.equals(contents, expected));
     final ShortCircuitCache cache =
-        fs.getClient().getClientContext().getShortCircuitCache(0);
+        fs.getClient().getClientContext().getShortCircuitCache();
     cache.close();
     Assert.assertTrue(cache.getDfsClientShmManager().
         getDomainSocketWatcher().isClosed());