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 su...@apache.org on 2011/03/02 02:21:36 UTC
svn commit: r1076075 - in /hadoop/hdfs/branches/HDFS-1052: ./
src/java/org/apache/hadoop/hdfs/server/datanode/
src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/
Author: suresh
Date: Wed Mar 2 01:21:35 2011
New Revision: 1076075
URL: http://svn.apache.org/viewvc?rev=1076075&view=rev
Log:
HDFS-1674. Rename BlockPool class to BlockPoolSlice. Contributed by Jakob Homan and Tanping Wang.
Added:
hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolSliceScanner.java
hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolSliceStorage.java
Removed:
hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolScanner.java
hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolStorage.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/DataStorage.java
hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java
hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/FSDataset.java
hadoop/hdfs/branches/HDFS-1052/src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
Modified: hadoop/hdfs/branches/HDFS-1052/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1052/CHANGES.txt?rev=1076075&r1=1076074&r2=1076075&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1052/CHANGES.txt (original)
+++ hadoop/hdfs/branches/HDFS-1052/CHANGES.txt Wed Mar 2 01:21:35 2011
@@ -110,7 +110,9 @@ Trunk (unreleased changes)
HDFS-1664. Add block pool storage usage to Namenode WebUI.
(Tanping via suresh)
-
+
+ HDFS-1674. Rename BlockPool class to BlockPoolSlice.
+ (jghoman, Tanping via suresh)
IMPROVEMENTS
Added: hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolSliceScanner.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolSliceScanner.java?rev=1076075&view=auto
==============================================================================
--- hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolSliceScanner.java (added)
+++ hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolSliceScanner.java Wed Mar 2 01:21:35 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 BlockPoolSliceScanner {
+
+ public static final Log LOG = LogFactory.getLog(BlockPoolSliceScanner.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));
+ }
+ }
+
+ BlockPoolSliceScanner(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.getBlockPoolSlice(blockPoolId).getDirectory();
+ if (LogFileHandler.isFilePresent(bpDir, verificationLogFile)) {
+ dir = bpDir;
+ break;
+ }
+ }
+ if (dir == null) {
+ dir = volumes[0].getBlockPoolSlice(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.getBlockPoolSlice(bpid).getDirectory(),
+ BlockPoolSliceScanner.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 scanBlockPoolSlice() {
+ 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;
+ }
+ }
+ }
+ }
+ }
+}
Added: hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolSliceStorage.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolSliceStorage.java?rev=1076075&view=auto
==============================================================================
--- hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolSliceStorage.java (added)
+++ hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/BlockPoolSliceStorage.java Wed Mar 2 01:21:35 2011
@@ -0,0 +1,522 @@
+/**
+ * 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.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Properties;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hdfs.protocol.FSConstants;
+import org.apache.hadoop.hdfs.server.common.InconsistentFSStateException;
+import org.apache.hadoop.hdfs.server.common.Storage;
+import org.apache.hadoop.hdfs.server.common.HdfsConstants.NodeType;
+import org.apache.hadoop.hdfs.server.common.HdfsConstants.StartupOption;
+import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo;
+import org.apache.hadoop.util.Daemon;
+
+/**
+ * Manages storage for the set of BlockPoolSlices which share a particular
+ * block pool id, on this DataNode.
+ *
+ * This class supports the following functionality:
+ * <ol>
+ * <li> Formatting a new block pool storage</li>
+ * <li> Recovering a storage state to a consistent state (if possible></li>
+ * <li> Taking a snapshot of the block pool during upgrade</li>
+ * <li> Rolling back a block pool to a previous snapshot</li>
+ * <li> Finalizing block storage by deletion of a snapshot</li>
+ * </ul>
+ *
+ * @see Storage
+ */
+@InterfaceAudience.Private
+public class BlockPoolSliceStorage extends Storage {
+ private static final Pattern BLOCK_POOL_PATH_PATTERN = Pattern
+ .compile("^(.*)"
+ + "(\\/BP-[0-9]+\\-\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\-[0-9]+\\/.*)$");
+
+ private String blockpoolID = ""; // id of the blockpool
+
+ BlockPoolSliceStorage() {
+ super(NodeType.DATA_NODE);
+ }
+
+ BlockPoolSliceStorage(int namespaceID, String bpID, long cTime, String clusterId) {
+ super(NodeType.DATA_NODE);
+ this.namespaceID = namespaceID;
+ this.blockpoolID = bpID;
+ this.cTime = cTime;
+ this.clusterID = clusterId;
+ }
+
+ /**
+ * Analyze storage directories. Recover from previous transitions if required.
+ *
+ * @param nsInfo namespace information
+ * @param dataDirs storage directories of block pool
+ * @param startOpt startup option
+ * @throws IOException on error
+ */
+ void recoverTransitionRead(NamespaceInfo nsInfo, Collection<File> dataDirs,
+ StartupOption startOpt) throws IOException {
+ assert FSConstants.LAYOUT_VERSION == nsInfo.getLayoutVersion()
+ : "Block-pool and name-node layout versions must be the same.";
+
+ // 1. For each BP data directory analyze the state and
+ // check whether all is consistent before transitioning.
+ this.storageDirs = new ArrayList<StorageDirectory>(dataDirs.size());
+ ArrayList<StorageState> dataDirStates = new ArrayList<StorageState>(
+ dataDirs.size());
+ for (Iterator<File> it = dataDirs.iterator(); it.hasNext();) {
+ File dataDir = it.next();
+ StorageDirectory sd = new StorageDirectory(dataDir, null, false);
+ StorageState curState;
+ try {
+ curState = sd.analyzeStorage(startOpt);
+ // sd is locked but not opened
+ switch (curState) {
+ case NORMAL:
+ break;
+ case NON_EXISTENT:
+ // ignore this storage
+ LOG.info("Storage directory " + dataDir + " does not exist.");
+ it.remove();
+ continue;
+ case NOT_FORMATTED: // format
+ LOG.info("Storage directory " + dataDir + " is not formatted.");
+ LOG.info("Formatting ...");
+ format(sd, nsInfo);
+ break;
+ default: // recovery part is common
+ sd.doRecover(curState);
+ }
+ } catch (IOException ioe) {
+ sd.unlock();
+ throw ioe;
+ }
+ // add to the storage list. This is inherited from parent class, Storage.
+ addStorageDir(sd);
+ dataDirStates.add(curState);
+ }
+
+ if (dataDirs.size() == 0) // none of the data dirs exist
+ throw new IOException(
+ "All specified directories are not accessible or do not exist.");
+
+ // 2. Do transitions
+ // Each storage directory is treated individually.
+ // During startup some of them can upgrade or roll back
+ // while others could be up-to-date for the regular startup.
+ for (int idx = 0; idx < getNumStorageDirs(); idx++) {
+ doTransition(getStorageDir(idx), nsInfo, startOpt);
+ assert getLayoutVersion() == nsInfo.getLayoutVersion()
+ : "Data-node and name-node layout versions must be the same.";
+ assert getCTime() == nsInfo.getCTime()
+ : "Data-node and name-node CTimes must be the same.";
+ }
+
+ // 3. Update all storages. Some of them might have just been formatted.
+ this.writeAll();
+ }
+
+ /**
+ * Format a block pool slice storage.
+ * @param dnCurDir DataStorage current directory
+ * @param nsInfo the name space info
+ * @throws IOException Signals that an I/O exception has occurred.
+ */
+ void format(File dnCurDir, NamespaceInfo nsInfo) throws IOException {
+ File curBpDir = getBpRoot(nsInfo.getBlockPoolID(), dnCurDir);
+ StorageDirectory bpSdir = new StorageDirectory(curBpDir);
+ format(bpSdir, nsInfo);
+ }
+
+ /**
+ * Format a block pool slice storage.
+ * @param sd the block pool storage
+ * @param nsInfo the name space info
+ * @throws IOException Signals that an I/O exception has occurred.
+ */
+ private void format(StorageDirectory bpSdir, NamespaceInfo nsInfo) throws IOException {
+ LOG.info("Formatting block pool " + blockpoolID + " directory "
+ + bpSdir.getCurrentDir());
+ bpSdir.clearDirectory(); // create directory
+ this.layoutVersion = FSConstants.LAYOUT_VERSION;
+ this.cTime = nsInfo.getCTime();
+ this.namespaceID = nsInfo.getNamespaceID();
+ this.blockpoolID = nsInfo.getBlockPoolID();
+ this.storageType = NodeType.DATA_NODE;
+ bpSdir.write();
+ }
+
+ /**
+ * Set layoutVersion, namespaceID and blockpoolID into block pool storage
+ * VERSION file
+ */
+ @Override
+ protected void setFields(Properties props, StorageDirectory sd)
+ throws IOException {
+ props.setProperty("layoutVersion", String.valueOf(layoutVersion));
+ props.setProperty("namespaceID", String.valueOf(namespaceID));
+ props.setProperty("blockpoolID", blockpoolID);
+ props.setProperty("cTime", String.valueOf(cTime));
+ }
+
+ /** Validate and set block pool ID */
+ private void setBlockPoolID(File storage, String bpid)
+ throws InconsistentFSStateException {
+ if (bpid == null || bpid.equals("")) {
+ throw new InconsistentFSStateException(storage, "file "
+ + STORAGE_FILE_VERSION + " is invalid.");
+ }
+
+ if (!blockpoolID.equals("") && !blockpoolID.equals(bpid)) {
+ throw new InconsistentFSStateException(storage,
+ "Unexepcted blockpoolID " + bpid + " . Expected " + blockpoolID);
+ }
+ blockpoolID = bpid;
+ }
+
+ @Override
+ protected void getFields(Properties props, StorageDirectory sd)
+ throws IOException {
+ setLayoutVersion(props, sd);
+ setNamespaceID(props, sd);
+ setcTime(props, sd);
+
+ String sbpid = props.getProperty("blockpoolID");
+ setBlockPoolID(sd.getRoot(), sbpid);
+ }
+
+ /**
+ * Analyze whether a transition of the BP state is required and
+ * perform it if necessary.
+ * <br>
+ * Rollback if previousLV >= LAYOUT_VERSION && prevCTime <= namenode.cTime.
+ * Upgrade if this.LV > LAYOUT_VERSION || this.cTime < namenode.cTime Regular
+ * startup if this.LV = LAYOUT_VERSION && this.cTime = namenode.cTime
+ *
+ * @param sd storage directory,
+ * @param nsInfo namespace info
+ * @param startOpt startup option
+ * @throws IOException
+ */
+ private void doTransition(StorageDirectory sd, // i.e. <SD>/current/<bpid>
+ NamespaceInfo nsInfo, StartupOption startOpt) throws IOException {
+ if (startOpt == StartupOption.ROLLBACK)
+ doRollback(sd, nsInfo); // rollback if applicable
+
+ sd.read();
+ checkVersionUpgradable(this.layoutVersion);
+ assert this.layoutVersion >= FSConstants.LAYOUT_VERSION
+ : "Future version is not allowed";
+ if (getNamespaceID() != nsInfo.getNamespaceID()) {
+ throw new IOException("Incompatible namespaceIDs in "
+ + sd.getRoot().getCanonicalPath() + ": namenode namespaceID = "
+ + nsInfo.getNamespaceID() + "; datanode namespaceID = "
+ + getNamespaceID());
+ }
+ if (!blockpoolID.equals(nsInfo.getBlockPoolID())) {
+ throw new IOException("Incompatible blockpoolIDs in "
+ + sd.getRoot().getCanonicalPath() + ": namenode blockpoolID = "
+ + nsInfo.getBlockPoolID() + "; datanode blockpoolID = "
+ + blockpoolID);
+ }
+ if (this.layoutVersion == FSConstants.LAYOUT_VERSION
+ && this.cTime == nsInfo.getCTime())
+ return; // regular startup
+
+ // verify necessity of a distributed upgrade
+ verifyDistributedUpgradeProgress(nsInfo);
+ if (this.layoutVersion > FSConstants.LAYOUT_VERSION
+ || this.cTime < nsInfo.getCTime()) {
+ doUpgrade(sd, nsInfo); // upgrade
+ return;
+ }
+ // layoutVersion == LAYOUT_VERSION && this.cTime > nsInfo.cTime
+ // must shutdown
+ throw new IOException("Datanode state: LV = " + this.getLayoutVersion()
+ + " CTime = " + this.getCTime()
+ + " is newer than the namespace state: LV = "
+ + nsInfo.getLayoutVersion() + " CTime = " + nsInfo.getCTime());
+ }
+
+ /**
+ * Upgrade to any release after 0.22 (0.22 included) release e.g. 0.22 => 0.23
+ * Upgrade procedure is as follows:
+ * <ol>
+ * <li>If <SD>/current/<bpid>/previous exists then delete it</li>
+ * <li>Rename <SD>/current/<bpid>/current to
+ * <SD>/current/bpid/current/previous.tmp</li>
+ * <li>Create new <SD>current/<bpid>/current directory</li>
+ * <ol>
+ * <li>Hard links for block files are created from previous.tmp to current</li>
+ * <li>Save new version file in current directory</li>
+ * </ol>
+ * <li>Rename previous.tmp to previous</li> </ol>
+ *
+ * @param bpSd storage directory <SD>/current/<bpid>
+ * @param nsInfo Namespace Info from the namenode
+ * @throws IOException on error
+ */
+ void doUpgrade(StorageDirectory bpSd, NamespaceInfo nsInfo) throws IOException {
+ // Upgrading is applicable only to release with federation or after
+ if (!(this.getLayoutVersion() < LAST_PRE_FEDERATION_LAYOUT_VERSION)) {
+ return;
+ }
+ LOG.info("Upgrading block pool storage directory " + bpSd.getRoot()
+ + ".\n old LV = " + this.getLayoutVersion() + "; old CTime = "
+ + this.getCTime() + ".\n new LV = " + nsInfo.getLayoutVersion()
+ + "; new CTime = " + nsInfo.getCTime());
+ // get <SD>/previous directory
+ String dnRoot = getDataNodeStorageRoot(bpSd.getRoot().getCanonicalPath());
+ StorageDirectory dnSdStorage = new StorageDirectory(new File(dnRoot));
+ File dnPrevDir = dnSdStorage.getPreviousDir();
+
+ // If <SD>/previous directory exists delete it
+ if (dnPrevDir.exists()) {
+ deleteDir(dnPrevDir);
+ }
+ File bpCurDir = bpSd.getCurrentDir();
+ File bpPrevDir = bpSd.getPreviousDir();
+ assert bpCurDir.exists() : "BP level current directory must exist.";
+ cleanupDetachDir(new File(bpCurDir, DataStorage.STORAGE_DIR_DETACHED));
+
+ // 1. Delete <SD>/current/<bpid>/previous dir before upgrading
+ if (bpPrevDir.exists()) {
+ deleteDir(bpPrevDir);
+ }
+ File bpTmpDir = bpSd.getPreviousTmp();
+ assert !bpTmpDir.exists() : "previous.tmp directory must not exist.";
+
+ // 2. Rename <SD>/curernt/<bpid>/current to <SD>/curernt/<bpid>/previous.tmp
+ rename(bpCurDir, bpTmpDir);
+
+ // 3. Create new <SD>/current with block files hardlinks and VERSION
+ linkAllBlocks(bpTmpDir, bpCurDir);
+ this.layoutVersion = FSConstants.LAYOUT_VERSION;
+ assert this.namespaceID == nsInfo.getNamespaceID()
+ : "Data-node and name-node layout versions must be the same.";
+ this.cTime = nsInfo.getCTime();
+ bpSd.write();
+
+ // 4.rename <SD>/curernt/<bpid>/previous.tmp to <SD>/curernt/<bpid>/previous
+ rename(bpTmpDir, bpPrevDir);
+ LOG.info("Upgrade of block pool " + blockpoolID + " at " + bpSd.getRoot()
+ + " is complete.");
+ }
+
+ /**
+ * Cleanup the detachDir.
+ *
+ * If the directory is not empty report an error; Otherwise remove the
+ * directory.
+ *
+ * @param detachDir detach directory
+ * @throws IOException if the directory is not empty or it can not be removed
+ */
+ private void cleanupDetachDir(File detachDir) throws IOException {
+ if (layoutVersion >= PRE_RBW_LAYOUT_VERSION && detachDir.exists()
+ && detachDir.isDirectory()) {
+
+ if (detachDir.list().length != 0) {
+ throw new IOException("Detached directory " + detachDir
+ + " is not empty. Please manually move each file under this "
+ + "directory to the finalized directory if the finalized "
+ + "directory tree does not have the file.");
+ } else if (!detachDir.delete()) {
+ throw new IOException("Cannot remove directory " + detachDir);
+ }
+ }
+ }
+
+ /*
+ * Roll back to old snapshot at the block pool level
+ * If previous directory exists:
+ * <ol>
+ * <li>Rename <SD>/current/<bpid>/current to removed.tmp</li>
+ * <li>Rename * <SD>/current/<bpid>/previous to current</li>
+ * <li>Remove removed.tmp</li>
+ * </ol>
+ *
+ * Do nothing if previous directory does not exist.
+ * @param bpSd Block pool storage directory at <SD>/current/<bpid>
+ */
+ void doRollback(StorageDirectory bpSd, NamespaceInfo nsInfo)
+ throws IOException {
+ File prevDir = bpSd.getPreviousDir();
+ // regular startup if previous dir does not exist
+ if (!prevDir.exists())
+ return;
+ // read attributes out of the VERSION file of previous directory
+ DataStorage prevInfo = new DataStorage();
+ StorageDirectory prevSD = prevInfo.new StorageDirectory(bpSd.getRoot());
+ prevSD.read(prevSD.getPreviousVersionFile());
+
+ // We allow rollback to a state, which is either consistent with
+ // the namespace state or can be further upgraded to it.
+ // In another word, we can only roll back when ( storedLV >= software LV)
+ // && ( DN.previousCTime <= NN.ctime)
+ if (!(prevInfo.getLayoutVersion() >= FSConstants.LAYOUT_VERSION &&
+ prevInfo.getCTime() <= nsInfo.getCTime())) { // cannot rollback
+ throw new InconsistentFSStateException(prevSD.getRoot(),
+ "Cannot rollback to a newer state.\nDatanode previous state: LV = "
+ + prevInfo.getLayoutVersion() + " CTime = " + prevInfo.getCTime()
+ + " is newer than the namespace state: LV = "
+ + nsInfo.getLayoutVersion() + " CTime = " + nsInfo.getCTime());
+ }
+
+ LOG.info("Rolling back storage directory " + bpSd.getRoot()
+ + ".\n target LV = " + nsInfo.getLayoutVersion()
+ + "; target CTime = " + nsInfo.getCTime());
+ File tmpDir = bpSd.getRemovedTmp();
+ assert !tmpDir.exists() : "removed.tmp directory must not exist.";
+ // 1. rename current to tmp
+ File curDir = bpSd.getCurrentDir();
+ assert curDir.exists() : "Current directory must exist.";
+ rename(curDir, tmpDir);
+
+ // 2. rename previous to current
+ rename(prevDir, curDir);
+
+ // 3. delete removed.tmp dir
+ deleteDir(tmpDir);
+ LOG.info("Rollback of " + bpSd.getRoot() + " is complete.");
+ }
+
+ /*
+ * Finalize the block pool storage by deleting <BP>/previous directory
+ * that holds the snapshot.
+ */
+ void doFinalize(File dnCurDir) throws IOException {
+ File bpRoot = getBpRoot(blockpoolID, dnCurDir);
+ StorageDirectory bpSd = new StorageDirectory(bpRoot);
+ // block pool level previous directory
+ File prevDir = bpSd.getPreviousDir();
+ if (!prevDir.exists()) {
+ return; // already finalized
+ }
+ final String dataDirPath = bpSd.getRoot().getCanonicalPath();
+ LOG.info("Finalizing upgrade for storage directory " + dataDirPath
+ + ".\n cur LV = " + this.getLayoutVersion() + "; cur CTime = "
+ + this.getCTime());
+ assert bpSd.getCurrentDir().exists() : "Current directory must exist.";
+
+ // rename previous to finalized.tmp
+ final File tmpDir = bpSd.getFinalizedTmp();
+ rename(prevDir, tmpDir);
+
+ // delete finalized.tmp dir in a separate thread
+ new Daemon(new Runnable() {
+ public void run() {
+ try {
+ deleteDir(tmpDir);
+ } catch (IOException ex) {
+ LOG.error("Finalize upgrade for " + dataDirPath + " failed.", ex);
+ }
+ LOG.info("Finalize upgrade for " + dataDirPath + " is complete.");
+ }
+
+ public String toString() {
+ return "Finalize " + dataDirPath;
+ }
+ }).start();
+ }
+
+ /**
+ * Hardlink all finalized and RBW blocks in fromDir to toDir
+ *
+ * @param fromDir directory where the snapshot is stored
+ * @param toDir the current data directory
+ * @throws IOException if error occurs during hardlink
+ */
+ private void linkAllBlocks(File fromDir, File toDir) throws IOException {
+ // do the link
+ int diskLayoutVersion = this.getLayoutVersion();
+ // hardlink finalized blocks in tmpDir
+ DataStorage.linkBlocks(fromDir, new File(toDir,
+ DataStorage.STORAGE_DIR_FINALIZED), diskLayoutVersion);
+ }
+
+ protected void corruptPreUpgradeStorage(File rootDir) throws IOException {
+ File oldF = new File(rootDir, "storage");
+ if (oldF.exists())
+ return;
+ // recreate old storage file to let pre-upgrade versions fail
+ if (!oldF.createNewFile())
+ throw new IOException("Cannot create file " + oldF);
+ RandomAccessFile oldFile = new RandomAccessFile(oldF, "rws");
+ // write new version into old storage file
+ try {
+ writeCorruptedData(oldFile);
+ } finally {
+ oldFile.close();
+ }
+ }
+
+ private void verifyDistributedUpgradeProgress(NamespaceInfo nsInfo)
+ throws IOException {
+ UpgradeManagerDatanode um = DataNode.getDataNode().upgradeManager;
+ assert um != null : "DataNode.upgradeManager is null.";
+ um.setUpgradeState(false, getLayoutVersion());
+ um.initializeUpgrade(nsInfo);
+ }
+
+ /**
+ * gets the data node storage directory based on block pool storage
+ *
+ * @param bpRoot
+ * @return
+ */
+ private static String getDataNodeStorageRoot(String bpRoot) {
+ Matcher matcher = BLOCK_POOL_PATH_PATTERN.matcher(bpRoot);
+ if (matcher.matches()) {
+ // return the data node root directory
+ return matcher.group(1);
+ }
+ return bpRoot;
+ }
+
+ @Override
+ public boolean isConversionNeeded(StorageDirectory sd) throws IOException {
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ return super.toString() + ";bpid=" + blockpoolID;
+ }
+
+ /**
+ * Get a block pool storage root based on data node storage root
+ * @param bpID block pool ID
+ * @param dnRoot data node storage root directory
+ * @return root directory for block pool storage
+ */
+ static File getBpRoot(String bpID, File dnCurDir) {
+ return new File(dnCurDir, bpID);
+ }
+}
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=1076075&r1=1076074&r2=1076075&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 01:21:35 2011
@@ -37,7 +37,7 @@ import org.apache.commons.logging.LogFac
/**
* DataBlockScanner manages block scanning for all the block pools. For each
- * block pool a {@link BlockPoolScanner} is created which runs in a separate
+ * block pool a {@link BlockPoolSliceScanner} is created which runs in a separate
* thread to scan the blocks for that block pool. When a {@link BPOfferService}
* becomes alive or dies, blockPoolScannerMap in this class is updated.
*/
@@ -52,8 +52,8 @@ public class DataBlockScanner implements
* 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>();
+ private final TreeMap<String, BlockPoolSliceScanner> blockPoolScannerMap =
+ new TreeMap<String, BlockPoolSliceScanner>();
Thread blockScannerThread = null;
DataBlockScanner(DataNode datanode, FSDataset dataset, Configuration conf) {
@@ -65,7 +65,7 @@ public class DataBlockScanner implements
public void run() {
String currentBpId = "";
while (datanode.shouldRun && !Thread.interrupted()) {
- BlockPoolScanner bpScanner = getNextBPScanner(currentBpId);
+ BlockPoolSliceScanner bpScanner = getNextBPScanner(currentBpId);
if (bpScanner == null) {
// Possible if thread is interrupted
continue;
@@ -78,7 +78,7 @@ public class DataBlockScanner implements
removeBlockPool(currentBpId);
continue;
}
- bpScanner.scanBlockPool();
+ bpScanner.scanBlockPoolSlice();
try {
Thread.sleep(5000);
} catch (InterruptedException ex) {
@@ -110,7 +110,7 @@ public class DataBlockScanner implements
* 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) {
+ private BlockPoolSliceScanner getNextBPScanner(String currentBpId) {
String nextBpId = null;
while ((nextBpId == null) && datanode.shouldRun
@@ -126,7 +126,7 @@ public class DataBlockScanner implements
String bpid = bpidIterator.next();
for (FSDataset.FSVolume vol : dataset.volumes.volumes) {
try {
- File currFile = BlockPoolScanner.getCurrentFile(vol, bpid);
+ File currFile = BlockPoolSliceScanner.getCurrentFile(vol, bpid);
if (currFile.exists()) {
long lastModified = currFile.lastModified();
if (lastScanTime < lastModified) {
@@ -173,7 +173,7 @@ public class DataBlockScanner implements
return blockPoolScannerMap.size();
}
- private synchronized BlockPoolScanner getBPScanner(String bpid) {
+ private synchronized BlockPoolSliceScanner getBPScanner(String bpid) {
return blockPoolScannerMap.get(bpid);
}
@@ -183,7 +183,7 @@ public class DataBlockScanner implements
}
public void addBlock(ExtendedBlock block) {
- BlockPoolScanner bpScanner = getBPScanner(block.getBlockPoolId());
+ BlockPoolSliceScanner bpScanner = getBPScanner(block.getBlockPoolId());
if (bpScanner != null) {
bpScanner.addBlock(block);
} else {
@@ -193,7 +193,7 @@ public class DataBlockScanner implements
}
public synchronized boolean isInitialized(String bpid) {
- BlockPoolScanner bpScanner = getBPScanner(bpid);
+ BlockPoolSliceScanner bpScanner = getBPScanner(bpid);
if (bpScanner != null) {
return bpScanner.isInitialized();
}
@@ -209,7 +209,7 @@ public class DataBlockScanner implements
return;
}
for (String bpid : bpIdList) {
- BlockPoolScanner bpScanner = getBPScanner(bpid);
+ BlockPoolSliceScanner bpScanner = getBPScanner(bpid);
buffer.append("\n\nBlock report for block pool: "+bpid + "\n");
bpScanner.printBlockReport(buffer, summary);
buffer.append("\n");
@@ -217,7 +217,7 @@ public class DataBlockScanner implements
}
public void deleteBlock(String poolId, Block toDelete) {
- BlockPoolScanner bpScanner = getBPScanner(poolId);
+ BlockPoolSliceScanner bpScanner = getBPScanner(poolId);
if (bpScanner != null) {
bpScanner.deleteBlock(toDelete);
} else {
@@ -227,7 +227,7 @@ public class DataBlockScanner implements
}
public void deleteBlocks(String poolId, Block[] toDelete) {
- BlockPoolScanner bpScanner = getBPScanner(poolId);
+ BlockPoolSliceScanner bpScanner = getBPScanner(poolId);
if (bpScanner != null) {
bpScanner.deleteBlocks(toDelete);
} else {
@@ -246,7 +246,7 @@ public class DataBlockScanner implements
if (blockPoolScannerMap.get(blockPoolId) != null) {
return;
}
- BlockPoolScanner bpScanner = new BlockPoolScanner(datanode, dataset,
+ BlockPoolSliceScanner bpScanner = new BlockPoolSliceScanner(datanode, dataset,
conf, blockPoolId);
try {
bpScanner.init();
@@ -266,7 +266,7 @@ public class DataBlockScanner implements
// This method is used for testing
long getBlocksScannedInLastRun(String bpid) throws IOException {
- BlockPoolScanner bpScanner = getBPScanner(bpid);
+ BlockPoolSliceScanner bpScanner = getBPScanner(bpid);
if (bpScanner == null) {
throw new IOException("Block Pool: "+bpid+" is not running");
} else {
Modified: hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataStorage.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataStorage.java?rev=1076075&r1=1076074&r2=1076075&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataStorage.java (original)
+++ hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DataStorage.java Wed Mar 2 01:21:35 2011
@@ -78,8 +78,8 @@ public class DataStorage extends Storage
private boolean initilized = false;
// BlockPoolStorage is map of <Block pool Id, BlockPoolStorage>
- private Map<String, BlockPoolStorage> bpStorageMap
- = new HashMap<String, BlockPoolStorage>();
+ private Map<String, BlockPoolSliceStorage> bpStorageMap
+ = new HashMap<String, BlockPoolSliceStorage>();
DataStorage() {
@@ -214,13 +214,13 @@ public class DataStorage extends Storage
Collection<File> bpDataDirs = new ArrayList<File>();
for(Iterator<File> it = dataDirs.iterator(); it.hasNext();) {
File dnRoot = it.next();
- File bpRoot = BlockPoolStorage.getBpRoot(bpID, new File(dnRoot,
+ File bpRoot = BlockPoolSliceStorage.getBpRoot(bpID, new File(dnRoot,
STORAGE_DIR_CURRENT));
bpDataDirs.add(bpRoot);
}
// mkdir for the list of BlockPoolStorage
makeBlockPoolDataDir(bpDataDirs, null);
- BlockPoolStorage bpStorage = new BlockPoolStorage(nsInfo.getNamespaceID(),
+ BlockPoolSliceStorage bpStorage = new BlockPoolSliceStorage(nsInfo.getNamespaceID(),
bpID, nsInfo.getCTime(), nsInfo.getClusterID());
bpStorage.recoverTransitionRead(nsInfo, bpDataDirs, startOpt);
@@ -439,8 +439,8 @@ public class DataStorage extends Storage
rename(curDir, tmpDir);
// 3. Format BP and hard link blocks from previous directory
- File curBpDir = BlockPoolStorage.getBpRoot(nsInfo.getBlockPoolID(), curDir);
- BlockPoolStorage bpStorage = new BlockPoolStorage(nsInfo.getNamespaceID(),
+ File curBpDir = BlockPoolSliceStorage.getBpRoot(nsInfo.getBlockPoolID(), curDir);
+ BlockPoolSliceStorage bpStorage = new BlockPoolSliceStorage(nsInfo.getNamespaceID(),
nsInfo.getBlockPoolID(), nsInfo.getCTime(), nsInfo.getClusterID());
bpStorage.format(curDir, nsInfo);
linkAllBlocks(tmpDir, new File(curBpDir, STORAGE_DIR_CURRENT));
@@ -583,7 +583,7 @@ public class DataStorage extends Storage
doFinalize(sd);
} else {
// block pool storage finalize using specific bpID
- BlockPoolStorage bpStorage = bpStorageMap.get(bpID);
+ BlockPoolSliceStorage bpStorage = bpStorageMap.get(bpID);
bpStorage.doFinalize(sd.getCurrentDir());
}
}
@@ -702,7 +702,7 @@ public class DataStorage extends Storage
/**
* Add bpStorage into bpStorageMap
*/
- private void addBlockPoolStorage(String bpID, BlockPoolStorage bpStorage)
+ private void addBlockPoolStorage(String bpID, BlockPoolSliceStorage bpStorage)
throws IOException {
if (!this.bpStorageMap.containsKey(bpID)) {
this.bpStorageMap.put(bpID, bpStorage);
Modified: hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java?rev=1076075&r1=1076074&r2=1076075&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java (original)
+++ hadoop/hdfs/branches/HDFS-1052/src/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java Wed Mar 2 01:21:35 2011
@@ -466,7 +466,7 @@ public class DirectoryScanner implements
ScanInfoPerBlockPool result = new ScanInfoPerBlockPool(bpList.length);
for (String bpid : bpList) {
LinkedList<ScanInfo> report = new LinkedList<ScanInfo>();
- File bpFinalizedDir = volume.getBlockPool(bpid).getFinalizedDir();
+ File bpFinalizedDir = volume.getBlockPoolSlice(bpid).getFinalizedDir();
result.put(bpid, compileReport(volume, bpFinalizedDir, report));
}
return result;
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=1076075&r1=1076074&r2=1076075&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 01:21:35 2011
@@ -301,15 +301,11 @@ public class FSDataset implements FSCons
}
/**
- * Manages block pool subdirectory under {@link FSVolume}
- *
- * TODO:FEDERATION This class should be renamed, perhaps to "BlockPoolSlice".
- * The real "block pool" is an abstraction that spans Volumes and
- * Datanodes. This "BlockPool" object doesn't represent that. Rather,
- * it only represents the concrete portion of a particular block pool
- * that resides on a particular Volume.
+ * A BlockPoolSlice represents a portion of a BlockPool stored on a volume.
+ * Taken together, all BlockPoolSlices sharing a block pool ID across a
+ * cluster represent a single block pool.
*/
- class BlockPool {
+ class BlockPoolSlice {
private final String bpid;
private final FSVolume volume; // volume to which this BlockPool belongs to
private final File currentDir; // StorageDirectory/current/bpid/current
@@ -328,7 +324,7 @@ public class FSDataset implements FSCons
* @param conf
* @throws IOException
*/
- BlockPool(String bpid, FSVolume volume, File bpDir, Configuration conf)
+ BlockPoolSlice(String bpid, FSVolume volume, File bpDir, Configuration conf)
throws IOException {
this.bpid = bpid;
this.volume = volume;
@@ -547,7 +543,7 @@ public class FSDataset implements FSCons
}
class FSVolume {
- private final Map<String, BlockPool> map = new HashMap<String, BlockPool>();
+ private final Map<String, BlockPoolSlice> map = new HashMap<String, BlockPoolSlice>();
private final File currentDir; // <StorageDirectory>/current
private final DF usage;
private final long reserved;
@@ -569,7 +565,7 @@ public class FSDataset implements FSCons
}
public File getRbwDir(String bpid) throws IOException {
- BlockPool bp = getBlockPool(bpid);
+ BlockPoolSlice bp = getBlockPoolSlice(bpid);
return bp.getRbwDir();
}
@@ -577,7 +573,7 @@ public class FSDataset implements FSCons
// The caller to this method (BlockFileDeleteTask.run()) does
// not have locked FSDataset.this yet.
synchronized(FSDataset.this) {
- BlockPool bp = map.get(bpid);
+ BlockPoolSlice bp = map.get(bpid);
if (bp != null) {
bp.decDfsUsed(value);
}
@@ -587,15 +583,15 @@ public class FSDataset implements FSCons
long getDfsUsed() throws IOException {
// TODO valid synchronization
long dfsUsed = 0;
- Set<Entry<String, BlockPool>> set = map.entrySet();
- for (Entry<String, BlockPool> entry : set) {
+ Set<Entry<String, BlockPoolSlice>> set = map.entrySet();
+ for (Entry<String, BlockPoolSlice> entry : set) {
dfsUsed += entry.getValue().getDfsUsed();
}
return dfsUsed;
}
long getBlockPoolUsed(String bpid) throws IOException {
- return getBlockPool(bpid).getDfsUsed();
+ return getBlockPoolSlice(bpid).getDfsUsed();
}
/**
@@ -625,8 +621,8 @@ public class FSDataset implements FSCons
return usage.getMount();
}
- BlockPool getBlockPool(String bpid) throws IOException {
- BlockPool bp = map.get(bpid);
+ BlockPoolSlice getBlockPoolSlice(String bpid) throws IOException {
+ BlockPoolSlice bp = map.get(bpid);
if (bp == null) {
// TODO:FEDERATION cleanup this exception
throw new IOException("block pool " + bpid + " not found");
@@ -648,7 +644,7 @@ public class FSDataset implements FSCons
* the block is finalized.
*/
File createTmpFile(String bpid, Block b) throws IOException {
- BlockPool bp = getBlockPool(bpid);
+ BlockPoolSlice bp = getBlockPoolSlice(bpid);
return bp.createTmpFile(b);
}
@@ -657,32 +653,32 @@ public class FSDataset implements FSCons
* the block is finalized.
*/
File createRbwFile(String bpid, Block b) throws IOException {
- BlockPool bp = getBlockPool(bpid);
+ BlockPoolSlice bp = getBlockPoolSlice(bpid);
return bp.createRbwFile(b);
}
File addBlock(String bpid, Block b, File f) throws IOException {
- BlockPool bp = getBlockPool(bpid);
+ BlockPoolSlice bp = getBlockPoolSlice(bpid);
return bp.addBlock(b, f);
}
void checkDirs() throws DiskErrorException {
// TODO:FEDERATION valid synchronization
- Set<Entry<String, BlockPool>> set = map.entrySet();
- for (Entry<String, BlockPool> entry : set) {
+ Set<Entry<String, BlockPoolSlice>> set = map.entrySet();
+ for (Entry<String, BlockPoolSlice> entry : set) {
entry.getValue().checkDirs();
}
}
void getVolumeMap(ReplicasMap volumeMap) throws IOException {
- Set<Entry<String, BlockPool>> set = map.entrySet();
- for (Entry<String, BlockPool> entry : set) {
+ Set<Entry<String, BlockPoolSlice>> set = map.entrySet();
+ for (Entry<String, BlockPoolSlice> entry : set) {
entry.getValue().getVolumeMap(volumeMap);
}
}
void getVolumeMap(String bpid, ReplicasMap volumeMap) throws IOException {
- BlockPool bp = getBlockPool(bpid);
+ BlockPoolSlice bp = getBlockPoolSlice(bpid);
bp.getVolumeMap(volumeMap);
}
@@ -696,14 +692,14 @@ public class FSDataset implements FSCons
*/
private void addToReplicasMap(String bpid, ReplicasMap volumeMap,
File dir, boolean isFinalized) throws IOException {
- BlockPool bp = getBlockPool(bpid);
+ BlockPoolSlice bp = getBlockPoolSlice(bpid);
// TODO move this up
// dfsUsage.incDfsUsed(b.getNumBytes()+metaFile.length());
bp.addToReplicasMap(volumeMap, dir, isFinalized);
}
void clearPath(String bpid, File f) throws IOException {
- BlockPool bp = getBlockPool(bpid);
+ BlockPoolSlice bp = getBlockPoolSlice(bpid);
bp.clearPath(f);
}
@@ -712,8 +708,8 @@ public class FSDataset implements FSCons
}
public void shutdown() {
- Set<Entry<String, BlockPool>> set = map.entrySet();
- for (Entry<String, BlockPool> entry : set) {
+ Set<Entry<String, BlockPoolSlice>> set = map.entrySet();
+ for (Entry<String, BlockPoolSlice> entry : set) {
entry.getValue().shutdown();
}
}
@@ -721,7 +717,7 @@ public class FSDataset implements FSCons
public void addBlockPool(String bpid, Configuration conf)
throws IOException {
File bpdir = new File(currentDir, bpid);
- BlockPool bp = new BlockPool(bpid, this, bpdir, conf);
+ BlockPoolSlice bp = new BlockPoolSlice(bpid, this, bpdir, conf);
map.put(bpid, bp);
}
}
Modified: hadoop/hdfs/branches/HDFS-1052/src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1052/src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java?rev=1076075&r1=1076074&r2=1076075&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1052/src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java (original)
+++ hadoop/hdfs/branches/HDFS-1052/src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java Wed Mar 2 01:21:35 2011
@@ -144,7 +144,7 @@ public class TestDirectoryScanner extend
FSVolume[] volumes = fds.volumes.volumes;
int index = rand.nextInt(volumes.length - 1);
long id = getFreeBlockId();
- File finalizedDir = volumes[index].getBlockPool(bpid).getFinalizedDir();
+ File finalizedDir = volumes[index].getBlockPoolSlice(bpid).getFinalizedDir();
File file = new File(finalizedDir, getBlockFile(id));
if (file.createNewFile()) {
LOG.info("Created block file " + file.getName());
@@ -157,7 +157,7 @@ public class TestDirectoryScanner extend
FSVolume[] volumes = fds.volumes.volumes;
int index = rand.nextInt(volumes.length - 1);
long id = getFreeBlockId();
- File finalizedDir = volumes[index].getBlockPool(bpid).getFinalizedDir();
+ File finalizedDir = volumes[index].getBlockPoolSlice(bpid).getFinalizedDir();
File file = new File(finalizedDir, getMetaFile(id));
if (file.createNewFile()) {
LOG.info("Created metafile " + file.getName());
@@ -170,7 +170,7 @@ public class TestDirectoryScanner extend
FSVolume[] volumes = fds.volumes.volumes;
int index = rand.nextInt(volumes.length - 1);
long id = getFreeBlockId();
- File finalizedDir = volumes[index].getBlockPool(bpid).getFinalizedDir();
+ File finalizedDir = volumes[index].getBlockPoolSlice(bpid).getFinalizedDir();
File file = new File(finalizedDir, getBlockFile(id));
if (file.createNewFile()) {
LOG.info("Created block file " + file.getName());