You are viewing a plain text version of this content. The canonical link for it is here.
Posted to hdfs-commits@hadoop.apache.org by ji...@apache.org on 2011/03/02 01:13:13 UTC
svn commit: r1076061 [1/2] - in /hadoop/hdfs/branches/HDFS-1052: ./
src/java/org/apache/hadoop/hdfs/server/datanode/
src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/
Author: jitendra
Date: Wed Mar 2 00:13:13 2011
New Revision: 1076061
URL: http://svn.apache.org/viewvc?rev=1076061&view=rev
Log:
Federation: DatablockScanner should scan blocks for all the block pools.
Added:
hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolScanner.java
hadoop/hdfs/branches/HDFS-1052/src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestMulitipleNNDataBlockScanner.java
Modified:
hadoop/hdfs/branches/HDFS-1052/CHANGES.txt
hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataBlockScanner.java
hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/FSDataset.java
hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/ReplicasMap.java
hadoop/hdfs/branches/HDFS-1052/src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestInterDatanodeProtocol.java
Modified: hadoop/hdfs/branches/HDFS-1052/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1052/CHANGES.txt?rev=1076061&r1=1076060&r2=1076061&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1052/CHANGES.txt (original)
+++ hadoop/hdfs/branches/HDFS-1052/CHANGES.txt Wed Mar 2 00:13:13 2011
@@ -105,6 +105,9 @@ Trunk (unreleased changes)
HDFS-1626. Make BLOCK_INVALIDATE_LIMIT configurable. (szetszwo)
+ HDFS-1655. Federation: DatablockScanner should scan blocks for
+ all the block pools. (jitendra)
+
IMPROVEMENTS
HDFS-1510. Added test-patch.properties required by test-patch.sh (nigel)
Added: hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolScanner.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolScanner.java?rev=1076061&view=auto
==============================================================================
--- hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolScanner.java (added)
+++ hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolScanner.java Wed Mar 2 00:13:13 2011
@@ -0,0 +1,928 @@
+/**
+ * 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.BufferedReader;
+import java.io.Closeable;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Random;
+import java.util.TreeSet;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+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.common.GenerationStamp;
+import org.apache.hadoop.hdfs.server.datanode.FSDataset.FSVolume;
+import org.apache.hadoop.hdfs.util.DataTransferThrottler;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.util.StringUtils;
+
+/**
+ * Performs two types of scanning:
+ * <li> Gets block files from the data directories and reconciles the
+ * difference between the blocks on the disk and in memory in
+ * {@link FSDataset}</li>
+ * <li> Scans the data directories for block files under a block pool
+ * and verifies that the files are not corrupt</li>
+ * This keeps track of blocks and their last verification times.
+ * Currently it does not modify the metadata for block.
+ */
+
+class BlockPoolScanner {
+
+ public static final Log LOG = LogFactory.getLog(BlockPoolScanner.class);
+
+ private static final int MAX_SCAN_RATE = 8 * 1024 * 1024; // 8MB per sec
+ private static final int MIN_SCAN_RATE = 1 * 1024 * 1024; // 1MB per sec
+
+ static final long DEFAULT_SCAN_PERIOD_HOURS = 21*24L; // three weeks
+ private final String blockPoolId;
+
+ private static final String dateFormatString = "yyyy-MM-dd HH:mm:ss,SSS";
+
+ static final String verificationLogFile = "dncp_block_verification.log";
+ static final int verficationLogLimit = 5; // * numBlocks.
+
+ private long scanPeriod = DEFAULT_SCAN_PERIOD_HOURS * 3600 * 1000;
+ private DataNode datanode;
+ private FSDataset dataset;
+
+ // sorted set
+ private TreeSet<BlockScanInfo> blockInfoSet;
+ private HashMap<Block, BlockScanInfo> blockMap;
+
+ // processedBlocks keeps track of which blocks are scanned
+ // since the last run.
+ private HashMap<Long, Integer> processedBlocks;
+
+ private long totalScans = 0;
+ private long totalScanErrors = 0;
+ private long totalTransientErrors = 0;
+ private long totalBlocksScannedInLastRun = 0; // Used for test only
+
+ private long currentPeriodStart = System.currentTimeMillis();
+ private long bytesLeft = 0; // Bytes to scan in this period
+ private long totalBytesToScan = 0;
+
+ private LogFileHandler verificationLog;
+
+ private Random random = new Random();
+
+ private DataTransferThrottler throttler = null;
+
+ private static enum ScanType {
+ VERIFICATION_SCAN, // scanned as part of periodic verfication
+ NONE,
+ }
+
+ static class BlockScanInfo implements Comparable<BlockScanInfo> {
+ Block block;
+ long lastScanTime = 0;
+ ScanType lastScanType = ScanType.NONE;
+ boolean lastScanOk = true;
+
+ BlockScanInfo(Block block) {
+ this.block = block;
+ }
+
+ public int hashCode() {
+ return block.hashCode();
+ }
+
+ public boolean equals(Object other) {
+ return other instanceof BlockScanInfo &&
+ compareTo((BlockScanInfo)other) == 0;
+ }
+
+ long getLastScanTime() {
+ return (lastScanType == ScanType.NONE) ? 0 : lastScanTime;
+ }
+
+ public int compareTo(BlockScanInfo other) {
+ long t1 = lastScanTime;
+ long t2 = other.lastScanTime;
+ return ( t1 < t2 ) ? -1 :
+ (( t1 > t2 ) ? 1 : block.compareTo(other.block));
+ }
+ }
+
+ BlockPoolScanner(DataNode datanode, FSDataset dataset, Configuration conf,
+ String bpid) {
+ this.datanode = datanode;
+ this.dataset = dataset;
+ this.blockPoolId = bpid;
+ scanPeriod = conf.getInt("dfs.datanode.scan.period.hours", 0);
+ if ( scanPeriod <= 0 ) {
+ scanPeriod = DEFAULT_SCAN_PERIOD_HOURS;
+ }
+ scanPeriod *= 3600 * 1000;
+ LOG.info("Periodic Block Verification scan initialized with interval " + scanPeriod + ".");
+ }
+
+ String getBlockPoolId() {
+ return blockPoolId;
+ }
+
+ synchronized boolean isInitialized() {
+ return throttler != null;
+ }
+
+ private void updateBytesToScan(long len, long lastScanTime) {
+ // len could be negative when a block is deleted.
+ totalBytesToScan += len;
+ if ( lastScanTime < currentPeriodStart ) {
+ bytesLeft += len;
+ }
+ // Should we change throttler bandwidth every time bytesLeft changes?
+ // not really required.
+ }
+
+ private synchronized void addBlockInfo(BlockScanInfo info) {
+ boolean added = blockInfoSet.add(info);
+ blockMap.put(info.block, info);
+
+ if (added) {
+ updateBytesToScan(info.block.getNumBytes(), info.lastScanTime);
+ }
+ }
+
+ private synchronized void delBlockInfo(BlockScanInfo info) {
+ boolean exists = blockInfoSet.remove(info);
+ blockMap.remove(info.block);
+
+ if (exists) {
+ updateBytesToScan(-info.block.getNumBytes(), info.lastScanTime);
+ }
+ }
+
+ /** Update blockMap by the given LogEntry */
+ private synchronized void updateBlockInfo(LogEntry e) {
+ BlockScanInfo info = blockMap.get(new Block(e.blockId, 0, e.genStamp));
+
+ if(info != null && e.verificationTime > 0 &&
+ info.lastScanTime < e.verificationTime) {
+ delBlockInfo(info);
+ info.lastScanTime = e.verificationTime;
+ info.lastScanType = ScanType.VERIFICATION_SCAN;
+ addBlockInfo(info);
+ }
+ }
+
+ void init() throws IOException {
+ // get the list of blocks and arrange them in random order
+ List<Block> arr = dataset.getFinalizedBlocks(blockPoolId);
+ Collections.shuffle(arr);
+
+ blockInfoSet = new TreeSet<BlockScanInfo>();
+ blockMap = new HashMap<Block, BlockScanInfo>();
+
+ long scanTime = -1;
+ for (Block block : arr) {
+ BlockScanInfo info = new BlockScanInfo( block );
+ info.lastScanTime = scanTime--;
+ //still keep 'info.lastScanType' to NONE.
+ addBlockInfo(info);
+ }
+
+ /* Pick the first directory that has any existing scanner log.
+ * otherwise, pick the first directory.
+ */
+ File dir = null;
+ FSDataset.FSVolume[] volumes = dataset.volumes.volumes;
+ for (FSDataset.FSVolume vol : volumes) {
+ File bpDir = vol.getBlockPool(blockPoolId).getDirectory();
+ if (LogFileHandler.isFilePresent(bpDir, verificationLogFile)) {
+ dir = bpDir;
+ break;
+ }
+ }
+ if (dir == null) {
+ dir = volumes[0].getBlockPool(blockPoolId).getDirectory();
+ }
+
+ try {
+ // max lines will be updated later during initialization.
+ verificationLog = new LogFileHandler(dir, verificationLogFile, 100);
+ } catch (IOException e) {
+ LOG.warn("Could not open verfication log. " +
+ "Verification times are not stored.");
+ }
+
+ synchronized (this) {
+ throttler = new DataTransferThrottler(200, MAX_SCAN_RATE);
+ }
+ }
+
+ private synchronized long getNewBlockScanTime() {
+ /* If there are a lot of blocks, this returns a random time with in
+ * the scan period. Otherwise something sooner.
+ */
+ long period = Math.min(scanPeriod,
+ Math.max(blockMap.size(),1) * 600 * 1000L);
+ return System.currentTimeMillis() - scanPeriod +
+ random.nextInt((int)period);
+ }
+
+ /** Adds block to list of blocks */
+ synchronized void addBlock(ExtendedBlock block) {
+ if (!isInitialized()) {
+ return;
+ }
+
+ BlockScanInfo info = blockMap.get(block.getLocalBlock());
+ if ( info != null ) {
+ LOG.warn("Adding an already existing block " + block);
+ delBlockInfo(info);
+ }
+
+ info = new BlockScanInfo(block.getLocalBlock());
+ info.lastScanTime = getNewBlockScanTime();
+
+ addBlockInfo(info);
+ adjustThrottler();
+ }
+
+ /** Deletes the block from internal structures */
+ synchronized void deleteBlock(Block block) {
+ if (!isInitialized()) {
+ return;
+ }
+ BlockScanInfo info = blockMap.get(block);
+ if ( info != null ) {
+ delBlockInfo(info);
+ }
+ }
+
+ /** @return the last scan time */
+ synchronized long getLastScanTime(Block block) {
+ if (!isInitialized()) {
+ return 0;
+ }
+ BlockScanInfo info = blockMap.get(block);
+ return info == null? 0: info.lastScanTime;
+ }
+
+ /** Deletes blocks from internal structures */
+ void deleteBlocks(Block[] blocks) {
+ for ( Block b : blocks ) {
+ deleteBlock(b);
+ }
+ }
+
+ private synchronized void updateScanStatus(Block block,
+ ScanType type,
+ boolean scanOk) {
+ if (!isInitialized()) {
+ return;
+ }
+ BlockScanInfo info = blockMap.get(block);
+
+ if ( info != null ) {
+ delBlockInfo(info);
+ } else {
+ // It might already be removed. Thats ok, it will be caught next time.
+ info = new BlockScanInfo(block);
+ }
+
+ long now = System.currentTimeMillis();
+ info.lastScanType = type;
+ info.lastScanTime = now;
+ info.lastScanOk = scanOk;
+ addBlockInfo(info);
+
+ // Don't update meta data if the verification failed.
+ if (!scanOk) {
+ return;
+ }
+
+ LogFileHandler log = verificationLog;
+ if (log != null) {
+ log.appendLine(now, block.getGenerationStamp(), block.getBlockId());
+ }
+ }
+
+ private void handleScanFailure(ExtendedBlock block) {
+ LOG.info("Reporting bad block " + block);
+ try {
+ datanode.reportBadBlocks(block);
+ } catch (IOException ie) {
+ // it is bad, but not bad enough to shutdown the scanner
+ LOG.warn("Cannot report bad block=" + block.getBlockId());
+ }
+ }
+
+ static private class LogEntry {
+ long blockId = -1;
+ long verificationTime = -1;
+ long genStamp = GenerationStamp.GRANDFATHER_GENERATION_STAMP;
+
+ /**
+ * The format consists of single line with multiple entries. each
+ * entry is in the form : name="value".
+ * This simple text and easily extendable and easily parseable with a
+ * regex.
+ */
+ private static Pattern entryPattern =
+ Pattern.compile("\\G\\s*([^=\\p{Space}]+)=\"(.*?)\"\\s*");
+
+ static LogEntry parseEntry(String line) {
+ LogEntry entry = new LogEntry();
+
+ Matcher matcher = entryPattern.matcher(line);
+ while (matcher.find()) {
+ String name = matcher.group(1);
+ String value = matcher.group(2);
+
+ try {
+ if (name.equals("id")) {
+ entry.blockId = Long.valueOf(value);
+ } else if (name.equals("time")) {
+ entry.verificationTime = Long.valueOf(value);
+ } else if (name.equals("genstamp")) {
+ entry.genStamp = Long.valueOf(value);
+ }
+ } catch(NumberFormatException nfe) {
+ LOG.warn("Cannot parse line: " + line, nfe);
+ return null;
+ }
+ }
+
+ return entry;
+ }
+ }
+
+ private synchronized void adjustThrottler() {
+ long timeLeft = currentPeriodStart+scanPeriod - System.currentTimeMillis();
+ long bw = Math.max(bytesLeft*1000/timeLeft, MIN_SCAN_RATE);
+ throttler.setBandwidth(Math.min(bw, MAX_SCAN_RATE));
+ }
+
+ private void verifyBlock(ExtendedBlock block) {
+ BlockSender blockSender = null;
+
+ /* In case of failure, attempt to read second time to reduce
+ * transient errors. How do we flush block data from kernel
+ * buffers before the second read?
+ */
+ for (int i=0; i<2; i++) {
+ boolean second = (i > 0);
+
+ try {
+ adjustThrottler();
+
+ blockSender = new BlockSender(block, 0, -1, false, false, true,
+ datanode);
+
+ DataOutputStream out =
+ new DataOutputStream(new IOUtils.NullOutputStream());
+
+ blockSender.sendBlock(out, null, throttler);
+
+ LOG.info((second ? "Second " : "") +
+ "Verification succeeded for " + block);
+
+ if ( second ) {
+ totalTransientErrors++;
+ }
+
+ updateScanStatus(block.getLocalBlock(), ScanType.VERIFICATION_SCAN, true);
+
+ return;
+ } catch (IOException e) {
+ updateScanStatus(block.getLocalBlock(), ScanType.VERIFICATION_SCAN, false);
+
+ // If the block does not exists anymore, then its not an error
+ if ( dataset.getFile(block.getBlockPoolId(), block.getLocalBlock()) == null ) {
+ LOG.info("Verification failed for " + block + ". Its ok since " +
+ "it not in datanode dataset anymore.");
+ deleteBlock(block.getLocalBlock());
+ return;
+ }
+
+ LOG.warn((second ? "Second " : "First ") +
+ "Verification failed for " + block + ". Exception : " +
+ StringUtils.stringifyException(e));
+
+ if (second) {
+ totalScanErrors++;
+ datanode.getMetrics().blockVerificationFailures.inc();
+ handleScanFailure(block);
+ return;
+ }
+ } finally {
+ IOUtils.closeStream(blockSender);
+ datanode.getMetrics().blocksVerified.inc();
+ totalScans++;
+ }
+ }
+ }
+
+ private synchronized long getEarliestScanTime() {
+ if ( blockInfoSet.size() > 0 ) {
+ return blockInfoSet.first().lastScanTime;
+ }
+ return Long.MAX_VALUE;
+ }
+
+ private synchronized boolean isFirstBlockProcessed() {
+ if (blockInfoSet.size() > 0 ) {
+ long blockId = blockInfoSet.first().block.getBlockId();
+ if ((processedBlocks.get(blockId) != null)
+ && (processedBlocks.get(blockId) == 1)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ // Picks one block and verifies it
+ private void verifyFirstBlock() {
+ Block block = null;
+ synchronized (this) {
+ if ( blockInfoSet.size() > 0 ) {
+ block = blockInfoSet.first().block;
+ }
+ }
+ if ( block != null ) {
+ verifyBlock(new ExtendedBlock(blockPoolId, block));
+ processedBlocks.put(block.getBlockId(), 1);
+ }
+ }
+
+ // Used for tests only
+ long getBlocksScannedInLastRun() {
+ return totalBlocksScannedInLastRun;
+ }
+
+ /**
+ * Reads the current and previous log files (if any) and marks the blocks
+ * processed if they were processed within last scan period. Copies the log
+ * records of recently scanned blocks from previous to current file.
+ * Returns false if the process was interrupted because the thread is marked
+ * to exit.
+ */
+ private boolean assignInitialVerificationTimes() {
+ int numBlocks = 1;
+ LogFileHandler log = null;
+ synchronized (this) {
+ log = verificationLog;
+ numBlocks = Math.max(blockMap.size(), 1);
+ }
+
+ long now = System.currentTimeMillis();
+ LogFileHandler.Reader logReader[] = new LogFileHandler.Reader[2];
+ try {
+ if (log != null) {
+ logReader[0] = log.getCurrentFileReader();
+ logReader[1] = log.getPreviousFileReader();
+ }
+ } catch (IOException e) {
+ LOG.warn("Could not read previous verification times : " +
+ StringUtils.stringifyException(e));
+ }
+
+ try {
+ for (LogFileHandler.Reader reader : logReader) {
+ // update verification times from the verificationLog.
+ while (logReader != null && reader.hasNext()) {
+ if (!datanode.shouldRun
+ || datanode.blockScanner.blockScannerThread.isInterrupted()) {
+ return false;
+ }
+ LogEntry entry = LogEntry.parseEntry(reader.next());
+ if (entry != null) {
+ updateBlockInfo(entry);
+ if (now - entry.verificationTime < scanPeriod) {
+ BlockScanInfo info = blockMap.get(new Block(entry.blockId, 0,
+ entry.genStamp));
+ if (info != null) {
+ if (processedBlocks.get(entry.blockId) == null) {
+ updateBytesLeft(-info.block.getNumBytes());
+ processedBlocks.put(entry.blockId, 1);
+ }
+ if (reader.file == log.prevFile) {
+ // write the log entry to current file
+ // so that the entry is preserved for later runs.
+ log.appendLine(entry.verificationTime, entry.genStamp,
+ entry.blockId);
+ }
+ }
+ }
+ }
+ }
+ }
+ } finally {
+ IOUtils.closeStream(logReader[0]);
+ IOUtils.closeStream(logReader[1]);
+ }
+
+ /* Initially spread the block reads over half of
+ * MIN_SCAN_PERIOD so that we don't keep scanning the
+ * blocks too quickly when restarted.
+ */
+ long verifyInterval = (long) (Math.min( scanPeriod/2.0/numBlocks,
+ 10*60*1000 ));
+ long lastScanTime = System.currentTimeMillis() - scanPeriod;
+
+ /* Before this loop, entries in blockInfoSet that are not
+ * updated above have lastScanTime of <= 0 . Loop until first entry has
+ * lastModificationTime > 0.
+ */
+ synchronized (this) {
+ if (blockInfoSet.size() > 0 ) {
+ BlockScanInfo info;
+ while ((info = blockInfoSet.first()).lastScanTime < 0) {
+ delBlockInfo(info);
+ info.lastScanTime = lastScanTime;
+ lastScanTime += verifyInterval;
+ addBlockInfo(info);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private synchronized void updateBytesLeft(long len) {
+ bytesLeft += len;
+ }
+
+ static File getCurrentFile(FSVolume vol, String bpid) throws IOException {
+ return LogFileHandler.getCurrentFile(vol.getBlockPool(bpid).getDirectory(),
+ BlockPoolScanner.verificationLogFile);
+ }
+
+ private synchronized void startNewPeriod() {
+ LOG.info("Starting a new period : work left in prev period : "
+ + String.format("%.2f%%", totalBytesToScan == 0 ? 0
+ : (bytesLeft * 100.0) / totalBytesToScan));
+
+ // reset the byte counts :
+ bytesLeft = totalBytesToScan;
+ currentPeriodStart = System.currentTimeMillis();
+ }
+
+ void scanBlockPool() {
+ startNewPeriod();
+ if (processedBlocks != null) {
+ totalBlocksScannedInLastRun = processedBlocks.size();
+ }
+ // Create a new processedBlocks structure
+ processedBlocks = new HashMap<Long, Integer>();
+ if (verificationLog != null) {
+ try {
+ verificationLog.openCurFile();
+ } catch (FileNotFoundException ex) {
+ LOG.warn("Could not open current file");
+ }
+ }
+ if (!assignInitialVerificationTimes()) {
+ return;
+ }
+ // Start scanning
+ scan();
+ }
+
+ public void scan() {
+ LOG.info("Starting to scan blockpool: " + blockPoolId);
+ try {
+ adjustThrottler();
+
+ while (datanode.shouldRun && !Thread.interrupted()
+ && datanode.isBPServiceAlive(blockPoolId)) {
+ long now = System.currentTimeMillis();
+ synchronized (this) {
+ if ( now >= (currentPeriodStart + scanPeriod)) {
+ startNewPeriod();
+ }
+ }
+ if (((now - getEarliestScanTime()) >= scanPeriod)
+ || (!(this.isFirstBlockProcessed()))) {
+ verifyFirstBlock();
+ } else {
+ LOG.info("All remaining blocks were processed recently, "
+ + "so this run is complete");
+ break;
+ }
+ }
+ } catch (RuntimeException e) {
+ LOG.warn("RuntimeException during BlockPoolScanner.scan() : " +
+ StringUtils.stringifyException(e));
+ throw e;
+ } finally {
+ cleanUp();
+ LOG.info("Done scanning block pool: " + blockPoolId);
+ }
+ }
+
+ private synchronized void cleanUp() {
+ if (verificationLog != null) {
+ try {
+ verificationLog.roll();
+ } catch (IOException ex) {
+ LOG.warn("Received exception: ", ex);
+ verificationLog.close();
+ }
+ }
+ }
+
+
+ synchronized void printBlockReport(StringBuilder buffer,
+ boolean summaryOnly) {
+ long oneHour = 3600*1000;
+ long oneDay = 24*oneHour;
+ long oneWeek = 7*oneDay;
+ long fourWeeks = 4*oneWeek;
+
+ int inOneHour = 0;
+ int inOneDay = 0;
+ int inOneWeek = 0;
+ int inFourWeeks = 0;
+ int inScanPeriod = 0;
+ int neverScanned = 0;
+
+ DateFormat dateFormat = new SimpleDateFormat(dateFormatString);
+
+ int total = blockInfoSet.size();
+
+ long now = System.currentTimeMillis();
+
+ Date date = new Date();
+
+ for(Iterator<BlockScanInfo> it = blockInfoSet.iterator(); it.hasNext();) {
+ BlockScanInfo info = it.next();
+
+ long scanTime = info.getLastScanTime();
+ long diff = now - scanTime;
+
+ if (diff <= oneHour) inOneHour++;
+ if (diff <= oneDay) inOneDay++;
+ if (diff <= oneWeek) inOneWeek++;
+ if (diff <= fourWeeks) inFourWeeks++;
+ if (diff <= scanPeriod) inScanPeriod++;
+ if (scanTime <= 0) neverScanned++;
+
+ if (!summaryOnly) {
+ date.setTime(scanTime);
+ String scanType =
+ (info.lastScanType == ScanType.VERIFICATION_SCAN) ? "local" : "none";
+ buffer.append(String.format("%-26s : status : %-6s type : %-6s" +
+ " scan time : " +
+ "%-15d %s\n", info.block,
+ (info.lastScanOk ? "ok" : "failed"),
+ scanType, scanTime,
+ (scanTime <= 0) ? "not yet verified" :
+ dateFormat.format(date)));
+ }
+ }
+
+ double pctPeriodLeft = (scanPeriod + currentPeriodStart - now)
+ *100.0/scanPeriod;
+ double pctProgress = (totalBytesToScan == 0) ? 100 :
+ (totalBytesToScan-bytesLeft)*100.0/totalBytesToScan;
+
+ buffer.append(String.format("\nTotal Blocks : %6d" +
+ "\nVerified in last hour : %6d" +
+ "\nVerified in last day : %6d" +
+ "\nVerified in last week : %6d" +
+ "\nVerified in last four weeks : %6d" +
+ "\nVerified in SCAN_PERIOD : %6d" +
+ "\nNot yet verified : %6d" +
+ "\nVerified since restart : %6d" +
+ "\nScans since restart : %6d" +
+ "\nScan errors since restart : %6d" +
+ "\nTransient scan errors : %6d" +
+ "\nCurrent scan rate limit KBps : %6d" +
+ "\nProgress this period : %6.0f%%" +
+ "\nTime left in cur period : %6.2f%%" +
+ "\n",
+ total, inOneHour, inOneDay, inOneWeek,
+ inFourWeeks, inScanPeriod, neverScanned,
+ totalScans, totalScans,
+ totalScanErrors, totalTransientErrors,
+ Math.round(throttler.getBandwidth()/1024.0),
+ pctProgress, pctPeriodLeft));
+ }
+
+ /**
+ * This class takes care of log file used to store the last verification
+ * times of the blocks. It rolls the current file when it is too big etc.
+ * If there is an error while writing, it stops updating with an error
+ * message.
+ */
+ private static class LogFileHandler {
+
+ private static final String curFileSuffix = ".curr";
+ private static final String prevFileSuffix = ".prev";
+ private final DateFormat dateFormat = new SimpleDateFormat(dateFormatString);
+
+ static File getCurrentFile(File dir, String filePrefix) {
+ return new File(dir, filePrefix + curFileSuffix);
+ }
+
+ public Reader getPreviousFileReader() throws IOException {
+ return new Reader(prevFile);
+ }
+
+ public Reader getCurrentFileReader() throws IOException {
+ return new Reader(curFile);
+ }
+
+ static boolean isFilePresent(File dir, String filePrefix) {
+ return new File(dir, filePrefix + curFileSuffix).exists() ||
+ new File(dir, filePrefix + prevFileSuffix).exists();
+ }
+ private File curFile;
+ private File prevFile;
+
+ private PrintStream out;
+
+ /**
+ * Opens the log file for appending.
+ * Note that rolling will happen only after "updateLineCount()" is
+ * called. This is so that line count could be updated in a separate
+ * thread without delaying start up.
+ *
+ * @param dir where the logs files are located.
+ * @param filePrefix prefix of the file.
+ * @param maxNumLines max lines in a file (its a soft limit).
+ * @throws IOException
+ */
+ LogFileHandler(File dir, String filePrefix, int maxNumLines)
+ throws IOException {
+ curFile = new File(dir, filePrefix + curFileSuffix);
+ prevFile = new File(dir, filePrefix + prevFileSuffix);
+ }
+
+ /**
+ * Append "\n" + line.
+ * If the log file need to be rolled, it will done after
+ * appending the text.
+ * This does not throw IOException when there is an error while
+ * appending. Currently does not throw an error even if rolling
+ * fails (may be it should?).
+ * return true if append was successful.
+ */
+ synchronized boolean appendLine(String line) {
+ if (out == null) {
+ return false;
+ }
+ out.println();
+ out.print(line);
+ return true;
+ }
+
+ boolean appendLine(long verificationTime, long genStamp, long blockId) {
+ return appendLine("date=\""
+ + dateFormat.format(new Date(verificationTime)) + "\"\t " + "time=\""
+ + verificationTime + "\"\t " + "genstamp=\"" + genStamp + "\"\t "
+ + "id=\"" + blockId + "\"");
+ }
+
+ private synchronized void openCurFile() throws FileNotFoundException {
+ close();
+ out = new PrintStream(new FileOutputStream(curFile, true));
+ }
+
+ private void roll() throws IOException {
+ LOG.info("Rolling current file: " + curFile.getAbsolutePath()
+ + " to previous file: " + prevFile.getAbsolutePath());
+
+ if (!prevFile.delete() && prevFile.exists()) {
+ throw new IOException("Could not delete " + prevFile);
+ }
+
+ close();
+
+ if (!curFile.renameTo(prevFile)) {
+ throw new IOException("Could not rename " + curFile +
+ " to " + prevFile);
+ }
+ }
+
+ synchronized void close() {
+ if (out != null) {
+ out.close();
+ out = null;
+ }
+ }
+
+ /**
+ * This is used to read the lines in order.
+ * If the data is not read completely (i.e, untill hasNext() returns
+ * false), it needs to be explicitly
+ */
+ private static class Reader implements Iterator<String>, Closeable {
+
+ BufferedReader reader;
+ File file;
+ String line;
+ boolean closed = false;
+
+ private Reader(File file) throws IOException {
+ reader = null;
+ this.file = file;
+ readNext();
+ }
+
+ private boolean openFile() throws IOException {
+ if (file == null) {
+ return false;
+ }
+ if (reader != null ) {
+ reader.close();
+ reader = null;
+ }
+ if (file.exists()) {
+ reader = new BufferedReader(new FileReader(file));
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ // read next line if possible.
+ private void readNext() throws IOException {
+ line = null;
+ if (reader == null) {
+ openFile();
+ }
+ try {
+ if (reader != null && (line = reader.readLine()) != null) {
+ return;
+ }
+ } finally {
+ if (!hasNext()) {
+ close();
+ }
+ }
+ }
+
+ public boolean hasNext() {
+ return line != null;
+ }
+
+ public String next() {
+ String curLine = line;
+ try {
+ readNext();
+ } catch (IOException e) {
+ LOG.info("Could not reade next line in LogHandler : " +
+ StringUtils.stringifyException(e));
+ }
+ return curLine;
+ }
+
+ public void remove() {
+ throw new RuntimeException("remove() is not supported.");
+ }
+
+ public void close() throws IOException {
+ if (!closed) {
+ try {
+ if (reader != null) {
+ reader.close();
+ }
+ } finally {
+ file = null;
+ reader = null;
+ closed = true;
+ }
+ }
+ }
+ }
+ }
+}
Modified: hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataBlockScanner.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataBlockScanner.java?rev=1076061&r1=1076060&r2=1076061&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataBlockScanner.java (original)
+++ hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataBlockScanner.java Wed Mar 2 00:13:13 2011
@@ -18,934 +18,266 @@
package org.apache.hadoop.hdfs.server.datanode;
-import java.io.BufferedReader;
-import java.io.Closeable;
-import java.io.DataOutputStream;
import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.FileOutputStream;
-import java.io.FileReader;
import java.io.IOException;
-import java.io.PrintStream;
-import java.text.DateFormat;
-import java.text.SimpleDateFormat;
-import java.util.Collections;
-import java.util.Date;
-import java.util.HashMap;
import java.util.Iterator;
-import java.util.List;
-import java.util.Random;
-import java.util.TreeSet;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
+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.common.GenerationStamp;
-import org.apache.hadoop.hdfs.util.DataTransferThrottler;
-import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.hdfs.server.datanode.DataNode.BPOfferService;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
/**
- * Performs two types of scanning:
- * <li> Gets block files from the data directories and reconciles the
- * difference between the blocks on the disk and in memory in
- * {@link FSDataset}</li>
- * <li> Scans the data directories for block files and verifies that
- * the files are not corrupt</li>
- * This keeps track of blocks and their last verification times.
- * Currently it does not modify the metadata for block.
+ * DataBlockScanner manages block scanning for all the block pools. For each
+ * block pool a {@link BlockPoolScanner} 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.
*/
-
-class DataBlockScanner implements Runnable {
-
+@InterfaceAudience.Private
+public class DataBlockScanner implements Runnable {
public static final Log LOG = LogFactory.getLog(DataBlockScanner.class);
+ private final DataNode datanode;
+ private final FSDataset dataset;
+ private final Configuration conf;
- private static final int MAX_SCAN_RATE = 8 * 1024 * 1024; // 8MB per sec
- private static final int MIN_SCAN_RATE = 1 * 1024 * 1024; // 1MB per sec
-
- static final long DEFAULT_SCAN_PERIOD_HOURS = 21*24L; // three weeks
- private static final long ONE_DAY = 24*3600*1000L;
-
- static final DateFormat dateFormat =
- new SimpleDateFormat("yyyy-MM-dd HH:mm:ss,SSS");
-
- static final String verificationLogFile = "dncp_block_verification.log";
- static final int verficationLogLimit = 5; // * numBlocks.
-
- private long scanPeriod = DEFAULT_SCAN_PERIOD_HOURS * 3600 * 1000;
- DataNode datanode;
- FSDataset dataset;
-
- // sorted set
- TreeSet<BlockScanInfo> blockInfoSet;
- HashMap<Block, BlockScanInfo> blockMap;
-
- long totalScans = 0;
- long totalVerifications = 0; // includes remote verification by clients.
- long totalScanErrors = 0;
- long totalTransientErrors = 0;
-
- long currentPeriodStart = System.currentTimeMillis();
- long bytesLeft = 0; // Bytes to scan in this period
- long totalBytesToScan = 0;
-
- private LogFileHandler verificationLog;
-
- Random random = new Random();
-
- DataTransferThrottler throttler = null;
-
- private static enum ScanType {
- REMOTE_READ, // Verified when a block read by a client etc
- VERIFICATION_SCAN, // scanned as part of periodic verfication
- NONE,
- }
-
- static class BlockScanInfo implements Comparable<BlockScanInfo> {
- Block block;
- long lastScanTime = 0;
- long lastLogTime = 0;
- ScanType lastScanType = ScanType.NONE;
- boolean lastScanOk = true;
-
- BlockScanInfo(Block block) {
- this.block = block;
- }
-
- public int hashCode() {
- return block.hashCode();
- }
-
- public boolean equals(Object other) {
- return other instanceof BlockScanInfo &&
- compareTo((BlockScanInfo)other) == 0;
- }
-
- long getLastScanTime() {
- return ( lastScanType == ScanType.NONE) ? 0 : lastScanTime;
- }
-
- public int compareTo(BlockScanInfo other) {
- long t1 = lastScanTime;
- long t2 = other.lastScanTime;
- return ( t1 < t2 ) ? -1 :
- (( t1 > t2 ) ? 1 : block.compareTo(other.block));
- }
- }
+ /**
+ * 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, BlockPoolScanner> blockPoolScannerMap =
+ new TreeMap<String, BlockPoolScanner>();
+ Thread blockScannerThread = null;
DataBlockScanner(DataNode datanode, FSDataset dataset, Configuration conf) {
this.datanode = datanode;
this.dataset = dataset;
- scanPeriod = conf.getInt("dfs.datanode.scan.period.hours", 0);
- if ( scanPeriod <= 0 ) {
- scanPeriod = DEFAULT_SCAN_PERIOD_HOURS;
- }
- scanPeriod *= 3600 * 1000;
- LOG.info("Periodic Block Verification scan initialized with interval " + scanPeriod + ".");
+ this.conf = conf;
}
- private synchronized boolean isInitialized() {
- return throttler != null;
- }
-
- private void updateBytesToScan(long len, long lastScanTime) {
- // len could be negative when a block is deleted.
- totalBytesToScan += len;
- if ( lastScanTime < currentPeriodStart ) {
- bytesLeft += len;
- }
- // Should we change throttler bandwidth every time bytesLeft changes?
- // not really required.
- }
-
- private synchronized void addBlockInfo(BlockScanInfo info) {
- boolean added = blockInfoSet.add(info);
- blockMap.put(info.block, info);
-
- if ( added ) {
- LogFileHandler log = verificationLog;
- if (log != null) {
- log.setMaxNumLines(blockMap.size() * verficationLogLimit);
+ public void run() {
+ String currentBpId = "";
+ while (datanode.shouldRun && !Thread.interrupted()) {
+ BlockPoolScanner 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.scanBlockPool();
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException ex) {
+ // Interrupt itself again to set the interrupt status
+ blockScannerThread.interrupt();
}
- updateBytesToScan(info.block.getNumBytes(), info.lastScanTime);
}
}
-
- private synchronized void delBlockInfo(BlockScanInfo info) {
- boolean exists = blockInfoSet.remove(info);
- blockMap.remove(info.block);
- if ( exists ) {
- LogFileHandler log = verificationLog;
- if (log != null) {
- log.setMaxNumLines(blockMap.size() * verficationLogLimit);
+
+ // Wait for at least one block pool to be up
+ private void waitForInit() {
+ while (! datanode.upgradeManager.isUpgradeCompleted()
+ || (getBlockPoolSetSize() < datanode.getAllBpOs().length)
+ || (getBlockPoolSetSize() < 1)) {
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ blockScannerThread.interrupt();
+ return;
}
- updateBytesToScan(-info.block.getNumBytes(), info.lastScanTime);
}
}
- /** Update blockMap by the given LogEntry */
- private synchronized void updateBlockInfo(LogEntry e) {
- BlockScanInfo info = blockMap.get(new Block(e.blockId, 0, e.genStamp));
-
- if(info != null && e.verificationTime > 0 &&
- info.lastScanTime < e.verificationTime) {
- delBlockInfo(info);
- info.lastScanTime = e.verificationTime;
- info.lastScanType = ScanType.VERIFICATION_SCAN;
- addBlockInfo(info);
- }
- }
-
- private void init() {
-
- // TODO:FEDERATION block scanner must work one BP at a time
- // get the list of blocks and arrange them in random order
- List<Block> arr = dataset.getFinalizedBlocks("TODO");
- Collections.shuffle(arr);
-
- blockInfoSet = new TreeSet<BlockScanInfo>();
- blockMap = new HashMap<Block, BlockScanInfo>();
+ /**
+ * 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 BlockPoolScanner getNextBPScanner(String currentBpId) {
- long scanTime = -1;
- for (Block block : arr) {
- BlockScanInfo info = new BlockScanInfo( block );
- info.lastScanTime = scanTime--;
- //still keep 'info.lastScanType' to NONE.
- addBlockInfo(info);
- }
-
- /* Pick the first directory that has any existing scanner log.
- * otherwise, pick the first directory.
- */
- // TODO:FEDERATION currently picking only one block pool directory
- // This needs to change to include all the block pool directories
- File dir = null;
- FSDataset.FSVolume[] volumes = dataset.volumes.volumes;
- for(FSDataset.FSVolume vol : volumes) {
- if (LogFileHandler.isFilePresent(vol.getDir(), verificationLogFile)) {
- dir = vol.getDir();
- break;
+ String nextBpId = null;
+ while ((nextBpId == null) && datanode.shouldRun
+ && !blockScannerThread.isInterrupted()) {
+ waitForInit();
+ synchronized (this) {
+ if (getBlockPoolSetSize() > 0) {
+ // Find nextBpId by finding the last modified current log file, if any
+ long lastScanTime = -1;
+ Iterator<String> bpidIterator = blockPoolScannerMap.keySet()
+ .iterator();
+ while (bpidIterator.hasNext()) {
+ String bpid = bpidIterator.next();
+ for (FSDataset.FSVolume vol : dataset.volumes.volumes) {
+ try {
+ File currFile = BlockPoolScanner.getCurrentFile(vol, bpid);
+ if (currFile.exists()) {
+ long lastModified = currFile.lastModified();
+ if (lastScanTime < lastModified) {
+ lastScanTime = lastModified;
+ nextBpId = bpid;
+ }
+ }
+ } catch (IOException e) {
+ LOG.warn("Received exception: ", e);
+ }
+ }
+ }
+
+ // nextBpId can still be null if no current log is found,
+ // find nextBpId sequentially.
+ if (nextBpId == null) {
+ if ("".equals(currentBpId)) {
+ nextBpId = blockPoolScannerMap.firstKey();
+ } else {
+ 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;
}
}
- if (dir == null) {
- dir = volumes[0].getDir();
- }
-
- try {
- // max lines will be updated later during initialization.
- verificationLog = new LogFileHandler(dir, verificationLogFile, 100);
- } catch (IOException e) {
- LOG.warn("Could not open verfication log. " +
- "Verification times are not stored.");
- }
-
- synchronized (this) {
- throttler = new DataTransferThrottler(200, MAX_SCAN_RATE);
- }
- }
-
- private synchronized long getNewBlockScanTime() {
- /* If there are a lot of blocks, this returns a random time with in
- * the scan period. Otherwise something sooner.
- */
- long period = Math.min(scanPeriod,
- Math.max(blockMap.size(),1) * 600 * 1000L);
- return System.currentTimeMillis() - scanPeriod +
- random.nextInt((int)period);
+ return null;
}
- /** Adds block to list of blocks */
- synchronized void addBlock(ExtendedBlock block) {
- if (!isInitialized()) {
- return;
- }
-
- // TODO:FEDERATION use ExtendedBlock
- BlockScanInfo info = blockMap.get(block.getLocalBlock());
- if ( info != null ) {
- LOG.warn("Adding an already existing block " + block);
- delBlockInfo(info);
- }
-
- // TODO:FEDERATION use ExtendedBlock
- info = new BlockScanInfo(block.getLocalBlock());
- info.lastScanTime = getNewBlockScanTime();
-
- addBlockInfo(info);
- adjustThrottler();
+ private synchronized int getBlockPoolSetSize() {
+ return blockPoolScannerMap.size();
}
- /** Deletes the block from internal structures */
- synchronized void deleteBlock(String bpid, Block block) {
- if (!isInitialized()) {
- return;
- }
- // FEDERATION:TODO use bpid
- BlockScanInfo info = blockMap.get(block);
- if ( info != null ) {
- delBlockInfo(info);
- }
- }
-
- /** @return the last scan time */
- synchronized long getLastScanTime(Block block) {
- if (!isInitialized()) {
- return 0;
- }
- BlockScanInfo info = blockMap.get(block);
- return info == null? 0: info.lastScanTime;
- }
-
- /** Deletes blocks from internal structures */
- void deleteBlocks(String bpid, Block[] blocks) {
- for ( Block b : blocks ) {
- deleteBlock(bpid, b);
- }
+ private synchronized BlockPoolScanner getBPScanner(String bpid) {
+ return blockPoolScannerMap.get(bpid);
}
- void verifiedByClient(Block block) {
- updateScanStatus(block, ScanType.REMOTE_READ, true);
+ private synchronized String[] getBpIdList() {
+ return blockPoolScannerMap.keySet().toArray(
+ new String[blockPoolScannerMap.keySet().size()]);
}
- private synchronized void updateScanStatus(Block block,
- ScanType type,
- boolean scanOk) {
- if (!isInitialized()) {
- return;
- }
- BlockScanInfo info = blockMap.get(block);
-
- if ( info != null ) {
- delBlockInfo(info);
+ public void addBlock(ExtendedBlock block) {
+ BlockPoolScanner bpScanner = getBPScanner(block.getBlockPoolId());
+ if (bpScanner != null) {
+ bpScanner.addBlock(block);
} else {
- // It might already be removed. Thats ok, it will be caught next time.
- info = new BlockScanInfo(block);
- }
-
- long now = System.currentTimeMillis();
- info.lastScanType = type;
- info.lastScanTime = now;
- info.lastScanOk = scanOk;
- addBlockInfo(info);
-
- if (type == ScanType.REMOTE_READ) {
- totalVerifications++;
- }
-
- // Don't update meta data too often in case of REMOTE_READ
- // of if the verification failed.
- long diff = now - info.lastLogTime;
- if (!scanOk || (type == ScanType.REMOTE_READ &&
- diff < scanPeriod/3 && diff < ONE_DAY)) {
- return;
- }
-
- info.lastLogTime = now;
- LogFileHandler log = verificationLog;
- if (log != null) {
- log.appendLine("date=\"" + dateFormat.format(new Date(now)) + "\"\t " +
- "time=\"" + now + "\"\t " +
- "genstamp=\"" + block.getGenerationStamp() + "\"\t " +
- "id=\"" + block.getBlockId() +"\"");
+ LOG.warn("No block pool scanner found for block pool id: "
+ + block.getBlockPoolId());
}
}
- private void handleScanFailure(ExtendedBlock block) {
- LOG.info("Reporting bad block " + block);
- try {
- datanode.reportBadBlocks(block);
- } catch (IOException ie) {
- // it is bad, but not bad enough to shutdown the scanner
- LOG.warn("Cannot report bad block=" + block.getBlockId());
+ public synchronized boolean isInitialized(String bpid) {
+ BlockPoolScanner bpScanner = getBPScanner(bpid);
+ if (bpScanner != null) {
+ return bpScanner.isInitialized();
}
+ return false;
}
-
- static private class LogEntry {
- long blockId = -1;
- long verificationTime = -1;
- long genStamp = GenerationStamp.GRANDFATHER_GENERATION_STAMP;
-
- /**
- * The format consists of single line with multiple entries. each
- * entry is in the form : name="value".
- * This simple text and easily extendable and easily parseable with a
- * regex.
- */
- private static Pattern entryPattern =
- Pattern.compile("\\G\\s*([^=\\p{Space}]+)=\"(.*?)\"\\s*");
-
- static LogEntry parseEntry(String line) {
- LogEntry entry = new LogEntry();
-
- Matcher matcher = entryPattern.matcher(line);
- while (matcher.find()) {
- String name = matcher.group(1);
- String value = matcher.group(2);
-
- try {
- if (name.equals("id")) {
- entry.blockId = Long.valueOf(value);
- } else if (name.equals("time")) {
- entry.verificationTime = Long.valueOf(value);
- } else if (name.equals("genstamp")) {
- entry.genStamp = Long.valueOf(value);
- }
- } catch(NumberFormatException nfe) {
- LOG.warn("Cannot parse line: " + line, nfe);
- return null;
- }
- }
-
- return entry;
+
+ 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) {
+ BlockPoolScanner bpScanner = getBPScanner(bpid);
+ buffer.append("\n\nBlock report for block pool: "+bpid + "\n");
+ bpScanner.printBlockReport(buffer, summary);
+ buffer.append("\n");
}
}
- private synchronized void adjustThrottler() {
- long timeLeft = currentPeriodStart+scanPeriod - System.currentTimeMillis();
- long bw = Math.max(bytesLeft*1000/timeLeft, MIN_SCAN_RATE);
- throttler.setBandwidth(Math.min(bw, MAX_SCAN_RATE));
+ public void deleteBlock(String poolId, Block toDelete) {
+ BlockPoolScanner bpScanner = getBPScanner(poolId);
+ if (bpScanner != null) {
+ bpScanner.deleteBlock(toDelete);
+ } else {
+ LOG.warn("No block pool scanner found for block pool id: "
+ + poolId);
+ }
}
-
- private void verifyBlock(ExtendedBlock block) {
- BlockSender blockSender = null;
-
- /* In case of failure, attempt to read second time to reduce
- * transient errors. How do we flush block data from kernel
- * buffers before the second read?
- */
- for (int i=0; i<2; i++) {
- boolean second = (i > 0);
-
- try {
- adjustThrottler();
-
- blockSender = new BlockSender(block, 0, -1, false, false, true,
- datanode);
-
- DataOutputStream out =
- new DataOutputStream(new IOUtils.NullOutputStream());
-
- blockSender.sendBlock(out, null, throttler);
-
- LOG.info((second ? "Second " : "") +
- "Verification succeeded for " + block);
-
- if ( second ) {
- totalTransientErrors++;
- }
-
- // TODO:FEDERATION use Extended block
- updateScanStatus(block.getLocalBlock(), ScanType.VERIFICATION_SCAN, true);
-
- return;
- } catch (IOException e) {
- // TODO:FEDERATION use Extended block
- updateScanStatus(block.getLocalBlock(), ScanType.VERIFICATION_SCAN, false);
-
- // If the block does not exists anymore, then its not an error
- if ( dataset.getFile(block.getBlockPoolId(), block.getLocalBlock()) == null ) {
- LOG.info("Verification failed for " + block + ". Its ok since " +
- "it not in datanode dataset anymore.");
- deleteBlock(block.getBlockPoolId(), block.getLocalBlock());
- return;
- }
- LOG.warn((second ? "Second " : "First ") +
- "Verification failed for " + block + ". Exception : " +
- StringUtils.stringifyException(e));
-
- if (second) {
- totalScanErrors++;
- datanode.getMetrics().blockVerificationFailures.inc();
- handleScanFailure(block);
- return;
- }
- } finally {
- IOUtils.closeStream(blockSender);
- datanode.getMetrics().blocksVerified.inc();
- totalScans++;
- totalVerifications++;
- }
+ public void deleteBlocks(String poolId, Block[] toDelete) {
+ BlockPoolScanner bpScanner = getBPScanner(poolId);
+ if (bpScanner != null) {
+ bpScanner.deleteBlocks(toDelete);
+ } else {
+ LOG.warn("No block pool scanner found for block pool id: "
+ + poolId);
}
}
- private synchronized long getEarliestScanTime() {
- if ( blockInfoSet.size() > 0 ) {
- return blockInfoSet.first().lastScanTime;
- }
- return Long.MAX_VALUE;
- }
-
- // Picks one block and verifies it
- private void verifyFirstBlock() {
- Block block = null;
- synchronized (this) {
- if ( blockInfoSet.size() > 0 ) {
- block = blockInfoSet.first().block;
- }
- }
-
- if ( block != null ) {
- // TODO:FEDERATION blockInfoSet should use ExtendedBlock
- verifyBlock(new ExtendedBlock("TODO", block));
+ public synchronized void shutdown() {
+ if (blockScannerThread != null) {
+ blockScannerThread.interrupt();
}
}
-
- /** returns false if the process was interrupted
- * because the thread is marked to exit.
- */
- private boolean assignInitialVerificationTimes() {
- int numBlocks = 1;
- LogFileHandler log = null;
- synchronized (this) {
- log = verificationLog;
- numBlocks = Math.max(blockMap.size(), 1);
- }
-
- //First udpates the last verification times from the log file.
- LogFileHandler.Reader logReader = null;
- try {
- if (log != null) {
- logReader = log.new Reader(false);
- }
- } catch (IOException e) {
- LOG.warn("Could not read previous verification times : " +
- StringUtils.stringifyException(e));
- }
-
- if (log != null) {
- log.updateCurNumLines();
- }
-
- try {
- // update verification times from the verificationLog.
- while (logReader != null && logReader.hasNext()) {
- if (!datanode.shouldRun || Thread.interrupted()) {
- return false;
- }
- LogEntry entry = LogEntry.parseEntry(logReader.next());
- if (entry != null) {
- updateBlockInfo(entry);
- }
- }
- } finally {
- IOUtils.closeStream(logReader);
- }
-
- /* Initially spread the block reads over half of
- * MIN_SCAN_PERIOD so that we don't keep scanning the
- * blocks too quickly when restarted.
- */
- long verifyInterval = (long) (Math.min( scanPeriod/2.0/numBlocks,
- 10*60*1000 ));
- long lastScanTime = System.currentTimeMillis() - scanPeriod;
-
- /* Before this loop, entries in blockInfoSet that are not
- * updated above have lastScanTime of <= 0 . Loop until first entry has
- * lastModificationTime > 0.
- */
- synchronized (this) {
- if (blockInfoSet.size() > 0 ) {
- BlockScanInfo info;
- while ((info = blockInfoSet.first()).lastScanTime < 0) {
- delBlockInfo(info);
- info.lastScanTime = lastScanTime;
- lastScanTime += verifyInterval;
- addBlockInfo(info);
- }
- }
+
+ public synchronized void addBlockPool(String blockPoolId) {
+ if (blockPoolScannerMap.get(blockPoolId) != null) {
+ return;
}
-
- return true;
- }
-
- private synchronized void startNewPeriod() {
- LOG.info("Starting a new period : work left in prev period : " +
- String.format("%.2f%%", (bytesLeft * 100.0)/totalBytesToScan));
- // reset the byte counts :
- bytesLeft = totalBytesToScan;
- currentPeriodStart = System.currentTimeMillis();
- }
-
- public void run() {
+ BlockPoolScanner bpScanner = new BlockPoolScanner(datanode, dataset,
+ conf, blockPoolId);
try {
-
- init();
-
- //Read last verification times
- if (!assignInitialVerificationTimes()) {
- return;
- }
-
- adjustThrottler();
-
- while (datanode.shouldRun && !Thread.interrupted()) {
- long now = System.currentTimeMillis();
- synchronized (this) {
- if ( now >= (currentPeriodStart + scanPeriod)) {
- startNewPeriod();
- }
- }
- if ( (now - getEarliestScanTime()) >= scanPeriod ) {
- verifyFirstBlock();
- } else {
- try {
- Thread.sleep(1000);
- } catch (InterruptedException ignored) {}
- }
- }
- } catch (RuntimeException e) {
- LOG.warn("RuntimeException during DataBlockScanner.run() : " +
- StringUtils.stringifyException(e));
- throw e;
- } finally {
- shutdown();
- LOG.info("Exiting DataBlockScanner thread.");
+ bpScanner.init();
+ } catch (IOException ex) {
+ LOG.warn("Failed to initialized block scanner for pool id="+blockPoolId);
+ return;
}
+ blockPoolScannerMap.put(blockPoolId, bpScanner);
+ LOG.info("Added bpid=" + blockPoolId + " to blockPoolScannerMap, new size="
+ + blockPoolScannerMap.size());
}
- synchronized void shutdown() {
- LogFileHandler log = verificationLog;
- verificationLog = null;
- if (log != null) {
- log.close();
- }
+ public synchronized void removeBlockPool(String blockPoolId) {
+ blockPoolScannerMap.remove(blockPoolId);
+ LOG.info("Removed bpid="+blockPoolId+" from blockPoolScannerMap");
}
- synchronized void printBlockReport(StringBuilder buffer,
- boolean summaryOnly) {
- long oneHour = 3600*1000;
- long oneDay = 24*oneHour;
- long oneWeek = 7*oneDay;
- long fourWeeks = 4*oneWeek;
-
- int inOneHour = 0;
- int inOneDay = 0;
- int inOneWeek = 0;
- int inFourWeeks = 0;
- int inScanPeriod = 0;
- int neverScanned = 0;
-
- int total = blockInfoSet.size();
-
- long now = System.currentTimeMillis();
-
- Date date = new Date();
-
- for(Iterator<BlockScanInfo> it = blockInfoSet.iterator(); it.hasNext();) {
- BlockScanInfo info = it.next();
-
- long scanTime = info.getLastScanTime();
- long diff = now - scanTime;
-
- if (diff <= oneHour) inOneHour++;
- if (diff <= oneDay) inOneDay++;
- if (diff <= oneWeek) inOneWeek++;
- if (diff <= fourWeeks) inFourWeeks++;
- if (diff <= scanPeriod) inScanPeriod++;
- if (scanTime <= 0) neverScanned++;
-
- if (!summaryOnly) {
- date.setTime(scanTime);
- String scanType =
- (info.lastScanType == ScanType.REMOTE_READ) ? "remote" :
- ((info.lastScanType == ScanType.VERIFICATION_SCAN) ? "local" :
- "none");
- buffer.append(String.format("%-26s : status : %-6s type : %-6s" +
- " scan time : " +
- "%-15d %s\n", info.block,
- (info.lastScanOk ? "ok" : "failed"),
- scanType, scanTime,
- (scanTime <= 0) ? "not yet verified" :
- dateFormat.format(date)));
- }
+ // This method is used for testing
+ long getBlocksScannedInLastRun(String bpid) throws IOException {
+ BlockPoolScanner bpScanner = getBPScanner(bpid);
+ if (bpScanner == null) {
+ throw new IOException("Block Pool: "+bpid+" is not running");
+ } else {
+ return bpScanner.getBlocksScannedInLastRun();
}
-
- double pctPeriodLeft = (scanPeriod + currentPeriodStart - now)
- *100.0/scanPeriod;
- double pctProgress = (totalBytesToScan == 0) ? 100 :
- (totalBytesToScan-bytesLeft)*10000.0/totalBytesToScan/
- (100-pctPeriodLeft+1e-10);
-
- buffer.append(String.format("\nTotal Blocks : %6d" +
- "\nVerified in last hour : %6d" +
- "\nVerified in last day : %6d" +
- "\nVerified in last week : %6d" +
- "\nVerified in last four weeks : %6d" +
- "\nVerified in SCAN_PERIOD : %6d" +
- "\nNot yet verified : %6d" +
- "\nVerified since restart : %6d" +
- "\nScans since restart : %6d" +
- "\nScan errors since restart : %6d" +
- "\nTransient scan errors : %6d" +
- "\nCurrent scan rate limit KBps : %6d" +
- "\nProgress this period : %6.0f%%" +
- "\nTime left in cur period : %6.2f%%" +
- "\n",
- total, inOneHour, inOneDay, inOneWeek,
- inFourWeeks, inScanPeriod, neverScanned,
- totalVerifications, totalScans,
- totalScanErrors, totalTransientErrors,
- Math.round(throttler.getBandwidth()/1024.0),
- pctProgress, pctPeriodLeft));
}
-
- /**
- * This class takes care of log file used to store the last verification
- * times of the blocks. It rolls the current file when it is too big etc.
- * If there is an error while writing, it stops updating with an error
- * message.
- */
- private static class LogFileHandler {
-
- private static final String curFileSuffix = ".curr";
- private static final String prevFileSuffix = ".prev";
-
- // Don't roll files more often than this
- private static final long minRollingPeriod = 6 * 3600 * 1000L; // 6 hours
- private static final long minWarnPeriod = minRollingPeriod;
- private static final int minLineLimit = 1000;
-
-
- static boolean isFilePresent(File dir, String filePrefix) {
- return new File(dir, filePrefix + curFileSuffix).exists() ||
- new File(dir, filePrefix + prevFileSuffix).exists();
- }
- private File curFile;
- private File prevFile;
-
- private int maxNumLines = -1; // not very hard limit on number of lines.
- private int curNumLines = -1;
-
- long lastWarningTime = 0;
-
- private PrintStream out;
-
- int numReaders = 0;
-
- /**
- * Opens the log file for appending.
- * Note that rolling will happen only after "updateLineCount()" is
- * called. This is so that line count could be updated in a separate
- * thread without delaying start up.
- *
- * @param dir where the logs files are located.
- * @param filePrefix prefix of the file.
- * @param maxNumLines max lines in a file (its a soft limit).
- * @throws IOException
- */
- LogFileHandler(File dir, String filePrefix, int maxNumLines)
- throws IOException {
- curFile = new File(dir, filePrefix + curFileSuffix);
- prevFile = new File(dir, filePrefix + prevFileSuffix);
- openCurFile();
- curNumLines = -1;
- setMaxNumLines(maxNumLines);
- }
-
- // setting takes affect when next entry is added.
- synchronized void setMaxNumLines(int maxNumLines) {
- this.maxNumLines = Math.max(maxNumLines, minLineLimit);
- }
-
- /**
- * Append "\n" + line.
- * If the log file need to be rolled, it will done after
- * appending the text.
- * This does not throw IOException when there is an error while
- * appending. Currently does not throw an error even if rolling
- * fails (may be it should?).
- * return true if append was successful.
- */
- synchronized boolean appendLine(String line) {
- out.println();
- out.print(line);
- curNumLines += (curNumLines < 0) ? -1 : 1;
- try {
- rollIfRequired();
- } catch (IOException e) {
- warn("Rolling failed for " + curFile + " : " + e.getMessage());
- return false;
- }
- return true;
- }
-
- //warns only once in a while
- synchronized private void warn(String msg) {
- long now = System.currentTimeMillis();
- if ((now - lastWarningTime) >= minWarnPeriod) {
- lastWarningTime = now;
- LOG.warn(msg);
- }
- }
-
- private synchronized void openCurFile() throws FileNotFoundException {
- close();
- out = new PrintStream(new FileOutputStream(curFile, true));
- }
-
- //This reads the current file and updates the count.
- void updateCurNumLines() {
- int count = 0;
- Reader it = null;
- try {
- for(it = new Reader(true); it.hasNext(); count++) {
- it.next();
- }
- } catch (IOException e) {
-
- } finally {
- synchronized (this) {
- curNumLines = count;
- }
- IOUtils.closeStream(it);
- }
- }
-
- private void rollIfRequired() throws IOException {
- if (curNumLines < maxNumLines || numReaders > 0) {
- return;
- }
-
- long now = System.currentTimeMillis();
- if (now < minRollingPeriod) {
- return;
- }
-
- if (!prevFile.delete() && prevFile.exists()) {
- throw new IOException("Could not delete " + prevFile);
- }
-
- close();
-
- if (!curFile.renameTo(prevFile)) {
- openCurFile();
- throw new IOException("Could not rename " + curFile +
- " to " + prevFile);
- }
-
- openCurFile();
- updateCurNumLines();
- }
-
- synchronized void close() {
- if (out != null) {
- out.close();
- out = null;
- }
- }
-
- /**
- * This is used to read the lines in order.
- * If the data is not read completely (i.e, untill hasNext() returns
- * false), it needs to be explicitly
- */
- private class Reader implements Iterator<String>, Closeable {
-
- BufferedReader reader;
- File file;
- String line;
- boolean closed = false;
-
- private Reader(boolean skipPrevFile) throws IOException {
- synchronized (LogFileHandler.this) {
- numReaders++;
- }
- reader = null;
- file = (skipPrevFile) ? curFile : prevFile;
- readNext();
- }
-
- private boolean openFile() throws IOException {
-
- for(int i=0; i<2; i++) {
- if (reader != null || i > 0) {
- // move to next file
- file = (file == prevFile) ? curFile : null;
- }
- if (file == null) {
- return false;
- }
- if (file.exists()) {
- break;
- }
- }
-
- if (reader != null ) {
- reader.close();
- reader = null;
- }
-
- reader = new BufferedReader(new FileReader(file));
- return true;
- }
-
- // read next line if possible.
- private void readNext() throws IOException {
- line = null;
- try {
- if (reader != null && (line = reader.readLine()) != null) {
- return;
- }
- if (line == null) {
- // move to the next file.
- if (openFile()) {
- readNext();
- }
- }
- } finally {
- if (!hasNext()) {
- close();
- }
- }
- }
-
- public boolean hasNext() {
- return line != null;
- }
-
- public String next() {
- String curLine = line;
- try {
- readNext();
- } catch (IOException e) {
- LOG.info("Could not reade next line in LogHandler : " +
- StringUtils.stringifyException(e));
- }
- return curLine;
- }
-
- public void remove() {
- throw new RuntimeException("remove() is not supported.");
- }
- public void close() throws IOException {
- if (!closed) {
- try {
- if (reader != null) {
- reader.close();
- }
- } finally {
- file = null;
- reader = null;
- closed = true;
- synchronized (LogFileHandler.this) {
- numReaders--;
- assert(numReaders >= 0);
- }
- }
- }
- }
- }
+ public void start() {
+ blockScannerThread = new Thread(this);
+ blockScannerThread.setDaemon(true);
+ blockScannerThread.start();
}
@InterfaceAudience.Private
@@ -954,25 +286,23 @@ class DataBlockScanner implements Runnab
public void doGet(HttpServletRequest request,
HttpServletResponse response) throws IOException {
-
response.setContentType("text/plain");
- DataBlockScanner blockScanner = (DataBlockScanner)
- getServletContext().getAttribute("datanode.blockScanner");
+ 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 if (blockScanner.isInitialized()) {
- blockScanner.printBlockReport(buffer, summary);
} else {
- buffer.append("Periodic block scanner is not yet initialized. " +
- "Please check back again after some time.");
+ blockScanner.printBlockReport(buffer, summary);
}
response.getWriter().write(buffer.toString()); // extra copy!
}
}
+
}
Modified: hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java?rev=1076061&r1=1076060&r2=1076061&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java (original)
+++ hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java Wed Mar 2 00:13:13 2011
@@ -277,6 +277,8 @@ public class DataNode extends Configured
void refreshNamenodes(Configuration conf)
throws IOException, InterruptedException {
+ LOG.info("Refresh request received for namnodes: "
+ + conf.get(DFSConfigKeys.DFS_FEDERATION_NAMENODES));
List<InetSocketAddress> newAddresses = DFSUtil.getNNAddresses(conf);
List<BPOfferService> toShutdown = new ArrayList<BPOfferService>();
List<InetSocketAddress> toStart = new ArrayList<InetSocketAddress>();
@@ -344,7 +346,6 @@ public class DataNode extends Configured
boolean isBlockTokenInitialized = false;
public DataBlockScanner blockScanner = null;
- public Daemon blockScannerThread = null;
private DirectoryScanner directoryScanner = null;
/** Activated plug-ins. */
@@ -457,7 +458,8 @@ public class DataNode extends Configured
this.infoServer.addInternalServlet(null, "/streamFile/*", StreamFile.class);
this.infoServer.addInternalServlet(null, "/getFileChecksum/*",
FileChecksumServlets.GetServlet.class);
- this.infoServer.setAttribute("datanode.blockScanner", blockScanner);
+
+ this.infoServer.setAttribute("datanode", this);
this.infoServer.setAttribute(JspHelper.CURRENT_CONF, conf);
this.infoServer.addServlet(null, "/blockScannerReport",
DataBlockScanner.Servlet.class);
@@ -519,7 +521,10 @@ public class DataNode extends Configured
/**
* See {@link DataBlockScanner}
*/
- private void initDataBlockScanner(Configuration conf) {
+ private synchronized void initDataBlockScanner(Configuration conf) {
+ if (blockScanner != null) {
+ return;
+ }
String reason = null;
if (conf.getInt(DFS_DATANODE_SCAN_PERIOD_HOURS_KEY,
DFS_DATANODE_SCAN_PERIOD_HOURS_DEFAULT) < 0) {
@@ -529,6 +534,7 @@ public class DataNode extends Configured
}
if (reason == null) {
blockScanner = new DataBlockScanner(this, (FSDataset)data, conf);
+ blockScanner.start();
} else {
LOG.info("Periodic Block Verification scan is disabled because " +
reason + ".");
@@ -536,19 +542,18 @@ public class DataNode extends Configured
}
private void shutdownDataBlockScanner() {
- if (blockScannerThread != null) {
- blockScannerThread.interrupt();
- try {
- blockScannerThread.join(3600000L); // wait for at most 1 hour
- } catch (InterruptedException ie) {
- }
+ if (blockScanner != null) {
+ blockScanner.shutdown();
}
}
/**
* See {@link DirectoryScanner}
*/
- private void initDirectoryScanner(Configuration conf) {
+ private synchronized void initDirectoryScanner(Configuration conf) {
+ if (directoryScanner != null) {
+ return;
+ }
String reason = null;
if (conf.getInt(DFS_DATANODE_DIRECTORYSCAN_INTERVAL_KEY,
DFS_DATANODE_DIRECTORYSCAN_INTERVAL_DEFAULT) < 0) {
@@ -753,6 +758,7 @@ public class DataNode extends Configured
}
initFsDataSet(conf, dataDirs);
data.addBlockPool(blockPoolId, conf);
+ initPeriodicScanners(conf);
}
/**
@@ -941,6 +947,9 @@ public class DataNode extends Configured
blockPoolManager.remove(this);
shouldServiceRun = false;
RPC.stopProxy(bpNamenode);
+ if (blockScanner != null) {
+ blockScanner.removeBlockPool(this.getBlockPoolId());
+ }
}
/**
@@ -984,22 +993,9 @@ public class DataNode extends Configured
DatanodeCommand cmd = blockReport();
processCommand(cmd);
- // start block scanner
- // TODO:FEDERATION - SHOULD BE MOVED OUT OF THE THREAD...
- // although it will work as is, since it checks (blockScannerThread == null)
- // under synchronized(blockScanner). See also {@link initDataBlockScanner()}
- //
- // Note that there may be a problem starting this before BPOfferService
- // is running; and we may need to kick it to restart within each BP subdir
- // as they come online, otherwise will wait 3 weeks for the first run.
+ // Now safe to start scanning the block pool
if (blockScanner != null) {
- synchronized(blockScanner) {
- if(blockScannerThread == null && upgradeManager.isUpgradeCompleted()) {
- LOG.info("Periodic Block Verification scan starting.");
- blockScannerThread = new Daemon(blockScanner);
- blockScannerThread.start();
- }
- }
+ blockScanner.addBlockPool(this.blockPoolId);
}
//
@@ -1158,7 +1154,7 @@ public class DataNode extends Configured
}
initialized = true; // bp is initialized;
-
+
while (shouldRun && shouldServiceRun) {
try {
// TODO:FEDERATION needs to be moved too
@@ -1928,7 +1924,6 @@ public class DataNode extends Configured
// start dataXceiveServer
dataXceiverServer.start();
ipcServer.start();
- initPeriodicScanners(conf);
startPlugins(conf);
// BlockTokenSecretManager is created here, but it shouldn't be
Modified: hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java?rev=1076061&r1=1076060&r2=1076061&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java (original)
+++ hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java Wed Mar 2 00:13:13 2011
@@ -178,17 +178,6 @@ class DataXceiver extends DataTransferPr
SUCCESS.write(out); // send op status
long read = blockSender.sendBlock(out, baseStream, null); // send data
-
- if (blockSender.isBlockReadFully()) {
- // See if client verification succeeded.
- // This is an optional response from client.
- try {
- if (DataTransferProtocol.Status.read(in) == CHECKSUM_OK
- && datanode.blockScanner != null) {
- datanode.blockScanner.verifiedByClient(block.getLocalBlock());
- }
- } catch (IOException ignored) {}
- }
datanode.myMetrics.bytesRead.inc((int) read);
datanode.myMetrics.blocksRead.inc();
Modified: hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/FSDataset.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/FSDataset.java?rev=1076061&r1=1076060&r2=1076061&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/FSDataset.java (original)
+++ hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/FSDataset.java Wed Mar 2 00:13:13 2011
@@ -2389,6 +2389,7 @@ public class FSDataset implements FSCons
throws IOException {
DataNode.LOG.info("Adding block pool " + bpid);
volumes.addBlockPool(bpid, conf);
+ volumeMap.initBlockPool(bpid);
volumes.getVolumeMap(bpid, volumeMap);
}
Modified: hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/ReplicasMap.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/ReplicasMap.java?rev=1076061&r1=1076060&r2=1076061&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/ReplicasMap.java (original)
+++ hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/ReplicasMap.java Wed Mar 2 00:13:13 2011
@@ -150,4 +150,14 @@ class ReplicasMap {
Map<Long, ReplicaInfo> m = map.get(bpid);
return m != null ? m.values() : null;
}
+
+ void initBlockPool(String bpid) {
+ checkBlockPool(bpid);
+ Map<Long, ReplicaInfo> m = map.get(bpid);
+ if (m == null) {
+ // Add an entry for block pool if it does not exist already
+ m = new HashMap<Long, ReplicaInfo>();
+ map.put(bpid, m);
+ }
+ }
}
\ No newline at end of file
Modified: hadoop/hdfs/branches/HDFS-1052/src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestInterDatanodeProtocol.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1052/src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestInterDatanodeProtocol.java?rev=1076061&r1=1076060&r2=1076061&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1052/src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestInterDatanodeProtocol.java (original)
+++ hadoop/hdfs/branches/HDFS-1052/src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestInterDatanodeProtocol.java Wed Mar 2 00:13:13 2011
@@ -98,7 +98,9 @@ public class TestInterDatanodeProtocol {
assertTrue(datanode != null);
//stop block scanner, so we could compare lastScanTime
- datanode.blockScannerThread.interrupt();
+ if (datanode.blockScanner != null) {
+ datanode.blockScanner.shutdown();
+ }
//verify BlockMetaDataInfo
ExtendedBlock b = locatedblock.getBlock();