fsdataset.java

hadoop:Nutch集群平台

/**
 * Copyright 2005 The Apache Software Foundation
 *
 * Licensed 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.dfs;

import java.io.*;
import java.util.*;

import org.apache.hadoop.fs.*;
import org.apache.hadoop.util.DiskChecker;
import org.apache.hadoop.util.DiskChecker.DiskErrorException;
import org.apache.hadoop.util.DiskChecker.DiskOutOfSpaceException;
import org.apache.hadoop.conf.*;

/**************************************************
 * FSDataset manages a set of data blocks.  Each block
 * has a unique name and an extent on disk.
 *
 * @author Mike Cafarella
 ***************************************************/
class FSDataset implements FSConstants {

	
  /**
     * A node type that can be built into a tree reflecting the
     * hierarchy of blocks on the local disk.
     */
    class FSDir {
        File dir;
        int numBlocks = 0;
        int myIdx = 0;
        FSDir children[];
        FSDir siblings[];

        /**
         */
        public FSDir(File dir, int myIdx, FSDir[] siblings) {
            this.dir = dir;
            this.myIdx = myIdx;
            this.siblings = siblings;
            this.children = null;
            if (! dir.exists()) {
              dir.mkdirs();
            } else {
              File[] files = dir.listFiles();
              int numChildren = 0;
              for (int idx = 0; idx < files.length; idx++) {
                if (files[idx].isDirectory()) {
                  numChildren++;
                } else if (Block.isBlockFilename(files[idx])) {
                  numBlocks++;
                }
              }
              if (numChildren > 0) {
                children = new FSDir[numChildren];
                int curdir = 0;
                for (int idx = 0; idx < files.length; idx++) {
                  if (files[idx].isDirectory()) {
                    children[curdir] = new FSDir(files[idx], curdir, children);
                    curdir++;
                  }
                }
              }
            }
        }

        /**
         */
        public File addBlock(Block b, File src) {
            if (numBlocks < maxBlocksPerDir) {
              File dest = new File(dir, b.getBlockName());
              src.renameTo(dest);
              numBlocks += 1;
              return dest;
            } else {
              if (siblings != null && myIdx != (siblings.length-1)) {
                File dest = siblings[myIdx+1].addBlock(b, src);
                if (dest != null) { return dest; }
              }
              if (children == null) {
                children = new FSDir[maxBlocksPerDir];
                for (int idx = 0; idx < maxBlocksPerDir; idx++) {
                  children[idx] = new FSDir(
                      new File(dir, "subdir"+idx), idx, children);
                }
              }
              return children[0].addBlock(b, src);
            }
        }

        /**
         * Populate the given blockSet with any child blocks
         * found at this node.
         */
        public void getBlockInfo(TreeSet<Block> blockSet) {
            if (children != null) {
                for (int i = 0; i < children.length; i++) {
                    children[i].getBlockInfo(blockSet);
                }
            }

            File blockFiles[] = dir.listFiles();
            for (int i = 0; i < blockFiles.length; i++) {
                if (Block.isBlockFilename(blockFiles[i])) {
                    blockSet.add(new Block(blockFiles[i], blockFiles[i].length()));
                }
            }
        }


        void getVolumeMap(HashMap<Block, FSVolume> volumeMap, FSVolume volume) {
          if (children != null) {
                for (int i = 0; i < children.length; i++) {
                    children[i].getVolumeMap(volumeMap, volume);
                }
            }

            File blockFiles[] = dir.listFiles();
            for (int i = 0; i < blockFiles.length; i++) {
                if (Block.isBlockFilename(blockFiles[i])) {
                    volumeMap.put(new Block(blockFiles[i], blockFiles[i].length()), volume);
                }
            }
        }
        
        void getBlockMap(HashMap<Block, File> blockMap) {
          if (children != null) {
                for (int i = 0; i < children.length; i++) {
                    children[i].getBlockMap(blockMap);
                }
            }

            File blockFiles[] = dir.listFiles();
            for (int i = 0; i < blockFiles.length; i++) {
                if (Block.isBlockFilename(blockFiles[i])) {
                    blockMap.put(new Block(blockFiles[i], blockFiles[i].length()), blockFiles[i]);
                }
            }
        }
        /**
         * check if a data diretory is healthy
         * @throws DiskErrorException
         * @author hairong
         */
        public void checkDirTree() throws DiskErrorException {
            DiskChecker.checkDir(dir);
            
            if (children != null) {
                for (int i = 0; i < children.length; i++) {
                    children[i].checkDirTree();
                }
            }
        }
        
        public String toString() {
          return "FSDir{" +
              "dir=" + dir +
              ", children=" + (children == null ? null : Arrays.asList(children)) +
              "}";
        }
    }

    class FSVolume {
      static final double USABLE_DISK_PCT_DEFAULT = 0.98f; 

      private File dir;
      private FSDir dataDir;
      private File tmpDir;
      private DF usage;
      private long reserved;
      private double usableDiskPct = USABLE_DISK_PCT_DEFAULT;
    
      FSVolume(File dir, Configuration conf) throws IOException {
        this.reserved = conf.getLong("dfs.datanode.du.reserved", 0);
        this.usableDiskPct = conf.getFloat("dfs.datanode.du.pct",
            (float) USABLE_DISK_PCT_DEFAULT);
        this.dir = dir;
        this.dataDir = new FSDir(new File(dir, "data"), 0, null);
        this.tmpDir = new File(dir, "tmp");
        if (tmpDir.exists()) {
          FileUtil.fullyDelete(tmpDir);
        }
        tmpDir.mkdirs();
        this.usage = new DF(dir, conf);
      }
      
      long getCapacity() throws IOException {
        return usage.getCapacity();
      }
      
      long getAvailable() throws IOException {
        return ((long) Math.round(usableDiskPct *
                usage.getAvailable()) - reserved);
      }
      
      String getMount() throws IOException {
        return usage.getMount();
      }
      
      File createTmpFile(Block b) throws IOException {
        File f = new File(tmpDir, b.getBlockName());
        try {
          if (f.exists()) {
            throw new IOException("Unexpected problem in creating temporary file for "+
                b + ".  File " + f + " should not be present, but is.");
          }
          // Create the zero-length temp file
          //
          if (!f.createNewFile()) {
            throw new IOException("Unexpected problem in creating temporary file for "+
                b + ".  File " + f + " should be creatable, but is already present.");
          }
        } catch (IOException ie) {
          System.out.println("Exception!  " + ie);
          throw ie;
        }
        reserved -= b.getNumBytes();
        return f;
      }
      
      File addBlock(Block b, File f) {
        return dataDir.addBlock(b, f);
      }
      
      void checkDirs() throws DiskErrorException {
        dataDir.checkDirTree();
        DiskChecker.checkDir(tmpDir);
      }
      
      void getBlockInfo(TreeSet<Block> blockSet) {
        dataDir.getBlockInfo(blockSet);
      }
      
      void getVolumeMap(HashMap<Block, FSVolume> volumeMap) {
        dataDir.getVolumeMap(volumeMap, this);
      }
      
      void getBlockMap(HashMap<Block, File> blockMap) {
        dataDir.getBlockMap(blockMap);
      }
      
      public String toString() {
        return dir.getAbsolutePath();
      }
    }
    
    class FSVolumeSet {
      FSVolume[] volumes = null;
      int curVolume = 0;
      HashMap<String,Long> mountMap = new HashMap<String,Long>();
      
      FSVolumeSet(FSVolume[] volumes) {
        this.volumes = volumes;
      }