HDFS-3754. BlockSender doesn't shutdown ReadaheadPool threads. Contributed by Eli Collins

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-1@1372187 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -161,6 +161,8 @@ Release 1.2.0 - unreleased
 
     MAPREDUCE-4036. Streaming TestUlimit fails on CentOS 6. (tucu)
 
+    HDFS-3754. BlockSender doesn't shutdown ReadaheadPool threads. (eli)
+
 Release 1.1.0 - unreleased
 
   INCOMPATIBLE CHANGES

src/core/org/apache/hadoop/io/ReadaheadPool.java

@@ -60,13 +60,14 @@ public class ReadaheadPool {
   }
 
   private ReadaheadPool() {
+    final ThreadFactory backingFactory = Executors.defaultThreadFactory();
     pool = new ThreadPoolExecutor(POOL_SIZE, MAX_POOL_SIZE, 3L, TimeUnit.SECONDS,
         new ArrayBlockingQueue<Runnable>(CAPACITY));
     pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
     pool.setThreadFactory(new ThreadFactory() {
       @Override
       public Thread newThread(Runnable runnable) {
-        Thread thread = Executors.defaultThreadFactory().newThread(runnable);
+        Thread thread = backingFactory.newThread(runnable);
         thread.setName(String.format("Readahead Thread #%d",
             THREAD_COUNTER.getAndIncrement()));
         thread.setDaemon(true);
@@ -246,4 +247,4 @@ public class ReadaheadPool {
           + ", off=" + off + ", len=" + len + "]";
     }
   }
-}
+}

src/hdfs/org/apache/hadoop/hdfs/server/datanode/BlockSender.java

@@ -86,6 +86,7 @@ class BlockSender implements java.io.Closeable, FSConstants {
   // Cache-management related fields
   private final long readaheadLength;
   private boolean shouldDropCacheBehindRead;
+  private ReadaheadPool readaheadPool;
   private ReadaheadRequest curReadahead;
   private long lastCacheDropOffset;
   private static final long CACHE_DROP_INTERVAL_BYTES = 1024 * 1024; // 1MB
@@ -95,8 +96,6 @@ class BlockSender implements java.io.Closeable, FSConstants {
    */
   private static final long LONG_READ_THRESHOLD_BYTES = 256 * 1024;
 
-  private static ReadaheadPool readaheadPool = ReadaheadPool.getInstance();
-  
   BlockSender(Block block, long startOffset, long length,
               boolean corruptChecksumOk, boolean chunkOffsetOK,
               boolean verifyChecksum, DataNode datanode) throws IOException {
@@ -117,6 +116,7 @@ class BlockSender implements java.io.Closeable, FSConstants {
       this.transferToAllowed = datanode.transferToAllowed;
       this.clientTraceFmt = clientTraceFmt;
       this.readaheadLength = datanode.getReadaheadLength();
+      this.readaheadPool = datanode.readaheadPool;
       this.shouldDropCacheBehindRead = datanode.shouldDropCacheBehindReads();
       
       if ( !corruptChecksumOk || datanode.data.metaFileExists(block) ) {

src/hdfs/org/apache/hadoop/hdfs/server/datanode/DataNode.java

@@ -107,6 +107,7 @@ import org.apache.hadoop.hdfs.web.WebHdfsFileSystem;
 import org.apache.hadoop.hdfs.web.resources.Param;
 import org.apache.hadoop.http.HttpServer;
 import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.io.ReadaheadPool;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.ipc.RemoteException;
@@ -270,6 +271,8 @@ public class DataNode extends Configured
   public Server ipcServer;
 
   private SecureResources secureResources = null;
+
+  ReadaheadPool readaheadPool;
   
   /**
    * Current system time.
@@ -469,6 +472,8 @@ public class DataNode extends Configured
                reason + ".");
     }
 
+    readaheadPool = ReadaheadPool.getInstance();
+
     this.connectToDnViaHostname = conf.getBoolean(
         DFSConfigKeys.DFS_DATANODE_USE_DN_HOSTNAME,
         DFSConfigKeys.DFS_DATANODE_USE_DN_HOSTNAME_DEFAULT);