hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cutt...@apache.org
Subject svn commit: r410635 [1/2] - in /lucene/hadoop/trunk: ./ src/java/org/apache/hadoop/dfs/ src/test/org/apache/hadoop/fs/ src/test/org/apache/hadoop/test/
Date Wed, 31 May 2006 18:52:50 GMT
Author: cutting
Date: Wed May 31 11:52:49 2006
New Revision: 410635

URL: http://svn.apache.org/viewvc?rev=410635&view=rev
Log:
HADOOP-124.  Change DFS so that datanodes are identified by a persistent ID rather than by host and port.  Contributed by Konstantin.

Added:
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/AlreadyBeingCreatedException.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataStorage.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeID.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeRegistration.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/IncorrectVersionException.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/LeaseExpiredException.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/NotReplicatedYetException.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/UnregisteredDatanodeException.java
Modified:
    lucene/hadoop/trunk/CHANGES.txt
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/BlockCommand.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSClient.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSck.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataNode.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeInfo.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeProtocol.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DistributedFileSystem.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDirectory.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/NameNode.java
    lucene/hadoop/trunk/src/test/org/apache/hadoop/fs/TestDFSIO.java
    lucene/hadoop/trunk/src/test/org/apache/hadoop/test/AllTestDriver.java

Modified: lucene/hadoop/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/CHANGES.txt?rev=410635&r1=410634&r2=410635&view=diff
==============================================================================
--- lucene/hadoop/trunk/CHANGES.txt (original)
+++ lucene/hadoop/trunk/CHANGES.txt Wed May 31 11:52:49 2006
@@ -85,6 +85,11 @@
     is closed, rather than as each block is written.
     (Milind Bhandarkar via cutting)
 
+23. HADOOP-124. Change DFS so that datanodes are identified by a
+    persistent ID rather than by host and port.  This solves a number
+    of filesystem integrity problems, when, e.g., datanodes are
+    restarted.  (Konstantin Shvachko via cutting)
+
 
 Release 0.2.1 - 2006-05-12
 

Added: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/AlreadyBeingCreatedException.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/AlreadyBeingCreatedException.java?rev=410635&view=auto
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/AlreadyBeingCreatedException.java (added)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/AlreadyBeingCreatedException.java Wed May 31 11:52:49 2006
@@ -0,0 +1,14 @@
+package org.apache.hadoop.dfs;
+
+import java.io.IOException;
+
+/**
+ * The exception that happens when you ask to create a file that already
+ * is being created, but is not closed yet.
+ * @author Owen O'Malley
+ */
+public class AlreadyBeingCreatedException extends IOException {
+  public AlreadyBeingCreatedException(String msg) {
+    super(msg);
+  }
+}

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/BlockCommand.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/BlockCommand.java?rev=410635&r1=410634&r2=410635&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/BlockCommand.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/BlockCommand.java Wed May 31 11:52:49 2006
@@ -40,12 +40,14 @@
   
     boolean transferBlocks = false;
     boolean invalidateBlocks = false;
+    boolean shutdown = false;
     Block blocks[];
     DatanodeInfo targets[][];
 
     public BlockCommand() {
         this.transferBlocks = false;
-        this.invalidateBlocks = false;        
+        this.invalidateBlocks = false;
+        this.shutdown = false;
         this.blocks = new Block[0];
         this.targets = new DatanodeInfo[0][];
     }
@@ -53,6 +55,7 @@
     public BlockCommand(Block blocks[], DatanodeInfo targets[][]) {
         this.transferBlocks = true;
         this.invalidateBlocks = false;
+        this.shutdown = false;
         this.blocks = blocks;
         this.targets = targets;
     }
@@ -60,10 +63,16 @@
     public BlockCommand(Block blocks[]) {
         this.transferBlocks = false;
         this.invalidateBlocks = true;
+        this.shutdown = false;
         this.blocks = blocks;
         this.targets = new DatanodeInfo[0][];
     }
 
+    public BlockCommand( boolean doShutdown ) {
+      this();
+      this.shutdown = doShutdown;
+    }
+
     public boolean transferBlocks() {
         return transferBlocks;
     }
@@ -72,6 +81,10 @@
         return invalidateBlocks;
     }
     
+    public boolean shutdownNode() {
+      return shutdown;
+  }
+  
     public Block[] getBlocks() {
         return blocks;
     }

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSClient.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSClient.java?rev=410635&r1=410634&r2=410635&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSClient.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSClient.java Wed May 31 11:52:49 2006
@@ -375,11 +375,7 @@
             if (deadNodes.contains(nodes[i])) {
                 continue;
             }
-            String nodename = nodes[i].getName().toString();
-            int colon = nodename.indexOf(':');
-            if (colon >= 0) {
-                nodename = nodename.substring(0, colon);
-            }
+            String nodename = nodes[i].getHost();
             if (localName.equals(nodename)) {
                 chosenNode = nodes[i];
                 break;
@@ -524,7 +520,7 @@
 
                 try {
                     chosenNode = bestNode(nodes[targetBlock], deadNodes);
-                    targetAddr = DataNode.createSocketAddr(chosenNode.getName().toString());
+                    targetAddr = DataNode.createSocketAddr(chosenNode.getName());
                 } catch (IOException ie) {
                     String blockInfo =
                       blocks[targetBlock]+" file="+src+" offset="+target;
@@ -758,12 +754,12 @@
                 //
                 // Connect to first DataNode in the list.  Abort if this fails.
                 //
-                InetSocketAddress target = DataNode.createSocketAddr(nodes[0].getName().toString());
+                InetSocketAddress target = DataNode.createSocketAddr(nodes[0].getName());
                 try {
                     s = new Socket();
                     s.connect(target, READ_TIMEOUT);
                     s.setSoTimeout(replication * READ_TIMEOUT);
-                    datanodeName = nodes[0].getName().toString();
+                    datanodeName = nodes[0].getName();
                 } catch (IOException ie) {
                     // Connection failed.  Let's wait a little bit and retry
                     try {
@@ -811,7 +807,7 @@
                     localName, overwrite, replication, blockSize);
               } catch (RemoteException e) {
                 if (--retries == 0 || 
-                    "org.apache.hadoop.dfs.NameNode.AlreadyBeingCreatedException".
+                    AlreadyBeingCreatedException.class.getName().
                         equals(e.getClassName())) {
                   throw e;
                 } else {
@@ -841,7 +837,7 @@
                                          clientName.toString());
               } catch (RemoteException e) {
                 if (--retries == 0 || 
-                    "org.apache.hadoop.dfs.NameNode.NotReplicatedYetException".
+                    NotReplicatedYetException.class.getName().
                         equals(e.getClassName())) {
                   throw e;
                 } else {

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSck.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSck.java?rev=410635&r1=410634&r2=410635&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSck.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSck.java Wed May 31 11:52:49 2006
@@ -266,7 +266,7 @@
 
         try {
             chosenNode = bestNode(lblock.getLocations(), deadNodes);
-            targetAddr = DataNode.createSocketAddr(chosenNode.getName().toString());
+            targetAddr = DataNode.createSocketAddr(chosenNode.getName());
         } catch (IOException ie) {
             if (failures >= DFSClient.MAX_BLOCK_ACQUIRE_FAILURES) {
                 throw new IOException("Could not obtain block " + lblock);
@@ -358,7 +358,7 @@
           if (deadNodes.contains(nodes[i])) {
               continue;
           }
-          String nodename = nodes[i].getName().toString();
+          String nodename = nodes[i].getName();
           int colon = nodename.indexOf(':');
           if (colon >= 0) {
               nodename = nodename.substring(0, colon);

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataNode.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataNode.java?rev=410635&r1=410634&r2=410635&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataNode.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataNode.java Wed May 31 11:52:49 2006
@@ -22,6 +22,7 @@
 
 import java.io.*;
 import java.net.*;
+import java.nio.channels.FileLock;
 import java.util.*;
 import java.util.logging.*;
 
@@ -59,7 +60,7 @@
  **********************************************************/
 public class DataNode implements FSConstants, Runnable {
     public static final Logger LOG = LogFormatter.getLogger("org.apache.hadoop.dfs.DataNode");
-  //
+    //
     // REMIND - mjc - I might bring "maxgigs" back so user can place 
     // artificial  limit on space
     //private static final long GIGABYTE = 1024 * 1024 * 1024;
@@ -87,12 +88,13 @@
     private static Vector subThreadList = null;
     DatanodeProtocol namenode;
     FSDataset data;
-    String localName;
+    DatanodeRegistration dnRegistration;
     boolean shouldRun = true;
     Vector receivedBlockList = new Vector();
     int xmitsInProgress = 0;
     Daemon dataXceiveServer = null;
     long blockReportInterval;
+    private DataStorage storage = null;
 
     /**
      * Create the DataNode given a configuration and a dataDir.
@@ -102,43 +104,80 @@
         this(InetAddress.getLocalHost().getHostName(), 
              new File(datadir),
              createSocketAddr(conf.get("fs.default.name", "local")), conf);
+        // register datanode
+        register();
     }
 
     /**
      * A DataNode can also be created with configuration information
      * explicitly given.
+     * 
+     * @see DataStorage
      */
-    public DataNode(String machineName, File datadir, InetSocketAddress nameNodeAddr, Configuration conf) throws IOException {
-        this.namenode = (DatanodeProtocol) RPC.getProxy(DatanodeProtocol.class, nameNodeAddr, conf);
-        this.data = new FSDataset(datadir, conf);
-
-        ServerSocket ss = null;
-        int tmpPort = conf.getInt("dfs.datanode.port", 50010);
-        while (ss == null) {
-            try {
-                ss = new ServerSocket(tmpPort);
-                LOG.info("Opened server at " + tmpPort);
-            } catch (IOException ie) {
-                LOG.info("Could not open server at " + tmpPort + ", trying new port");
-                tmpPort++;
-            }
+    private DataNode(String machineName, 
+                    File datadir, 
+                    InetSocketAddress nameNodeAddr, 
+                    Configuration conf ) throws IOException {
+      // get storage info and lock the data dir
+      storage = new DataStorage( datadir );
+      // connect to name node
+      this.namenode = (DatanodeProtocol) RPC.getProxy(DatanodeProtocol.class, 
+                                                      nameNodeAddr, 
+                                                      conf);
+      // find free port
+      ServerSocket ss = null;
+      int tmpPort = conf.getInt("dfs.datanode.port", 50010);
+      while (ss == null) {
+        try {
+          ss = new ServerSocket(tmpPort);
+          LOG.info("Opened server at " + tmpPort);
+        } catch (IOException ie) {
+          LOG.info("Could not open server at " + tmpPort + ", trying new port");
+          tmpPort++;
         }
-        this.localName = machineName + ":" + tmpPort;
-        this.dataXceiveServer = new Daemon(new DataXceiveServer(ss));
-        this.dataXceiveServer.start();
-
-        long blockReportIntervalBasis =
-          conf.getLong("dfs.blockreport.intervalMsec", BLOCKREPORT_INTERVAL);
-        this.blockReportInterval =
-          blockReportIntervalBasis - new Random().nextInt((int)(blockReportIntervalBasis/10));
+      }
+      // construct registration
+      this.dnRegistration = new DatanodeRegistration(
+                                        DFS_CURRENT_VERSION, 
+                                        machineName + ":" + tmpPort, 
+                                        storage.getStorageID(),
+                                        "" );
+      // initialize data node internal structure
+      this.data = new FSDataset(datadir, conf);
+      this.dataXceiveServer = new Daemon(new DataXceiveServer(ss));
+      this.dataXceiveServer.start();
+
+      long blockReportIntervalBasis =
+        conf.getLong("dfs.blockreport.intervalMsec", BLOCKREPORT_INTERVAL);
+      this.blockReportInterval =
+        blockReportIntervalBasis - new Random().nextInt((int)(blockReportIntervalBasis/10));
     }
 
     /**
      * Return the namenode's identifier
      */
     public String getNamenode() {
-        //return namenode.toString();
-	return "<namenode>";
+      //return namenode.toString();
+      return "<namenode>";
+    }
+
+    /**
+     * Register datanode
+     * <p>
+     * The datanode needs to register with the namenode on startup in order
+     * 1) to report which storage it is serving now and 
+     * 2) to receive a registrationID 
+     * issued by the namenode to recognize registered datanodes.
+     * 
+     * @see FSNamesystem#registerDatanode(DatanodeRegistration)
+     * @throws IOException
+     */
+    private void register() throws IOException {
+      dnRegistration = namenode.register( dnRegistration );
+      if( storage.getStorageID().equals("") ) {
+        storage.setStorageID( dnRegistration.getStorageID());
+        storage.write();
+      }
     }
 
     /**
@@ -152,13 +191,17 @@
             this.dataXceiveServer.join();
         } catch (InterruptedException ie) {
         }
+        try {
+          this.storage.close();
+        } catch (IOException ie) {
+        }
     }
 
     void handleDiskError( String errMsgr ) {
         LOG.warning( "Shuting down DataNode because "+errMsgr );
         try {
             namenode.errorReport(
-                    localName, DatanodeProtocol.DISK_ERROR, errMsgr);
+                    dnRegistration, DatanodeProtocol.DISK_ERROR, errMsgr);
         } catch( IOException ignored) {              
         }
         shutdown();
@@ -169,116 +212,122 @@
      * forever calling remote NameNode functions.
      */
     public void offerService() throws Exception {
-        long lastHeartbeat = 0, lastBlockReport = 0;
-        LOG.info("using BLOCKREPORT_INTERVAL of " + blockReportInterval + "msec");
+      long lastHeartbeat = 0, lastBlockReport = 0;
+      LOG.info("using BLOCKREPORT_INTERVAL of " + blockReportInterval + "msec");
 
-        //
-        // Now loop for a long time....
-        //
+      //
+      // Now loop for a long time....
+      //
 
-        try {
-            while (shouldRun) {
-                long now = System.currentTimeMillis();
-    
+      try {
+        while (shouldRun) {
+          long now = System.currentTimeMillis();
+
+          //
+          // Every so often, send heartbeat or block-report
+          //
+          if (now - lastHeartbeat > HEARTBEAT_INTERVAL) {
+            //
+            // All heartbeat messages include following info:
+            // -- Datanode name
+            // -- data transfer port
+            // -- Total capacity
+            // -- Bytes remaining
+            //
+            BlockCommand cmd = namenode.sendHeartbeat(dnRegistration, 
+                                                      data.getCapacity(), 
+                                                      data.getRemaining(), 
+                                                      xmitsInProgress);
+            //LOG.info("Just sent heartbeat, with name " + localName);
+            lastHeartbeat = now;
+
+            if( cmd != null ) {
+              data.checkDataDir();
+              if (cmd.transferBlocks()) {
                 //
-                // Every so often, send heartbeat or block-report
+                // Send a copy of a block to another datanode
                 //
-                if (now - lastHeartbeat > HEARTBEAT_INTERVAL) {
-                    //
-                    // All heartbeat messages include following info:
-                    // -- Datanode name
-                    // -- data transfer port
-                    // -- Total capacity
-                    // -- Bytes remaining
-                    //
-                    BlockCommand cmd = namenode.sendHeartbeat(localName, 
-                            data.getCapacity(), data.getRemaining(), xmitsInProgress);
-                    //LOG.info("Just sent heartbeat, with name " + localName);
-                    lastHeartbeat = now;
-    
-                    if( cmd != null ) {
-                        data.checkDataDir();
-                        if (cmd.transferBlocks()) {
-                            //
-                            // Send a copy of a block to another datanode
-                            //
-                            Block blocks[] = cmd.getBlocks();
-                            DatanodeInfo xferTargets[][] = cmd.getTargets();
-                        
-                            for (int i = 0; i < blocks.length; i++) {
-                                if (!data.isValidBlock(blocks[i])) {
-                                    String errStr = "Can't send invalid block " + blocks[i];
-                                    LOG.info(errStr);
-                                    namenode.errorReport(
-                                        localName, 
-                                        DatanodeProtocol.INVALID_BLOCK, 
-                                        errStr);
-                                    break;
-                                } else {
-                                    if (xferTargets[i].length > 0) {
-                                        LOG.info("Starting thread to transfer block " + blocks[i] + " to " + xferTargets[i]);
-                                        new Daemon(new DataTransfer(xferTargets[i], blocks[i])).start();
-                                    }
-                                }
-                            }
-                         } else if (cmd.invalidateBlocks()) {
-                            //
-                            // Some local block(s) are obsolete and can be 
-                            // safely garbage-collected.
-                            //
-                            data.invalidate(cmd.getBlocks());
-                         }
-                    }
-                }
-                
-                // send block report
-                if (now - lastBlockReport > blockReportInterval) {
-                    // before send block report, check if data directory is healthy
-                    data.checkDataDir();
+                Block blocks[] = cmd.getBlocks();
+                DatanodeInfo xferTargets[][] = cmd.getTargets();
                     
-                     //
-                     // Send latest blockinfo report if timer has expired.
-                     // Get back a list of local block(s) that are obsolete
-                     // and can be safely GC'ed.
-                     //
-                     Block toDelete[] = namenode.blockReport(localName, data.getBlockReport());
-                     data.invalidate(toDelete);
-                     lastBlockReport = now;
-                     continue;
-                }
-                
-                // check if there are newly received blocks
-                Block [] blockArray=null;
-                synchronized( receivedBlockList ) {
-                    if (receivedBlockList.size() > 0) {
-                        //
-                        // Send newly-received blockids to namenode
-                        //
-                        blockArray = (Block[]) receivedBlockList.toArray(new Block[receivedBlockList.size()]);
-                        receivedBlockList.removeAllElements();
+                for (int i = 0; i < blocks.length; i++) {
+                  if (!data.isValidBlock(blocks[i])) {
+                    String errStr = "Can't send invalid block " + blocks[i];
+                    LOG.info(errStr);
+                    namenode.errorReport( dnRegistration, 
+                                DatanodeProtocol.INVALID_BLOCK, 
+                                errStr);
+                    break;
+                  } else {
+                    if (xferTargets[i].length > 0) {
+                        LOG.info("Starting thread to transfer block " + blocks[i] + " to " + xferTargets[i]);
+                        new Daemon(new DataTransfer(xferTargets[i], blocks[i])).start();
                     }
+                  }
                 }
-                if( blockArray != null ) {
-                    namenode.blockReceived(localName, blockArray);
-                }
-                
+              } else if (cmd.invalidateBlocks()) {
                 //
-                // There is no work to do;  sleep until hearbeat timer elapses, 
-                // or work arrives, and then iterate again.
+                // Some local block(s) are obsolete and can be 
+                // safely garbage-collected.
                 //
-                long waitTime = HEARTBEAT_INTERVAL - (System.currentTimeMillis() - lastHeartbeat);
-                synchronized( receivedBlockList ) {
-                    if (waitTime > 0 && receivedBlockList.size() == 0) {
-                        try {
-                            receivedBlockList.wait(waitTime);
-                        } catch (InterruptedException ie) {
-                        }
-                    }
-                } // synchronized
-            } // while (shouldRun)
-        } catch(DiskErrorException e) {
-            handleDiskError(e.getMessage());
-        }
+                data.invalidate(cmd.getBlocks());
+              } else if( cmd.shutdownNode()) {
+                // shut down the data node
+                this.shutdown();
+                continue;
+              }
+            }
+          }
+            
+          // send block report
+          if (now - lastBlockReport > blockReportInterval) {
+            // before send block report, check if data directory is healthy
+            data.checkDataDir();
+                
+            //
+            // Send latest blockinfo report if timer has expired.
+            // Get back a list of local block(s) that are obsolete
+            // and can be safely GC'ed.
+            //
+            Block toDelete[] = namenode.blockReport(dnRegistration,
+                                                    data.getBlockReport());
+            data.invalidate(toDelete);
+            lastBlockReport = now;
+            continue;
+          }
+            
+          // check if there are newly received blocks
+          Block [] blockArray=null;
+          synchronized( receivedBlockList ) {
+            if (receivedBlockList.size() > 0) {
+              //
+              // Send newly-received blockids to namenode
+              //
+              blockArray = (Block[]) receivedBlockList.toArray(new Block[receivedBlockList.size()]);
+              receivedBlockList.removeAllElements();
+            }
+          }
+          if( blockArray != null ) {
+            namenode.blockReceived( dnRegistration, blockArray );
+          }
+            
+          //
+          // There is no work to do;  sleep until hearbeat timer elapses, 
+          // or work arrives, and then iterate again.
+          //
+          long waitTime = HEARTBEAT_INTERVAL - (System.currentTimeMillis() - lastHeartbeat);
+          synchronized( receivedBlockList ) {
+            if (waitTime > 0 && receivedBlockList.size() == 0) {
+              try {
+                receivedBlockList.wait(waitTime);
+              } catch (InterruptedException ie) {
+              }
+            }
+          } // synchronized
+        } // while (shouldRun)
+      } catch(DiskErrorException e) {
+        handleDiskError(e.getMessage());
+      }
     } // offerService
 
     /**
@@ -503,7 +552,7 @@
               DataOutputStream out2 = null;
               if (targets.length > 1) {
                 // Connect to backup machine
-                mirrorNode = targets[1].getName().toString();
+                mirrorNode = targets[1].getName();
                 mirrorTarget = createSocketAddr(mirrorNode);
                 try {
                   Socket s2 = new Socket();
@@ -699,7 +748,7 @@
          * entire target list, the block, and the data.
          */
         public DataTransfer(DatanodeInfo targets[], Block b) throws IOException {
-            this.curTarget = createSocketAddr(targets[0].getName().toString());
+            this.curTarget = createSocketAddr(targets[0].getName());
             this.targets = targets;
             this.b = b;
             this.buf = new byte[BUFFER_SIZE];
@@ -709,7 +758,7 @@
          * Do the deed, write the bytes
          */
         public void run() {
-	    xmitsInProgress++;
+      xmitsInProgress++;
             try {
                 Socket s = new Socket();
                 s.connect(curTarget, READ_TIMEOUT);
@@ -750,8 +799,8 @@
             } catch (IOException ie) {
               LOG.log(Level.WARNING, "Failed to transfer "+b+" to "+curTarget, ie);
             } finally {
-		xmitsInProgress--;
-	    }
+    xmitsInProgress--;
+      }
         }
     }
 
@@ -821,7 +870,6 @@
     }
   }
 
-
   /**
    * Make an instance of DataNode after ensuring that given data directory
    * (and parent directories, if necessary) can be created.
@@ -847,17 +895,18 @@
   public String toString() {
     return "DataNode{" +
         "data=" + data +
-        ", localName='" + localName + "'" +
+        ", localName='" + dnRegistration.getName() + "'" +
+        ", storageID='" + dnRegistration.getStorageID() + "'" +
         ", xmitsInProgress=" + xmitsInProgress +
         "}";
   }
 
-    /**
-     */
-    public static void main(String args[]) throws IOException {
-        Configuration conf = new Configuration();
-        LogFormatter.setShowThreadIDs(true);
-        LogFormatter.initFileHandler(conf, "datanode");
-        runAndWait(conf);
-    }
+  /**
+   */
+  public static void main(String args[]) throws IOException {
+    Configuration conf = new Configuration();
+    LogFormatter.setShowThreadIDs(true);
+    LogFormatter.initFileHandler(conf, "datanode");
+    runAndWait(conf);
+  }
 }

Added: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataStorage.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataStorage.java?rev=410635&view=auto
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataStorage.java (added)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataStorage.java Wed May 31 11:52:49 2006
@@ -0,0 +1,148 @@
+package org.apache.hadoop.dfs;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.channels.FileLock;
+
+import org.apache.hadoop.io.UTF8;
+
+/** 
+ * Data storage information file.
+ * <p>
+ * During startup the datanode reads its data storage file.
+ * The data storage file is stored in the dfs.data.dir directory.
+ * It contains version and storageID.
+ * Datanode holds a lock on the dataStorage file while it runs so that other 
+ * datanodes were not able to start working with the same data storage.
+ * The lock is released when the datanode stops (normally or abnormally).
+ * 
+ * @author Konstantin Shvachko
+ */
+class DataStorage {
+  public static final String STORAGE_INFO_FILE_NAME = "storage";
+
+  // persistent fields
+  private int version = 0;  /// stored version
+  private String storageID; /// unique per cluster storageID
+  
+  // non persistent fields
+  private RandomAccessFile storageFile = null;
+  private FileLock storageLock = null;
+  
+  /**
+   * Create DataStorage and verify its version.
+   * 
+   * @param datadir data storage directory
+   * @throws IOException
+   */
+  public DataStorage( File datadir ) throws IOException {
+    this( DataNode.DFS_CURRENT_VERSION, datadir );
+    
+    if( version != DataNode.DFS_CURRENT_VERSION )
+      throw new IncorrectVersionException( version, "data storage" );
+  }
+  
+  /**
+   * Create DataStorage.
+   * 
+   * Read data storage file if exists or create it if not.
+   * Lock the file.
+   * 
+   * @param curVersion can be used to read file saved with a previous version.
+   * @param datadir data storage directory
+   * @throws IOException
+   */
+  public DataStorage( int curVersion, File datadir ) throws IOException {
+    this.version = curVersion;
+    storageFile = new RandomAccessFile( 
+                        new File(datadir, STORAGE_INFO_FILE_NAME ), 
+                        "rws" );
+    lock();
+    boolean needToSave;
+    try {
+      needToSave = read();
+    } catch( java.io.EOFException e ) {
+      storageID = "";
+      needToSave = true;
+    }
+    
+    if( needToSave )
+      write();
+  }
+  
+  public int getVersion() {
+    return version;
+  }
+
+  public String getStorageID() {
+    return storageID;
+  }
+  
+  public void setStorageID( String newStorageID ) {
+    this.storageID = newStorageID;
+  }
+  
+  public void setVersion( int newVersion ) {
+    this.version = newVersion;
+  }
+  
+  /**
+   * Lock datastoarge file.
+   * 
+   * @throws IOException
+   */
+  public void lock() throws IOException {
+    storageLock = storageFile.getChannel().tryLock();
+    if( storageLock == null )
+      throw new IOException( "Cannot start multiple Datanode instances "
+                              + "sharing the same data directory.\n" 
+                              + STORAGE_INFO_FILE_NAME + " is locked. ");
+  }
+  
+  /**
+   * Unlock datastoarge file.
+   * 
+   * @throws IOException
+   */
+  public void unlock() throws IOException {
+    storageLock.release();
+  }
+  
+  /**
+   * Close datastoarge file.
+   * 
+   * @throws IOException
+   */
+  public void close() throws IOException {
+    storageLock.release();
+    storageFile.close();
+  }
+  
+  /**
+   * Read data storage file.
+   * 
+   * @return whether the data storage file need to be updated.
+   * @throws IOException
+   */
+  public boolean read() throws IOException {
+    storageFile.seek(0);
+    this.version = storageFile.readInt();
+    UTF8 uID = new UTF8();
+    uID.readFields( storageFile );
+    this.storageID = uID.toString();
+    return false;
+  }
+
+  /**
+   * Write data storage file.
+   * 
+   * @throws IOException
+   */
+  public void write() throws IOException {
+    storageFile.seek(0);
+    storageFile.writeInt( this.version );
+    UTF8 uID = new UTF8( this.storageID );
+    uID.write( storageFile );
+  }
+}

Added: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeID.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeID.java?rev=410635&view=auto
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeID.java (added)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeID.java Wed May 31 11:52:49 2006
@@ -0,0 +1,55 @@
+package org.apache.hadoop.dfs;
+
+/**
+ * DatanodeID is composed of the data node 
+ * name (hostname:portNumber) and the data storage ID, 
+ * which it currently represents.
+ * 
+ * @author Konstantin Shvachko
+ */
+class DatanodeID {
+
+  protected String name;      /// hostname:portNumber
+  protected String storageID; /// unique per cluster storageID
+  
+  /**
+   * Create DatanodeID
+   * 
+   * @param nodeName (hostname:portNumber) 
+   * @param storageID data storage ID
+   */
+  public DatanodeID( String nodeName, String storageID ) {
+    this.name = nodeName;
+    this.storageID = storageID;
+  }
+  
+  /**
+   * @return hostname:portNumber.
+   */
+  public String getName() {
+    return name;
+  }
+  
+  /**
+   * @return data storage ID.
+   */
+  public String getStorageID() {
+    return this.storageID;
+  }
+
+  /**
+   * @return hostname and no :portNumber.
+   */
+  public String getHost() {
+    int colon = name.indexOf(":");
+    if (colon < 0) {
+      return name;
+    } else {
+      return name.substring(0, colon);
+    }
+  }
+  
+  public String toString() {
+    return name;
+  }
+}

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeInfo.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeInfo.java?rev=410635&r1=410634&r2=410635&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeInfo.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeInfo.java Wed May 31 11:52:49 2006
@@ -27,7 +27,7 @@
  *
  * @author Mike Cafarella
  **************************************************/
-class DatanodeInfo implements Writable, Comparable {
+public class DatanodeInfo extends DatanodeID implements Writable, Comparable {
 
     static {                                      // register a ctor
       WritableFactories.setFactory
@@ -37,30 +37,36 @@
          });
     }
 
-    private UTF8 name;
     private long capacityBytes, remainingBytes, lastUpdate;
     private volatile TreeSet blocks;
 
     /** Create an empty DatanodeInfo.
      */
     public DatanodeInfo() {
-        this(new UTF8(), 0, 0);
+        this(new String(), new String(), 0, 0);
     }
 
+    public DatanodeInfo( DatanodeID nodeID ) {
+      this( nodeID.getName(), nodeID.getStorageID(), 0, 0);
+    }
+    
    /**
-    * @param name hostname:portNumber as UTF8 object.
+    * Create an empty DatanodeInfo.
     */
-    public DatanodeInfo(UTF8 name) {
-        this.name = name;
-        this.blocks = new TreeSet();
-        updateHeartbeat(0, 0);        
+    public DatanodeInfo(DatanodeID nodeID, 
+                        long capacity, 
+                        long remaining) {
+      this( nodeID.getName(), nodeID.getStorageID(), capacity, remaining );
     }
 
    /**
-    * @param name hostname:portNumber as UTF8 object.
+    * @param name hostname:portNumber as String object.
     */
-    public DatanodeInfo(UTF8 name, long capacity, long remaining) {
-        this.name = name;
+    public DatanodeInfo(String name, 
+                        String storageID, 
+                        long capacity, 
+                        long remaining) {
+        super( name, storageID );
         this.blocks = new TreeSet();
         updateHeartbeat(capacity, remaining);
     }
@@ -88,28 +94,6 @@
         this.lastUpdate = System.currentTimeMillis();
     }
 
-    /**
-     * @return hostname:portNumber as UTF8 object.
-     */
-    public UTF8 getName() {
-        return name;
-    }
-
-    /**
-     * @return hostname and no :portNumber as UTF8 object.
-     */
-    public UTF8 getHost() {
-        String nameStr = name.toString();
-        int colon = nameStr.indexOf(":");
-        if (colon < 0) {
-            return name;
-        } else {
-            return new UTF8(nameStr.substring(0, colon));
-        }
-    }
-    public String toString() {
-        return name.toString();
-    }
     public Block[] getBlocks() {
         return (Block[]) blocks.toArray(new Block[blocks.size()]);
     }
@@ -127,7 +111,7 @@
     }
 
   /** Comparable.
-   * Basis of compare is the UTF8 name (host:portNumber) only.
+   * Basis of compare is the String name (host:portNumber) only.
    * @param o
    * @return as specified by Comparable.
    */
@@ -142,7 +126,8 @@
     /**
      */
     public void write(DataOutput out) throws IOException {
-        name.write(out);
+        new UTF8( this.name ).write(out);
+        new UTF8( this.storageID ).write(out);
         out.writeLong(capacityBytes);
         out.writeLong(remainingBytes);
         out.writeLong(lastUpdate);
@@ -158,8 +143,11 @@
     /**
      */
     public void readFields(DataInput in) throws IOException {
-        this.name = new UTF8();
-        this.name.readFields(in);
+        UTF8 uStr = new UTF8();
+        uStr.readFields(in);
+        this.name = uStr.toString();
+        uStr.readFields(in);
+        this.storageID = uStr.toString();
         this.capacityBytes = in.readLong();
         this.remainingBytes = in.readLong();
         this.lastUpdate = in.readLong();

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeProtocol.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeProtocol.java?rev=410635&r1=410634&r2=410635&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeProtocol.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeProtocol.java Wed May 31 11:52:49 2006
@@ -28,9 +28,21 @@
  * @author Michael Cafarella
  **********************************************************************/
 interface DatanodeProtocol {
-    // error code
-    final static int DISK_ERROR = 1;
-    final static int INVALID_BLOCK = 2;
+  // error code
+  final static int DISK_ERROR = 1;
+  final static int INVALID_BLOCK = 2;
+  /** 
+   * Register Datanode.
+   *
+   * @see DataNode#register()
+   * @see FSNamesystem#registerDatanode(DatanodeRegistration)
+   * 
+   * @return updated {@link DatanodeRegistration}, which contains 
+   * new storageID if the datanode did not have one and
+   * registration ID for further communication.
+   */
+    public DatanodeRegistration register( DatanodeRegistration registration
+                                        ) throws IOException;
     /**
      * sendHeartbeat() tells the NameNode that the DataNode is still
      * alive and well.  Includes some status info, too. 
@@ -38,9 +50,9 @@
      * A BlockCommand tells the DataNode to invalidate local block(s), 
      * or to copy them to other DataNodes, etc.
      */
-    public BlockCommand sendHeartbeat(String sender, 
-            long capacity, long remaining,
-            int xmitsInProgress) throws IOException;
+    public BlockCommand sendHeartbeat(DatanodeRegistration registration,
+                                      long capacity, long remaining,
+                                      int xmitsInProgress) throws IOException;
 
     /**
      * blockReport() tells the NameNode about all the locally-stored blocks.
@@ -49,7 +61,8 @@
      * the locally-stored blocks.  It's invoked upon startup and then
      * infrequently afterwards.
      */
-    public Block[] blockReport(String sender, Block blocks[]) throws IOException;
+    public Block[] blockReport( DatanodeRegistration registration,
+                                Block blocks[]) throws IOException;
     
     /**
      * blockReceived() allows the DataNode to tell the NameNode about
@@ -57,11 +70,14 @@
      * writes a new Block here, or another DataNode copies a Block to
      * this DataNode, it will call blockReceived().
      */
-    public void blockReceived(String sender, Block blocks[]) throws IOException;
+    public void blockReceived(DatanodeRegistration registration,
+                              Block blocks[]) throws IOException;
 
     /**
      * errorReport() tells the NameNode about something that has gone
      * awry.  Useful for debugging.
      */
-    public void errorReport(String sender, int errorCode, String msg) throws IOException;
+    public void errorReport(DatanodeRegistration registration,
+                            int errorCode, 
+                            String msg) throws IOException;
 }

Added: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeRegistration.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeRegistration.java?rev=410635&view=auto
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeRegistration.java (added)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeRegistration.java Wed May 31 11:52:49 2006
@@ -0,0 +1,87 @@
+package org.apache.hadoop.dfs;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.UTF8;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableFactories;
+import org.apache.hadoop.io.WritableFactory;
+
+/** 
+ * DatanodeRegistration class conatins all information the Namenode needs
+ * to identify and verify a Datanode when it contacts the Namenode.
+ * This information is sent by Datanode with each communication request.
+ * 
+ * @author Konstantin Shvachko
+ */
+class DatanodeRegistration extends DatanodeID implements Writable {
+  static {                                      // register a ctor
+    WritableFactories.setFactory
+      (DatanodeRegistration.class,
+       new WritableFactory() {
+         public Writable newInstance() { return new DatanodeRegistration(); }
+       });
+  }
+
+  int version;            /// current Datanode version
+  String registrationID;  /// a unique per namenode id; indicates   
+                          /// the namenode the datanode is registered with
+
+  /**
+   * Default constructor.
+   */
+  public DatanodeRegistration() {
+    this( 0, null, null, null );
+  }
+  
+  /**
+   * Create DatanodeRegistration
+   */
+  public DatanodeRegistration(int version, 
+                              String nodeName, 
+                              String storageID,
+                              String registrationID ) {
+    super(nodeName, storageID);
+    this.version = version;
+    this.registrationID = registrationID;
+  }
+
+  /**
+   */
+  public int getVersion() {
+    return version;
+  }
+  
+  /**
+   */
+  public String getRegistrationID() {
+    return registrationID;
+  }
+
+  /////////////////////////////////////////////////
+  // Writable
+  /////////////////////////////////////////////////
+  /**
+   */
+  public void write(DataOutput out) throws IOException {
+    out.writeInt(this.version);
+    new UTF8( this.name ).write(out);
+    new UTF8( this.storageID ).write(out);
+    new UTF8( this.registrationID ).write(out);   
+  }
+
+  /**
+   */
+  public void readFields(DataInput in) throws IOException {
+    this.version = in.readInt();
+    UTF8 uStr = new UTF8();
+    uStr.readFields(in);
+    this.name = uStr.toString();
+    uStr.readFields(in);
+    this.storageID = uStr.toString();
+    uStr.readFields(in);
+    this.registrationID = uStr.toString();   
+  }
+}

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DistributedFileSystem.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DistributedFileSystem.java?rev=410635&r1=410634&r2=410635&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DistributedFileSystem.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DistributedFileSystem.java Wed May 31 11:52:49 2006
@@ -133,7 +133,7 @@
         }
 
         DFSFileInfo info[] = dfs.listPaths(getPath(f));
-        return info[0].getLen();
+        return (info == null) ? 0 : info[0].getLen();
     }
 
     public short getReplication(Path f) throws IOException {
@@ -247,8 +247,8 @@
 
       for (int i = 0; i < dnReport.length; i++) {
         reports[i] = new DataNodeReport();
-        reports[i].name = dnReport[i].getName().toString();
-        reports[i].host = dnReport[i].getHost().toString();
+        reports[i].name = dnReport[i].getName();
+        reports[i].host = dnReport[i].getHost();
         reports[i].capacity = dnReport[i].getCapacity();
         reports[i].remaining = dnReport[i].getRemaining();
         reports[i].lastUpdate = dnReport[i].lastUpdate();

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java?rev=410635&r1=410634&r2=410635&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java Wed May 31 11:52:49 2006
@@ -22,7 +22,7 @@
  *
  * @author Mike Cafarella
  ************************************/
-interface FSConstants {
+public interface FSConstants {
     public static int MIN_BLOCKS_FOR_WRITE = 5;
 
     public static final long WRITE_COMPLETE = 0xcafae11a;
@@ -109,5 +109,9 @@
     //TODO mb@media-style.com: should be conf injected?
     public static final int BUFFER_SIZE = new Configuration().getInt("io.file.buffer.size", 4096);
 
+    // Version is reflected in the dfs image and edit log files.
+    // Version is reflected in the data storage file.
+    // Versions are negative.
+    // Decrement DFS_CURRENT_VERSION to define a new version.
+    public static final int DFS_CURRENT_VERSION = -2;
 }
-

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDirectory.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDirectory.java?rev=410635&r1=410634&r2=410635&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDirectory.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDirectory.java Wed May 31 11:52:49 2006
@@ -35,10 +35,6 @@
  * @author Mike Cafarella
  *************************************************/
 class FSDirectory implements FSConstants {
-    // Version is reflected in the dfs image and edit log files.
-    // Versions are negative. 
-    // Decrement DFS_CURRENT_VERSION to define a new version.
-    private static final int DFS_CURRENT_VERSION = -1;
     private static final String FS_IMAGE = "fsimage";
     private static final String NEW_FS_IMAGE = "fsimage.new";
     private static final String OLD_FS_IMAGE = "fsimage.old";
@@ -295,6 +291,7 @@
     TreeMap activeLocks = new TreeMap();
     DataOutputStream editlog = null;
     boolean ready = false;
+    int namespaceID = 0;  /// a persistent attribute of the namespace
 
     /** Access an existing dfs name directory. */
     public FSDirectory(File dir, Configuration conf) throws IOException {
@@ -329,6 +326,27 @@
           throw new IOException("Unable to format: "+dir);
         }
     }
+    
+    /**
+     * Generate new namespaceID.
+     * 
+     * namespaceID is a persistent attribute of the namespace.
+     * It is generated when the namenode is formatted and remains the same
+     * during the life cycle of the namenode.
+     * When a datanodes register they receive it as the registrationID,
+     * which is checked every time the datanode is communicating with the 
+     * namenode. Datanodes that do not 'know' the namespaceID are rejected.
+     * 
+     * @return new namespaceID
+     */
+    private int newNamespaceID() {
+      Random r = new Random();
+      r.setSeed( System.currentTimeMillis() );
+      int newID = 0;
+      while( newID == 0)
+        newID = r.nextInt();
+      return newID;
+    }
 
     /**
      * Shutdown the filestore
@@ -387,11 +405,16 @@
         //
         // Load in bits
         //
+        boolean needToSave = true;
+        int imgVersion = DFS_CURRENT_VERSION;
         if (curFile.exists()) {
             DataInputStream in = new DataInputStream(new BufferedInputStream(new FileInputStream(curFile)));
             try {
-                // read image version
-                int imgVersion = in.readInt();
+                // read image version: first appeared in version -1
+                imgVersion = in.readInt();
+                // read namespaceID: first appeared in version -2
+                if( imgVersion <= -2 )
+                  namespaceID = in.readInt();
                 // read number of files
                 int numFiles = 0;
                 // version 0 does not store version #
@@ -402,6 +425,7 @@
                 } else 
                   numFiles = in.readInt();
                   
+                needToSave = ( imgVersion != DFS_CURRENT_VERSION );
                 if( imgVersion < DFS_CURRENT_VERSION ) // future version
                   throw new IOException(
                               "Unsupported version of the file system image: "
@@ -436,11 +460,10 @@
             }
         }
 
-        if (edits.exists() && loadFSEdits(edits, conf) > 0) {
-            return true;
-        } else {
-            return false;
-        }
+        if( namespaceID == 0 )
+          namespaceID = newNamespaceID();
+        
+        return needToSave || ( edits.exists() && loadFSEdits(edits, conf) > 0 );
     }
 
     /**
@@ -584,6 +607,7 @@
         DataOutputStream out = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(newFile)));
         try {
             out.writeInt(DFS_CURRENT_VERSION);
+            out.writeInt(this.namespaceID);
             out.writeInt(rootDir.numItemsInTree() - 1);
             rootDir.saveImage("", out);
         } finally {
@@ -967,9 +991,9 @@
                 lastSuccess = false;
             }
         }
-        if( !lastSuccess )
+/*        if( !lastSuccess )
             NameNode.stateChangeLog.warning("DIR* FSDirectory.mkdirs: "
-                    +"failed to create directory "+src );
+                    +"failed to create directory "+src );*/
         return lastSuccess;
     }
 

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java?rev=410635&r1=410634&r2=410635&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java Wed May 31 11:52:49 2006
@@ -105,7 +105,7 @@
             } else if (lu1 > lu2) {
                 return 1;
             } else {
-                return d1.getName().compareTo(d2.getName());
+                return d1.getStorageID().compareTo(d2.getStorageID());
             }
         }
     });
@@ -325,7 +325,7 @@
             +src+" for "+holder+" at "+clientMachine);
       try {
         if (pendingCreates.get(src) != null) {
-           throw new NameNode.AlreadyBeingCreatedException(
+           throw new AlreadyBeingCreatedException(
                    "failed to create file " + src + " for " + holder +
                    " on client " + clientMachine + 
                    " because pendingCreates is non-null.");
@@ -410,10 +410,10 @@
           (FileUnderConstruction) pendingCreates.get(src);
         // make sure that we still have the lease on this file
         if (pendingFile == null) {
-          throw new NameNode.LeaseExpiredException("No lease on " + src);
+          throw new LeaseExpiredException("No lease on " + src);
         }
         if (!pendingFile.getClientName().equals(clientName)) {
-          throw new NameNode.LeaseExpiredException("Lease mismatch on " + src + 
+          throw new LeaseExpiredException("Lease mismatch on " + src + 
               " owned by " + pendingFile.getClientName() + 
               " and appended by " + clientName);
         }
@@ -421,6 +421,12 @@
           throw new IOException("File " + src + " created during write");
         }
 
+        //
+        // If we fail this, bad things happen!
+        //
+        if (!checkFileProgress(src)) {
+          throw new NotReplicatedYetException("Not replicated yet");
+        }
         
         // Get the array of replication targets 
         DatanodeInfo targets[] = chooseTargets(pendingFile.getReplication(), 
@@ -660,10 +666,10 @@
                 if (containingNodes != null) {
                     for (Iterator it = containingNodes.iterator(); it.hasNext(); ) {
                         DatanodeInfo node = (DatanodeInfo) it.next();
-                        Vector invalidateSet = (Vector) recentInvalidateSets.get(node.getName());
+                        Vector invalidateSet = (Vector) recentInvalidateSets.get(node.getStorageID());
                         if (invalidateSet == null) {
                             invalidateSet = new Vector();
-                            recentInvalidateSets.put(node.getName(), invalidateSet);
+                            recentInvalidateSets.put(node.getStorageID(), invalidateSet);
                         }
                         invalidateSet.add(b);
                         NameNode.stateChangeLog.finer("BLOCK* NameSystem.delete: "
@@ -755,7 +761,7 @@
                 if (containingNodes != null) {
                   for (Iterator it =containingNodes.iterator(); it.hasNext();) {
                     DatanodeInfo cur = (DatanodeInfo) it.next();
-                    v.add(cur.getHost());
+                    v.add(new UTF8( cur.getHost() ));
                   }
                 }
                 hosts[i-startBlock] = (UTF8[]) v.toArray(new UTF8[v.size()]);
@@ -973,35 +979,149 @@
     //
     /////////////////////////////////////////////////////////
     /**
+     * Register Datanode.
+     * <p>
+     * The purpose of registration is to identify whether the new datanode
+     * serves a new data storage, and will report new data block copies,
+     * which the namenode was not aware of; or the datanode is a replacement
+     * node for the data storage that was previously served by a different
+     * or the same (in terms of host:port) datanode.
+     * The data storages are distinguished by their storageIDs. When a new
+     * data storage is reported the namenode issues a new unique storageID.
+     * <p>
+     * Finally, the namenode returns its namespaceID as the registrationID
+     * for the datanodes. 
+     * namespaceID is a persistent attribute of the name space.
+     * The registrationID is checked every time the datanode is communicating
+     * with the namenode. 
+     * Datanodes with inappropriate registrationID are rejected.
+     * If the namenode stops, and then restarts it can restore its 
+     * namespaceID and will continue serving the datanodes that has previously
+     * registered with the namenode without restarting the whole cluster.
+     * 
+     * @see DataNode#register()
+     * @author Konstantin Shvachko
+     */
+    public synchronized void registerDatanode( DatanodeRegistration nodeReg 
+                                              ) throws IOException {
+      NameNode.stateChangeLog.fine(
+          "BLOCK* NameSystem.registerDatanode: "
+          + "node registration from " + nodeReg.getName()
+          + " storage " + nodeReg.getStorageID() );
+
+      nodeReg.registrationID = getRegistrationID();
+      DatanodeInfo nodeS = (DatanodeInfo)datanodeMap.get(nodeReg.getStorageID());
+      DatanodeInfo nodeN = getDatanodeByName( nodeReg.getName() );
+      
+      if( nodeN != null && nodeS != null && nodeN == nodeS ) {
+        // The same datanode has been just restarted to serve the same data 
+        // storage. We do not need to remove old data blocks, the delta will  
+        // be calculated on the next block report from the datanode
+        NameNode.stateChangeLog.fine(
+            "BLOCK* NameSystem.registerDatanode: "
+            + "node restarted." );
+        return;
+      }
+      
+      if( nodeN != null ) {
+        // nodeN previously served a different data storage, 
+        // which is not served by anybody anymore.
+        removeDatanode( nodeN );
+        nodeN = null;
+      }
+      
+      // nodeN is not found
+      if( nodeS == null ) {
+        // this is a new datanode serving a new data storage
+        if( nodeReg.getStorageID().equals("") ) {
+          // this data storage has never registered
+          // it is either empty or was created by previous version of DFS
+          nodeReg.storageID = newStorageID();
+          NameNode.stateChangeLog.fine(
+              "BLOCK* NameSystem.registerDatanode: "
+              + "new storageID " + nodeReg.getStorageID() + " assigned." );
+        }
+        // register new datanode
+        datanodeMap.put(nodeReg.getStorageID(), 
+                        new DatanodeInfo( nodeReg ) );
+        NameNode.stateChangeLog.fine(
+            "BLOCK* NameSystem.registerDatanode: "
+            + "node registered." );
+        return;
+      }
+
+      // nodeS is found
+      // The registering datanode is a replacement node for the existing 
+      // data storage, which from now on will be served by a new node.
+      NameNode.stateChangeLog.fine(
+          "BLOCK* NameSystem.registerDatanode: "
+          + "node " + nodeS.name
+          + " is replaced by " + nodeReg.getName() + "." );
+      nodeS.name = nodeReg.getName();
+      return;
+    }
+    
+    /**
+     * Get registrationID for datanodes based on the namespaceID.
+     * 
+     * @see #registerDatanode(DatanodeRegistration)
+     * @see FSDirectory#newNamespaceID()
+     * @return registration ID
+     */
+    public String getRegistrationID() {
+      return "NS" + Integer.toString( dir.namespaceID );
+    }
+    
+    /**
+     * Generate new storage ID.
+     * 
+     * @return unique storage ID
+     * 
+     * Note: that collisions are still possible if somebody will try 
+     * to bring in a data storage from a different cluster.
+     */
+    private String newStorageID() {
+      String newID = null;
+      while( newID == null ) {
+        newID = "DS" + Integer.toString( r.nextInt() );
+        if( datanodeMap.get( newID ) != null )
+          newID = null;
+      }
+      return newID;
+    }
+    
+    /**
      * The given node has reported in.  This method should:
      * 1) Record the heartbeat, so the datanode isn't timed out
      * 2) Adjust usage stats for future block allocation
      */
-    public synchronized void gotHeartbeat(UTF8 name, long capacity, long remaining) {
-        synchronized (heartbeats) {
-            synchronized (datanodeMap) {
-                long capacityDiff = 0;
-                long remainingDiff = 0;
-                DatanodeInfo nodeinfo = (DatanodeInfo) datanodeMap.get(name);
-
-                if (nodeinfo == null) {
-                    NameNode.stateChangeLog.fine("BLOCK* NameSystem.gotHeartbeat: "
-                            +"brand-new heartbeat from "+name );
-                     nodeinfo = new DatanodeInfo(name, capacity, remaining);
-                    datanodeMap.put(name, nodeinfo);
-                    capacityDiff = capacity;
-                    remainingDiff = remaining;
-                } else {
-                    capacityDiff = capacity - nodeinfo.getCapacity();
-                    remainingDiff = remaining - nodeinfo.getRemaining();
-                    heartbeats.remove(nodeinfo);
-                    nodeinfo.updateHeartbeat(capacity, remaining);
-                }
-                heartbeats.add(nodeinfo);
-                totalCapacity += capacityDiff;
-                totalRemaining += remainingDiff;
-            }
+    public synchronized void gotHeartbeat(DatanodeID nodeID,
+                                          long capacity, 
+                                          long remaining) throws IOException {
+      synchronized (heartbeats) {
+        synchronized (datanodeMap) {
+          long capacityDiff = 0;
+          long remainingDiff = 0;
+          DatanodeInfo nodeinfo = getDatanode( nodeID );
+
+          if (nodeinfo == null) {
+            NameNode.stateChangeLog.fine("BLOCK* NameSystem.gotHeartbeat: "
+                    +"brand-new heartbeat from "+nodeID.getName() );
+            nodeinfo = new DatanodeInfo(nodeID, capacity, remaining);
+            datanodeMap.put(nodeinfo.getStorageID(), nodeinfo);
+            capacityDiff = capacity;
+            remainingDiff = remaining;
+          } else {
+            capacityDiff = capacity - nodeinfo.getCapacity();
+            remainingDiff = remaining - nodeinfo.getRemaining();
+            heartbeats.remove(nodeinfo);
+            nodeinfo.updateHeartbeat(capacity, remaining);
+          }
+          heartbeats.add(nodeinfo);
+          totalCapacity += capacityDiff;
+          totalRemaining += remainingDiff;
         }
+      }
     }
 
     /**
@@ -1020,77 +1140,73 @@
             }
         }
     }
-    
-    /**
-     * remove a datanode info
-     * @param name: datanode name
-     * @author hairong
-     */
 
-    synchronized void rmDataNodeByName( UTF8 name ) {
-        DatanodeInfo nodeInfo = (DatanodeInfo) datanodeMap.get(name);
-        if (nodeInfo != null) {
-            rmDataNode( nodeInfo );
-        } else {
-            NameNode.stateChangeLog.warning("BLOCK* NameSystem.rmDataNodeByName: "
-                    + nodeInfo.getName() + " does not exist");
-        }
-    }
-    
     /**
      * remove a datanode info
-     * @param nodeInfo: datanode info
+     * @param name: datanode name
      * @author hairong
      */
-    private synchronized void rmDataNode( DatanodeInfo nodeInfo ) {
-        heartbeats.remove( nodeInfo );
-        synchronized (datanodeMap) {
-            datanodeMap.remove(nodeInfo.getName());
-            NameNode.stateChangeLog.finer("BLOCK* NameSystem.heartbeatCheck: "
-                    + nodeInfo.getName() + " is removed from datanodeMap");
-        }
-        totalCapacity -= nodeInfo.getCapacity();
-        totalRemaining -= nodeInfo.getRemaining();
-
-        Block deadblocks[] = nodeInfo.getBlocks();
-        if (deadblocks != null) {
-            for (int i = 0; i < deadblocks.length; i++) {
-                removeStoredBlock(deadblocks[i], nodeInfo);
-            }
-        }
+    synchronized public void removeDatanode( DatanodeID nodeID ) 
+    throws IOException {
+      DatanodeInfo nodeInfo = getDatanode( nodeID );
+      if (nodeInfo != null) {
+        removeDatanode( nodeInfo );
+      } else {
+          NameNode.stateChangeLog.warning("BLOCK* NameSystem.removeDatanode: "
+                  + nodeInfo.getName() + " does not exist");
+      }
+  }
+  
+  /**
+   * remove a datanode info
+   * @param nodeInfo: datanode info
+   * @author hairong
+   */
+    private void removeDatanode( DatanodeInfo nodeInfo ) {
+      heartbeats.remove(nodeInfo);
+      datanodeMap.remove(nodeInfo.getStorageID());
+      NameNode.stateChangeLog.finer("BLOCK* NameSystem.removeDatanode: "
+              + nodeInfo.getName() + " is removed from datanodeMap");
+      totalCapacity -= nodeInfo.getCapacity();
+      totalRemaining -= nodeInfo.getRemaining();
+
+      Block deadblocks[] = nodeInfo.getBlocks();
+      if( deadblocks != null )
+        for( int i = 0; i < deadblocks.length; i++ )
+          removeStoredBlock(deadblocks[i], nodeInfo);
     }
-        
+
     /**
      * Check if there are any expired heartbeats, and if so,
      * whether any blocks have to be re-replicated.
      */
     synchronized void heartbeatCheck() {
-        synchronized (heartbeats) {
-            DatanodeInfo nodeInfo = null;
+      synchronized (heartbeats) {
+        DatanodeInfo nodeInfo = null;
 
-            while ((heartbeats.size() > 0) &&
-                   ((nodeInfo = (DatanodeInfo) heartbeats.first()) != null) &&
-                   (nodeInfo.lastUpdate() < System.currentTimeMillis() - EXPIRE_INTERVAL)) {
-                NameNode.stateChangeLog.info("BLOCK* NameSystem.heartbeatCheck: "
-                           + "lost heartbeat from " + nodeInfo.getName());
-                rmDataNode(nodeInfo);
-            }
+        while ((heartbeats.size() > 0) &&
+               ((nodeInfo = (DatanodeInfo) heartbeats.first()) != null) &&
+               (nodeInfo.lastUpdate() < System.currentTimeMillis() - EXPIRE_INTERVAL)) {
+          NameNode.stateChangeLog.info("BLOCK* NameSystem.heartbeatCheck: "
+              + "lost heartbeat from " + nodeInfo.getName());
+          removeDatanode( nodeInfo );
+          if (heartbeats.size() > 0) {
+              nodeInfo = (DatanodeInfo) heartbeats.first();
+          }
         }
+      }
     }
     
     /**
      * The given node is reporting all its blocks.  Use this info to 
      * update the (machine-->blocklist) and (block-->machinelist) tables.
      */
-    public synchronized Block[] processReport(Block newReport[], UTF8 name) {
+    public synchronized Block[] processReport(DatanodeID nodeID, 
+                                              Block newReport[]
+                                            ) throws IOException {
         NameNode.stateChangeLog.fine("BLOCK* NameSystem.processReport: "
-                +"from "+name+" "+newReport.length+" blocks" );
-        DatanodeInfo node = (DatanodeInfo) datanodeMap.get(name);
-        if (node == null) {
-            NameNode.stateChangeLog.severe("BLOCK* NameSystem.processReport: "
-                    +"from "+name+" but can not find its info" );
-            throw new IllegalArgumentException("Unexpected exception.  Received block report from node " + name + ", but there is no info for " + name);
-        }
+          +"from "+nodeID.getName()+" "+newReport.length+" blocks" );
+        DatanodeInfo node = getDatanode( nodeID );
 
         //
         // Modify the (block-->datanode) map, according to the difference
@@ -1150,7 +1266,7 @@
             if (! dir.isValidBlock(b) && ! pendingCreateBlocks.contains(b)) {
                 obsolete.add(b);
                 NameNode.stateChangeLog.info("BLOCK* NameSystem.processReport: "
-                        +"ask "+name+" to delete "+b.getBlockName() );
+                        +"ask "+nodeID.getName()+" to delete "+b.getBlockName() );
             }
         }
         return (Block[]) obsolete.toArray(new Block[obsolete.size()]);
@@ -1215,7 +1331,7 @@
       Vector nonExcess = new Vector();
       for (Iterator it = containingNodes.iterator(); it.hasNext(); ) {
           DatanodeInfo cur = (DatanodeInfo) it.next();
-          TreeSet excessBlocks = (TreeSet) excessReplicateMap.get(cur.getName());
+          TreeSet excessBlocks = (TreeSet) excessReplicateMap.get(cur.getStorageID());
           if (excessBlocks == null || ! excessBlocks.contains(block)) {
               nonExcess.add(cur);
           }
@@ -1238,10 +1354,10 @@
             DatanodeInfo cur = (DatanodeInfo) nonExcess.elementAt(chosenNode);
             nonExcess.removeElementAt(chosenNode);
 
-            TreeSet excessBlocks = (TreeSet) excessReplicateMap.get(cur.getName());
+            TreeSet excessBlocks = (TreeSet) excessReplicateMap.get(cur.getStorageID());
             if (excessBlocks == null) {
                 excessBlocks = new TreeSet();
-                excessReplicateMap.put(cur.getName(), excessBlocks);
+                excessReplicateMap.put(cur.getStorageID(), excessBlocks);
             }
             excessBlocks.add(b);
             NameNode.stateChangeLog.finer("BLOCK* NameSystem.chooseExcessReplicates: "
@@ -1256,10 +1372,10 @@
             // should be deleted.  Items are removed from the invalidate list
             // upon giving instructions to the namenode.
             //
-            Vector invalidateSet = (Vector) recentInvalidateSets.get(cur.getName());
+            Vector invalidateSet = (Vector) recentInvalidateSets.get(cur.getStorageID());
             if (invalidateSet == null) {
                 invalidateSet = new Vector();
-                recentInvalidateSets.put(cur.getName(), invalidateSet);
+                recentInvalidateSets.put(cur.getStorageID(), invalidateSet);
             }
             invalidateSet.add(b);
             NameNode.stateChangeLog.finer("BLOCK* NameSystem.chooseExcessReplicates: "
@@ -1299,13 +1415,13 @@
         // We've removed a block from a node, so it's definitely no longer
         // in "excess" there.
         //
-        TreeSet excessBlocks = (TreeSet) excessReplicateMap.get(node.getName());
+        TreeSet excessBlocks = (TreeSet) excessReplicateMap.get(node.getStorageID());
         if (excessBlocks != null) {
             excessBlocks.remove(block);
             NameNode.stateChangeLog.finer("BLOCK* NameSystem.removeStoredBlock: "
                     +block.getBlockName()+" is removed from excessBlocks" );
             if (excessBlocks.size() == 0) {
-                excessReplicateMap.remove(node.getName());
+                excessReplicateMap.remove(node.getStorageID());
             }
         }
     }
@@ -1313,15 +1429,20 @@
     /**
      * The given node is reporting that it received a certain block.
      */
-    public synchronized void blockReceived(Block block, UTF8 name) {
-        DatanodeInfo node = (DatanodeInfo) datanodeMap.get(name);
+    public synchronized void blockReceived( DatanodeID nodeID,  
+                                            Block block
+                                          ) throws IOException {
+        DatanodeInfo node = getDatanode( nodeID );
         if (node == null) {
             NameNode.stateChangeLog.warning("BLOCK* NameSystem.blockReceived: "
-                    +block.getBlockName()+" is received from an unrecorded node " + name );
-            throw new IllegalArgumentException("Unexpected exception.  Got blockReceived message from node " + name + ", but there is no info for " + name);
+             + block.getBlockName() + " is received from an unrecorded node " 
+             + nodeID.getName() );
+            throw new IllegalArgumentException(
+                "Unexpected exception.  Got blockReceived message from node " 
+                + block.getBlockName() + ", but there is no info for it");
         }
         NameNode.stateChangeLog.fine("BLOCK* NameSystem.blockReceived: "
-                +block.getBlockName()+" is received from " + name );
+                +block.getBlockName()+" is received from " + nodeID.getName() );
         //
         // Modify the blocks->datanode map
         // 
@@ -1374,8 +1495,9 @@
     /**
      * Check if there are any recently-deleted blocks a datanode should remove.
      */
-    public synchronized Block[] blocksToInvalidate(UTF8 sender) {
-        Vector invalidateSet = (Vector) recentInvalidateSets.remove(sender);
+    public synchronized Block[] blocksToInvalidate( DatanodeID nodeID ) {
+        Vector invalidateSet = (Vector) recentInvalidateSets.remove( 
+                                                      nodeID.getStorageID() );
  
         if (invalidateSet == null ) 
             return null;
@@ -1387,7 +1509,7 @@
                 blockList.append(((Block)invalidateSet.elementAt(i)).getBlockName());
             }
             NameNode.stateChangeLog.info("BLOCK* NameSystem.blockToInvalidate: "
-                   +"ask "+sender+" to delete " + blockList );
+                   +"ask "+nodeID.getName()+" to delete " + blockList );
         }
         return (Block[]) invalidateSet.toArray(new Block[invalidateSet.size()]);
     }
@@ -1402,114 +1524,117 @@
      *     target sequence for the Block at the appropriate index.
      *
      */
-    public synchronized Object[] pendingTransfers(DatanodeInfo srcNode, int xmitsInProgress) {
-        synchronized (neededReplications) {
-            Object results[] = null;
-            int scheduledXfers = 0;
-
-            if (neededReplications.size() > 0) {
-                //
-                // Go through all blocks that need replications.  See if any
-                // are present at the current node.  If so, ask the node to
-                // replicate them.
-                //
-                Vector replicateBlocks = new Vector();
-                Vector replicateTargetSets = new Vector();
-                for (Iterator it = neededReplications.iterator(); it.hasNext(); ) {
-                    //
-                    // We can only reply with 'maxXfers' or fewer blocks
-                    //
-                    if (scheduledXfers >= this.maxReplicationStreams - xmitsInProgress) {
-                        break;
-                    }
-
-                    Block block = (Block) it.next();
-                    long blockSize = block.getNumBytes();
-                    FSDirectory.INode fileINode = dir.getFileByBlock(block);
-                    if( fileINode == null ) { // block does not belong to any file
-                        it.remove();
-                    } else {
-                        TreeSet containingNodes = (TreeSet) blocksMap.get(block);
-                        TreeSet excessBlocks = (TreeSet) excessReplicateMap.get(srcNode.getName());
-                        // srcNode must contain the block, and the block must 
-                        // not be scheduled for removal on that node
-                        if (containingNodes.contains(srcNode) 
-                              && ( excessBlocks == null 
-                               || ! excessBlocks.contains(block))) {
-                            DatanodeInfo targets[] = 
-                              chooseTargets(Math.min(fileINode.getReplication() 
-                                                       - containingNodes.size(), 
-                                                     maxReplicationStreams
-                                                       - xmitsInProgress), 
-                                                     containingNodes, null, 
-                                                     blockSize);
-                            if (targets.length > 0) {
-                                // Build items to return
-                                replicateBlocks.add(block);
-                                replicateTargetSets.add(targets);
-                                scheduledXfers += targets.length;
-                            }
-                        }
-                    }
-                }
+    public synchronized Object[] pendingTransfers(DatanodeInfo srcNode,
+                                                  int xmitsInProgress) {
+    synchronized (neededReplications) {
+      Object results[] = null;
+      int scheduledXfers = 0;
+
+      if (neededReplications.size() > 0) {
+        //
+        // Go through all blocks that need replications. See if any
+        // are present at the current node. If so, ask the node to
+        // replicate them.
+        //
+        Vector replicateBlocks = new Vector();
+        Vector replicateTargetSets = new Vector();
+        for (Iterator it = neededReplications.iterator(); it.hasNext();) {
+          //
+          // We can only reply with 'maxXfers' or fewer blocks
+          //
+          if (scheduledXfers >= this.maxReplicationStreams - xmitsInProgress) {
+            break;
+          }
 
-                //
-                // Move the block-replication into a "pending" state.
-                // The reason we use 'pending' is so we can retry
-                // replications that fail after an appropriate amount of time.  
-                // (REMIND - mjc - this timer is not yet implemented.)
-                //
-                if (replicateBlocks.size() > 0) {
-                    int i = 0;
-                    for (Iterator it = replicateBlocks.iterator(); it.hasNext(); i++) {
-                        Block block = (Block) it.next();
-                        DatanodeInfo targets[] = (DatanodeInfo[]) replicateTargetSets.elementAt(i);
-                        TreeSet containingNodes = (TreeSet) blocksMap.get(block);
-
-                        if (containingNodes.size() + targets.length >= dir.getFileByBlock(block).getReplication()) {
-                            neededReplications.remove(block);
-                            pendingReplications.add(block);
-                            NameNode.stateChangeLog.finer("BLOCK* NameSystem.pendingTransfer: "
-                                    +block.getBlockName()
-                                    +" is removed from neededReplications to pendingReplications" );
-                        }
+          Block block = (Block) it.next();
+          long blockSize = block.getNumBytes();
+          FSDirectory.INode fileINode = dir.getFileByBlock(block);
+          if (fileINode == null) { // block does not belong to any file
+            it.remove();
+          } else {
+            TreeSet containingNodes = (TreeSet) blocksMap.get(block);
+            TreeSet excessBlocks = (TreeSet) excessReplicateMap.get( 
+                                                      srcNode.getStorageID() );
+            // srcNode must contain the block, and the block must
+            // not be scheduled for removal on that node
+            if (containingNodes.contains(srcNode)
+                && (excessBlocks == null || ! excessBlocks.contains(block))) {
+              DatanodeInfo targets[] = chooseTargets(
+                  Math.min( fileINode.getReplication() - containingNodes.size(),
+                            this.maxReplicationStreams - xmitsInProgress), 
+                  containingNodes, null, blockSize);
+              if (targets.length > 0) {
+                // Build items to return
+                replicateBlocks.add(block);
+                replicateTargetSets.add(targets);
+                scheduledXfers += targets.length;
+              }
+            }
+          }
+        }
 
-                        if(NameNode.stateChangeLog.isLoggable(Level.INFO)) {
-                            StringBuffer targetList = new StringBuffer( "datanode(s)");
-                            for(int k=0; k<targets.length; k++) {
-                               targetList.append(' ');
-                               targetList.append(targets[k].getName());
-                            }
-                            NameNode.stateChangeLog.info("BLOCK* NameSystem.pendingTransfer: "
-                                    +"ask "+srcNode.getName()
-                                    +" to replicate "+block.getBlockName()
-                                    +" to "+targetList);
-                        }
-                    }
+        //
+        // Move the block-replication into a "pending" state.
+        // The reason we use 'pending' is so we can retry
+        // replications that fail after an appropriate amount of time.
+        // (REMIND - mjc - this timer is not yet implemented.)
+        //
+        if (replicateBlocks.size() > 0) {
+          int i = 0;
+          for (Iterator it = replicateBlocks.iterator(); it.hasNext(); i++) {
+            Block block = (Block) it.next();
+            DatanodeInfo targets[] = 
+                      (DatanodeInfo[]) replicateTargetSets.elementAt(i);
+            TreeSet containingNodes = (TreeSet) blocksMap.get(block);
+
+            if (containingNodes.size() + targets.length >= 
+                    dir.getFileByBlock( block).getReplication() ) {
+              neededReplications.remove(block);
+              pendingReplications.add(block);
+              NameNode.stateChangeLog.finer(
+                "BLOCK* NameSystem.pendingTransfer: "
+                + block.getBlockName()
+                + " is removed from neededReplications to pendingReplications");
+            }
+
+            if (NameNode.stateChangeLog.isLoggable(Level.INFO)) {
+              StringBuffer targetList = new StringBuffer("datanode(s)");
+              for (int k = 0; k < targets.length; k++) {
+                targetList.append(' ');
+                targetList.append(targets[k].getName());
+              }
+              NameNode.stateChangeLog.info(
+                      "BLOCK* NameSystem.pendingTransfer: " + "ask "
+                      + srcNode.getName() + " to replicate "
+                      + block.getBlockName() + " to " + targetList);
+            }
+          }
 
-                    //
-                    // Build returned objects from above lists
-                    //
-                    DatanodeInfo targetMatrix[][] = new DatanodeInfo[replicateTargetSets.size()][];
-                    for (i = 0; i < targetMatrix.length; i++) {
-                        targetMatrix[i] = (DatanodeInfo[]) replicateTargetSets.elementAt(i);
-                    }
+          //
+          // Build returned objects from above lists
+          //
+          DatanodeInfo targetMatrix[][] = 
+                        new DatanodeInfo[replicateTargetSets.size()][];
+          for (i = 0; i < targetMatrix.length; i++) {
+            targetMatrix[i] = (DatanodeInfo[]) replicateTargetSets.elementAt(i);
+          }
 
-                    results = new Object[2];
-                    results[0] = replicateBlocks.toArray(new Block[replicateBlocks.size()]);
-                    results[1]  = targetMatrix;
-                }
-            }
-            return results;
+          results = new Object[2];
+          results[0] = replicateBlocks.toArray(new Block[replicateBlocks.size()]);
+          results[1] = targetMatrix;
         }
+      }
+      return results;
     }
+  }
 
     /**
      * Get a certain number of targets, if possible.
      * If not, return as many as we can.
-     * @param desiredReplicates number of duplicates wanted.
-     * @param forbiddenNodes of DatanodeInfo instances that should not be
-     * considered targets.
+     * @param desiredReplicates
+     *          number of duplicates wanted.
+     * @param forbiddenNodes
+     *          of DatanodeInfo instances that should not be considered targets.
      * @return array of DatanodeInfo instances uses as targets.
      */
     DatanodeInfo[] chooseTargets(int desiredReplicates, TreeSet forbiddenNodes,
@@ -1684,5 +1809,46 @@
       public UTF8 getClientMachine() {
         return clientMachine;
       }
+    }
+
+    /**
+     * Get data node by storage ID.
+     * 
+     * @param nodeID
+     * @return DatanodeInfo or null if the node is not found.
+     * @throws IOException
+     */
+    public DatanodeInfo getDatanode( DatanodeID nodeID ) throws IOException {
+      UnregisteredDatanodeException e = null;
+      DatanodeInfo node = (DatanodeInfo) datanodeMap.get(nodeID.getStorageID());
+      if (node == null) 
+        return null;
+      if (!node.getName().equals(nodeID.getName())) {
+        e = new UnregisteredDatanodeException( nodeID, node );
+        NameNode.stateChangeLog.severe("BLOCK* NameSystem.getDatanode: "
+            + e.getLocalizedMessage() );
+        throw e;
+      }
+      return node;
+    }
+    
+    /**
+     * Find data node by its name.
+     * 
+     * This method is called when the node is registering.
+     * Not performance critical.
+     * Otherwise an additional tree-like structure will be required.
+     * 
+     * @param name
+     * @return DatanodeInfo if found or null otherwise 
+     * @throws IOException
+     */
+    public DatanodeInfo getDatanodeByName( String name ) throws IOException {
+      for (Iterator it = datanodeMap.values().iterator(); it.hasNext(); ) {
+        DatanodeInfo node = (DatanodeInfo) it.next();
+        if( node.getName().equals(name) )
+           return node;
+      }
+      return null;
     }
 }

Added: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/IncorrectVersionException.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/IncorrectVersionException.java?rev=410635&view=auto
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/IncorrectVersionException.java (added)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/IncorrectVersionException.java Wed May 31 11:52:49 2006
@@ -0,0 +1,19 @@
+package org.apache.hadoop.dfs;
+
+import java.io.IOException;
+
+/**
+ * The exception is thrown when external version does not match 
+ * current version of the appication.
+ * 
+ * @author Konstantin Shvachko
+ */
+class IncorrectVersionException extends IOException {
+
+  public IncorrectVersionException( int version, String ofWhat ) {
+    super( "Unexpected version " 
+        + (ofWhat==null ? "" : "of " + ofWhat) + " reported: "
+        + version + ". Expecting = " + FSConstants.DFS_CURRENT_VERSION + "." );
+  }
+
+}

Added: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/LeaseExpiredException.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/LeaseExpiredException.java?rev=410635&view=auto
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/LeaseExpiredException.java (added)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/LeaseExpiredException.java Wed May 31 11:52:49 2006
@@ -0,0 +1,13 @@
+package org.apache.hadoop.dfs;
+
+import java.io.IOException;
+
+/**
+ * The lease that was being used to create this file has expired.
+ * @author Owen O'Malley
+ */
+public class LeaseExpiredException extends IOException {
+  public LeaseExpiredException(String msg) {
+    super(msg);
+  }
+}



Mime
View raw message