fsdirectory.java

hadoop:Nutch集群平台

    
    /**
     */
    boolean unprotectedAddFile(UTF8 path, INode newNode) {
      synchronized (rootDir) {
         try {
            if( rootDir.addNode(path.toString(), newNode ) != null ) {
                int nrBlocks = (newNode.blocks == null) ? 0 : newNode.blocks.length;
                // Add file->block mapping
                for (int i = 0; i < nrBlocks; i++)
                    activeBlocks.put(newNode.blocks[i], newNode);
                return true;
            } else {
                return false;
            }
        } catch (FileNotFoundException e ) {
            return false;
        }
      }
    }
    
    boolean unprotectedAddFile(UTF8 path, Block[] blocks, short replication ) {
      return unprotectedAddFile( path,  
                    new INode( path.toString(), blocks, replication ));
    }

    /**
     * Change the filename
     */
    public boolean renameTo(UTF8 src, UTF8 dst) {
      NameNode.stateChangeLog.debug("DIR* FSDirectory.renameTo: "
          +src+" to "+dst );
      waitForReady();
      if( ! unprotectedRenameTo(src, dst) )
        return false;
      fsImage.getEditLog().logRename(src, dst);
      return true;
    }

    /**
     */
    boolean unprotectedRenameTo(UTF8 src, UTF8 dst) {
        synchronized(rootDir) {
          String srcStr = src.toString();
          String dstStr = dst.toString();
            INode renamedNode = rootDir.getNode(srcStr);
            if (renamedNode == null) {
                NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
                        +"failed to rename "+src+" to "+dst+ " because source does not exist" );
                return false;
            }
            if (isDir(dst)) {
              dstStr += "/" + new File(srcStr).getName();
            }
            if( rootDir.getNode(dstStr.toString()) != null ) {
                NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
                        +"failed to rename "+src+" to "+dstStr+ " because destination exists" );
                return false;
            }
            renamedNode.removeNode();
            
            // the renamed node can be reused now
            try {
                if( rootDir.addNode(dstStr, renamedNode ) != null ) {
                    NameNode.stateChangeLog.debug("DIR* FSDirectory.unprotectedRenameTo: "
                        +src+" is renamed to "+dst );
                    return true;
                }
            } catch (FileNotFoundException e ) {
                NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
                        +"failed to rename "+src+" to "+dst );
                try {
                    rootDir.addNode(srcStr, renamedNode); // put it back
                }catch(FileNotFoundException e2) {                
                }
            }

            return false;
        }
    }

    /**
     * Set file replication
     * 
     * @param src file name
     * @param replication new replication
     * @param oldReplication old replication - output parameter
     * @return array of file blocks
     * @throws IOException
     */
    Block[] setReplication( String src, 
                            short replication,
                            Vector oldReplication
                           ) throws IOException {
      waitForReady();
      Block[] fileBlocks = unprotectedSetReplication(src, replication, oldReplication );
      if( fileBlocks != null )  // log replication change
        fsImage.getEditLog().logSetReplication( src, replication );
      return fileBlocks;
    }

    Block[] unprotectedSetReplication(  String src, 
                                        short replication,
                                        Vector oldReplication
                                      ) throws IOException {
      if( oldReplication == null )
        oldReplication = new Vector();
      oldReplication.setSize(1);
      oldReplication.set( 0, new Integer(-1) );
      Block[] fileBlocks = null;
      synchronized(rootDir) {
        INode fileNode = rootDir.getNode(src);
        if (fileNode == null)
          return null;
        if( fileNode.isDir() )
          return null;
        oldReplication.set( 0, new Integer( fileNode.blockReplication ));
        fileNode.blockReplication = replication;
        fileBlocks = fileNode.blocks;
      }
      return fileBlocks;
    }

    /**
     * Get the blocksize of a file
     * @param filename the filename
     * @return the number of bytes in the first block
     * @throws IOException if it is a directory or does not exist.
     */
    public long getBlockSize(String filename) throws IOException {
      synchronized (rootDir) {
        INode fileNode = rootDir.getNode(filename);
        if (fileNode == null) {
          throw new IOException("Unknown file: " + filename);
        }
        if (fileNode.isDir()) {
          throw new IOException("Getting block size of a directory: " + 
                                filename);
        }
        return fileNode.getBlockSize();
      }
    }
    
    /**
     * Remove the file from management, return blocks
     */
    public Block[] delete(UTF8 src) {
        NameNode.stateChangeLog.debug("DIR* FSDirectory.delete: "
                +src );
        waitForReady();
        Block[] blocks = unprotectedDelete(src); 
        if( blocks != null )
          fsImage.getEditLog().logDelete( src );
        return blocks;
    }

    /**
     */
    Block[] unprotectedDelete(UTF8 src) {
        synchronized (rootDir) {
            INode targetNode = rootDir.getNode(src.toString());
            if (targetNode == null) {
                NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedDelete: "
                        +"failed to remove "+src+" because it does not exist" );
                return null;
            } else {
                //
                // Remove the node from the namespace and GC all
                // the blocks underneath the node.
                //
                if (! targetNode.removeNode()) {
                    NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedDelete: "
                            +"failed to remove "+src+" because it does not have a parent" );
                    return null;
                } else {
                    NameNode.stateChangeLog.debug("DIR* FSDirectory.unprotectedDelete: "
                            +src+" is removed" );
                    Vector v = new Vector();
                    targetNode.collectSubtreeBlocks(v);
                    for (Iterator it = v.iterator(); it.hasNext(); ) {
                        Block b = (Block) it.next();
                        activeBlocks.remove(b);
                    }
                    return (Block[]) v.toArray(new Block[v.size()]);
                }
            }
        }
    }

    /**
     */
    public int obtainLock(UTF8 src, UTF8 holder, boolean exclusive) {
        TreeSet holders = (TreeSet) activeLocks.get(src);
        if (holders == null) {
            holders = new TreeSet();
            activeLocks.put(src, holders);
        }
        if (exclusive && holders.size() > 0) {
            return STILL_WAITING;
        } else {
            holders.add(holder);
            return COMPLETE_SUCCESS;
        }
    }

    /**
     */
    public int releaseLock(UTF8 src, UTF8 holder) {
        TreeSet holders = (TreeSet) activeLocks.get(src);
        if (holders != null && holders.contains(holder)) {
            holders.remove(holder);
            if (holders.size() == 0) {
                activeLocks.remove(src);
            }
            return COMPLETE_SUCCESS;
        } else {
            return OPERATION_FAILED;
        }
    }

    /**
     * Get a listing of files given path 'src'
     *
     * This function is admittedly very inefficient right now.  We'll
     * make it better later.
     */
    public DFSFileInfo[] getListing(UTF8 src) {
        String srcs = normalizePath(src);

        synchronized (rootDir) {
            INode targetNode = rootDir.getNode(srcs);
            if (targetNode == null) {
                return null;
            } else {
                Vector contents = new Vector();
                targetNode.listContents(contents);

                DFSFileInfo listing[] = new DFSFileInfo[contents.size()];
                int i = 0;
                for (Iterator it = contents.iterator(); it.hasNext(); i++) {
                    listing[i] = new DFSFileInfo( (INode) it.next() );
                }
                return listing;
            }
        }
    }

    /**
     * Get the blocks associated with the file
     */
    public Block[] getFile(UTF8 src) {
        waitForReady();
        synchronized (rootDir) {
            INode targetNode = rootDir.getNode(src.toString());
            if (targetNode == null) {
                return null;
            } else {
                return targetNode.blocks;
            }
        }
    }

    /** 
     * Check whether the filepath could be created
     */
    public boolean isValidToCreate(UTF8 src) {
        String srcs = normalizePath(src);
        synchronized (rootDir) {
            if (srcs.startsWith("/") && 
                ! srcs.endsWith("/") && 
                rootDir.getNode(srcs) == null) {
                return true;
            } else {
                return false;
            }
        }
    }

    /**
     * Check whether the path specifies a directory
     */
    public boolean isDir(UTF8 src) {
        synchronized (rootDir) {
            INode node = rootDir.getNode(normalizePath(src));
            return node != null && node.isDir();
        }
    }

    /**
     * @deprecated use @link #mkdirs(String) instead
     */
    public boolean mkdirs(UTF8 src) {
        return mkdirs(src.toString());
    }

    /**
     * Create directory entries for every item
     */
    boolean mkdirs(String src) {
        src = normalizePath(new UTF8(src));

        // Use this to collect all the dirs we need to construct
        Vector v = new Vector();

        // The dir itself
        v.add(src);

        // All its parents
        Path parent = new Path(src).getParent();
        while (parent != null) {
            v.add(parent.toString());
            parent = parent.getParent();
        }

        // Now go backwards through list of dirs, creating along
        // the way
        int numElts = v.size();
        for (int i = numElts - 1; i >= 0; i--) {
            String cur = (String) v.elementAt(i);
            try {
               INode inserted = unprotectedMkdir(cur);
               if (inserted != null) {
                   NameNode.stateChangeLog.debug("DIR* FSDirectory.mkdirs: "
                        +"created directory "+cur );
                   fsImage.getEditLog().logMkDir( inserted );
               } // otherwise cur exists, continue
            } catch (FileNotFoundException e ) {
                NameNode.stateChangeLog.debug("DIR* FSDirectory.mkdirs: "
                        +"failed to create directory "+src);
                return false;
            }
        }
        return true;
    }

    /**
     */
    INode unprotectedMkdir(String src) throws FileNotFoundException {
        synchronized (rootDir) {
            return rootDir.addNode(src, new INode(new File(src).getName()));
        }
    }

    /**
     */
    String normalizePath(UTF8 src) {
        String srcs = src.toString();
        if (srcs.length() > 1 && srcs.endsWith("/")) {
            srcs = srcs.substring(0, srcs.length() - 1);
        }
        return srcs;
    }

    /**
     * Returns whether the given block is one pointed-to by a file.
     */
    public boolean isValidBlock(Block b) {
        synchronized (rootDir) {
            if (activeBlocks.containsKey(b)) {
                return true;
            } else {
                return false;
            }
        }
    }

    /**
     * Returns whether the given block is one pointed-to by a file.
     */
    public INode getFileByBlock(Block b) {
      synchronized (rootDir) {
        return (INode)activeBlocks.get(b);
      }
    }
}