hadoop-hdfs-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jiten...@apache.org
Subject 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/
Date Wed, 02 Mar 2011 00:13:13 GMT
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();



Mime
View raw message