branch-2.0.4-alpha/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataBlockScanner.java - hadoop - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.hadoop.hdfs.server.datanode;

 import java.io.IOException;
 import java.util.TreeMap;

 import javax.servlet.http.HttpServlet;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;

 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.protocol.Block;
 import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
 import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
 import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi;

 import com.google.common.annotations.VisibleForTesting;

 /**
  * DataBlockScanner manages block scanning for all the block pools. For each
  * block pool a {@link BlockPoolSliceScanner} is created which runs in a separate
  * thread to scan the blocks for that block pool. When a {@link BPOfferService}
  * becomes alive or dies, blockPoolScannerMap in this class is updated.
  */
 @InterfaceAudience.Private
 public class DataBlockScanner implements Runnable {
   public static final Log LOG = LogFactory.getLog(DataBlockScanner.class);
   private final DataNode datanode;
   private final FsDatasetSpi<? extends FsVolumeSpi> dataset;
   private final Configuration conf;

   static final int SLEEP_PERIOD_MS = 5 * 1000;

   /**
    * Map to find the BlockPoolScanner for a given block pool id. This is updated
    * when a BPOfferService becomes alive or dies.
    */
   private final TreeMap<String, BlockPoolSliceScanner> blockPoolScannerMap =
     new TreeMap<String, BlockPoolSliceScanner>();
   Thread blockScannerThread = null;

   DataBlockScanner(DataNode datanode,
       FsDatasetSpi<? extends FsVolumeSpi> dataset,
       Configuration conf) {
     this.datanode = datanode;
     this.dataset = dataset;
     this.conf = conf;
   }

   @Override
   public void run() {
     String currentBpId = "";
     boolean firstRun = true;
     while (datanode.shouldRun && !Thread.interrupted()) {
       //Sleep everytime except in the first iteration.
       if (!firstRun) {
         try {
           Thread.sleep(SLEEP_PERIOD_MS);
         } catch (InterruptedException ex) {
           // Interrupt itself again to set the interrupt status
           blockScannerThread.interrupt();
           continue;
         }
       } else {
         firstRun = false;
       }

       BlockPoolSliceScanner bpScanner = getNextBPScanner(currentBpId);
       if (bpScanner == null) {
         // Possible if thread is interrupted
         continue;
       }
       currentBpId = bpScanner.getBlockPoolId();
       // If BPOfferService for this pool is not alive, don't process it
       if (!datanode.isBPServiceAlive(currentBpId)) {
         LOG.warn("Block Pool " + currentBpId + " is not alive");
         // Remove in case BP service died abruptly without proper shutdown
         removeBlockPool(currentBpId);
         continue;
       }
       bpScanner.scanBlockPoolSlice();
     }

     // Call shutdown for each allocated BlockPoolSliceScanner.
     for (BlockPoolSliceScanner bpss: blockPoolScannerMap.values()) {
       bpss.shutdown();
     }
   }

   // Wait for at least one block pool to be up
   private void waitForInit() {
     while ((getBlockPoolSetSize() < datanode.getAllBpOs().length)
         || (getBlockPoolSetSize() < 1)) {
       try {
         Thread.sleep(SLEEP_PERIOD_MS);
       } catch (InterruptedException e) {
         blockScannerThread.interrupt();
         return;
       }
     }
   }

   /**
    * Find next block pool id to scan. There should be only one current
    * verification log file. Find which block pool contains the current
    * verification log file and that is used as the starting block pool id. If no
    * current files are found start with first block-pool in the blockPoolSet.
    * However, if more than one current files are found, the one with latest
    * modification time is used to find the next block pool id.
    */
   private BlockPoolSliceScanner getNextBPScanner(String currentBpId) {

     String nextBpId = null;
     while ((nextBpId == null) && datanode.shouldRun
         && !blockScannerThread.isInterrupted()) {
       waitForInit();
       synchronized (this) {
         if (getBlockPoolSetSize() > 0) {
           // Find nextBpId by the minimum of the last scan time
           long lastScanTime = 0;
           for (String bpid : blockPoolScannerMap.keySet()) {
             final long t = getBPScanner(bpid).getLastScanTime();
             if (t != 0L) {
               if (bpid == null || t < lastScanTime) {
                 lastScanTime =  t;
                 nextBpId = bpid;
               }
             }
           }

           // nextBpId can still be null if no current log is found,
           // find nextBpId sequentially.
           if (nextBpId == null) {
             nextBpId = blockPoolScannerMap.higherKey(currentBpId);
             if (nextBpId == null) {
               nextBpId = blockPoolScannerMap.firstKey();
             }
           }
           if (nextBpId != null) {
             return getBPScanner(nextBpId);
           }
         }
       }
       LOG.warn("No block pool is up, going to wait");
       try {
         Thread.sleep(5000);
       } catch (InterruptedException ex) {
         LOG.warn("Received exception: " + ex);
         blockScannerThread.interrupt();
         return null;
       }
     }
     return null;
   }

   private synchronized int getBlockPoolSetSize() {
     return blockPoolScannerMap.size();
   }

   @VisibleForTesting
   synchronized BlockPoolSliceScanner getBPScanner(String bpid) {
     return blockPoolScannerMap.get(bpid);
   }

   private synchronized String[] getBpIdList() {
     return blockPoolScannerMap.keySet().toArray(
         new String[blockPoolScannerMap.keySet().size()]);
   }

   public void addBlock(ExtendedBlock block) {
     BlockPoolSliceScanner bpScanner = getBPScanner(block.getBlockPoolId());
     if (bpScanner != null) {
       bpScanner.addBlock(block);
     } else {
       LOG.warn("No block pool scanner found for block pool id: "
           + block.getBlockPoolId());
     }
   }

   boolean isInitialized(String bpid) {
     return getBPScanner(bpid) != null;
   }

   public synchronized void printBlockReport(StringBuilder buffer,
       boolean summary) {
     String[] bpIdList = getBpIdList();
     if (bpIdList == null || bpIdList.length == 0) {
       buffer.append("Periodic block scanner is not yet initialized. "
           + "Please check back again after some time.");
       return;
     }
     for (String bpid : bpIdList) {
       BlockPoolSliceScanner bpScanner = getBPScanner(bpid);
       buffer.append("\n\nBlock report for block pool: "+bpid + "\n");
       bpScanner.printBlockReport(buffer, summary);
       buffer.append("\n");
     }
   }

   public void deleteBlock(String poolId, Block toDelete) {
     BlockPoolSliceScanner bpScanner = getBPScanner(poolId);
     if (bpScanner != null) {
       bpScanner.deleteBlock(toDelete);
     } else {
       LOG.warn("No block pool scanner found for block pool id: "
           + poolId);
     }
   }

   public void deleteBlocks(String poolId, Block[] toDelete) {
     BlockPoolSliceScanner bpScanner = getBPScanner(poolId);
     if (bpScanner != null) {
       bpScanner.deleteBlocks(toDelete);
     } else {
       LOG.warn("No block pool scanner found for block pool id: "
           + poolId);
     }
   }

   public void shutdown() {
     synchronized (this) {
       if (blockScannerThread != null) {
         blockScannerThread.interrupt();
       }
     }

     // We cannot join within the synchronized block, because it would create a
     // deadlock situation.  blockScannerThread calls other synchronized methods.
     if (blockScannerThread != null) {
       try {
         blockScannerThread.join();
       } catch (InterruptedException e) {
         // shutting down anyway
       }
     }
   }

   public synchronized void addBlockPool(String blockPoolId) {
     if (blockPoolScannerMap.get(blockPoolId) != null) {
       return;
     }
     BlockPoolSliceScanner bpScanner = new BlockPoolSliceScanner(blockPoolId,
         datanode, dataset, conf);
     blockPoolScannerMap.put(blockPoolId, bpScanner);
     LOG.info("Added bpid=" + blockPoolId + " to blockPoolScannerMap, new size="
         + blockPoolScannerMap.size());
   }

   public synchronized void removeBlockPool(String blockPoolId) {
     blockPoolScannerMap.remove(blockPoolId);
     LOG.info("Removed bpid="+blockPoolId+" from blockPoolScannerMap");
   }

   @VisibleForTesting
   long getBlocksScannedInLastRun(String bpid) throws IOException {
     BlockPoolSliceScanner bpScanner = getBPScanner(bpid);
     if (bpScanner == null) {
       throw new IOException("Block Pool: "+bpid+" is not running");
     } else {
       return bpScanner.getBlocksScannedInLastRun();
     }
   }

   @VisibleForTesting
   long getTotalScans(String bpid) throws IOException {
     BlockPoolSliceScanner bpScanner = getBPScanner(bpid);
     if (bpScanner == null) {
       throw new IOException("Block Pool: "+bpid+" is not running");
     } else {
       return bpScanner.getTotalScans();
     }
   }

   public void start() {
     blockScannerThread = new Thread(this);
     blockScannerThread.setDaemon(true);
     blockScannerThread.start();
   }

   @InterfaceAudience.Private
   public static class Servlet extends HttpServlet {
     private static final long serialVersionUID = 1L;

     @Override
     public void doGet(HttpServletRequest request,
                       HttpServletResponse response) throws IOException {
       response.setContentType("text/plain");

       DataNode datanode = (DataNode) getServletContext().getAttribute("datanode");
       DataBlockScanner blockScanner = datanode.blockScanner;

       boolean summary = (request.getParameter("listblocks") == null);

       StringBuilder buffer = new StringBuilder(8*1024);
       if (blockScanner == null) {
         LOG.warn("Periodic block scanner is not running");
         buffer.append("Periodic block scanner is not running. " +
                       "Please check the datanode log if this is unexpected.");
       } else {
         blockScanner.printBlockReport(buffer, summary);
       }
       response.getWriter().write(buffer.toString()); // extra copy!
     }
   }

 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.hadoop.hdfs.server.datanode;

	import java.io.IOException;
	import java.util.TreeMap;

	import javax.servlet.http.HttpServlet;
	import javax.servlet.http.HttpServletRequest;
	import javax.servlet.http.HttpServletResponse;

	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;
	import org.apache.hadoop.classification.InterfaceAudience;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.hdfs.protocol.Block;
	import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
	import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
	import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi;

	import com.google.common.annotations.VisibleForTesting;

	/**
	* DataBlockScanner manages block scanning for all the block pools. For each
	* block pool a {@link BlockPoolSliceScanner} is created which runs in a separate
	* thread to scan the blocks for that block pool. When a {@link BPOfferService}
	* becomes alive or dies, blockPoolScannerMap in this class is updated.
	*/
	@InterfaceAudience.Private
	public class DataBlockScanner implements Runnable {
	public static final Log LOG = LogFactory.getLog(DataBlockScanner.class);
	private final DataNode datanode;
	private final FsDatasetSpi<? extends FsVolumeSpi> dataset;
	private final Configuration conf;

	static final int SLEEP_PERIOD_MS = 5 * 1000;

	/**
	* Map to find the BlockPoolScanner for a given block pool id. This is updated
	* when a BPOfferService becomes alive or dies.
	*/
	private final TreeMap<String, BlockPoolSliceScanner> blockPoolScannerMap =
	new TreeMap<String, BlockPoolSliceScanner>();
	Thread blockScannerThread = null;

	DataBlockScanner(DataNode datanode,
	FsDatasetSpi<? extends FsVolumeSpi> dataset,
	Configuration conf) {
	this.datanode = datanode;
	this.dataset = dataset;
	this.conf = conf;
	}

	@Override
	public void run() {
	String currentBpId = "";
	boolean firstRun = true;
	while (datanode.shouldRun && !Thread.interrupted()) {
	//Sleep everytime except in the first iteration.
	if (!firstRun) {
	try {
	Thread.sleep(SLEEP_PERIOD_MS);
	} catch (InterruptedException ex) {
	// Interrupt itself again to set the interrupt status
	blockScannerThread.interrupt();
	continue;
	}
	} else {
	firstRun = false;
	}

	BlockPoolSliceScanner bpScanner = getNextBPScanner(currentBpId);
	if (bpScanner == null) {
	// Possible if thread is interrupted
	continue;
	}
	currentBpId = bpScanner.getBlockPoolId();
	// If BPOfferService for this pool is not alive, don't process it
	if (!datanode.isBPServiceAlive(currentBpId)) {
	LOG.warn("Block Pool " + currentBpId + " is not alive");
	// Remove in case BP service died abruptly without proper shutdown
	removeBlockPool(currentBpId);
	continue;
	}
	bpScanner.scanBlockPoolSlice();
	}

	// Call shutdown for each allocated BlockPoolSliceScanner.
	for (BlockPoolSliceScanner bpss: blockPoolScannerMap.values()) {
	bpss.shutdown();
	}
	}

	// Wait for at least one block pool to be up
	private void waitForInit() {
	while ((getBlockPoolSetSize() < datanode.getAllBpOs().length)
	\|\| (getBlockPoolSetSize() < 1)) {
	try {
	Thread.sleep(SLEEP_PERIOD_MS);
	} catch (InterruptedException e) {
	blockScannerThread.interrupt();
	return;
	}
	}
	}

	/**
	* Find next block pool id to scan. There should be only one current
	* verification log file. Find which block pool contains the current
	* verification log file and that is used as the starting block pool id. If no
	* current files are found start with first block-pool in the blockPoolSet.
	* However, if more than one current files are found, the one with latest
	* modification time is used to find the next block pool id.
	*/
	private BlockPoolSliceScanner getNextBPScanner(String currentBpId) {

	String nextBpId = null;
	while ((nextBpId == null) && datanode.shouldRun
	&& !blockScannerThread.isInterrupted()) {
	waitForInit();
	synchronized (this) {
	if (getBlockPoolSetSize() > 0) {
	// Find nextBpId by the minimum of the last scan time
	long lastScanTime = 0;
	for (String bpid : blockPoolScannerMap.keySet()) {
	final long t = getBPScanner(bpid).getLastScanTime();
	if (t != 0L) {
	if (bpid == null \|\| t < lastScanTime) {
	lastScanTime = t;
	nextBpId = bpid;
	}
	}
	}

	// nextBpId can still be null if no current log is found,
	// find nextBpId sequentially.
	if (nextBpId == null) {
	nextBpId = blockPoolScannerMap.higherKey(currentBpId);
	if (nextBpId == null) {
	nextBpId = blockPoolScannerMap.firstKey();
	}
	}
	if (nextBpId != null) {
	return getBPScanner(nextBpId);
	}
	}
	}
	LOG.warn("No block pool is up, going to wait");
	try {
	Thread.sleep(5000);
	} catch (InterruptedException ex) {
	LOG.warn("Received exception: " + ex);
	blockScannerThread.interrupt();
	return null;
	}
	}
	return null;
	}

	private synchronized int getBlockPoolSetSize() {
	return blockPoolScannerMap.size();
	}

	@VisibleForTesting
	synchronized BlockPoolSliceScanner getBPScanner(String bpid) {
	return blockPoolScannerMap.get(bpid);
	}

	private synchronized String[] getBpIdList() {
	return blockPoolScannerMap.keySet().toArray(
	new String[blockPoolScannerMap.keySet().size()]);
	}

	public void addBlock(ExtendedBlock block) {
	BlockPoolSliceScanner bpScanner = getBPScanner(block.getBlockPoolId());
	if (bpScanner != null) {
	bpScanner.addBlock(block);
	} else {
	LOG.warn("No block pool scanner found for block pool id: "
	+ block.getBlockPoolId());
	}
	}

	boolean isInitialized(String bpid) {
	return getBPScanner(bpid) != null;
	}

	public synchronized void printBlockReport(StringBuilder buffer,
	boolean summary) {
	String[] bpIdList = getBpIdList();
	if (bpIdList == null \|\| bpIdList.length == 0) {
	buffer.append("Periodic block scanner is not yet initialized. "
	+ "Please check back again after some time.");
	return;
	}
	for (String bpid : bpIdList) {
	BlockPoolSliceScanner bpScanner = getBPScanner(bpid);
	buffer.append("\n\nBlock report for block pool: "+bpid + "\n");
	bpScanner.printBlockReport(buffer, summary);
	buffer.append("\n");
	}
	}

	public void deleteBlock(String poolId, Block toDelete) {
	BlockPoolSliceScanner bpScanner = getBPScanner(poolId);
	if (bpScanner != null) {
	bpScanner.deleteBlock(toDelete);
	} else {
	LOG.warn("No block pool scanner found for block pool id: "
	+ poolId);
	}
	}

	public void deleteBlocks(String poolId, Block[] toDelete) {
	BlockPoolSliceScanner bpScanner = getBPScanner(poolId);
	if (bpScanner != null) {
	bpScanner.deleteBlocks(toDelete);
	} else {
	LOG.warn("No block pool scanner found for block pool id: "
	+ poolId);
	}
	}

	public void shutdown() {
	synchronized (this) {
	if (blockScannerThread != null) {
	blockScannerThread.interrupt();
	}
	}

	// We cannot join within the synchronized block, because it would create a
	// deadlock situation. blockScannerThread calls other synchronized methods.
	if (blockScannerThread != null) {
	try {
	blockScannerThread.join();
	} catch (InterruptedException e) {
	// shutting down anyway
	}
	}
	}

	public synchronized void addBlockPool(String blockPoolId) {
	if (blockPoolScannerMap.get(blockPoolId) != null) {
	return;
	}
	BlockPoolSliceScanner bpScanner = new BlockPoolSliceScanner(blockPoolId,
	datanode, dataset, conf);
	blockPoolScannerMap.put(blockPoolId, bpScanner);
	LOG.info("Added bpid=" + blockPoolId + " to blockPoolScannerMap, new size="
	+ blockPoolScannerMap.size());
	}

	public synchronized void removeBlockPool(String blockPoolId) {
	blockPoolScannerMap.remove(blockPoolId);
	LOG.info("Removed bpid="+blockPoolId+" from blockPoolScannerMap");
	}

	@VisibleForTesting
	long getBlocksScannedInLastRun(String bpid) throws IOException {
	BlockPoolSliceScanner bpScanner = getBPScanner(bpid);
	if (bpScanner == null) {
	throw new IOException("Block Pool: "+bpid+" is not running");
	} else {
	return bpScanner.getBlocksScannedInLastRun();
	}
	}

	@VisibleForTesting
	long getTotalScans(String bpid) throws IOException {
	BlockPoolSliceScanner bpScanner = getBPScanner(bpid);
	if (bpScanner == null) {
	throw new IOException("Block Pool: "+bpid+" is not running");
	} else {
	return bpScanner.getTotalScans();
	}
	}

	public void start() {
	blockScannerThread = new Thread(this);
	blockScannerThread.setDaemon(true);
	blockScannerThread.start();
	}

	@InterfaceAudience.Private
	public static class Servlet extends HttpServlet {
	private static final long serialVersionUID = 1L;

	@Override
	public void doGet(HttpServletRequest request,
	HttpServletResponse response) throws IOException {
	response.setContentType("text/plain");

	DataNode datanode = (DataNode) getServletContext().getAttribute("datanode");
	DataBlockScanner blockScanner = datanode.blockScanner;

	boolean summary = (request.getParameter("listblocks") == null);

	StringBuilder buffer = new StringBuilder(8*1024);
	if (blockScanner == null) {
	LOG.warn("Periodic block scanner is not running");
	buffer.append("Periodic block scanner is not running. " +
	"Please check the datanode log if this is unexpected.");
	} else {
	blockScanner.printBlockReport(buffer, summary);
	}
	response.getWriter().write(buffer.toString()); // extra copy!
	}
	}

	}