namenode.java

hadoop:Nutch集群平台

            return new String[0][];
        } else {
            String results[][] = new String[hosts.length][];
            for (int i = 0; i < hosts.length; i++) {
                results[i] = new String[hosts[i].length];
                for (int j = 0; j < results[i].length; j++) {
                    results[i][j] = hosts[i][j].toString();
                }
            }
            return results;
        }
    }
    
    public long getBlockSize(String filename) throws IOException {
      return namesystem.getBlockSize(filename);
    }
    
    /**
     */
    public boolean rename(String src, String dst) throws IOException {
        stateChangeLog.debug("*DIR* NameNode.rename: " + src + " to " + dst );
        if (!checkPathLength(dst)) {
            throw new IOException("rename: Pathname too long.  Limit " 
                + MAX_PATH_LENGTH + " characters, " + MAX_PATH_DEPTH + " levels.");
        }
        boolean ret = namesystem.renameTo(new UTF8(src), new UTF8(dst));
        if (ret) {
            myMetrics.renameFile();
        }
        return ret;
    }

    /**
     */
    public boolean delete(String src) throws IOException {
        stateChangeLog.debug("*DIR* NameNode.delete: " + src );
        return namesystem.delete(new UTF8(src));
    }

    /**
     */
    public boolean exists(String src) throws IOException {
        return namesystem.exists(new UTF8(src));
    }

    /**
     */
    public boolean isDir(String src) throws IOException {
        return namesystem.isDir(new UTF8(src));
    }

    /**
     * Check path length does not exceed maximum.  Returns true if
     * length and depth are okay.  Returns false if length is too long 
     * or depth is too great.
     * 
     */
    private boolean checkPathLength(String src) {
        Path srcPath = new Path(src);
        return (src.length() <= MAX_PATH_LENGTH &&
                srcPath.depth() <= MAX_PATH_DEPTH);
    }
    
    /**
     */
    public boolean mkdirs(String src) throws IOException {
        stateChangeLog.debug("*DIR* NameNode.mkdirs: " + src );
        if (!checkPathLength(src)) {
            throw new IOException("mkdirs: Pathname too long.  Limit " 
                + MAX_PATH_LENGTH + " characters, " + MAX_PATH_DEPTH + " levels.");
        }
        return namesystem.mkdirs( src );
    }

    /**
     */
    public boolean obtainLock(String src, String clientName, boolean exclusive) throws IOException {
        int returnCode = namesystem.obtainLock(new UTF8(src), new UTF8(clientName), exclusive);
        if (returnCode == COMPLETE_SUCCESS) {
            return true;
        } else if (returnCode == STILL_WAITING) {
            return false;
        } else {
            throw new IOException("Failure when trying to obtain lock on " + src);
        }
    }

    /**
     */
    public boolean releaseLock(String src, String clientName) throws IOException {
        int returnCode = namesystem.releaseLock(new UTF8(src), new UTF8(clientName));
        if (returnCode == COMPLETE_SUCCESS) {
            return true;
        } else if (returnCode == STILL_WAITING) {
            return false;
        } else {
            throw new IOException("Failure when trying to release lock on " + src);
        }
    }

    /**
     */
    public void renewLease(String clientName) throws IOException {
        namesystem.renewLease(new UTF8(clientName));        
    }

    /**
     */
    public DFSFileInfo[] getListing(String src) throws IOException {
        DFSFileInfo[] files = namesystem.getListing(new UTF8(src));
        if (files != null) {
            myMetrics.listFile(files.length);
        }
        return files;
    }

    /**
     */
    public long[] getStats() throws IOException {
        long results[] = new long[2];
        results[0] = namesystem.totalCapacity();
        results[1] = namesystem.totalCapacity() - namesystem.totalRemaining();
        return results;
    }

    /**
     */
    public DatanodeInfo[] getDatanodeReport() throws IOException {
        DatanodeInfo results[] = namesystem.datanodeReport();
        if (results == null || results.length == 0) {
            throw new IOException("Cannot find datanode report");
        }
        return results;
    }
    
    /**
     * @inheritDoc
     */
    public boolean setSafeMode( SafeModeAction action ) throws IOException {
      switch( action ) {
      case SAFEMODE_LEAVE: // leave safe mode
        namesystem.leaveSafeMode();
        break;
      case SAFEMODE_ENTER: // enter safe mode
        namesystem.enterSafeMode();
        break;
      case SAFEMODE_GET: // get safe mode
      }
      return namesystem.isInSafeMode();
    }

    /**
     * Is the cluster currently in safe mode?
     */
    boolean isInSafeMode() {
      return namesystem.isInSafeMode();
    }
    
    ////////////////////////////////////////////////////////////////
    // DatanodeProtocol
    ////////////////////////////////////////////////////////////////
    /** 
     */
    public DatanodeRegistration register( DatanodeRegistration nodeReg
                                        ) throws IOException {
      verifyVersion( nodeReg.getVersion() );
      namesystem.registerDatanode( nodeReg );
      return nodeReg;
    }
    
    /**
     * Data node notify the name node that it is alive 
     * Return a block-oriented command for the datanode to execute.
     * This will be either a transfer or a delete operation.
     */
    public BlockCommand sendHeartbeat(DatanodeRegistration nodeReg,
                                      long capacity, 
                                      long remaining,
                                      int xmitsInProgress,
                                      int xceiverCount) throws IOException {
        verifyRequest( nodeReg );
        namesystem.gotHeartbeat( nodeReg, capacity, remaining, xceiverCount );
        
        //
        // Ask to perform pending transfers, if any
        //
        Object xferResults[] = namesystem.pendingTransfers( nodeReg,
                                                            xmitsInProgress );
        if (xferResults != null) {
            return new BlockCommand((Block[]) xferResults[0], (DatanodeInfo[][]) xferResults[1]);
        }

        //
        // If there are no transfers, check for recently-deleted blocks that
        // should be removed.  This is not a full-datanode sweep, as is done during
        // a block report.  This is just a small fast removal of blocks that have
        // just been removed.
        //
        Block blocks[] = namesystem.blocksToInvalidate( nodeReg );
        if (blocks != null) {
            return new BlockCommand(blocks);
        }
        return null;
    }

    public Block[] blockReport( DatanodeRegistration nodeReg,
                                Block blocks[]) throws IOException {
        verifyRequest( nodeReg );
        stateChangeLog.debug("*BLOCK* NameNode.blockReport: "
                +"from "+nodeReg.getName()+" "+blocks.length+" blocks" );

        return namesystem.processReport( nodeReg, blocks );
     }

    public void blockReceived(DatanodeRegistration nodeReg, 
                              Block blocks[]) throws IOException {
        verifyRequest( nodeReg );
        stateChangeLog.debug("*BLOCK* NameNode.blockReceived: "
                +"from "+nodeReg.getName()+" "+blocks.length+" blocks." );
        for (int i = 0; i < blocks.length; i++) {
            namesystem.blockReceived( nodeReg, blocks[i] );
        }
    }

    /**
     */
    public void errorReport(DatanodeRegistration nodeReg,
                            int errorCode, 
                            String msg) throws IOException {
      // Log error message from datanode
      verifyRequest( nodeReg );
      LOG.warn("Report from " + nodeReg.getName() + ": " + msg);
      if( errorCode == DatanodeProtocol.DISK_ERROR ) {
          namesystem.removeDatanode( nodeReg );            
      }
    }

    /** 
     * Verify request.
     * 
     * Verifies correctness of the datanode version and registration ID.
     * 
     * @param nodeReg data node registration
     * @throws IOException
     */
    public void verifyRequest( DatanodeRegistration nodeReg ) throws IOException {
      verifyVersion( nodeReg.getVersion() );
      if( ! namesystem.getRegistrationID().equals( nodeReg.getRegistrationID() ))
          throw new UnregisteredDatanodeException( nodeReg );
    }
    
    /**
     * Verify version.
     * 
     * @param version
     * @throws IOException
     */
    public void verifyVersion( int version ) throws IOException {
      if( version != DFS_CURRENT_VERSION )
        throw new IncorrectVersionException( version, "data node" );
    }

    /**
     */
    public static void main(String argv[]) throws Exception {
        Configuration conf = new Configuration();

        if (argv.length == 1 && argv[0].equals("-format")) {
          File dir = getDir(conf);
          if (dir.exists()) {
            System.err.print("Re-format filesystem in " + dir +" ? (Y or N) ");
            if (!(System.in.read() == 'Y')) {
              System.err.println("Format aborted.");
              System.exit(1);
            }
          }
          format(conf);
          System.err.println("Formatted "+dir);
          System.exit(0);
        }
        
        NameNode namenode = new NameNode(conf);
        namenode.join();
    }
}