开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > Internet/网络编程 > 网格计算 > 查看源码
DataNode.java
资源名称：hadoop-0.20.0.tar.gz　[点击查看]
上传用户：quxuerui
上传日期：2018-01-08
资源大小：41811k
文件大小：58k
源码类别：

网格计算

开发平台：

Java

DataNode.java：源码内容
  1. /**
  2.  * Licensed to the Apache Software Foundation (ASF) under one
  3.  * or more contributor license agreements.  See the NOTICE file
  4.  * distributed with this work for additional information
  5.  * regarding copyright ownership.  The ASF licenses this file
  6.  * to you under the Apache License, Version 2.0 (the
  7.  * "License"); you may not use this file except in compliance
  8.  * with the License.  You may obtain a copy of the License at
  9.  *
  10.  *     http://www.apache.org/licenses/LICENSE-2.0
  11.  *
  12.  * Unless required by applicable law or agreed to in writing, software
  13.  * distributed under the License is distributed on an "AS IS" BASIS,
  14.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15.  * See the License for the specific language governing permissions and
  16.  * limitations under the License.
  17.  */
  18. package org.apache.hadoop.hdfs.server.datanode;
  20. import java.io.BufferedOutputStream;
  21. import java.io.DataOutputStream;
  22. import java.io.File;
  23. import java.io.IOException;
  24. import java.io.OutputStream;
  25. import java.net.InetSocketAddress;
  26. import java.net.ServerSocket;
  27. import java.net.Socket;
  28. import java.net.SocketTimeoutException;
  29. import java.net.UnknownHostException;
  30. import java.nio.channels.ServerSocketChannel;
  31. import java.nio.channels.SocketChannel;
  32. import java.security.NoSuchAlgorithmException;
  33. import java.security.SecureRandom;
  34. import java.util.AbstractList;
  35. import java.util.ArrayList;
  36. import java.util.Arrays;
  37. import java.util.HashMap;
  38. import java.util.LinkedList;
  39. import java.util.List;
  40. import java.util.Map;
  41. import java.util.Random;
  42. import java.util.concurrent.atomic.AtomicInteger;
  44. import org.apache.commons.logging.Log;
  45. import org.apache.commons.logging.LogFactory;
  46. import org.apache.hadoop.conf.Configuration;
  47. import org.apache.hadoop.conf.Configured;
  48. import org.apache.hadoop.hdfs.HDFSPolicyProvider;
  49. import org.apache.hadoop.hdfs.protocol.Block;
  50. import org.apache.hadoop.hdfs.protocol.BlockListAsLongs;
  51. import org.apache.hadoop.hdfs.protocol.ClientDatanodeProtocol;
  52. import org.apache.hadoop.hdfs.protocol.DataTransferProtocol;
  53. import org.apache.hadoop.hdfs.protocol.DatanodeID;
  54. import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
  55. import org.apache.hadoop.hdfs.protocol.FSConstants;
  56. import org.apache.hadoop.hdfs.protocol.LocatedBlock;
  57. import org.apache.hadoop.hdfs.protocol.UnregisteredDatanodeException;
  58. import org.apache.hadoop.hdfs.server.common.HdfsConstants.StartupOption;
  59. import org.apache.hadoop.hdfs.server.common.HdfsConstants;
  60. import org.apache.hadoop.hdfs.server.common.GenerationStamp;
  61. import org.apache.hadoop.hdfs.server.common.IncorrectVersionException;
  62. import org.apache.hadoop.hdfs.server.common.Storage;
  63. import org.apache.hadoop.hdfs.server.datanode.metrics.DataNodeMetrics;
  64. import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
  65. import org.apache.hadoop.hdfs.server.namenode.FileChecksumServlets;
  66. import org.apache.hadoop.hdfs.server.namenode.NameNode;
  67. import org.apache.hadoop.hdfs.server.namenode.StreamFile;
  68. import org.apache.hadoop.hdfs.server.protocol.BlockCommand;
  69. import org.apache.hadoop.hdfs.server.protocol.BlockMetaDataInfo;
  70. import org.apache.hadoop.hdfs.server.protocol.DatanodeCommand;
  71. import org.apache.hadoop.hdfs.server.protocol.DatanodeProtocol;
  72. import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration;
  73. import org.apache.hadoop.hdfs.server.protocol.DisallowedDatanodeException;
  74. import org.apache.hadoop.hdfs.server.protocol.InterDatanodeProtocol;
  75. import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo;
  76. import org.apache.hadoop.hdfs.server.protocol.UpgradeCommand;
  77. import org.apache.hadoop.http.HttpServer;
  78. import org.apache.hadoop.io.IOUtils;
  79. import org.apache.hadoop.io.Text;
  80. import org.apache.hadoop.ipc.RPC;
  81. import org.apache.hadoop.ipc.RemoteException;
  82. import org.apache.hadoop.ipc.Server;
  83. import org.apache.hadoop.net.DNS;
  84. import org.apache.hadoop.net.NetUtils;
  85. import org.apache.hadoop.security.SecurityUtil;
  86. import org.apache.hadoop.security.authorize.ConfiguredPolicy;
  87. import org.apache.hadoop.security.authorize.PolicyProvider;
  88. import org.apache.hadoop.security.authorize.ServiceAuthorizationManager;
  89. import org.apache.hadoop.util.Daemon;
  90. import org.apache.hadoop.util.DiskChecker;
  91. import org.apache.hadoop.util.ReflectionUtils;
  92. import org.apache.hadoop.util.StringUtils;
  93. import org.apache.hadoop.util.DiskChecker.DiskErrorException;
  94. import org.apache.hadoop.util.DiskChecker.DiskOutOfSpaceException;
  96. /**********************************************************
  97.  * DataNode is a class (and program) that stores a set of
  98.  * blocks for a DFS deployment.  A single deployment can
  99.  * have one or many DataNodes.  Each DataNode communicates
  100.  * regularly with a single NameNode.  It also communicates
  101.  * with client code and other DataNodes from time to time.
  102.  *
  103.  * DataNodes store a series of named blocks.  The DataNode
  104.  * allows client code to read these blocks, or to write new
  105.  * block data.  The DataNode may also, in response to instructions
  106.  * from its NameNode, delete blocks or copy blocks to/from other
  107.  * DataNodes.
  108.  *
  109.  * The DataNode maintains just one critical table:
  110.  *   block-> stream of bytes (of BLOCK_SIZE or less)
  111.  *
  112.  * This info is stored on a local disk.  The DataNode
  113.  * reports the table's contents to the NameNode upon startup
  114.  * and every so often afterwards.
  115.  *
  116.  * DataNodes spend their lives in an endless loop of asking
  117.  * the NameNode for something to do.  A NameNode cannot connect
  118.  * to a DataNode directly; a NameNode simply returns values from
  119.  * functions invoked by a DataNode.
  120.  *
  121.  * DataNodes maintain an open server socket so that client code 
  122.  * or other DataNodes can read/write data.  The host/port for
  123.  * this server is reported to the NameNode, which then sends that
  124.  * information to clients or other DataNodes that might be interested.
  125.  *
  126.  **********************************************************/
  127. public class DataNode extends Configured 
  128.     implements InterDatanodeProtocol, ClientDatanodeProtocol, FSConstants, Runnable {
  129.   public static final Log LOG = LogFactory.getLog(DataNode.class);
  130.   
  131.   static{
  132.     Configuration.addDefaultResource("hdfs-default.xml");
  133.     Configuration.addDefaultResource("hdfs-site.xml");
  134.   }
  136.   public static final String DN_CLIENTTRACE_FORMAT =
  137.         "src: %s" +      // src IP
  138.         ", dest: %s" +   // dst IP
  139.         ", bytes: %s" +  // byte count
  140.         ", op: %s" +     // operation
  141.         ", cliID: %s" +  // DFSClient id
  142.         ", srvID: %s" +  // DatanodeRegistration
  143.         ", blockid: %s"; // block id
  144.   static final Log ClientTraceLog =
  145.     LogFactory.getLog(DataNode.class.getName() + ".clienttrace");
  147.   /**
  148.    * Use {@link NetUtils#createSocketAddr(String)} instead.
  149.    */
  150.   @Deprecated
  151.   public static InetSocketAddress createSocketAddr(String target
  152.                                                    ) throws IOException {
  153.     return NetUtils.createSocketAddr(target);
  154.   }
  155.   
  156.   public DatanodeProtocol namenode = null;
  157.   public FSDatasetInterface data = null;
  158.   public DatanodeRegistration dnRegistration = null;
  160.   volatile boolean shouldRun = true;
  161.   private LinkedList<Block> receivedBlockList = new LinkedList<Block>();
  162.   /** list of blocks being recovered */
  163.   private final Map<Block, Block> ongoingRecovery = new HashMap<Block, Block>();
  164.   private LinkedList<String> delHints = new LinkedList<String>();
  165.   public final static String EMPTY_DEL_HINT = "";
  166.   AtomicInteger xmitsInProgress = new AtomicInteger();
  167.   Daemon dataXceiverServer = null;
  168.   ThreadGroup threadGroup = null;
  169.   long blockReportInterval;
  170.   //disallow the sending of BR before instructed to do so
  171.   long lastBlockReport = 0;
  172.   boolean resetBlockReportTime = true;
  173.   long initialBlockReportDelay = BLOCKREPORT_INITIAL_DELAY * 1000L;
  174.   long lastHeartbeat = 0;
  175.   long heartBeatInterval;
  176.   private DataStorage storage = null;
  177.   private HttpServer infoServer = null;
  178.   DataNodeMetrics myMetrics;
  179.   private static InetSocketAddress nameNodeAddr;
  180.   private InetSocketAddress selfAddr;
  181.   private static DataNode datanodeObject = null;
  182.   private Thread dataNodeThread = null;
  183.   String machineName;
  184.   private static String dnThreadName;
  185.   int socketTimeout;
  186.   int socketWriteTimeout = 0;  
  187.   boolean transferToAllowed = true;
  188.   int writePacketSize = 0;
  189.   
  190.   public DataBlockScanner blockScanner = null;
  191.   public Daemon blockScannerThread = null;
  192.   
  193.   private static final Random R = new Random();
  194.   
  195.   // For InterDataNodeProtocol
  196.   public Server ipcServer;
  198.   /**
  199.    * Current system time.
  200.    * @return current time in msec.
  201.    */
  202.   static long now() {
  203.     return System.currentTimeMillis();
  204.   }
  206.   /**
  207.    * Create the DataNode given a configuration and an array of dataDirs.
  208.    * 'dataDirs' is where the blocks are stored.
  209.    */
  210.   DataNode(Configuration conf, 
  211.            AbstractList<File> dataDirs) throws IOException {
  212.     super(conf);
  213.     datanodeObject = this;
  215.     try {
  216.       startDataNode(conf, dataDirs);
  217.     } catch (IOException ie) {
  218.       shutdown();
  219.       throw ie;
  220.     }
  221.   }
  222.     
  223.   
  224.   /**
  225.    * This method starts the data node with the specified conf.
  226.    * 
  227.    * @param conf - the configuration
  228.    *  if conf's CONFIG_PROPERTY_SIMULATED property is set
  229.    *  then a simulated storage based data node is created.
  230.    * 
  231.    * @param dataDirs - only for a non-simulated storage data node
  232.    * @throws IOException
  233.    */
  234.   void startDataNode(Configuration conf, 
  235.                      AbstractList<File> dataDirs
  236.                      ) throws IOException {
  237.     // use configured nameserver & interface to get local hostname
  238.     if (conf.get("slave.host.name") != null) {
  239.       machineName = conf.get("slave.host.name");   
  240.     }
  241.     if (machineName == null) {
  242.       machineName = DNS.getDefaultHost(
  243.                                      conf.get("dfs.datanode.dns.interface","default"),
  244.                                      conf.get("dfs.datanode.dns.nameserver","default"));
  245.     }
  246.     InetSocketAddress nameNodeAddr = NameNode.getAddress(conf);
  247.     
  248.     this.socketTimeout =  conf.getInt("dfs.socket.timeout",
  249.                                       HdfsConstants.READ_TIMEOUT);
  250.     this.socketWriteTimeout = conf.getInt("dfs.datanode.socket.write.timeout",
  251.                                           HdfsConstants.WRITE_TIMEOUT);
  252.     /* Based on results on different platforms, we might need set the default 
  253.      * to false on some of them. */
  254.     this.transferToAllowed = conf.getBoolean("dfs.datanode.transferTo.allowed", 
  255.                                              true);
  256.     this.writePacketSize = conf.getInt("dfs.write.packet.size", 64*1024);
  257.     String address = 
  258.       NetUtils.getServerAddress(conf,
  259.                                 "dfs.datanode.bindAddress", 
  260.                                 "dfs.datanode.port",
  261.                                 "dfs.datanode.address");
  262.     InetSocketAddress socAddr = NetUtils.createSocketAddr(address);
  263.     int tmpPort = socAddr.getPort();
  264.     storage = new DataStorage();
  265.     // construct registration
  266.     this.dnRegistration = new DatanodeRegistration(machineName + ":" + tmpPort);
  268.     // connect to name node
  269.     this.namenode = (DatanodeProtocol) 
  270.       RPC.waitForProxy(DatanodeProtocol.class,
  271.                        DatanodeProtocol.versionID,
  272.                        nameNodeAddr, 
  273.                        conf);
  274.     // get version and id info from the name-node
  275.     NamespaceInfo nsInfo = handshake();
  276.     StartupOption startOpt = getStartupOption(conf);
  277.     assert startOpt != null : "Startup option must be set.";
  278.     
  279.     boolean simulatedFSDataset = 
  280.         conf.getBoolean("dfs.datanode.simulateddatastorage", false);
  281.     if (simulatedFSDataset) {
  282.         setNewStorageID(dnRegistration);
  283.         dnRegistration.storageInfo.layoutVersion = FSConstants.LAYOUT_VERSION;
  284.         dnRegistration.storageInfo.namespaceID = nsInfo.namespaceID;
  285.         // it would have been better to pass storage as a parameter to
  286.         // constructor below - need to augment ReflectionUtils used below.
  287.         conf.set("StorageId", dnRegistration.getStorageID());
  288.         try {
  289.           //Equivalent of following (can't do because Simulated is in test dir)
  290.           //  this.data = new SimulatedFSDataset(conf);
  291.           this.data = (FSDatasetInterface) ReflectionUtils.newInstance(
  292.               Class.forName("org.apache.hadoop.hdfs.server.datanode.SimulatedFSDataset"), conf);
  293.         } catch (ClassNotFoundException e) {
  294.           throw new IOException(StringUtils.stringifyException(e));
  295.         }
  296.     } else { // real storage
  297.       // read storage info, lock data dirs and transition fs state if necessary
  298.       storage.recoverTransitionRead(nsInfo, dataDirs, startOpt);
  299.       // adjust
  300.       this.dnRegistration.setStorageInfo(storage);
  301.       // initialize data node internal structure
  302.       this.data = new FSDataset(storage, conf);
  303.     }
  305.       
  306.     // find free port
  307.     ServerSocket ss = (socketWriteTimeout > 0) ? 
  308.           ServerSocketChannel.open().socket() : new ServerSocket();
  309.     Server.bind(ss, socAddr, 0);
  310.     ss.setReceiveBufferSize(DEFAULT_DATA_SOCKET_SIZE); 
  311.     // adjust machine name with the actual port
  312.     tmpPort = ss.getLocalPort();
  313.     selfAddr = new InetSocketAddress(ss.getInetAddress().getHostAddress(),
  314.                                      tmpPort);
  315.     this.dnRegistration.setName(machineName + ":" + tmpPort);
  316.     LOG.info("Opened info server at " + tmpPort);
  317.       
  318.     this.threadGroup = new ThreadGroup("dataXceiverServer");
  319.     this.dataXceiverServer = new Daemon(threadGroup, 
  320.         new DataXceiverServer(ss, conf, this));
  321.     this.threadGroup.setDaemon(true); // auto destroy when empty
  323.     this.blockReportInterval =
  324.       conf.getLong("dfs.blockreport.intervalMsec", BLOCKREPORT_INTERVAL);
  325.     this.initialBlockReportDelay = conf.getLong("dfs.blockreport.initialDelay",
  326.                                             BLOCKREPORT_INITIAL_DELAY)* 1000L; 
  327.     if (this.initialBlockReportDelay >= blockReportInterval) {
  328.       this.initialBlockReportDelay = 0;
  329.       LOG.info("dfs.blockreport.initialDelay is greater than " +
  330.         "dfs.blockreport.intervalMsec." + " Setting initial delay to 0 msec:");
  331.     }
  332.     this.heartBeatInterval = conf.getLong("dfs.heartbeat.interval", HEARTBEAT_INTERVAL) * 1000L;
  333.     DataNode.nameNodeAddr = nameNodeAddr;
  335.     //initialize periodic block scanner
  336.     String reason = null;
  337.     if (conf.getInt("dfs.datanode.scan.period.hours", 0) < 0) {
  338.       reason = "verification is turned off by configuration";
  339.     } else if ( !(data instanceof FSDataset) ) {
  340.       reason = "verifcation is supported only with FSDataset";
  341.     } 
  342.     if ( reason == null ) {
  343.       blockScanner = new DataBlockScanner(this, (FSDataset)data, conf);
  344.     } else {
  345.       LOG.info("Periodic Block Verification is disabled because " +
  346.                reason + ".");
  347.     }
  349.     //create a servlet to serve full-file content
  350.     String infoAddr = 
  351.       NetUtils.getServerAddress(conf, 
  352.                               "dfs.datanode.info.bindAddress", 
  353.                               "dfs.datanode.info.port",
  354.                               "dfs.datanode.http.address");
  355.     InetSocketAddress infoSocAddr = NetUtils.createSocketAddr(infoAddr);
  356.     String infoHost = infoSocAddr.getHostName();
  357.     int tmpInfoPort = infoSocAddr.getPort();
  358.     this.infoServer = new HttpServer("datanode", infoHost, tmpInfoPort,
  359.         tmpInfoPort == 0, conf);
  360.     if (conf.getBoolean("dfs.https.enable", false)) {
  361.       boolean needClientAuth = conf.getBoolean("dfs.https.need.client.auth", false);
  362.       InetSocketAddress secInfoSocAddr = NetUtils.createSocketAddr(conf.get(
  363.           "dfs.datanode.https.address", infoHost + ":" + 0));
  364.       Configuration sslConf = new Configuration(false);
  365.       sslConf.addResource(conf.get("dfs.https.server.keystore.resource",
  366.           "ssl-server.xml"));
  367.       this.infoServer.addSslListener(secInfoSocAddr, sslConf, needClientAuth);
  368.     }
  369.     this.infoServer.addInternalServlet(null, "/streamFile/*", StreamFile.class);
  370.     this.infoServer.addInternalServlet(null, "/getFileChecksum/*",
  371.         FileChecksumServlets.GetServlet.class);
  372.     this.infoServer.setAttribute("datanode.blockScanner", blockScanner);
  373.     this.infoServer.addServlet(null, "/blockScannerReport", 
  374.                                DataBlockScanner.Servlet.class);
  375.     this.infoServer.start();
  376.     // adjust info port
  377.     this.dnRegistration.setInfoPort(this.infoServer.getPort());
  378.     myMetrics = new DataNodeMetrics(conf, dnRegistration.getStorageID());
  379.     
  380.     // set service-level authorization security policy
  381.     if (conf.getBoolean(
  382.           ServiceAuthorizationManager.SERVICE_AUTHORIZATION_CONFIG, false)) {
  383.       PolicyProvider policyProvider = 
  384.         (PolicyProvider)(ReflectionUtils.newInstance(
  385.             conf.getClass(PolicyProvider.POLICY_PROVIDER_CONFIG, 
  386.                 HDFSPolicyProvider.class, PolicyProvider.class), 
  387.             conf));
  388.       SecurityUtil.setPolicy(new ConfiguredPolicy(conf, policyProvider));
  389.     }
  391.     //init ipc server
  392.     InetSocketAddress ipcAddr = NetUtils.createSocketAddr(
  393.         conf.get("dfs.datanode.ipc.address"));
  394.     ipcServer = RPC.getServer(this, ipcAddr.getHostName(), ipcAddr.getPort(), 
  395.         conf.getInt("dfs.datanode.handler.count", 3), false, conf);
  396.     ipcServer.start();
  397.     dnRegistration.setIpcPort(ipcServer.getListenerAddress().getPort());
  399.     LOG.info("dnRegistration = " + dnRegistration);
  400.   }
  402.   /**
  403.    * Creates either NIO or regular depending on socketWriteTimeout.
  404.    */
  405.   protected Socket newSocket() throws IOException {
  406.     return (socketWriteTimeout > 0) ? 
  407.            SocketChannel.open().socket() : new Socket();                                   
  408.   }
  409.   
  410.   private NamespaceInfo handshake() throws IOException {
  411.     NamespaceInfo nsInfo = new NamespaceInfo();
  412.     while (shouldRun) {
  413.       try {
  414.         nsInfo = namenode.versionRequest();
  415.         break;
  416.       } catch(SocketTimeoutException e) {  // namenode is busy
  417.         LOG.info("Problem connecting to server: " + getNameNodeAddr());
  418.         try {
  419.           Thread.sleep(1000);
  420.         } catch (InterruptedException ie) {}
  421.       }
  422.     }
  423.     String errorMsg = null;
  424.     // verify build version
  425.     if( ! nsInfo.getBuildVersion().equals( Storage.getBuildVersion() )) {
  426.       errorMsg = "Incompatible build versions: namenode BV = " 
  427.         + nsInfo.getBuildVersion() + "; datanode BV = "
  428.         + Storage.getBuildVersion();
  429.       LOG.fatal( errorMsg );
  430.       try {
  431.         namenode.errorReport( dnRegistration,
  432.                               DatanodeProtocol.NOTIFY, errorMsg );
  433.       } catch( SocketTimeoutException e ) {  // namenode is busy
  434.         LOG.info("Problem connecting to server: " + getNameNodeAddr());
  435.       }
  436.       throw new IOException( errorMsg );
  437.     }
  438.     assert FSConstants.LAYOUT_VERSION == nsInfo.getLayoutVersion() :
  439.       "Data-node and name-node layout versions must be the same."
  440.       + "Expected: "+ FSConstants.LAYOUT_VERSION + " actual "+ nsInfo.getLayoutVersion();
  441.     return nsInfo;
  442.   }
  444.   /** Return the DataNode object
  445.    * 
  446.    */
  447.   public static DataNode getDataNode() {
  448.     return datanodeObject;
  449.   } 
  451.   public static InterDatanodeProtocol createInterDataNodeProtocolProxy(
  452.       DatanodeID datanodeid, Configuration conf) throws IOException {
  453.     InetSocketAddress addr = NetUtils.createSocketAddr(
  454.         datanodeid.getHost() + ":" + datanodeid.getIpcPort());
  455.     if (InterDatanodeProtocol.LOG.isDebugEnabled()) {
  456.       InterDatanodeProtocol.LOG.info("InterDatanodeProtocol addr=" + addr);
  457.     }
  458.     return (InterDatanodeProtocol)RPC.getProxy(InterDatanodeProtocol.class,
  459.         InterDatanodeProtocol.versionID, addr, conf);
  460.   }
  462.   public InetSocketAddress getNameNodeAddr() {
  463.     return nameNodeAddr;
  464.   }
  465.   
  466.   public InetSocketAddress getSelfAddr() {
  467.     return selfAddr;
  468.   }
  469.     
  470.   DataNodeMetrics getMetrics() {
  471.     return myMetrics;
  472.   }
  473.   
  474.   /**
  475.    * Return the namenode's identifier
  476.    */
  477.   public String getNamenode() {
  478.     //return namenode.toString();
  479.     return "<namenode>";
  480.   }
  482.   public static void setNewStorageID(DatanodeRegistration dnReg) {
  483.     /* Return 
  484.      * "DS-randInt-ipaddr-currentTimeMillis"
  485.      * It is considered extermely rare for all these numbers to match
  486.      * on a different machine accidentally for the following 
  487.      * a) SecureRandom(INT_MAX) is pretty much random (1 in 2 billion), and
  488.      * b) Good chance ip address would be different, and
  489.      * c) Even on the same machine, Datanode is designed to use different ports.
  490.      * d) Good chance that these are started at different times.
  491.      * For a confict to occur all the 4 above have to match!.
  492.      * The format of this string can be changed anytime in future without
  493.      * affecting its functionality.
  494.      */
  495.     String ip = "unknownIP";
  496.     try {
  497.       ip = DNS.getDefaultIP("default");
  498.     } catch (UnknownHostException ignored) {
  499.       LOG.warn("Could not find ip address of "default" inteface.");
  500.     }
  501.     
  502.     int rand = 0;
  503.     try {
  504.       rand = SecureRandom.getInstance("SHA1PRNG").nextInt(Integer.MAX_VALUE);
  505.     } catch (NoSuchAlgorithmException e) {
  506.       LOG.warn("Could not use SecureRandom");
  507.       rand = R.nextInt(Integer.MAX_VALUE);
  508.     }
  509.     dnReg.storageID = "DS-" + rand + "-"+ ip + "-" + dnReg.getPort() + "-" + 
  510.                       System.currentTimeMillis();
  511.   }
  512.   /**
  513.    * Register datanode
  514.    * <p>
  515.    * The datanode needs to register with the namenode on startup in order
  516.    * 1) to report which storage it is serving now and 
  517.    * 2) to receive a registrationID 
  518.    * issued by the namenode to recognize registered datanodes.
  519.    * 
  520.    * @see FSNamesystem#registerDatanode(DatanodeRegistration)
  521.    * @throws IOException
  522.    */
  523.   private void register() throws IOException {
  524.     if (dnRegistration.getStorageID().equals("")) {
  525.       setNewStorageID(dnRegistration);
  526.     }
  527.     while(shouldRun) {
  528.       try {
  529.         // reset name to machineName. Mainly for web interface.
  530.         dnRegistration.name = machineName + ":" + dnRegistration.getPort();
  531.         dnRegistration = namenode.register(dnRegistration);
  532.         break;
  533.       } catch(SocketTimeoutException e) {  // namenode is busy
  534.         LOG.info("Problem connecting to server: " + getNameNodeAddr());
  535.         try {
  536.           Thread.sleep(1000);
  537.         } catch (InterruptedException ie) {}
  538.       }
  539.     }
  540.     assert ("".equals(storage.getStorageID()) 
  541.             && !"".equals(dnRegistration.getStorageID()))
  542.             || storage.getStorageID().equals(dnRegistration.getStorageID()) :
  543.             "New storageID can be assigned only if data-node is not formatted";
  544.     if (storage.getStorageID().equals("")) {
  545.       storage.setStorageID(dnRegistration.getStorageID());
  546.       storage.writeAll();
  547.       LOG.info("New storage id " + dnRegistration.getStorageID()
  548.           + " is assigned to data-node " + dnRegistration.getName());
  549.     }
  550.     if(! storage.getStorageID().equals(dnRegistration.getStorageID())) {
  551.       throw new IOException("Inconsistent storage IDs. Name-node returned "
  552.           + dnRegistration.getStorageID() 
  553.           + ". Expecting " + storage.getStorageID());
  554.     }
  555.     
  556.     // random short delay - helps scatter the BR from all DNs
  557.     scheduleBlockReport(initialBlockReportDelay);
  558.   }
  560.   /**
  561.    * Shut down this instance of the datanode.
  562.    * Returns only after shutdown is complete.
  563.    * This method can only be called by the offerService thread.
  564.    * Otherwise, deadlock might occur.
  565.    */
  566.   public void shutdown() {
  567.     if (infoServer != null) {
  568.       try {
  569.         infoServer.stop();
  570.       } catch (Exception e) {
  571.         LOG.warn("Exception shutting down DataNode", e);
  572.       }
  573.     }
  574.     if (ipcServer != null) {
  575.       ipcServer.stop();
  576.     }
  577.     this.shouldRun = false;
  578.     if (dataXceiverServer != null) {
  579.       ((DataXceiverServer) this.dataXceiverServer.getRunnable()).kill();
  580.       this.dataXceiverServer.interrupt();
  582.       // wait for all data receiver threads to exit
  583.       if (this.threadGroup != null) {
  584.         while (true) {
  585.           this.threadGroup.interrupt();
  586.           LOG.info("Waiting for threadgroup to exit, active threads is " +
  587.                    this.threadGroup.activeCount());
  588.           if (this.threadGroup.activeCount() == 0) {
  589.             break;
  590.           }
  591.           try {
  592.             Thread.sleep(1000);
  593.           } catch (InterruptedException e) {}
  594.         }
  595.       }
  596.       // wait for dataXceiveServer to terminate
  597.       try {
  598.         this.dataXceiverServer.join();
  599.       } catch (InterruptedException ie) {
  600.       }
  601.     }
  602.     
  603.     RPC.stopProxy(namenode); // stop the RPC threads
  604.     
  605.     if(upgradeManager != null)
  606.       upgradeManager.shutdownUpgrade();
  607.     if (blockScannerThread != null) { 
  608.       blockScannerThread.interrupt();
  609.       try {
  610.         blockScannerThread.join(3600000L); // wait for at most 1 hour
  611.       } catch (InterruptedException ie) {
  612.       }
  613.     }
  614.     if (storage != null) {
  615.       try {
  616.         this.storage.unlockAll();
  617.       } catch (IOException ie) {
  618.       }
  619.     }
  620.     if (dataNodeThread != null) {
  621.       dataNodeThread.interrupt();
  622.       try {
  623.         dataNodeThread.join();
  624.       } catch (InterruptedException ie) {
  625.       }
  626.     }
  627.     if (data != null) {
  628.       data.shutdown();
  629.     }
  630.     if (myMetrics != null) {
  631.       myMetrics.shutdown();
  632.     }
  633.   }
  634.   
  635.   
  636.   /* Check if there is no space in disk or the disk is read-only
  637.    *  when IOException occurs. 
  638.    * If so, handle the error */
  639.   protected void checkDiskError( IOException e ) throws IOException {
  640.     if (e.getMessage() != null && 
  641.         e.getMessage().startsWith("No space left on device")) {
  642.       throw new DiskOutOfSpaceException("No space left on device");
  643.     } else {
  644.       checkDiskError();
  645.     }
  646.   }
  647.   
  648.   /* Check if there is no disk space and if so, handle the error*/
  649.   protected void checkDiskError( ) throws IOException {
  650.     try {
  651.       data.checkDataDir();
  652.     } catch(DiskErrorException de) {
  653.       handleDiskError(de.getMessage());
  654.     }
  655.   }
  656.   
  657.   private void handleDiskError(String errMsgr) {
  658.     LOG.warn("DataNode is shutting down.n" + errMsgr);
  659.     shouldRun = false;
  660.     try {
  661.       namenode.errorReport(
  662.                            dnRegistration, DatanodeProtocol.DISK_ERROR, errMsgr);
  663.     } catch(IOException ignored) {              
  664.     }
  665.   }
  666.     
  667.   /** Number of concurrent xceivers per node. */
  668.   int getXceiverCount() {
  669.     return threadGroup == null ? 0 : threadGroup.activeCount();
  670.   }
  671.     
  672.   /**
  673.    * Main loop for the DataNode.  Runs until shutdown,
  674.    * forever calling remote NameNode functions.
  675.    */
  676.   public void offerService() throws Exception {
  677.      
  678.     LOG.info("using BLOCKREPORT_INTERVAL of " + blockReportInterval + "msec" + 
  679.        " Initial delay: " + initialBlockReportDelay + "msec");
  681.     //
  682.     // Now loop for a long time....
  683.     //
  685.     while (shouldRun) {
  686.       try {
  687.         long startTime = now();
  689.         //
  690.         // Every so often, send heartbeat or block-report
  691.         //
  692.         
  693.         if (startTime - lastHeartbeat > heartBeatInterval) {
  694.           //
  695.           // All heartbeat messages include following info:
  696.           // -- Datanode name
  697.           // -- data transfer port
  698.           // -- Total capacity
  699.           // -- Bytes remaining
  700.           //
  701.           lastHeartbeat = startTime;
  702.           DatanodeCommand[] cmds = namenode.sendHeartbeat(dnRegistration,
  703.                                                        data.getCapacity(),
  704.                                                        data.getDfsUsed(),
  705.                                                        data.getRemaining(),
  706.                                                        xmitsInProgress.get(),
  707.                                                        getXceiverCount());
  708.           myMetrics.heartbeats.inc(now() - startTime);
  709.           //LOG.info("Just sent heartbeat, with name " + localName);
  710.           if (!processCommand(cmds))
  711.             continue;
  712.         }
  713.             
  714.         // check if there are newly received blocks
  715.         Block [] blockArray=null;
  716.         String [] delHintArray=null;
  717.         synchronized(receivedBlockList) {
  718.           synchronized(delHints) {
  719.             int numBlocks = receivedBlockList.size();
  720.             if (numBlocks > 0) {
  721.               if(numBlocks!=delHints.size()) {
  722.                 LOG.warn("Panic: receiveBlockList and delHints are not of the same length" );
  723.               }
  724.               //
  725.               // Send newly-received blockids to namenode
  726.               //
  727.               blockArray = receivedBlockList.toArray(new Block[numBlocks]);
  728.               delHintArray = delHints.toArray(new String[numBlocks]);
  729.             }
  730.           }
  731.         }
  732.         if (blockArray != null) {
  733.           if(delHintArray == null || delHintArray.length != blockArray.length ) {
  734.             LOG.warn("Panic: block array & delHintArray are not the same" );
  735.           }
  736.           namenode.blockReceived(dnRegistration, blockArray, delHintArray);
  737.           synchronized (receivedBlockList) {
  738.             synchronized (delHints) {
  739.               for(int i=0; i<blockArray.length; i++) {
  740.                 receivedBlockList.remove(blockArray[i]);
  741.                 delHints.remove(delHintArray[i]);
  742.               }
  743.             }
  744.           }
  745.         }
  747.         // send block report
  748.         if (startTime - lastBlockReport > blockReportInterval) {
  749.           //
  750.           // Send latest blockinfo report if timer has expired.
  751.           // Get back a list of local block(s) that are obsolete
  752.           // and can be safely GC'ed.
  753.           //
  754.           long brStartTime = now();
  755.           Block[] bReport = data.getBlockReport();
  756.           DatanodeCommand cmd = namenode.blockReport(dnRegistration,
  757.                   BlockListAsLongs.convertToArrayLongs(bReport));
  758.           long brTime = now() - brStartTime;
  759.           myMetrics.blockReports.inc(brTime);
  760.           LOG.info("BlockReport of " + bReport.length +
  761.               " blocks got processed in " + brTime + " msecs");
  762.           //
  763.           // If we have sent the first block report, then wait a random
  764.           // time before we start the periodic block reports.
  765.           //
  766.           if (resetBlockReportTime) {
  767.             lastBlockReport = startTime - R.nextInt((int)(blockReportInterval));
  768.             resetBlockReportTime = false;
  769.           } else {
  770.             /* say the last block report was at 8:20:14. The current report 
  771.              * should have started around 9:20:14 (default 1 hour interval). 
  772.              * If current time is :
  773.              *   1) normal like 9:20:18, next report should be at 10:20:14
  774.              *   2) unexpected like 11:35:43, next report should be at 12:20:14
  775.              */
  776.             lastBlockReport += (now() - lastBlockReport) / 
  777.                                blockReportInterval * blockReportInterval;
  778.           }
  779.           processCommand(cmd);
  780.         }
  782.         // start block scanner
  783.         if (blockScanner != null && blockScannerThread == null &&
  784.             upgradeManager.isUpgradeCompleted()) {
  785.           LOG.info("Starting Periodic block scanner.");
  786.           blockScannerThread = new Daemon(blockScanner);
  787.           blockScannerThread.start();
  788.         }
  789.             
  790.         //
  791.         // There is no work to do;  sleep until hearbeat timer elapses, 
  792.         // or work arrives, and then iterate again.
  793.         //
  794.         long waitTime = heartBeatInterval - (System.currentTimeMillis() - lastHeartbeat);
  795.         synchronized(receivedBlockList) {
  796.           if (waitTime > 0 && receivedBlockList.size() == 0) {
  797.             try {
  798.               receivedBlockList.wait(waitTime);
  799.             } catch (InterruptedException ie) {
  800.             }
  801.           }
  802.         } // synchronized
  803.       } catch(RemoteException re) {
  804.         String reClass = re.getClassName();
  805.         if (UnregisteredDatanodeException.class.getName().equals(reClass) ||
  806.             DisallowedDatanodeException.class.getName().equals(reClass) ||
  807.             IncorrectVersionException.class.getName().equals(reClass)) {
  808.           LOG.warn("DataNode is shutting down: " + 
  809.                    StringUtils.stringifyException(re));
  810.           shutdown();
  811.           return;
  812.         }
  813.         LOG.warn(StringUtils.stringifyException(re));
  814.       } catch (IOException e) {
  815.         LOG.warn(StringUtils.stringifyException(e));
  816.       }
  817.     } // while (shouldRun)
  818.   } // offerService
  820.   /**
  821.    * Process an array of datanode commands
  822.    * 
  823.    * @param cmds an array of datanode commands
  824.    * @return true if further processing may be required or false otherwise. 
  825.    */
  826.   private boolean processCommand(DatanodeCommand[] cmds) {
  827.     if (cmds != null) {
  828.       for (DatanodeCommand cmd : cmds) {
  829.         try {
  830.           if (processCommand(cmd) == false) {
  831.             return false;
  832.           }
  833.         } catch (IOException ioe) {
  834.           LOG.warn("Error processing datanode Command", ioe);
  835.         }
  836.       }
  837.     }
  838.     return true;
  839.   }
  840.   
  841.     /**
  842.      * 
  843.      * @param cmd
  844.      * @return true if further processing may be required or false otherwise. 
  845.      * @throws IOException
  846.      */
  847.   private boolean processCommand(DatanodeCommand cmd) throws IOException {
  848.     if (cmd == null)
  849.       return true;
  850.     final BlockCommand bcmd = cmd instanceof BlockCommand? (BlockCommand)cmd: null;
  852.     switch(cmd.getAction()) {
  853.     case DatanodeProtocol.DNA_TRANSFER:
  854.       // Send a copy of a block to another datanode
  855.       transferBlocks(bcmd.getBlocks(), bcmd.getTargets());
  856.       myMetrics.blocksReplicated.inc(bcmd.getBlocks().length);
  857.       break;
  858.     case DatanodeProtocol.DNA_INVALIDATE:
  859.       //
  860.       // Some local block(s) are obsolete and can be 
  861.       // safely garbage-collected.
  862.       //
  863.       Block toDelete[] = bcmd.getBlocks();
  864.       try {
  865.         if (blockScanner != null) {
  866.           blockScanner.deleteBlocks(toDelete);
  867.         }
  868.         data.invalidate(toDelete);
  869.       } catch(IOException e) {
  870.         checkDiskError();
  871.         throw e;
  872.       }
  873.       myMetrics.blocksRemoved.inc(toDelete.length);
  874.       break;
  875.     case DatanodeProtocol.DNA_SHUTDOWN:
  876.       // shut down the data node
  877.       this.shutdown();
  878.       return false;
  879.     case DatanodeProtocol.DNA_REGISTER:
  880.       // namenode requested a registration - at start or if NN lost contact
  881.       LOG.info("DatanodeCommand action: DNA_REGISTER");
  882.       if (shouldRun) {
  883.         register();
  884.       }
  885.       break;
  886.     case DatanodeProtocol.DNA_FINALIZE:
  887.       storage.finalizeUpgrade();
  888.       break;
  889.     case UpgradeCommand.UC_ACTION_START_UPGRADE:
  890.       // start distributed upgrade here
  891.       processDistributedUpgradeCommand((UpgradeCommand)cmd);
  892.       break;
  893.     case DatanodeProtocol.DNA_RECOVERBLOCK:
  894.       recoverBlocks(bcmd.getBlocks(), bcmd.getTargets());
  895.       break;
  896.     default:
  897.       LOG.warn("Unknown DatanodeCommand action: " + cmd.getAction());
  898.     }
  899.     return true;
  900.   }
  902.   // Distributed upgrade manager
  903.   UpgradeManagerDatanode upgradeManager = new UpgradeManagerDatanode(this);
  905.   private void processDistributedUpgradeCommand(UpgradeCommand comm
  906.                                                ) throws IOException {
  907.     assert upgradeManager != null : "DataNode.upgradeManager is null.";
  908.     upgradeManager.processUpgradeCommand(comm);
  909.   }
  912.   /**
  913.    * Start distributed upgrade if it should be initiated by the data-node.
  914.    */
  915.   private void startDistributedUpgradeIfNeeded() throws IOException {
  916.     UpgradeManagerDatanode um = DataNode.getDataNode().upgradeManager;
  917.     assert um != null : "DataNode.upgradeManager is null.";
  918.     if(!um.getUpgradeState())
  919.       return;
  920.     um.setUpgradeState(false, um.getUpgradeVersion());
  921.     um.startUpgrade();
  922.     return;
  923.   }
  925.   private void transferBlock( Block block, 
  926.                               DatanodeInfo xferTargets[] 
  927.                               ) throws IOException {
  928.     if (!data.isValidBlock(block)) {
  929.       // block does not exist or is under-construction
  930.       String errStr = "Can't send invalid block " + block;
  931.       LOG.info(errStr);
  932.       namenode.errorReport(dnRegistration, 
  933.                            DatanodeProtocol.INVALID_BLOCK, 
  934.                            errStr);
  935.       return;
  936.     }
  938.     // Check if NN recorded length matches on-disk length 
  939.     long onDiskLength = data.getLength(block);
  940.     if (block.getNumBytes() > onDiskLength) {
  941.       // Shorter on-disk len indicates corruption so report NN the corrupt block
  942.       namenode.reportBadBlocks(new LocatedBlock[]{
  943.           new LocatedBlock(block, new DatanodeInfo[] {
  944.               new DatanodeInfo(dnRegistration)})});
  945.       LOG.info("Can't replicate block " + block
  946.           + " because on-disk length " + onDiskLength 
  947.           + " is shorter than NameNode recorded length " + block.getNumBytes());
  948.       return;
  949.     }
  950.     
  951.     int numTargets = xferTargets.length;
  952.     if (numTargets > 0) {
  953.       if (LOG.isInfoEnabled()) {
  954.         StringBuilder xfersBuilder = new StringBuilder();
  955.         for (int i = 0; i < numTargets; i++) {
  956.           xfersBuilder.append(xferTargets[i].getName());
  957.           xfersBuilder.append(" ");
  958.         }
  959.         LOG.info(dnRegistration + " Starting thread to transfer block " + 
  960.                  block + " to " + xfersBuilder);                       
  961.       }
  963.       new Daemon(new DataTransfer(xferTargets, block, this)).start();
  964.     }
  965.   }
  967.   private void transferBlocks( Block blocks[], 
  968.                                DatanodeInfo xferTargets[][] 
  969.                                ) {
  970.     for (int i = 0; i < blocks.length; i++) {
  971.       try {
  972.         transferBlock(blocks[i], xferTargets[i]);
  973.       } catch (IOException ie) {
  974.         LOG.warn("Failed to transfer block " + blocks[i], ie);
  975.       }
  976.     }
  977.   }
  979.   /*
  980.    * Informing the name node could take a long long time! Should we wait
  981.    * till namenode is informed before responding with success to the
  982.    * client? For now we don't.
  983.    */
  984.   protected void notifyNamenodeReceivedBlock(Block block, String delHint) {
  985.     if(block==null || delHint==null) {
  986.       throw new IllegalArgumentException(block==null?"Block is null":"delHint is null");
  987.     }
  988.     synchronized (receivedBlockList) {
  989.       synchronized (delHints) {
  990.         receivedBlockList.add(block);
  991.         delHints.add(delHint);
  992.         receivedBlockList.notifyAll();
  993.       }
  994.     }
  995.   }
  997.   
  1000.   /* ********************************************************************
  1001.   Protocol when a client reads data from Datanode (Cur Ver: 9):
  1002.   
  1003.   Client's Request :
  1004.   =================
  1005.    
  1006.      Processed in DataXceiver:
  1007.      +----------------------------------------------+
  1008.      | Common Header   | 1 byte OP == OP_READ_BLOCK |
  1009.      +----------------------------------------------+
  1010.      
  1011.      Processed in readBlock() :
  1012.      +-------------------------------------------------------------------------+
  1013.      | 8 byte Block ID | 8 byte genstamp | 8 byte start offset | 8 byte length |
  1014.      +-------------------------------------------------------------------------+
  1015.      |   vInt length   |  <DFSClient id> |
  1016.      +-----------------------------------+
  1017.      
  1018.      Client sends optional response only at the end of receiving data.
  1019.        
  1020.   DataNode Response :
  1021.   ===================
  1022.    
  1023.     In readBlock() :
  1024.     If there is an error while initializing BlockSender :
  1025.        +---------------------------+
  1026.        | 2 byte OP_STATUS_ERROR    | and connection will be closed.
  1027.        +---------------------------+
  1028.     Otherwise
  1029.        +---------------------------+
  1030.        | 2 byte OP_STATUS_SUCCESS  |
  1031.        +---------------------------+
  1032.        
  1033.     Actual data, sent by BlockSender.sendBlock() :
  1034.     
  1035.       ChecksumHeader :
  1036.       +--------------------------------------------------+
  1037.       | 1 byte CHECKSUM_TYPE | 4 byte BYTES_PER_CHECKSUM |
  1038.       +--------------------------------------------------+
  1039.       Followed by actual data in the form of PACKETS: 
  1040.       +------------------------------------+
  1041.       | Sequence of data PACKETs ....      |
  1042.       +------------------------------------+
  1043.     
  1044.     A "PACKET" is defined further below.
  1045.     
  1046.     The client reads data until it receives a packet with 
  1047.     "LastPacketInBlock" set to true or with a zero length. If there is 
  1048.     no checksum error, it replies to DataNode with OP_STATUS_CHECKSUM_OK:
  1049.     
  1050.     Client optional response at the end of data transmission :
  1051.       +------------------------------+
  1052.       | 2 byte OP_STATUS_CHECKSUM_OK |
  1053.       +------------------------------+
  1054.     
  1055.     PACKET : Contains a packet header, checksum and data. Amount of data
  1056.     ======== carried is set by BUFFER_SIZE.
  1057.     
  1058.       +-----------------------------------------------------+
  1059.       | 4 byte packet length (excluding packet header)      |
  1060.       +-----------------------------------------------------+
  1061.       | 8 byte offset in the block | 8 byte sequence number |
  1062.       +-----------------------------------------------------+
  1063.       | 1 byte isLastPacketInBlock                          |
  1064.       +-----------------------------------------------------+
  1065.       | 4 byte Length of actual data                        |
  1066.       +-----------------------------------------------------+
  1067.       | x byte checksum data. x is defined below            |
  1068.       +-----------------------------------------------------+
  1069.       | actual data ......                                  |
  1070.       +-----------------------------------------------------+
  1071.       
  1072.       x = (length of data + BYTE_PER_CHECKSUM - 1)/BYTES_PER_CHECKSUM *
  1073.           CHECKSUM_SIZE
  1074.           
  1075.       CHECKSUM_SIZE depends on CHECKSUM_TYPE (usually, 4 for CRC32)
  1076.       
  1077.       The above packet format is used while writing data to DFS also.
  1078.       Not all the fields might be used while reading.
  1079.     
  1080.    ************************************************************************ */
  1081.   
  1082.   /** Header size for a packet */
  1083.   public static final int PKT_HEADER_LEN = ( 4 + /* Packet payload length */
  1084.                                       8 + /* offset in block */
  1085.                                       8 + /* seqno */
  1086.                                       1   /* isLastPacketInBlock */);
  1087.   
  1090.   /**
  1091.    * Used for transferring a block of data.  This class
  1092.    * sends a piece of data to another DataNode.
  1093.    */
  1094.   class DataTransfer implements Runnable {
  1095.     DatanodeInfo targets[];
  1096.     Block b;
  1097.     DataNode datanode;
  1099.     /**
  1100.      * Connect to the first item in the target list.  Pass along the 
  1101.      * entire target list, the block, and the data.
  1102.      */
  1103.     public DataTransfer(DatanodeInfo targets[], Block b, DataNode datanode) throws IOException {
  1104.       this.targets = targets;
  1105.       this.b = b;
  1106.       this.datanode = datanode;
  1107.     }
  1109.     /**
  1110.      * Do the deed, write the bytes
  1111.      */
  1112.     public void run() {
  1113.       xmitsInProgress.getAndIncrement();
  1114.       Socket sock = null;
  1115.       DataOutputStream out = null;
  1116.       BlockSender blockSender = null;
  1117.       
  1118.       try {
  1119.         InetSocketAddress curTarget = 
  1120.           NetUtils.createSocketAddr(targets[0].getName());
  1121.         sock = newSocket();
  1122.         NetUtils.connect(sock, curTarget, socketTimeout);
  1123.         sock.setSoTimeout(targets.length * socketTimeout);
  1125.         long writeTimeout = socketWriteTimeout + 
  1126.                             HdfsConstants.WRITE_TIMEOUT_EXTENSION * (targets.length-1);
  1127.         OutputStream baseStream = NetUtils.getOutputStream(sock, writeTimeout);
  1128.         out = new DataOutputStream(new BufferedOutputStream(baseStream, 
  1129.                                                             SMALL_BUFFER_SIZE));
  1131.         blockSender = new BlockSender(b, 0, b.getNumBytes(), false, false, false, 
  1132.             datanode);
  1133.         DatanodeInfo srcNode = new DatanodeInfo(dnRegistration);
  1135.         //
  1136.         // Header info
  1137.         //
  1138.         out.writeShort(DataTransferProtocol.DATA_TRANSFER_VERSION);
  1139.         out.writeByte(DataTransferProtocol.OP_WRITE_BLOCK);
  1140.         out.writeLong(b.getBlockId());
  1141.         out.writeLong(b.getGenerationStamp());
  1142.         out.writeInt(0);           // no pipelining
  1143.         out.writeBoolean(false);   // not part of recovery
  1144.         Text.writeString(out, ""); // client
  1145.         out.writeBoolean(true); // sending src node information
  1146.         srcNode.write(out); // Write src node DatanodeInfo
  1147.         // write targets
  1148.         out.writeInt(targets.length - 1);
  1149.         for (int i = 1; i < targets.length; i++) {
  1150.           targets[i].write(out);
  1151.         }
  1152.         // send data & checksum
  1153.         blockSender.sendBlock(out, baseStream, null);
  1155.         // no response necessary
  1156.         LOG.info(dnRegistration + ":Transmitted block " + b + " to " + curTarget);
  1158.       } catch (IOException ie) {
  1159.         LOG.warn(dnRegistration + ":Failed to transfer " + b + " to " + targets[0].getName()
  1160.             + " got " + StringUtils.stringifyException(ie));
  1161.       } finally {
  1162.         xmitsInProgress.getAndDecrement();
  1163.         IOUtils.closeStream(blockSender);
  1164.         IOUtils.closeStream(out);
  1165.         IOUtils.closeSocket(sock);
  1166.       }
  1167.     }
  1168.   }
  1170.   /**
  1171.    * No matter what kind of exception we get, keep retrying to offerService().
  1172.    * That's the loop that connects to the NameNode and provides basic DataNode
  1173.    * functionality.
  1174.    *
  1175.    * Only stop when "shouldRun" is turned off (which can only happen at shutdown).
  1176.    */
  1177.   public void run() {
  1178.     LOG.info(dnRegistration + "In DataNode.run, data = " + data);
  1180.     // start dataXceiveServer
  1181.     dataXceiverServer.start();
  1182.         
  1183.     while (shouldRun) {
  1184.       try {
  1185.         startDistributedUpgradeIfNeeded();
  1186.         offerService();
  1187.       } catch (Exception ex) {
  1188.         LOG.error("Exception: " + StringUtils.stringifyException(ex));
  1189.         if (shouldRun) {
  1190.           try {
  1191.             Thread.sleep(5000);
  1192.           } catch (InterruptedException ie) {
  1193.           }
  1194.         }
  1195.       }
  1196.     }
  1197.         
  1198.     LOG.info(dnRegistration + ":Finishing DataNode in: "+data);
  1199.     shutdown();
  1200.   }
  1201.     
  1202.   /** Start a single datanode daemon and wait for it to finish.
  1203.    *  If this thread is specifically interrupted, it will stop waiting.
  1204.    */
  1205.   public static void runDatanodeDaemon(DataNode dn) throws IOException {
  1206.     if (dn != null) {
  1207.       //register datanode
  1208.       dn.register();
  1209.       dn.dataNodeThread = new Thread(dn, dnThreadName);
  1210.       dn.dataNodeThread.setDaemon(true); // needed for JUnit testing
  1211.       dn.dataNodeThread.start();
  1212.     }
  1213.   }
  1214.   
  1215.   static boolean isDatanodeUp(DataNode dn) {
  1216.     return dn.dataNodeThread != null && dn.dataNodeThread.isAlive();
  1217.   }
  1219.   /** Instantiate a single datanode object. This must be run by invoking
  1220.    *  {@link DataNode#runDatanodeDaemon(DataNode)} subsequently. 
  1221.    */
  1222.   public static DataNode instantiateDataNode(String args[],
  1223.                                       Configuration conf) throws IOException {
  1224.     if (conf == null)
  1225.       conf = new Configuration();
  1226.     if (!parseArguments(args, conf)) {
  1227.       printUsage();
  1228.       return null;
  1229.     }
  1230.     if (conf.get("dfs.network.script") != null) {
  1231.       LOG.error("This configuration for rack identification is not supported" +
  1232.           " anymore. RackID resolution is handled by the NameNode.");
  1233.       System.exit(-1);
  1234.     }
  1235.     String[] dataDirs = conf.getStrings("dfs.data.dir");
  1236.     dnThreadName = "DataNode: [" +
  1237.                         StringUtils.arrayToString(dataDirs) + "]";
  1238.     return makeInstance(dataDirs, conf);
  1239.   }
  1241.   /** Instantiate & Start a single datanode daemon and wait for it to finish.
  1242.    *  If this thread is specifically interrupted, it will stop waiting.
  1243.    */
  1244.   public static DataNode createDataNode(String args[],
  1245.                                  Configuration conf) throws IOException {
  1246.     DataNode dn = instantiateDataNode(args, conf);
  1247.     runDatanodeDaemon(dn);
  1248.     return dn;
  1249.   }
  1251.   void join() {
  1252.     if (dataNodeThread != null) {
  1253.       try {
  1254.         dataNodeThread.join();
  1255.       } catch (InterruptedException e) {}
  1256.     }
  1257.   }
  1259.   /**
  1260.    * Make an instance of DataNode after ensuring that at least one of the
  1261.    * given data directories (and their parent directories, if necessary)
  1262.    * can be created.
  1263.    * @param dataDirs List of directories, where the new DataNode instance should
  1264.    * keep its files.
  1265.    * @param conf Configuration instance to use.
  1266.    * @return DataNode instance for given list of data dirs and conf, or null if
  1267.    * no directory from this directory list can be created.
  1268.    * @throws IOException
  1269.    */
  1270.   public static DataNode makeInstance(String[] dataDirs, Configuration conf)
  1271.     throws IOException {
  1272.     ArrayList<File> dirs = new ArrayList<File>();
  1273.     for (int i = 0; i < dataDirs.length; i++) {
  1274.       File data = new File(dataDirs[i]);
  1275.       try {
  1276.         DiskChecker.checkDir(data);
  1277.         dirs.add(data);
  1278.       } catch(DiskErrorException e) {
  1279.         LOG.warn("Invalid directory in dfs.data.dir: " + e.getMessage());
  1280.       }
  1281.     }
  1282.     if (dirs.size() > 0) 
  1283.       return new DataNode(conf, dirs);
  1284.     LOG.error("All directories in dfs.data.dir are invalid.");
  1285.     return null;
  1286.   }
  1288.   @Override
  1289.   public String toString() {
  1290.     return "DataNode{" +
  1291.       "data=" + data +
  1292.       ", localName='" + dnRegistration.getName() + "'" +
  1293.       ", storageID='" + dnRegistration.getStorageID() + "'" +
  1294.       ", xmitsInProgress=" + xmitsInProgress.get() +
  1295.       "}";
  1296.   }
  1297.   
  1298.   private static void printUsage() {
  1299.     System.err.println("Usage: java DataNode");
  1300.     System.err.println("           [-rollback]");
  1301.   }
  1303.   /**
  1304.    * Parse and verify command line arguments and set configuration parameters.
  1305.    *
  1306.    * @return false if passed argements are incorrect
  1307.    */
  1308.   private static boolean parseArguments(String args[], 
  1309.                                         Configuration conf) {
  1310.     int argsLen = (args == null) ? 0 : args.length;
  1311.     StartupOption startOpt = StartupOption.REGULAR;
  1312.     for(int i=0; i < argsLen; i++) {
  1313.       String cmd = args[i];
  1314.       if ("-r".equalsIgnoreCase(cmd) || "--rack".equalsIgnoreCase(cmd)) {
  1315.         LOG.error("-r, --rack arguments are not supported anymore. RackID " +
  1316.             "resolution is handled by the NameNode.");
  1317.         System.exit(-1);
  1318.       } else if ("-rollback".equalsIgnoreCase(cmd)) {
  1319.         startOpt = StartupOption.ROLLBACK;
  1320.       } else if ("-regular".equalsIgnoreCase(cmd)) {
  1321.         startOpt = StartupOption.REGULAR;
  1322.       } else
  1323.         return false;
  1324.     }
  1325.     setStartupOption(conf, startOpt);
  1326.     return true;
  1327.   }
  1329.   private static void setStartupOption(Configuration conf, StartupOption opt) {
  1330.     conf.set("dfs.datanode.startup", opt.toString());
  1331.   }
  1333.   static StartupOption getStartupOption(Configuration conf) {
  1334.     return StartupOption.valueOf(conf.get("dfs.datanode.startup",
  1335.                                           StartupOption.REGULAR.toString()));
  1336.   }
  1338.   /**
  1339.    * This methods  arranges for the data node to send the block report at the next heartbeat.
  1340.    */
  1341.   public void scheduleBlockReport(long delay) {
  1342.     if (delay > 0) { // send BR after random delay
  1343.       lastBlockReport = System.currentTimeMillis()
  1344.                             - ( blockReportInterval - R.nextInt((int)(delay)));
  1345.     } else { // send at next heartbeat
  1346.       lastBlockReport = lastHeartbeat - blockReportInterval;
  1347.     }
  1348.     resetBlockReportTime = true; // reset future BRs for randomness
  1349.   }
  1350.   
  1351.   
  1352.   /**
  1353.    * This method is used for testing. 
  1354.    * Examples are adding and deleting blocks directly.
  1355.    * The most common usage will be when the data node's storage is similated.
  1356.    * 
  1357.    * @return the fsdataset that stores the blocks
  1358.    */
  1359.   public FSDatasetInterface getFSDataset() {
  1360.     return data;
  1361.   }
  1363.   /**
  1364.    */
  1365.   public static void main(String args[]) {
  1366.     try {
  1367.       StringUtils.startupShutdownMessage(DataNode.class, args, LOG);
  1368.       DataNode datanode = createDataNode(args, null);
  1369.       if (datanode != null)
  1370.         datanode.join();
  1371.     } catch (Throwable e) {
  1372.       LOG.error(StringUtils.stringifyException(e));
  1373.       System.exit(-1);
  1374.     }
  1375.   }
  1377.   // InterDataNodeProtocol implementation
  1378.   /** {@inheritDoc} */
  1379.   public BlockMetaDataInfo getBlockMetaDataInfo(Block block
  1380.       ) throws IOException {
  1381.     if (LOG.isDebugEnabled()) {
  1382.       LOG.debug("block=" + block);
  1383.     }
  1384.     Block stored = data.getStoredBlock(block.getBlockId());
  1386.     if (stored == null) {
  1387.       return null;
  1388.     }
  1389.     BlockMetaDataInfo info = new BlockMetaDataInfo(stored,
  1390.                                  blockScanner.getLastScanTime(stored));
  1391.     if (LOG.isDebugEnabled()) {
  1392.       LOG.debug("getBlockMetaDataInfo successful block=" + stored +
  1393.                 " length " + stored.getNumBytes() +
  1394.                 " genstamp " + stored.getGenerationStamp());
  1395.     }
  1397.     // paranoia! verify that the contents of the stored block
  1398.     // matches the block file on disk.
  1399.     data.validateBlockMetadata(stored);
  1400.     return info;
  1401.   }
  1403.   public Daemon recoverBlocks(final Block[] blocks, final DatanodeInfo[][] targets) {
  1404.     Daemon d = new Daemon(threadGroup, new Runnable() {
  1405.       /** Recover a list of blocks. It is run by the primary datanode. */
  1406.       public void run() {
  1407.         for(int i = 0; i < blocks.length; i++) {
  1408.           try {
  1409.             logRecoverBlock("NameNode", blocks[i], targets[i]);
  1410.             recoverBlock(blocks[i], false, targets[i], true);
  1411.           } catch (IOException e) {
  1412.             LOG.warn("recoverBlocks FAILED, blocks[" + i + "]=" + blocks[i], e);
  1413.           }
  1414.         }
  1415.       }
  1416.     });
  1417.     d.start();
  1418.     return d;
  1419.   }
  1421.   /** {@inheritDoc} */
  1422.   public void updateBlock(Block oldblock, Block newblock, boolean finalize) throws IOException {
  1423.     LOG.info("oldblock=" + oldblock + "(length=" + oldblock.getNumBytes()
  1424.         + "), newblock=" + newblock + "(length=" + newblock.getNumBytes()
  1425.         + "), datanode=" + dnRegistration.getName());
  1426.     data.updateBlock(oldblock, newblock);
  1427.     if (finalize) {
  1428.       data.finalizeBlock(newblock);
  1429.       myMetrics.blocksWritten.inc(); 
  1430.       notifyNamenodeReceivedBlock(newblock, EMPTY_DEL_HINT);
  1431.       LOG.info("Received block " + newblock +
  1432.                 " of size " + newblock.getNumBytes() +
  1433.                 " as part of lease recovery.");
  1434.     }
  1435.   }
  1437.   /** {@inheritDoc} */
  1438.   public long getProtocolVersion(String protocol, long clientVersion
  1439.       ) throws IOException {
  1440.     if (protocol.equals(InterDatanodeProtocol.class.getName())) {
  1441.       return InterDatanodeProtocol.versionID; 
  1442.     } else if (protocol.equals(ClientDatanodeProtocol.class.getName())) {
  1443.       return ClientDatanodeProtocol.versionID; 
  1444.     }
  1445.     throw new IOException("Unknown protocol to " + getClass().getSimpleName()
  1446.         + ": " + protocol);
  1447.   }
  1449.   /** A convenient class used in lease recovery */
  1450.   private static class BlockRecord { 
  1451.     final DatanodeID id;
  1452.     final InterDatanodeProtocol datanode;
  1453.     final Block block;
  1454.     
  1455.     BlockRecord(DatanodeID id, InterDatanodeProtocol datanode, Block block) {
  1456.       this.id = id;
  1457.       this.datanode = datanode;
  1458.       this.block = block;
  1459.     }
  1461.     /** {@inheritDoc} */
  1462.     public String toString() {
  1463.       return "block:" + block + " node:" + id;
  1464.     }
  1465.   }
  1467.   /** Recover a block */
  1468.   private LocatedBlock recoverBlock(Block block, boolean keepLength,
  1469.       DatanodeID[] datanodeids, boolean closeFile) throws IOException {
  1471.     // If the block is already being recovered, then skip recovering it.
  1472.     // This can happen if the namenode and client start recovering the same
  1473.     // file at the same time.
  1474.     synchronized (ongoingRecovery) {
  1475.       Block tmp = new Block();
  1476.       tmp.set(block.getBlockId(), block.getNumBytes(), GenerationStamp.WILDCARD_STAMP);
  1477.       if (ongoingRecovery.get(tmp) != null) {
  1478.         String msg = "Block " + block + " is already being recovered, " +
  1479.                      " ignoring this request to recover it.";
  1480.         LOG.info(msg);
  1481.         throw new IOException(msg);
  1482.       }
  1483.       ongoingRecovery.put(block, block);
  1484.     }
  1485.     try {
  1486.       List<BlockRecord> syncList = new ArrayList<BlockRecord>();
  1487.       long minlength = Long.MAX_VALUE;
  1488.       int errorCount = 0;
  1490.       //check generation stamps
  1491.       for(DatanodeID id : datanodeids) {
  1492.         try {
  1493.           InterDatanodeProtocol datanode = dnRegistration.equals(id)?
  1494.               this: DataNode.createInterDataNodeProtocolProxy(id, getConf());
  1495.           BlockMetaDataInfo info = datanode.getBlockMetaDataInfo(block);
  1496.           if (info != null && info.getGenerationStamp() >= block.getGenerationStamp()) {
  1497.             if (keepLength) {
  1498.               if (info.getNumBytes() == block.getNumBytes()) {
  1499.                 syncList.add(new BlockRecord(id, datanode, new Block(info)));
  1500.               }
  1501.             }
  1502.             else {
  1503.               syncList.add(new BlockRecord(id, datanode, new Block(info)));
  1504.               if (info.getNumBytes() < minlength) {
  1505.                 minlength = info.getNumBytes();
  1506.               }
  1507.             }
  1508.           }
  1509.         } catch (IOException e) {
  1510.           ++errorCount;
  1511.           InterDatanodeProtocol.LOG.warn(
  1512.               "Failed to getBlockMetaDataInfo for block (=" + block 
  1513.               + ") from datanode (=" + id + ")", e);
  1514.         }
  1515.       }
  1517.       if (syncList.isEmpty() && errorCount > 0) {
  1518.         throw new IOException("All datanodes failed: block=" + block
  1519.             + ", datanodeids=" + Arrays.asList(datanodeids));
  1520.       }
  1521.       if (!keepLength) {
  1522.         block.setNumBytes(minlength);
  1523.       }
  1524.       return syncBlock(block, syncList, closeFile);
  1525.     } finally {
  1526.       synchronized (ongoingRecovery) {
  1527.         ongoingRecovery.remove(block);
  1528.       }
  1529.     }
  1530.   }
  1532.   /** Block synchronization */
  1533.   private LocatedBlock syncBlock(Block block, List<BlockRecord> syncList,
  1534.       boolean closeFile) throws IOException {
  1535.     if (LOG.isDebugEnabled()) {
  1536.       LOG.debug("block=" + block + ", (length=" + block.getNumBytes()
  1537.           + "), syncList=" + syncList + ", closeFile=" + closeFile);
  1538.     }
  1540.     //syncList.isEmpty() that all datanodes do not have the block
  1541.     //so the block can be deleted.
  1542.     if (syncList.isEmpty()) {
  1543.       namenode.commitBlockSynchronization(block, 0, 0, closeFile, true,
  1544.           DatanodeID.EMPTY_ARRAY);
  1545.       return null;
  1546.     }
  1548.     List<DatanodeID> successList = new ArrayList<DatanodeID>();
  1550.     long generationstamp = namenode.nextGenerationStamp(block);
  1551.     Block newblock = new Block(block.getBlockId(), block.getNumBytes(), generationstamp);
  1553.     for(BlockRecord r : syncList) {
  1554.       try {
  1555.         r.datanode.updateBlock(r.block, newblock, closeFile);
  1556.         successList.add(r.id);
  1557.       } catch (IOException e) {
  1558.         InterDatanodeProtocol.LOG.warn("Failed to updateBlock (newblock="
  1559.             + newblock + ", datanode=" + r.id + ")", e);
  1560.       }
  1561.     }
  1563.     if (!successList.isEmpty()) {
  1564.       DatanodeID[] nlist = successList.toArray(new DatanodeID[successList.size()]);
  1566.       namenode.commitBlockSynchronization(block,
  1567.           newblock.getGenerationStamp(), newblock.getNumBytes(), closeFile, false,
  1568.           nlist);
  1569.       DatanodeInfo[] info = new DatanodeInfo[nlist.length];
  1570.       for (int i = 0; i < nlist.length; i++) {
  1571.         info[i] = new DatanodeInfo(nlist[i]);
  1572.       }
  1573.       return new LocatedBlock(newblock, info); // success
  1574.     }
  1576.     //failed
  1577.     StringBuilder b = new StringBuilder();
  1578.     for(BlockRecord r : syncList) {
  1579.       b.append("n  " + r.id);
  1580.     }
  1581.     throw new IOException("Cannot recover " + block + ", none of these "
  1582.         + syncList.size() + " datanodes success {" + b + "n}");
  1583.   }
  1584.   
  1585.   // ClientDataNodeProtocol implementation
  1586.   /** {@inheritDoc} */
  1587.   public LocatedBlock recoverBlock(Block block, boolean keepLength, DatanodeInfo[] targets
  1588.       ) throws IOException {
  1589.     logRecoverBlock("Client", block, targets);
  1590.     return recoverBlock(block, keepLength, targets, false);
  1591.   }
  1593.   private static void logRecoverBlock(String who,
  1594.       Block block, DatanodeID[] targets) {
  1595.     StringBuilder msg = new StringBuilder(targets[0].getName());
  1596.     for (int i = 1; i < targets.length; i++) {
  1597.       msg.append(", " + targets[i].getName());
  1598.     }
  1599.     LOG.info(who + " calls recoverBlock(block=" + block
  1600.         + ", targets=[" + msg + "])");
  1601.   }
  1602. }