Namenode源代码分析

<一>.namenode结构简介：
1.基本概念：
|-->一个hdfs cluster包含一个NameNode和若干的DataNode，NameNode（以下简称nn）是master，主要负责管理hdfs文件系
统，具体地包括namespace管理（其实就是目录结构），block管理（其中包括 filename->block，block->ddatanode list的
对应关 系）。nn提供的是始终被动接收服务的server，主要有三类协议接口：ClientProtocol接口、DatanodeProtocol接口、
NamenodeProtocol接口。


2.second namenode:
|-->该部分主要是定时对NameNode进行数据snapshots进行备份，这样尽量降低NameNode崩溃之后，导致数据的丢失，其实所作的工
作就是从nn获得fsimage和edits把二者重新合并然后发给nn，这样，既能减轻nn的负担又能保险地备份。


3.FSnamesystem:
|-->client或者datanode的消息发到nn后最终都会落到FSnamesystem身上，这是一个重量级家伙，如图，对各种服务请求的处理
都转交给它完成，它提供了对各种数据结构操作的接口，这些数据结构共同维护了整个namenode的元数据信息。
4.Namenode数据结构

FSDirectory:
|--->存储整个文件系统的目录状态，对整个目录结构的管理通过调用FSImage和FSEditLog的方法从namenode本地磁盘读取元数据
信息和向本地磁盘写入元数据信息，并登记对目录结构所作的修改到日志文件。另外，FSDirectory保存了文件名和数据块的映射
关系。


INode：类似Linux的INode节点，又分为目录节点和文件节点。
|-->INodeFile:文件节点， BlockInfo blocks[], 它记录了一个文件所包含的所有block
|-->INodefileUnderConstruction:用于表示正在创建中的文件
|-->INodeDeirectory:主要记录一个INodeFile的List集合
|-->InodeDirectoryWithQuota：表示有限制有配额的目录，比如根目录


FSimage:
|-->用于持久化保存节点数的内存状况，启动时从fsimage中加载数据至内存
|-->启动后操作纪录在editlog中定期将edit和fsimage进行合并刷新的fsimage
|-->namenode通过loadImage从fsimage文件中读取内存信息，通过saveImage从内存中刷节点信息至文件
|-->fsimage二进制文件，保存hdfs所有的文件和目录的元数据信息，格式如下

BlockMap:
|-->block->datanode的信息没有持久化存储，而是namenode通过datanode的blockreport获取block->
blockList的列表
|-->BlocksMap负责维护了三种信息：block->datanode list 、block->INodeFile、datanode->blocks
　　 |-->Object[] triplets三元组，block有几个副本就保存几个三元组，triplets的3i保存block对应的datanode信息，第3i+1,3i+2二个节点当前
block对应的前一block和后一 个block信息



<二> 源代码分析
1.INode*
|-->基本信息：
|-->name |名称
|-->parent |父目录
|-->modificationTime |修改时间
|-->accessTime |访问时间
|-->权限控制：针对组和用户分配id,修改permission的状态，权限最终用一个long型表示
|-->setPermissionStatus(PermissionStatus ps)
|-->setUser(ps.getUserName());
|-->setGroup(ps.getGroupName());
|-->int n = SerialNumberManager.INSTANCE.getGroupSerialNumber(group);
|-->updatePermissionStatus(PermissionStatusFormat. GROUP, n);
|-->permission = f.combine(n, permission); |最终保存long型权限值
|-->setPermission(ps.getPermission());
|-->get/set()方法获取信息
|-->computeContentSummary() |用于递归计算文件大小和命中情况
|-->return new ContentSummary (a[0], a[1], a[2], getNsQuota(), a[3], getDsQuota());


|-->InodeDirectory即hdfs中的目录，主要是管理子目录或者节点
|-->private List<INode> children; |获取所有子节点和目录的信息
|-->addNode() |添加新的INode节点，其它对节点和目录操作类似


|-->InodeFile：即节点信息
|-->BlockInfo List |用于保存路径节点上的block列表。获取InodeFile结点路径，即可取得当前节点对
应的block list信息集
|-->this .triplets = new Object[3*replication]; |三元组保存block对应的datanode中前后缀信息
|-->INodeFileUnderConstruction |保存正在构建的文件信息，涉及文件租约


2. Datanode*
|-->DatanodeId：主要保存datanode基本信息
|-->name
|-->storageID
|-->infoPort
|-->ipcPort


|-->DatanodeInfo：继承自DatanodeId，增加了动态信息
|-->capacity
|-->dfsUsed
|-->remaining
|-->xceiverCount
|-->location |机架位置


|-->DatanodeDescriptor: 主要保存block与对应的DatanodeDiscriptor的关联关系
|-->blockList |blockInfo信息
|-->replicateBlocks |三个block队列，分别保存当前datanode的复制、恢复，不合法的block队列
|-->recoverBlocks
|-->invalidateBlocks
|-->addBlock(BlockInfo b) |首先添加至block对应的triplets数祖中，再设置对应的datanode中blocklist中的前后缀信息中
|-->b.addNode(this) |BlockInfo继承至block，管理对应的DataNode节点链接
|-->findDatanode(node) |确认当前node是否存在
|-->ensureCapacity(1) |确认存在容量存在，且返回最后一个last位置
|-->setDatanode(lastNode, node) |triplets数祖最后三个位置保存DatanodeDiscriptor节点
|-->setNext(lastNode, null ) |分别设置此时节点的前、后缀
|-->setPrevious(lastNode, null )
|-->b.listInsert(blockList, this ) |进一步确认DatanodeDiscriptor位置，加至blocklist的头位置
|-->int dnIndex = this.findDatanode(dn);
|-->this .setPrevious(dnIndex, null);
|-->this .setNext(dnIndex, head); |设置为blocklist的头位置
|-->head.setPrevious(head. findDatanode(dn), this ); |原有头位置指向新加节点
### |-->reportDiff(blocksMap , newReport, toAdd, toRemove, toInvalidate) |datanode向namenode汇
报时调用，分析现有namenode信差别


3.FsImage
|-->NameNode存储在Disk磁盘上的信息 dfs/name目录
|-->current
|-->edits |日志
|-->fsimage |内存镜像
|-->fstime |镜像时间
|-->VERSION |版本信息
|-->image
|-->in_use.lock
|-->FsImage:用于管理Storage，用于保存内存镜像持久化进磁盘，常用操作如下
|-->loadFSImage() |用于保存内存中INode节点树镜像
|-->saveFSImage()
|-->recoverTransitionRead()


|-->loadFSImage(File curFile) |首先获取文件流，再读取File文件信息
|-->DataInputStream in = new DataInputStream( new
BufferedInputStream(new FileInputStream(curFile)));
|-->逐步读取in内容
|-->imageVersion
|-->namespaceID
|-->numFiles
|-->genstamp
|-->needToSave = (imgVersion != FSConstants.LAYOUT_VERSION); |设置是否要与FSEdit合并后保存
|-->循环读取numFile个节点信息
|-->path
|-->replication
|-->modificationTime
|-->atime
|-->blockSize
|-->numBlocks
|-->for (int j = 0; j < numBlocks; j++) ｜循环读取blocks信息
|-->blocks[j] = new Block();
|-->if (-14 < imgVersion)
|-->blocks[j].set(in.readLong(), in.readLong(),
Block.GRANDFATHER_GENERATION_STAMP);
|-->blocks[j].readFields(in);
|-->blockSize |设置blockSize大小
|-->if (imgVersion <= -16 && blocks == null)
|-->nsQuota
|-->dsQuota
|-->parentINode = fsDir.addToParent(path, parentINode, permissions, ｜设置新的INode节点
blocks, replication, modificationTime,
atime, nsQuota, dsQuota, blockSize);


|-->this.loadDatanodes(imgVersion, in); |加载DataNode信息
|-->this.loadFilesUnderConstruction(imgVersion, in, fsNamesys); ｜加载正在构建的节点信息

|-->loadFSEdits(StorageDirectory sd)
|-->new EditLogFileInputStream(getImageFile(sd, NameNodeFile.EDITS)); |构建EditLog输出流
|-->FSEditLog.loadFSEdits(edits) ; |加载FSEditlog日记信息，并写入内存结构当中
|-->FSNamesystem.getFSNamesystem().dir.updateCountForINodeWithQuota();



|-->FSImage结构图如下





4.FSEditLog: |日志记录器
|-->FSEditLog.loadFSEdits(edits)
|-->DataInputStream in = new DataInputStream( new BufferedInputStream(edits));
|-->logVersion = in.readByte(); |获取版本号
|-->while ( true)
|-->opcode = in.readByte();
|-->switch (opcode) |通过操作码，处理各种不同的日志操作，如添加、删除、重命名等
|-->case OP_ADD:
|-->case OP_CLOSE:
|-->case OP_SET_REPLICATION:
|-->case OP_RENAME:
|-->case OP_DELETE
|-->FSEditLog.logOpenFile
|-->nameReplicationPair |构建Pair对，存储以下信
|-->newNode.getReplication()
|-->newNode.getModificationTime()
|-->newNode.getAccessTime()
|-->newNode.getPreferredBlockSize()
|-->logEdit(OP_ADD, Writable[])
|-->for ( int idx = 0; idx < editStreams .size(); idx++)
|-->EditLogOutputStream eStream = editStreams.get(idx);
|-->eStream.write(op, writables);
|-->TransactionId id = myTransactionId .get(); |获取新txid ，每次TransactionId更新时均需改变




5.LeaseManager：租约管理器，用于管理租约，租约相当于锁，包含路径，时间等信息
|-->addLease(String holder, string src)
|-->lease = getLease(holder)
|-->if(lease == null)
|-->leases.put(holder, lease) |存入一个map当中
|-->sortedLeases.add(lease) |加入一个sortedSet的lease集合当中，排序规则为最后更新时间
|-->else
|-->renewLease(lease)
|-->sortedLeaseByPath.put(src, lease)
|-->lease.paths.add(src) |保存一个src路径到租约的map集合

|-->checkLeases():监控器用于监听租约是否过期,通过sortedSet来判断
|-->for(; sortedLeases.size() > 0; )
|-->Lease oldest = sortedLeases.first();
|--oldest.getPaths().toArray(leasePaths);
|-->for(String p : leasePaths)
|-->fsnamesystem.internalReleaseLease(oldest, p); |过期则释放租约
|-->for(String p : removing)
|-->removeLease(oldest, p);





6.FSDirectory ：用于存储文件目录状态
|-->主要成员变量
|-->final FSNamesystem namesystem ;
|-->final INodeDirectoryWithQuota rootDir ;
|-->FSImage fsImage;
|-->loadFSImage
|-->if (startOpt == StartupOption.FORMAT)
|-->fsImage.format();
|-->startOpt = StartupOption.REGULAR;
|-->if (fsImage.recoverTransitionRead(dataDirs, editsDirs, startOpt) |分析目录，如有必要从前一事物中恢复
|-->fsImage.saveFSImage();
|-->FSEditLog editLog = fsImage.getEditLog();
|-->if (!editLog.isOpen())
|-->editLog.open();
|-->synchronized (this)
|-->this.ready = true;
|-->this.notifyAll();


|-->addFile() |返回INodeFileUnderConstruction正在构建的对象, namenode创建文件时被调用
|-->waitForReady(); ｜等待被唤醒
|-->newNode = new INodeFileUnderConstruction( |构建新的INodeFile结点
permissions,replication,
preferredBlockSize, modTime, clientName,
clientMachine, clientNode);
|-->synchronized (rootDir)
|-->newNode = addNode(path, newNode, -1, false); ｜添加INode结点至目录树中
|-->fsImage.getEditLog().logOpenFile(path, newNode); ｜写入日志文件
|-->return newNode;|-->


|-->addBlock() |添加block至src路径下
|-->INodeFile fileNode = (INodeFile) inodes[inodes.length-1]; |获取src的INodes节点中最后一个INode文件进行添加
|-->updateCount(inodes, inodes.length-1, 0,...) |用于设置quota
|-->namesystem.blocksMap.addINode(block, fileNode); |将block添加至fileNode节点中
|-->BlockInfo blockInfo = namesystem.blocksMap.getStoredBlock(block); |从blockmap中获取block对应的BlockInfo信息，主要是DatanodeDescriptor信息链
|-->fileNode.addBlock(blockInfo); |在INode节点下添加block，可参见DatanodeDescriptor的addBlock()




7.FSNamesystem
|-->主要成员变量
|-->Daemon hbthread = null; // 心跳监控
|-->public Daemon lmthread = null; // 租约管理
|-->Daemon smmthread = null; // 安全模式管理
|-->public Daemon replthread = null; // 复制模块管理
|-->public FsDirectory dir; //指向系统使用的目录结构对象
|-->BlockMaps blockmap = new Blockmap() //保存INode到DataNode的指向关系
|-->CorruptReplicationsMap corruptReplicas; //保存较验未通过块
|-->HttpServer infoServer //保存httpserver
|-->private PendingReplicationBlocks pendingReplications; //用于保存正在复制的数据块信息


|-->initialize() |FSnamesystem初始化时被调用
|-->this.dir = new FSDirectory(this, conf); |初始化目录
|-->StartupOption startOpt = NameNode.getStartupOption(conf); |配置文件中读取dfs.namenode.startup
|-->this.dir.loadFSImage(getNamespaceDirs(conf), |从目录中加载FSImage至内存，形成目录树
getNamespaceEditsDirs(conf), startOpt);
|-->this.safeMode = new SafeModeInfo(conf); |安全模式
|-->setBlockTotal()
|-->pendingReplications = new PendingReplicationBlocks( |保存正在复制的block信息
conf.getInt("dfs.replication.pending.timeout.sec",
-1) * 1000L);
|-->hbthread.start(); |分别启动心跳进程
|-->lmthread.start(); |租约进程
|-->replthread.start();
|-->this.hostsReader = new HostsFileReader(conf.get("dfs.hosts",""),
conf.get("dfs.hosts.exclude",""));
|-->dnthread.start();
|-->this.dnsToSwitchMapping = ReflectionUtils.newInstance( |获取网络拓朴结构
conf.getClass("topology.node.switch.mapping.impl", ScriptBasedMapping.class,
DNSToSwitchMapping.class), conf);


|-->resolveNetworkLocation(nodeS) |解析节点所在的rack机架信息，脚本不一定精确，应以交换机为真实机架
|-->names.add(node.getHost());
|-->List<String> rName = dnsToSwitchMapping.resolve(names); |CachedDNSToSwitchMapping.resolve(names)
|-->names = NetUtils.normalizeHostNames(names);
|-->List<String> rNames = rawMapping.resolve(unCachedHosts);
|-->for (String name : names)
|-->result.add(networkLocation);
|-->RawScriptBasedMapping.resolve()
|-->if (scriptName == null)
|-->m.add(NetworkTopology.DEFAULT_RACK);
|-->String output = runResolveCommand(names); |读取topology.script.file.name属性值，执行脚本获取rack信息
|-->while (numProcessed != args.size())
|-->cmdList.add(scriptName);
|-->for (numProcessed = start; numProcessed < (start + maxArgs) &&
numProcessed < args.size(); numProcessed++)
|-->cmdList.add(args.get(numProcessed));
|-->ShellCommandExecutor s = new ShellCommandExecutor(cmdList.toArray(new String[0]), dir);
|-->s.execute();
|-->allOutput.append(s.getOutput() + " ");


8.SafeModeInfo: 安全模式，不允许对命名空间任何改动，包括对block块的复制、删除等操作
|-->enter()
|-->this.reached = 0; |设置状态位
|-->leave()
|-->needUpgrade = startDistributedUpgradeIfNeeded();
|-->processMisReplicatedBlocks();
|-->reached = -1;
|-->canLeave()
|-->needEnter()
|-->needEnter()
|-->getSafeBlockRatio() < threshold; |threadhold代表合格的DataNode百分比，dfs.safemode.threshold.pct控制
|-->checkMode()
|-->SafeModeMonitor() |在FsNamesystem启动中就有一个实例，用于对SafeModeInfo进行监控，是否进入安全模式


9.NetworkTopology |用于构造datanode的网络拓朴结构，加载至内存，形成Datanode节点网络，实际操作在InnerNode中进行，InnerNode指向服务器或者交换机节点
|-->add(Node node)
|-->netlock.writeLock().lock(); |读写锁的性能消耗
|-->Node rack = getNode(node.getNetworkLocation());
|-->if (clusterMap.add(node)) |内部类InnerNode，继承自Nodebase，作为服务器节点
|-->numOfRacks++;
|-->netlock.writeLock().unlock();
|-->remove(Node node)
|-->netlock.writeLock().lock();
|-->if (clusterMap.remove(node))
|-->InnerNode rack = (InnerNode)getNode(node.getNetworkLocation()); |此时node被删除，获取地址应为null
|--> netlock.writeLock().unlock();
|-->isOnSameRack(Node node1, Node node2)
|-->netlock.readLock().lock();
|-->return node1.getParent()==node2.getParent();
|-->netlock.readLock().unlock();
|-->chooseRandom(String scope, String excludedScope)
|-->Node node = getNode(scope);
|-->if (!(node instanceof InnerNode)) |获取的是节点信息
|-->return node;
|-->InnerNode innerNode = (InnerNode)node; |非服务器节点、继续往下找
|-->int numOfDatanodes = innerNode.getNumOfLeaves();
|-->if (excludedScope == null)
|-->node = null;
|-->else
|-->node = getNode(excludedScope);
|-->if (!(node instanceof InnerNode))
|-->numOfDatanodes -= 1;
|-->else
|-->numOfDatanodes -= ((InnerNode)node).getNumOfLeaves()
|-->int leaveIndex = r.nextInt(numOfDatanodes);
|-->return innerNode.getLeaf(leaveIndex, node); |从InnerNode的叶子节点中获取Node




10.ReplicationTargetChooser replicator |用于返回复制块的信息，最重要的方法是chooseTarget()
|获取复制的目标Datanode节点，获取Datanode集合后建立连接进行复制
|-->chooseTarget(int numOfReplicas, DatanodeDescriptor writer, List<Node> excludedNodes,
long blocksize,
int maxNodesPerRack,
List<DatanodeDescriptor> results)


|-->int numOfResults = results.size();
|-->boolean newBlock = (numOfResults==0);
|--> switch(numOfResults)
|-->case 0:
|-->writer = chooseLocalNode(writer, excludedNodes,
blocksize, maxNodesPerRack, results);
|-->case 1:
|-->chooseRemoteRack(1, results.get(0), excludedNodes,
blocksize, maxNodesPerRack, results);
|-->case 2
|-->if (clusterMap.isOnSameRack(results.get(0), results.get(1)))
|-->chooseRemoteRack(1, results.get(0), excludedNodes,
blocksize, maxNodesPerRack, results);
|-->else if (newBlock)
|-->chooseLocalRack(results.get(1), excludedNodes, blocksize,
maxNodesPerRack, results);
|-->else
|-->chooseLocalRack(writer, excludedNodes, blocksize,
maxNodesPerRack, results);


11.Namenode的clientProtocol接口分析：用于客户端调用
|-->getBlockLocations() |用于获取datanode具体位置信息，将从这些信息中获取读取实际数据
|-->namesystem.getBlockLocations(getClientMachine(), src, offset, length);


|-->getBlockLocations(String clientMachine, String src,long offset, long length)
|-->INodeFile inode = dir.getFileINode(src); |根据src路径，获取INode节点
|-->Block[] blocks = inode.getBlocks(); |INode对应的blocklist集合列表
|-->int curBlk = 0;
|-->long curPos = 0, blkSize = 0;
|-->for (curBlk = 0; curBlk < nrBlocks; curBlk++) |根据offset来确认当前位置
|-->blkSize = blocks[curBlk].getNumBytes()
|-->if (curPos + blkSize > offset)
|-->break;
|-->curPos += blkSize;
|-->while (curPos < endOff && curBlk < blocks.length && results.size() < nrBlocksToReturn); |确认从offset开头，至length结束
|-->int numNodes = blocksMap.numNodes(blocks[curBlk]);
|-->int numCorruptNodes = countNodes(blocks[curBlk]).corruptReplicas();
|-->int numCorruptReplicas = corruptReplicas.numCorruptReplicas(blocks[curBlk]);
|-->boolean blockCorrupt = (numCorruptNodes == numNodes);
|-->int numMachineSet = blockCorrupt ? numNodes : (numNodes - numCorruptNodes);
|-->DatanodeDescriptor[] machineSet = new DatanodeDescriptor[numMachineSet];
|-->if (numMachineSet > 0)
|-->for(Iterator<DatanodeDescriptor> it = blocksMap.nodeIterator(blocks[curBlk]); it.hasNext();) |获取blocks对应的DatanodeDescriptor
|-->DatanodeDescriptor dn = it.next();
|-->boolean replicaCorrupt = corruptReplicas.isReplicaCorrupt(blocks[curBlk], dn);
|-->if (blockCorrupt || (!blockCorrupt && !replicaCorrupt))
|-->machineSet[numNodes++] = dn;
|-->results.add(new LocatedBlock(blocks[curBlk], machineSet, curPos,blockCorrupt)); |构建LocatedBlock
|-->curPos += blocks[curBlk].getNumBytes();
|-->curBlk++;
|-->return inode.createLocatedBlocks(results); |创建inode节点的LocatedBlocks列表


|-->create(String src, FsPermission masked, String clientName, boolean overwrite, short replication,long blockSize )
|-->namesystem.startFile(src,
new PermissionStatus(UserGroupInformation.getCurrentUGI().getUserName(),
null , masked),
clientName, clientMachine, overwrite, replication, blockSize);
|-->startFileInternal(src, permissions, holder, clientMachine, overwrite, false,
replication, blockSize);
|-->verify safemode/permission/path
|-->isInSafeMode()
|-->pathExists && dir.isDir(src
|-->isPermissionEnabled
|-->checkPathAccess(src, FsAction.WRITE);
|-->checkAncestorAccess(src, FsAction.WRITE);
|-->INode myFile = dir.getFileINode(src);
|-->verifyReplication(src, replication, clientMachine);
|-->DatanodeDescriptor clientNode = host2DataNodeMap.getDatanodeByHost(clientMachine);
|-->if (append) |如果为append，则添加Inode节点,创建InodeFileUnderConstruction表明正在创建对象
|-->INodeFile node = (INodeFile) myFile;
|-->INodeFileUnderConstruction cons = new INodeFileUnderConstruction()
|-->dir.replaceNode(src, node, cons);
|-->leaseManager.addLease(cons.clientName, src);
|-->else
|-->checkFsObjectLimit();
|-->INodeFileUnderConstruction newNode = dir.addFile()
|-->leaseManager.addLease(newNode.clientName, src);




|-->markBlockAsCorrupt(Block blk, DatanodeInfo dn)
|-->DatanodeDescriptor node = getDatanode(dn);
|-->final BlockInfo storedBlockInfo = blocksMap.getStoredBlock(blk);
|-->if (storedBlockInfo == null)
|-->corruptReplicas.addToCorruptReplicasMap(storedBlockInfo, node);
|-->if (countNodes(storedBlockInfo).liveReplicas()>inode.getReplication())
|-->invalidateBlock(storedBlockInfo, node);
|-->count = countNodes(blk).liveReplicas();
|-->if (count > 1)
|-->addToInvalidates(blk, dn);
|-->addToInvalidatesNoLog(b, n);
|-->removeStoredBlock(blk, node);
|-->else
|-->updateNeededReplications(storedBlockInfo, -1, 0);
|-->else
|-->INodeFile inode = storedBlockInfo.getINode();
|-->if (inode == null)
|-->addToInvalidates(storedBlockInfo, node);
|-->addToInvalidatesNoLog(Block b, DatanodeInfo n)
|-->recentInvalidateSets.put(n.getStorageID(), invalidateSet);
|-->invalidateSet.add(b)




|-->blockReport(DatanodeRegistration nodeReg, long[] blocks) |datanode与namenode建立RPC连接后，由datanode的offerService方法向namenode汇报blocks状态
|-->verifyRequest(nodeReg); |验证请求信息
|-->BlockListAsLongs blist = new BlockListAsLongs(blocks);
|-->namesystem.processReport(nodeReg, blist);
|-->DatanodeDescriptor node = getDatanode(nodeID);
|-->node.reportDiff(blocksMap, newReport, toAdd, toRemove, toInvalidate); |此时比较block不同较为耗时
|-->for (Block b : toRemove) ｜分别处理删除，新增，失效的块文件
|-->removeStoredBlock(b, node);
|-->for (Block b : toAdd)
|-->addStoredBlock(b, node, null); |详见下面分解
|-->for (Block b : toInvalidate)
|-->addToInvalidates(b, node);
|-->NameNode.getNameNodeMetrics().blockReport.inc((int) (now() - startTime));
|-->if (getFSImage().isUpgradeFinalized())
|-->return DatanodeCommand.FINALIZE;




|-->addStoredBlock(Block block, DatanodeDescriptor node, DatanodeDescriptor delNodeHint) |添加block块
|-->BlockInfo storedBlock = blocksMap.getStoredBlock(block); |获取block在内存中对应的datanode的list列表blockinfo
|-->boolean added = node.addBlock(storedBlock); |将当前的datanode添加至blockinfo的列表当中
|-->if (block != storedBlock) ｜之前汇报的block跟现有block存在不同
|-->if (cursize == 0)
|-->storedBlock.setNumBytes(block.getNumBytes());
|-->else if (cursize != block.getNumBytes())
|-->try
|-->if (cursize > block.getNumBytes())
|-->markBlockAsCorrupt(block, node);
|-->else
|-->int numNodes = blocksMap.numNodes(block);
|-->for (int j = 0; j < count; j++)
|-->markBlockAsCorrupt(block, nodes[j]);
|-->if (diff > 0 && file.isUnderConstruction() && cursize < storedBlock.getNumBytes())
|-->String path = leaseManager.findPath((INodeFileUnderConstruction)file);
|-->dir.updateSpaceConsumed(path, 0, -diff*file.getReplication());
|-->NumberReplicas num = countNodes(storedBlock); |开始整理block的复制数，比较已存在的复制数，正在复制数，文档需要复制数三者关系进行复制
|-->int numLiveReplicas = num.liveReplicas();
|-->int numCurrentReplica = numLiveReplicas + pendingReplications.getNumReplicas(block);
|-->incrementSafeBlockCount(numCurrentReplica);
|-->if (fileINode.isUnderConstruction())
|-->return block;
|-->if (numCurrentReplica >= fileReplication)
|-->neededReplications.remove(block, numCurrentReplica,
num.decommissionedReplicas, fileReplication);
|-->else
|-->updateNeededReplications(block, curReplicaDelta, 0);
|-->if (numCurrentReplica > fileReplication)
|-->processOverReplicatedBlock(block, fileReplication, node, delNodeHint);
|-->if ((corruptReplicasCount > 0) && (numLiveReplicas >= fileReplication))
|-->invalidateCorruptReplicas(block);