fsdirectory.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 org.apache.hadoop.io.*;

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

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.metrics.MetricsRecord;
import org.apache.hadoop.metrics.Metrics;

/*************************************************
 * FSDirectory stores the filesystem directory state.
 * It handles writing/loading values to disk, and logging
 * changes as we go.
 *
 * It keeps the filename->blockset mapping always-current
 * and logged to disk.
 * 
 * @author Mike Cafarella
 *************************************************/
class FSDirectory implements FSConstants {

    /******************************************************
     * We keep an in-memory representation of the file/block
     * hierarchy.
     * 
     * TODO: Factor out INode to a standalone class.
     ******************************************************/
    class INode {
        private String name;
        private INode parent;
        private TreeMap children = new TreeMap();
        private Block blocks[];
        private short blockReplication;

        /**
         */
        INode(String name, Block blocks[], short replication) {
            this.name = name;
            this.parent = null;
            this.blocks = blocks;
            this.blockReplication = replication;
        }

        /**
         */
        INode(String name) {
            this.name = name;
            this.parent = null;
            this.blocks = null;
            this.blockReplication = 0;
        }

        /**
         * Check whether it's a directory
         */
        synchronized public boolean isDir() {
          return (blocks == null);
        }
        
        /**
         * Get block replication for the file 
         * @return block replication
         */
        public short getReplication() {
          return this.blockReplication;
        }
        
        /**
         * Get local file name
         * @return local file name
         */
        String getLocalName() {
          return name;
        }

        /**
         * Get file blocks 
         * @return file blocks
         */
        Block[] getBlocks() {
          return this.blocks;
        }
        
        /**
         * Get parent directory 
         * @return parent INode
         */
        INode getParent() {
          return this.parent;
        }

        /**
         * Get children 
         * @return TreeMap of children
         */
        TreeMap getChildren() {
          return this.children;
        }

        /**
         * This is the external interface
         */
        INode getNode(String target) {
            if ( target == null || 
                ! target.startsWith("/") || target.length() == 0) {
                return null;
            } else if (parent == null && "/".equals(target)) {
                return this;
            } else {
                Vector components = new Vector();
                int start = 0;
                int slashid = 0;
                while (start < target.length() && (slashid = target.indexOf('/', start)) >= 0) {
                    components.add(target.substring(start, slashid));
                    start = slashid + 1;
                }
                if (start < target.length()) {
                    components.add(target.substring(start));
                }
                return getNode(components, 0);
            }
        }

        /**
         */
        INode getNode(Vector components, int index) {
            if (! name.equals((String) components.elementAt(index))) {
                return null;
            }
            if (index == components.size()-1) {
                return this;
            }

            // Check with children
            INode child = this.getChild((String)components.elementAt(index+1));
            if (child == null) {
                return null;
            } else {
                return child.getNode(components, index+1);
            }
        }
        
        INode getChild( String name) {
          return (INode) children.get( name );
        }

        /**
         * Add new INode to the file tree.
         * Find the parent and insert 
         * 
         * @param path file path
         * @param newNode INode to be added
         * @return null if the node already exists; inserted INode, otherwise
         * @throws FileNotFoundException 
         * @author shv
         */
        INode addNode(String path, INode newNode) throws FileNotFoundException {
          File target = new File( path );
          // find parent
          Path parent = new Path(path).getParent();
          if (parent == null) { // add root
              return null;
          }
          INode parentNode = getNode(parent.toString());
          if (parentNode == null) {
              throw new FileNotFoundException(
                      "Parent path does not exist: "+path);
          }
          if (!parentNode.isDir()) {
        	  throw new FileNotFoundException(
        			  "Parent path is not a directory: "+path);
          }
           // check whether the parent already has a node with that name
          String name = newNode.name = target.getName();
          if( parentNode.getChild( name ) != null )
            return null;
          // insert into the parent children list
          parentNode.children.put(name, newNode);
          newNode.parent = parentNode;
          return newNode;
        }

        /**
         */
        boolean removeNode() {
            if (parent == null) {
                return false;
            } else {
                parent.children.remove(name);
                return true;
            }
        }
          
        /**
         * Collect all the blocks at this INode and all its children.
         * This operation is performed after a node is removed from the tree,
         * and we want to GC all the blocks at this node and below.
         */
        void collectSubtreeBlocks(Vector v) {
            if (blocks != null) {
                for (int i = 0; i < blocks.length; i++) {
                    v.add(blocks[i]);
                }
            }
            Metrics.report(metricsRecord, "files-deleted", ++numFilesDeleted);
            for (Iterator it = children.values().iterator(); it.hasNext(); ) {
                INode child = (INode) it.next();
                child.collectSubtreeBlocks(v);
            }
        }

        /**
         */
        int numItemsInTree() {
            int total = 0;
            for (Iterator it = children.values().iterator(); it.hasNext(); ) {
                INode child = (INode) it.next();
                total += child.numItemsInTree();
            }
            return total + 1;
        }

        /**
         */
        String computeName() {
            if (parent != null) {
                return parent.computeName() + "/" + name;
            } else {
                return name;
            }
        }

        /**
         */
        long computeFileLength() {
            long total = 0;
            if (blocks != null) {
                for (int i = 0; i < blocks.length; i++) {
                    total += blocks[i].getNumBytes();
                }
            }
            return total;
        }

        /**
         */
        long computeContentsLength() {
            long total = computeFileLength();
            for (Iterator it = children.values().iterator(); it.hasNext(); ) {
                INode child = (INode) it.next();
                total += child.computeContentsLength();
            }
            return total;
        }

        /**
         * Get the block size of the first block
         * @return the number of bytes
         */
        public long getBlockSize() {
          if (blocks == null || blocks.length == 0) {
            return 0;
          } else {
            return blocks[0].getNumBytes();
          }
        }
        
        /**
         */
        void listContents(Vector v) {
            if (parent != null && blocks != null) {
                v.add(this);
            }

            for (Iterator it = children.values().iterator(); it.hasNext(); ) {
                INode child = (INode) it.next();
                v.add(child);
            }
        }
    }

    
    INode rootDir = new INode("");
    Map activeBlocks = new HashMap();
    TreeMap activeLocks = new TreeMap();
    FSImage fsImage;  
    boolean ready = false;
    int namespaceID = 0;    // TODO: move to FSImage class, it belongs there
    // Metrics members
    private MetricsRecord metricsRecord = null;
    private int numFilesDeleted = 0;
    
    /** Access an existing dfs name directory. */
    public FSDirectory(File dir) throws IOException {
      this.fsImage = new FSImage( dir );
    }
    
    void loadFSImage( Configuration conf ) throws IOException {
      fsImage.loadFSImage( conf );
      synchronized (this) {
        this.ready = true;
        this.notifyAll();
        fsImage.getEditLog().create();
      }
      metricsRecord = Metrics.createRecord("dfs", "namenode");
    }

    /** Create a new dfs name directory.  Caution: this destroys all files
     * in this filesystem.
     * @deprecated use @link FSImage#format(File, Configuration) instead */
    public static void format(File dir, Configuration conf) throws IOException {
      FSImage.format( dir, conf );
    }
    
    /**
     * Shutdown the filestore
     */
    public void close() throws IOException {
        fsImage.getEditLog().close();
    }

    /**
     * Block until the object is ready to be used.
     */
    void waitForReady() {
        if (! ready) {
            synchronized (this) {
                while (!ready) {
                    try {
                        this.wait(5000);
                    } catch (InterruptedException ie) {
                    }
                }
            }
        }
    }

    /**
     * Add the given filename to the fs.
     */
    public boolean addFile(UTF8 path, Block[] blocks, short replication) {
        waitForReady();

        // Always do an implicit mkdirs for parent directory tree
        String pathString = path.toString();
        mkdirs(new Path(pathString).getParent().toString());
        INode newNode = new INode( new File(pathString).getName(), blocks, replication);
        if( ! unprotectedAddFile(path, newNode) ) {
           NameNode.stateChangeLog.info("DIR* FSDirectory.addFile: "
                    +"failed to add "+path+" with "
                    +blocks.length+" blocks to the file system" );
           return false;
        }
        // add create file record to log
        fsImage.getEditLog().logCreateFile( newNode );
        NameNode.stateChangeLog.debug("DIR* FSDirectory.addFile: "
                +path+" with "+blocks.length+" blocks is added to the file system" );
        return true;
    }