您的位置: 首页  >  文章  >  Hadoop Balancer源码解读

Hadoop Balancer源码解读

分类: 文章 • 2023-11-18 12:54:58

前言

最近在做一些Hadoop运维的相关工作，发现了一个有趣的问题，我们公司的Hadoop集群磁盘占比数值参差不齐，高的接近80%，低的接近40%，并没有充分利用好上面的资源，但是balance的操作跑的也是正常的啊，所以打算看一下Hadoop的balance的源代码，更深层次的去了解Hadoop Balance的机制。

Balancer和Distpatch

上面2个类的设计就是与Hadoop Balance操作最紧密联系的类，Balancer类负载找出<source, target>这样的起始，目标结果对，然后存入到Distpatch中，然后通过Distpatch进行分发，不过在分发之前，会进行block的验证，判断此block是否能被移动，这里会涉及到一些条件的判断，具体的可以通过后面的代码中的注释去发现。

在Balancer的最后阶段，会将source和Target加入到dispatcher中，详见下面的代码：

[java]view plaincopyprint?
  1. /**
  2. *根据源节点和目标节点,构造任务对
  3. */
  4. privatevoidmatchSourceWithTargetToMove(Sourcesource,StorageGrouptarget){
  5. longsize=Math.min(source.availableSizeToMove(),target.availableSizeToMove());
  6. finalTasktask=newTask(target,size);
  7. source.addTask(task);
  8. target.incScheduledSize(task.getSize());
  9. //加入分发器中
  10. dispatcher.add(source,target);
  11. LOG.info("Decidedtomove"+StringUtils.byteDesc(size)+"bytesfrom"
  12. +source.getDisplayName()+"to"+target.getDisplayName());
  13. }
然后就是dispather类中的代码。然后进行block的分发操作：

[java]view plaincopyprint?
  1. /**
  2. *Foreachdatanode,choosematchingnodesfromthecandidates.Eitherthe
  3. *datanodesorthecandidatesaresourcenodeswith(utilization>Avg),and
  4. *theothersaretargetnodeswith(utilization<Avg).
  5. */
  6. private<GextendsStorageGroup,CextendsStorageGroup>
  7. voidchooseStorageGroups(Collection<G>groups,Collection<C>candidates,
  8. Matchermatcher){
  9. for(finalIterator<G>i=groups.iterator();i.hasNext();){
  10. finalGg=i.next();
  11. for(;choose4One(g,candidates,matcher););
  12. if(!g.hasSpaceForScheduling()){
  13. //如果候选节点没有空间调度,则直接移除掉
  14. i.remove();
  15. }
  16. }
  17. }
继续调用后面的方法

[java]view plaincopyprint?
  1. /**
  2. *Decideall<source,target>pairsand
  3. *thenumberofbytestomovefromasourcetoatarget
  4. *Maximumbytestobemovedperstoragegroupis
  5. *min(1Bandworthofbytes,MAX_SIZE_TO_MOVE).
  6. *从源节点列表和目标节点列表中各自选择节点组成一个个对,选择顺序优先为同节点组,同机架,然后是针对所有
  7. *@returntotalnumberofbytestomoveinthisiteration
  8. */
  9. privatelongchooseStorageGroups(){
  10. //First,matchnodesonthesamenodegroupifclusterisnodegroupaware
  11. if(dispatcher.getCluster().isNodeGroupAware()){
  12. //首先匹配的条件是同节点组
  13. chooseStorageGroups(Matcher.SAME_NODE_GROUP);
  14. }
  16. //Then,matchnodesonthesamerack
  17. //然后是同机架
  18. chooseStorageGroups(Matcher.SAME_RACK);
  19. //Atlast,matchallremainingnodes
  20. //最后是匹配所有的节点
  21. chooseStorageGroups(Matcher.ANY_OTHER);
  23. returndispatcher.bytesToMove();
  24. }
然后再通过调用Dispatch的层层方法，下面是简图：

Hadoop Balancer源码解读

最后核心的检验block块是否合适的代码为下面这个：

[java]view plaincopyprint?
  1. /**
  2. *Decideiftheblockisagoodcandidatetobemovedfromsourcetotarget.
  3. *Ablockisagoodcandidateif
  4. *1.theblockisnotintheprocessofbeingmoved/hasnotbeenmoved;
  5. *移动的块不是正在被移动的块
  6. *2.theblockdoesnothaveareplicaonthetarget;
  7. *在目标节点上没有移动的block块
  8. *3.doingthemovedoesnotreducethenumberofracksthattheblockhas
  9. *移动之后,不同机架上的block块的数量应该是不变的.
  10. */
  11. privatebooleanisGoodBlockCandidate(Sourcesource,StorageGrouptarget,
  12. DBlockblock){
  13. if(source.storageType!=target.storageType){
  14. returnfalse;
  15. }
  16. //checkiftheblockismovedornot
  17. //如果所要移动的块是存在于正在被移动的块列表,则返回false
  18. if(movedBlocks.contains(block.getBlock())){
  19. returnfalse;
  20. }
  21. //如果移动的块已经存在于目标节点上,则返回false,将不会予以移动
  22. if(block.isLocatedOn(target)){
  23. returnfalse;
  24. }
  26. //如果开启了机架感知的配置,则目标节点不应该有相同的block
  27. if(cluster.isNodeGroupAware()
  28. &&isOnSameNodeGroupWithReplicas(target,block,source)){
  29. returnfalse;
  30. }
  32. //需要维持机架上的block块数量不变
  33. if(reduceNumOfRacks(source,target,block)){
  34. returnfalse;
  35. }
  36. returntrue;
  37. }

下面是Balancer.java和Dispatch.java类的完整代码解析：

Balancer.java:

[java]view plaincopyprint?
  1. /**
  2. *LicensedtotheApacheSoftwareFoundation(ASF)underone
  3. *ormorecontributorlicenseagreements.SeetheNOTICEfile
  4. *distributedwiththisworkforadditionalinformation
  5. *regardingcopyrightownership.TheASFlicensesthisfile
  6. *toyouundertheApacheLicense,Version2.0(the
  7. *"License");youmaynotusethisfileexceptincompliance
  8. *withtheLicense.YoumayobtainacopyoftheLicenseat
  9. *
  10. *http://www.apache.org/licenses/LICENSE-2.0
  11. *
  12. *Unlessrequiredbyapplicablelaworagreedtoinwriting,software
  13. *distributedundertheLicenseisdistributedonan"ASIS"BASIS,
  14. *WITHOUTWARRANTIESORCONDITIONSOFANYKIND,eitherexpressorimplied.
  15. *SeetheLicenseforthespecificlanguagegoverningpermissionsand
  16. *limitationsundertheLicense.
  17. */
  18. packageorg.apache.hadoop.hdfs.server.balancer;
  20. importstaticcom.google.common.base.Preconditions.checkArgument;
  22. importjava.io.IOException;
  23. importjava.io.PrintStream;
  24. importjava.net.URI;
  25. importjava.text.DateFormat;
  26. importjava.util.ArrayList;
  27. importjava.util.Arrays;
  28. importjava.util.Collection;
  29. importjava.util.Collections;
  30. importjava.util.Date;
  31. importjava.util.Formatter;
  32. importjava.util.Iterator;
  33. importjava.util.LinkedList;
  34. importjava.util.List;
  35. importjava.util.Set;
  37. importorg.apache.commons.logging.Log;
  38. importorg.apache.commons.logging.LogFactory;
  39. importorg.apache.hadoop.classification.InterfaceAudience;
  40. importorg.apache.hadoop.conf.Configuration;
  41. importorg.apache.hadoop.conf.Configured;
  42. importorg.apache.hadoop.fs.Path;
  43. importorg.apache.hadoop.hdfs.DFSConfigKeys;
  44. importorg.apache.hadoop.hdfs.DFSUtil;
  45. importorg.apache.hadoop.hdfs.HdfsConfiguration;
  46. importorg.apache.hadoop.hdfs.StorageType;
  47. importorg.apache.hadoop.hdfs.server.balancer.Dispatcher.DDatanode;
  48. importorg.apache.hadoop.hdfs.server.balancer.Dispatcher.DDatanode.StorageGroup;
  49. importorg.apache.hadoop.hdfs.server.balancer.Dispatcher.Source;
  50. importorg.apache.hadoop.hdfs.server.balancer.Dispatcher.Task;
  51. importorg.apache.hadoop.hdfs.server.balancer.Dispatcher.Util;
  52. importorg.apache.hadoop.hdfs.server.blockmanagement.BlockPlacementPolicy;
  53. importorg.apache.hadoop.hdfs.server.blockmanagement.BlockPlacementPolicyDefault;
  54. importorg.apache.hadoop.hdfs.server.namenode.UnsupportedActionException;
  55. importorg.apache.hadoop.hdfs.server.protocol.DatanodeStorageReport;
  56. importorg.apache.hadoop.hdfs.server.protocol.StorageReport;
  57. importorg.apache.hadoop.util.StringUtils;
  58. importorg.apache.hadoop.util.Time;
  59. importorg.apache.hadoop.util.Tool;
  60. importorg.apache.hadoop.util.ToolRunner;
  62. importcom.google.common.base.Preconditions;
  64. /**<p>ThebalancerisatoolthatbalancesdiskspaceusageonanHDFScluster
  65. *whensomedatanodesbecomefullorwhennewemptynodesjointhecluster.
  66. *Thetoolisdeployedasanapplicationprogramthatcanberunbythe
  67. *clusteradministratoronaliveHDFSclusterwhileapplications
  68. *addinganddeletingfiles.
  69. *
  70. *<p>SYNOPSIS
  71. *<pre>
  72. *Tostart:
  73. *bin/start-balancer.sh[-threshold<threshold>]
  74. *Example:bin/start-balancer.sh
  75. *startthebalancerwithadefaultthresholdof10%
  76. *bin/start-balancer.sh-threshold5
  77. *startthebalancerwithathresholdof5%
  78. *Tostop:
  79. *bin/stop-balancer.sh
  80. *</pre>
  81. *
  82. *<p>DESCRIPTION
  83. *<p>Thethresholdparameterisafractionintherangeof(1%,100%)witha
  84. *defaultvalueof10%.Thethresholdsetsatargetforwhetherthecluster
  85. *isbalanced.Aclusterisbalancedifforeachdatanode,theutilization
  86. *ofthenode(ratioofusedspaceatthenodetototalcapacityofthenode)
  87. *differsfromtheutilizationofthe(ratioofusedspaceinthecluster
  88. *tototalcapacityofthecluster)bynomorethanthethresholdvalue.
  89. *Thesmallerthethreshold,themorebalancedaclusterwillbecome.
  90. *Ittakesmoretimetorunthebalancerforsmallthresholdvalues.
  91. *Alsoforaverysmallthresholdtheclustermaynotbeabletoreachthe
  92. *balancedstatewhenapplicationswriteanddeletefilesconcurrently.
  93. *
  94. *<p>Thetoolmovesblocksfromhighlyutilizeddatanodestopoorly
  95. *utilizeddatanodesiteratively.Ineachiterationadatanodemovesor
  96. *receivesnomorethanthelesserof10Gbytesorthethresholdfraction
  97. *ofitscapacity.Eachiterationrunsnomorethan20minutes.
  98. *每次移动不超过10G大小，每次移动不超过20分钟。
  99. *Attheendofeachiteration,thebalancerobtainsupdateddatanodes
  100. *informationfromthenamenode.
  101. *
  102. *<p>Asystempropertythatlimitsthebalancer'suseofbandwidthis
  103. *definedinthedefaultconfigurationfile:
  104. *<pre>
  105. *<property>
  106. *<name>dfs.balance.bandwidthPerSec</name>
  107. *<value>1048576</value>
  108. *<description>Specifiesthemaximumbandwidththateachdatanode
  109. *canutilizeforthebalancingpurposeintermofthenumberofbytes
  110. *persecond.</description>
  111. *</property>
  112. *</pre>
  113. *
  114. *<p>Thispropertydeterminesthemaximumspeedatwhichablockwillbe
  115. *movedfromonedatanodetoanother.Thedefaultvalueis1MB/s.Thehigher
  116. *thebandwidth,thefasteraclustercanreachthebalancedstate,
  117. *butwithgreatercompetitionwithapplicationprocesses.Ifan
  118. *administratorchangesthevalueofthispropertyintheconfiguration
  119. *file,thechangeisobservedwhenHDFSisnextrestarted.
  120. *
  121. *<p>MONITERINGBALANCERPROGRESS
  122. *<p>Afterthebalancerisstarted,anoutputfilenamewherethebalancer
  123. *progresswillberecordedisprintedonthescreen.Theadministrator
  124. *canmonitortherunningofthebalancerbyreadingtheoutputfile.
  125. *Theoutputshowsthebalancer'sstatusiterationbyiteration.Ineach
  126. *iterationitprintsthestartingtime,theiterationnumber,thetotal
  127. *numberofbytesthathavebeenmovedinthepreviousiterations,
  128. *thetotalnumberofbytesthatarelefttomoveinorderforthecluster
  129. *tobebalanced,andthenumberofbytesthatarebeingmovedinthis
  130. *iteration.Normally"BytesAlreadyMoved"isincreasingwhile"BytesLeft
  131. *ToMove"isdecreasing.
  132. *
  133. *<p>RunningmultipleinstancesofthebalancerinanHDFSclusteris
  134. *prohibitedbythetool.
  135. *
  136. *<p>Thebalancerautomaticallyexitswhenanyofthefollowingfive
  137. *conditionsissatisfied:
  138. *<ol>
  139. *<li>Theclusterisbalanced;
  140. *<li>Noblockcanbemoved;
  141. *<li>Noblockhasbeenmovedforfiveconsecutive(连续)iterations;
  142. *<li>AnIOExceptionoccurswhilecommunicatingwiththenamenode;
  143. *<li>Anotherbalancerisrunning.
  144. *</ol>
  145. *下面5种情况会导致Balance操作的失败
  146. *1、整个集群已经达到平衡状态
  147. *2、经过计算发现没有可以被移动的block块
  148. *3、在连续5次的迭代中，没有block块被移动
  149. *4、当datanode节点与namenode节点通信的时候，发生IO异常
  150. *5、已经存在一个Balance操作
  151. *
  152. *<p>Uponexit,abalancerreturnsanexitcodeandprintsoneofthe
  153. *followingmessagestotheoutputfileincorrespondingtotheaboveexit
  154. *reasons:
  155. *<ol>
  156. *<li>Theclusterisbalanced.Exiting
  157. *<li>Noblockcanbemoved.Exiting...
  158. *<li>Noblockhasbeenmovedfor5iterations.Exiting...
  159. *<li>ReceivedanIOexception:failurereason.Exiting...
  160. *<li>Anotherbalancerisrunning.Exiting...
  161. *</ol>
  162. *在下面的5种情况下，balancer操作会自动退出
  163. *1、整个集群已经达到平衡的状态
  164. *2、经过计算发现没有可以被移动block块
  165. *3、在5论的迭代没有block被移动
  166. *4、接收端发生了I异常
  167. *5、已经存在一个balanr进程在工作
  168. *
  169. *<p>Theadministratorcaninterrupttheexecutionofthebalanceratany
  170. *timebyrunningthecommand"stop-balancer.sh"onthemachinewherethe
  171. *balancerisrunning.
  172. */
  174. @InterfaceAudience.Private
  175. publicclassBalancer{
  176. staticfinalLogLOG=LogFactory.getLog(Balancer.class);
  178. privatestaticfinalPathBALANCER_ID_PATH=newPath("/system/balancer.id");
  180. privatestaticfinallongGB=1L<<30;//1GB
  181. privatestaticfinallongMAX_SIZE_TO_MOVE=10*GB;
  183. privatestaticfinalStringUSAGE="Usage:java"
  184. +Balancer.class.getSimpleName()
  185. +"\n\t[-policy<policy>]\tthebalancingpolicy:"
  186. +BalancingPolicy.Node.INSTANCE.getName()+"or"
  187. +BalancingPolicy.Pool.INSTANCE.getName()
  188. +"\n\t[-threshold<threshold>]\tPercentageofdiskcapacity"
  189. +"\n\t[-exclude[-f<hosts-file>|comma-speratedlistofhosts]]"
  190. +"\tExcludesthespecifieddatanodes."
  191. +"\n\t[-include[-f<hosts-file>|comma-speratedlistofhosts]]"
  192. +"\tIncludesonlythespecifieddatanodes.";
  194. privatefinalDispatcherdispatcher;
  195. privatefinalBalancingPolicypolicy;
  196. privatefinaldoublethreshold;
  198. //alldatanodelists
  199. //四种datanode节点类型
  200. privatefinalCollection<Source>overUtilized=newLinkedList<Source>();
  201. privatefinalCollection<Source>aboveAvgUtilized=newLinkedList<Source>();
  202. privatefinalCollection<StorageGroup>belowAvgUtilized
  203. =newLinkedList<StorageGroup>();
  204. privatefinalCollection<StorageGroup>underUtilized
  205. =newLinkedList<StorageGroup>();
  207. /*CheckthatthisBalanceriscompatible(兼容)withtheBlockPlacementPolicy
  208. *usedbytheNamenode.
  209. *检测此balancer均衡工具是否于与目前的namenode节点所用的block存放策略相兼容
  210. */
  211. privatestaticvoidcheckReplicationPolicyCompatibility(Configurationconf
  212. )throwsUnsupportedActionException{
  213. if(!(BlockPlacementPolicy.getInstance(conf,null,null,null)instanceof
  214. BlockPlacementPolicyDefault)){
  215. thrownewUnsupportedActionException(
  216. //如果不兼容则抛异常
  217. "BalancerwithoutBlockPlacementPolicyDefault");
  218. }
  219. }
  221. /**
  222. *Constructabalancer.
  223. *Initializebalancer.Itsetsthevalueofthethreshold,and
  224. *buildsthecommunicationproxiesto
  225. *namenodeasaclientandasecondarynamenodeandretryproxies
  226. *whenconnectionfails.
  227. *
  228. *构造一个balancer均衡器,设置threshold值,读取配置中的分发线程数的值
  229. */
  230. Balancer(NameNodeConnectortheblockpool,Parametersp,Configurationconf){
  231. finallongmovedWinWidth=conf.getLong(
  232. DFSConfigKeys.DFS_BALANCER_MOVEDWINWIDTH_KEY,
  233. DFSConfigKeys.DFS_BALANCER_MOVEDWINWIDTH_DEFAULT);
  234. //移动线程数
  235. finalintmoverThreads=conf.getInt(
  236. DFSConfigKeys.DFS_BALANCER_MOVERTHREADS_KEY,
  237. DFSConfigKeys.DFS_BALANCER_MOVERTHREADS_DEFAULT);
  238. //分发线程数
  239. finalintdispatcherThreads=conf.getInt(
  240. DFSConfigKeys.DFS_BALANCER_DISPATCHERTHREADS_KEY,
  241. DFSConfigKeys.DFS_BALANCER_DISPATCHERTHREADS_DEFAULT);
  242. finalintmaxConcurrentMovesPerNode=conf.getInt(
  243. DFSConfigKeys.DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY,
  244. DFSConfigKeys.DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_DEFAULT);
  246. this.dispatcher=newDispatcher(theblockpool,p.nodesToBeIncluded,
  247. p.nodesToBeExcluded,movedWinWidth,moverThreads,dispatcherThreads,
  248. maxConcurrentMovesPerNode,conf);
  249. this.threshold=p.threshold;
  250. this.policy=p.policy;
  251. }
  253. /**
  254. *获取节点总容量大小
  255. */
  256. privatestaticlonggetCapacity(DatanodeStorageReportreport,StorageTypet){
  257. longcapacity=0L;
  258. for(StorageReportr:report.getStorageReports()){
  259. if(r.getStorage().getStorageType()==t){
  260. capacity+=r.getCapacity();
  261. }
  262. }
  263. returncapacity;
  264. }
  266. /**
  267. *获取节点剩余可用容量大小
  268. */
  269. privatestaticlonggetRemaining(DatanodeStorageReportreport,StorageTypet){
  270. longremaining=0L;
  271. for(StorageReportr:report.getStorageReports()){
  272. if(r.getStorage().getStorageType()==t){
  273. remaining+=r.getRemaining();
  274. }
  275. }
  276. returnremaining;
  277. }
  279. /**
  280. *Givenadatanodestorageset,buildanetworktopologyanddecide
  281. *over-utilizedstorages,aboveaverageutilizedstorages,
  282. *belowaverageutilizedstorages,andunderutilizedstorages.
  283. *Theinputdatanodestoragesetisshuffledinordertorandomize
  284. *tothestoragematchinglateron.
  285. *给定一个datanode集合,创建一个网络拓扑逻辑并划分出过度使用,使用率超出平均值,
  286. *低于平均值的,以及未能充分利用资源的4种类型
  287. *
  288. *@returnthenumberofbytesneededtomoveinordertobalancethecluster.
  289. */
  290. privatelonginit(List<DatanodeStorageReport>reports){
  291. //计算平均使用率
  292. for(DatanodeStorageReportr:reports){
  293. //累加每个节点上的使用空间
  294. policy.accumulateSpaces(r);
  295. }
  296. //计算出平均值
  297. policy.initAvgUtilization();
  299. longoverLoadedBytes=0L,underLoadedBytes=0L;
  300. //进行使用率等级的划分,总共4种,over-utilized,above-average,below-averageandunder-utilized
  301. for(DatanodeStorageReportr:reports){
  302. finalDDatanodedn=dispatcher.newDatanode(r);
  303. for(StorageTypet:StorageType.asList()){
  304. finalDoubleutilization=policy.getUtilization(r,t);
  305. if(utilization==null){//datanodedoesnothavesuchstoragetype
  306. continue;
  307. }
  309. finallongcapacity=getCapacity(r,t);
  310. finaldoubleutilizationDiff=utilization-policy.getAvgUtilization(t);
  311. finaldoublethresholdDiff=Math.abs(utilizationDiff)-threshold;
  312. //计算理论上最大移动空间
  313. finallongmaxSize2Move=computeMaxSize2Move(capacity,
  314. getRemaining(r,t),utilizationDiff,threshold);
  316. finalStorageGroupg;
  317. if(utilizationDiff>0){
  318. //使用率超出平均值,加入发送节点列表中
  319. finalSources=dn.addSource(t,maxSize2Move,dispatcher);
  320. if(thresholdDiff<=0){//withinthreshold
  321. //如果在threshold范围之内,则加入aboveAvgUtilized
  322. aboveAvgUtilized.add(s);
  323. }else{
  324. //否则加入overUtilized,并计算超出空间
  325. overLoadedBytes+=precentage2bytes(thresholdDiff,capacity);
  326. overUtilized.add(s);
  327. }
  328. g=s;
  329. }else{
  330. //与上面的相反
  331. g=dn.addStorageGroup(t,maxSize2Move);
  332. if(thresholdDiff<=0){//withinthreshold
  333. belowAvgUtilized.add(g);
  334. }else{
  335. underLoadedBytes+=precentage2bytes(thresholdDiff,capacity);
  336. underUtilized.add(g);
  337. }
  338. }
  339. dispatcher.getStorageGroupMap().put(g);
  340. }
  341. }
  343. logUtilizationCollections();
  345. Preconditions.checkState(dispatcher.getStorageGroupMap().size()
  346. ==overUtilized.size()+underUtilized.size()+aboveAvgUtilized.size()
  347. +belowAvgUtilized.size(),
  348. "Mismatchednumberofstoragegroups");
  350. //returnnumberofbytestobemovedinordertomaketheclusterbalanced
  351. returnMath.max(overLoadedBytes,underLoadedBytes);
  352. }
  354. privatestaticlongcomputeMaxSize2Move(finallongcapacity,finallongremaining,
  355. finaldoubleutilizationDiff,finaldoublethreshold){
  356. finaldoublediff=Math.min(threshold,Math.abs(utilizationDiff));
  357. longmaxSizeToMove=precentage2bytes(diff,capacity);
  358. if(utilizationDiff<0){
  359. maxSizeToMove=Math.min(remaining,maxSizeToMove);
  360. }
  361. returnMath.min(MAX_SIZE_TO_MOVE,maxSizeToMove);
  362. }
  364. privatestaticlongprecentage2bytes(doubleprecentage,longcapacity){
  365. Preconditions.checkArgument(precentage>=0,
  366. "precentage="+precentage+"<0");
  367. return(long)(precentage*capacity/100.0);
  368. }
  370. /*logtheoverutilized&underutilizednodes*/
  371. privatevoidlogUtilizationCollections(){
  372. logUtilizationCollection("over-utilized",overUtilized);
  373. if(LOG.isTraceEnabled()){
  374. logUtilizationCollection("above-average",aboveAvgUtilized);
  375. logUtilizationCollection("below-average",belowAvgUtilized);
  376. }
  377. logUtilizationCollection("underutilized",underUtilized);
  378. }
  380. privatestatic<TextendsStorageGroup>
  381. voidlogUtilizationCollection(Stringname,Collection<T>items){
  382. LOG.info(items.size()+""+name+":"+items);
  383. }
  385. /**
  386. *Decideall<source,target>pairsand
  387. *thenumberofbytestomovefromasourcetoatarget
  388. *Maximumbytestobemovedperstoragegroupis
  389. *min(1Bandworthofbytes,MAX_SIZE_TO_MOVE).
  390. *从源节点列表和目标节点列表中各自选择节点组成一个个对,选择顺序优先为同节点组,同机架,然后是针对所有
  391. *@returntotalnumberofbytestomoveinthisiteration
  392. */
  393. privatelongchooseStorageGroups(){
  394. //First,matchnodesonthesamenodegroupifclusterisnodegroupaware
  395. if(dispatcher.getCluster().isNodeGroupAware()){
  396. //首先匹配的条件是同节点组
  397. chooseStorageGroups(Matcher.SAME_NODE_GROUP);
  398. }
  400. //Then,matchnodesonthesamerack
  401. //然后是同机架
  402. chooseStorageGroups(Matcher.SAME_RACK);
  403. //Atlast,matchallremainingnodes
  404. //最后是匹配所有的节点
  405. chooseStorageGroups(Matcher.ANY_OTHER);
  407. returndispatcher.bytesToMove();
  408. }
  410. /**Decideall<source,target>pairsaccordingtothematcher.*/
  411. privatevoidchooseStorageGroups(finalMatchermatcher){
  412. /*firststep:matcheachoverUtilizeddatanode(source)to
  413. *oneormoreunderUtilizeddatanodes(targets).
  414. *
  415. *把over组的数据移动under组中
  416. */
  417. chooseStorageGroups(overUtilized,underUtilized,matcher);
  419. /*matcheachremainingoverutilizeddatanode(source)to
  420. *belowaverageutilizeddatanodes(targets).
  421. *Noteonlyoverutilizeddatanodesthathaven'thadthatmaxbytestomove
  422. *satisfiedinstep1areselected
  424. *把over组的数据移动到below
  425. */
  426. chooseStorageGroups(overUtilized,belowAvgUtilized,matcher);
  428. /*matcheachremainingunderutilizeddatanode(target)to
  429. *aboveaverageutilizeddatanodes(source).
  430. *Noteonlyunderutilizeddatanodesthathavenothadthatmaxbytesto
  431. *movesatisfiedinstep1areselected.
  432. *
  433. *然后,再把under组的数据移动一部分到above组中
  434. */
  435. chooseStorageGroups(underUtilized,aboveAvgUtilized,matcher);
  436. }
  438. /**
  439. *Foreachdatanode,choosematchingnodesfromthecandidates.Eitherthe
  440. *datanodesorthecandidatesaresourcenodeswith(utilization>Avg),and
  441. *theothersaretargetnodeswith(utilization<Avg).
  442. */
  443. private<GextendsStorageGroup,CextendsStorageGroup>
  444. voidchooseStorageGroups(Collection<G>groups,Collection<C>candidates,
  445. Matchermatcher){
  446. for(finalIterator<G>i=groups.iterator();i.hasNext();){
  447. finalGg=i.next();
  448. for(;choose4One(g,candidates,matcher););
  449. if(!g.hasSpaceForScheduling()){
  450. //如果候选节点没有空间调度,则直接移除掉
  451. i.remove();
  452. }
  453. }
  454. }
  456. /**
  457. *Forthegivendatanode,chooseacandidateandthenscheduleit.
  458. *@returntrueifacandidateischosen;falseifnocandidatesischosen.
  459. */
  460. private<CextendsStorageGroup>booleanchoose4One(StorageGroupg,
  461. Collection<C>candidates,Matchermatcher){
  462. finalIterator<C>i=candidates.iterator();
  463. finalCchosen=chooseCandidate(g,i,matcher);
  465. if(chosen==null){
  466. returnfalse;
  467. }
  468. if(ginstanceofSource){
  469. matchSourceWithTargetToMove((Source)g,chosen);
  470. }else{
  471. matchSourceWithTargetToMove((Source)chosen,g);
  472. }
  473. if(!chosen.hasSpaceForScheduling()){
  474. i.remove();
  475. }
  476. returntrue;
  477. }
  479. /**
  480. *根据源节点和目标节点,构造任务对
  481. */
  482. privatevoidmatchSourceWithTargetToMove(Sourcesource,StorageGrouptarget){
  483. longsize=Math.min(source.availableSizeToMove(),target.availableSizeToMove());
  484. finalTasktask=newTask(target,size);
  485. source.addTask(task);
  486. target.incScheduledSize(task.getSize());
  487. //加入分发器中
  488. dispatcher.add(source,target);
  489. LOG.info("Decidedtomove"+StringUtils.byteDesc(size)+"bytesfrom"
  490. +source.getDisplayName()+"to"+target.getDisplayName());
  491. }
  493. /**Chooseacandidateforthegivendatanode.*/
  494. private<GextendsStorageGroup,CextendsStorageGroup>
  495. CchooseCandidate(Gg,Iterator<C>candidates,Matchermatcher){
  496. if(g.hasSpaceForScheduling()){
  497. for(;candidates.hasNext();){
  498. finalCc=candidates.next();
  499. if(!c.hasSpaceForScheduling()){
  500. candidates.remove();
  501. }elseif(matcher.match(dispatcher.getCluster(),
  502. g.getDatanodeInfo(),c.getDatanodeInfo())){
  503. //如果满足匹配的条件,则返回值
  504. returnc;
  505. }
  506. }
  507. }
  508. returnnull;
  509. }
  511. /*resetallfieldsinabalancerpreparingforthenextiteration*/
  512. privatevoidresetData(Configurationconf){
  513. this.overUtilized.clear();
  514. this.aboveAvgUtilized.clear();
  515. this.belowAvgUtilized.clear();
  516. this.underUtilized.clear();
  517. this.policy.reset();
  518. dispatcher.reset(conf);;
  519. }
  521. /**Runaniterationforalldatanodes.*/
  522. privateExitStatusrun(intiteration,Formatterformatter,
  523. Configurationconf){
  524. try{
  525. finalList<DatanodeStorageReport>reports=dispatcher.init();
  526. finallongbytesLeftToMove=init(reports);
  527. if(bytesLeftToMove==0){
  528. System.out.println("Theclusterisbalanced.Exiting...");
  529. returnExitStatus.SUCCESS;
  530. }else{
  531. LOG.info("Needtomove"+StringUtils.byteDesc(bytesLeftToMove)
  532. +"tomaketheclusterbalanced.");
  533. }
  535. /*Decideallthenodesthatwillparticipateintheblockmoveand
  536. *thenumberofbytesthatneedtobemovedfromonenodetoanother
  537. *inthisiteration.Maximumbytestobemovedpernodeis
  538. *Min(1Bandworthofbytes,MAX_SIZE_TO_MOVE).
  539. */
  540. finallongbytesToMove=chooseStorageGroups();
  541. if(bytesToMove==0){
  542. System.out.println("Noblockcanbemoved.Exiting...");
  543. returnExitStatus.NO_MOVE_BLOCK;
  544. }else{
  545. LOG.info("Willmove"+StringUtils.byteDesc(bytesToMove)+
  546. "inthisiteration");
  547. }
  549. formatter.format("%-24s%10d%19s%18s%17s%n",
  550. DateFormat.getDateTimeInstance().format(newDate()),
  551. iteration,
  552. StringUtils.byteDesc(dispatcher.getBytesMoved()),
  553. StringUtils.byteDesc(bytesLeftToMove),
  554. StringUtils.byteDesc(bytesToMove)
  555. );
  557. /*Foreachpairof<source,target>,startathreadthatrepeatedly
  558. *decideablocktobemovedanditsproxysource,
  559. *theninitiatesthemoveuntilallbytesaremovedornomoreblock
  560. *availabletomove.
  561. *Exitnobytehasbeenmovedfor5consecutiveiterations.
  562. *
  563. *如果发现在5次连续的迭代中还是没有字节被移动,则退出
  564. */
  565. if(!dispatcher.dispatchAndCheckContinue()){
  566. returnExitStatus.NO_MOVE_PROGRESS;
  567. }
  569. returnExitStatus.IN_PROGRESS;
  570. }catch(IllegalArgumentExceptione){
  571. System.out.println(e+".Exiting...");
  572. returnExitStatus.ILLEGAL_ARGUMENTS;
  573. }catch(IOExceptione){
  574. System.out.println(e+".Exiting...");
  575. returnExitStatus.IO_EXCEPTION;
  576. }catch(InterruptedExceptione){
  577. System.out.println(e+".Exiting...");
  578. returnExitStatus.INTERRUPTED;
  579. }finally{
  580. dispatcher.shutdownNow();
  581. }
  582. }
  584. /**
  585. *Balanceallnamenodes.
  586. *Foreachiteration,
  587. *foreachnamenode,
  588. *executea{@linkBalancer}toworkthroughalldatanodesonce.
  589. *
  590. *开放给外部调用的run方法
  591. */
  592. staticintrun(Collection<URI>namenodes,finalParametersp,
  593. Configurationconf)throwsIOException,InterruptedException{
  594. finallongsleeptime=2000*conf.getLong(
  595. DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY,
  596. DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT);
  597. LOG.info("namenodes="+namenodes);
  598. LOG.info("parameters="+p);
  600. finalFormatterformatter=newFormatter(System.out);
  601. System.out.println("TimeStampIteration#BytesAlreadyMovedBytesLeftToMoveBytesBeingMoved");
  603. finalList<NameNodeConnector>connectors
  604. =newArrayList<NameNodeConnector>(namenodes.size());
  605. try{
  606. for(URIuri:namenodes){
  607. finalNameNodeConnectornnc=newNameNodeConnector(
  608. Balancer.class.getSimpleName(),uri,BALANCER_ID_PATH,conf);
  609. nnc.getKeyManager().startBlockKeyUpdater();
  610. connectors.add(nnc);
  611. }
  613. booleandone=false;
  614. for(intiteration=0;!done;iteration++){
  615. done=true;
  616. Collections.shuffle(connectors);
  617. for(NameNodeConnectornnc:connectors){
  618. //初始化均衡器具
  619. finalBalancerb=newBalancer(nnc,p,conf);
  620. //均衡器执行balance操作
  621. finalExitStatusr=b.run(iteration,formatter,conf);
  622. //cleanalllists
  623. b.resetData(conf);
  624. if(r==ExitStatus.IN_PROGRESS){
  625. done=false;
  626. }elseif(r!=ExitStatus.SUCCESS){
  627. //mustbeanerrorstatue,return.
  628. returnr.getExitCode();
  629. }
  630. }
  632. if(!done){
  633. Thread.sleep(sleeptime);
  634. }
  635. }
  636. }finally{
  637. for(NameNodeConnectornnc:connectors){
  638. nnc.close();
  639. }
  640. }
  641. returnExitStatus.SUCCESS.getExitCode();
  642. }
  644. /*GivenelaspedTimeinms,returnaprintablestring*/
  645. privatestaticStringtime2Str(longelapsedTime){
  646. Stringunit;
  647. doubletime=elapsedTime;
  648. if(elapsedTime<1000){
  649. unit="milliseconds";
  650. }elseif(elapsedTime<60*1000){
  651. unit="seconds";
  652. time=time/1000;
  653. }elseif(elapsedTime<3600*1000){
  654. unit="minutes";
  655. time=time/(60*1000);
  656. }else{
  657. unit="hours";
  658. time=time/(3600*1000);
  659. }
  661. returntime+""+unit;
  662. }
  664. staticclassParameters{
  665. staticfinalParametersDEFAULT=newParameters(
  666. BalancingPolicy.Node.INSTANCE,10.0,
  667. Collections.<String>emptySet(),Collections.<String>emptySet());
  669. finalBalancingPolicypolicy;
  670. finaldoublethreshold;
  671. //excludethenodesinthissetfrombalancingoperations
  672. Set<String>nodesToBeExcluded;
  673. //includeonlythesenodesinbalancingoperations
  674. Set<String>nodesToBeIncluded;
  676. Parameters(BalancingPolicypolicy,doublethreshold,
  677. Set<String>nodesToBeExcluded,Set<String>nodesToBeIncluded){
  678. this.policy=policy;
  679. this.threshold=threshold;
  680. this.nodesToBeExcluded=nodesToBeExcluded;
  681. this.nodesToBeIncluded=nodesToBeIncluded;
  682. }
  684. @Override
  685. publicStringtoString(){
  686. returnBalancer.class.getSimpleName()+"."+getClass().getSimpleName()
  687. +"["+policy+",threshold="+threshold+
  688. ",numberofnodestobeexcluded="+nodesToBeExcluded.size()+
  689. ",numberofnodestobeincluded="+nodesToBeIncluded.size()+"]";
  690. }
  691. }
  693. staticclassCliextendsConfiguredimplementsTool{
  694. /**
  695. *ParseargumentsandthenrunBalancer.
  696. *
  697. *@paramargscommandspecificarguments.
  698. *@returnexitcode.0indicatessuccess,non-zeroindicatesfailure.
  699. */
  700. @Override
  701. publicintrun(String[]args){
  702. finallongstartTime=Time.now();
  703. finalConfigurationconf=getConf();
  705. try{
  706. checkReplicationPolicyCompatibility(conf);
  708. finalCollection<URI>namenodes=DFSUtil.getNsServiceRpcUris(conf);
  709. returnBalancer.run(namenodes,parse(args),conf);
  710. }catch(IOExceptione){
  711. System.out.println(e+".Exiting...");
  712. returnExitStatus.IO_EXCEPTION.getExitCode();
  713. }catch(InterruptedExceptione){
  714. System.out.println(e+".Exiting...");
  715. returnExitStatus.INTERRUPTED.getExitCode();
  716. }finally{
  717. System.out.format("%-24s",DateFormat.getDateTimeInstance().format(newDate()));
  718. System.out.println("Balancingtook"+time2Str(Time.now()-startTime));
  719. }
  720. }
  722. /**parsecommandlinearguments*/
  723. staticParametersparse(String[]args){
  724. BalancingPolicypolicy=Parameters.DEFAULT.policy;
  725. doublethreshold=Parameters.DEFAULT.threshold;
  726. Set<String>nodesTobeExcluded=Parameters.DEFAULT.nodesToBeExcluded;
  727. Set<String>nodesTobeIncluded=Parameters.DEFAULT.nodesToBeIncluded;
  729. if(args!=null){
  730. try{
  731. for(inti=0;i<args.length;i++){
  732. if("-threshold".equalsIgnoreCase(args[i])){
  733. checkArgument(++i<args.length,
  734. "Thresholdvalueismissing:args="+Arrays.toString(args));
  735. try{
  736. threshold=Double.parseDouble(args[i]);
  737. if(threshold<1||threshold>100){
  738. thrownewIllegalArgumentException(
  739. "Numberoutofrange:threshold="+threshold);
  740. }
  741. LOG.info("Usingathresholdof"+threshold);
  742. }catch(IllegalArgumentExceptione){
  743. System.err.println(
  744. "Expectinganumberintherangeof[1.0,100.0]:"
  745. +args[i]);
  746. throwe;
  747. }
  748. }elseif("-policy".equalsIgnoreCase(args[i])){
  749. checkArgument(++i<args.length,
  750. "Policyvalueismissing:args="+Arrays.toString(args));
  751. try{
  752. policy=BalancingPolicy.parse(args[i]);
  753. }catch(IllegalArgumentExceptione){
  754. System.err.println("Illegalpolicyname:"+args[i]);
  755. throwe;
  756. }
  757. }elseif("-exclude".equalsIgnoreCase(args[i])){
  758. checkArgument(++i<args.length,
  759. "Listofnodestoexclude|-f<filename>ismissing:args="
  760. +Arrays.toString(args));
  761. if("-f".equalsIgnoreCase(args[i])){
  762. checkArgument(++i<args.length,
  763. "Filecontainingnodestoexcludeisnotspecified:args="
  764. +Arrays.toString(args));
  765. nodesTobeExcluded=Util.getHostListFromFile(args[i],"exclude");
  766. }else{
  767. nodesTobeExcluded=Util.parseHostList(args[i]);
  768. }
  769. }elseif("-include".equalsIgnoreCase(args[i])){
  770. checkArgument(++i<args.length,
  771. "Listofnodestoinclude|-f<filename>ismissing:args="
  772. +Arrays.toString(args));
  773. if("-f".equalsIgnoreCase(args[i])){
  774. checkArgument(++i<args.length,
  775. "Filecontainingnodestoincludeisnotspecified:args="
  776. +Arrays.toString(args));
  777. nodesTobeIncluded=Util.getHostListFromFile(args[i],"include");
  778. }else{
  779. nodesTobeIncluded=Util.parseHostList(args[i]);
  780. }
  781. }else{
  782. thrownewIllegalArgumentException("args="
  783. +Arrays.toString(args));
  784. }
  785. }
  786. checkArgument(nodesTobeExcluded.isEmpty()||nodesTobeIncluded.isEmpty(),
  787. "-excludeand-includeoptionscannotbespecifiedtogether.");
  788. }catch(RuntimeExceptione){
  789. printUsage(System.err);
  790. throwe;
  791. }
  792. }
  794. returnnewParameters(policy,threshold,nodesTobeExcluded,nodesTobeIncluded);
  795. }
  797. privatestaticvoidprintUsage(PrintStreamout){
  798. out.println(USAGE+"\n");
  799. }
  800. }
  802. /**
  803. *Runabalancer
  804. *@paramargsCommandlinearguments
  805. */
  806. publicstaticvoidmain(String[]args){
  807. if(DFSUtil.parseHelpArgument(args,USAGE,System.out,true)){
  808. System.exit(0);
  809. }
  811. try{
  812. System.exit(ToolRunner.run(newHdfsConfiguration(),newCli(),args));
  813. }catch(Throwablee){
  814. LOG.error("Exitingbalancerdueanexception",e);
  815. System.exit(-1);
  816. }
  817. }
  818. }
下面是Dispatch.java类的解析：

[java]view plaincopyprint?
  1. /**
  2. *LicensedtotheApacheSoftwareFoundation(ASF)underone
  3. *ormorecontributorlicenseagreements.SeetheNOTICEfile
  4. *distributedwiththisworkforadditionalinformation
  5. *regardingcopyrightownership.TheASFlicensesthisfile
  6. *toyouundertheApacheLicense,Version2.0(the
  7. *"License");youmaynotusethisfileexceptincompliance
  8. *withtheLicense.YoumayobtainacopyoftheLicenseat
  9. *
  10. *http://www.apache.org/licenses/LICENSE-2.0
  11. *
  12. *Unlessrequiredbyapplicablelaworagreedtoinwriting,software
  13. *distributedundertheLicenseisdistributedonan"ASIS"BASIS,
  14. *WITHOUTWARRANTIESORCONDITIONSOFANYKIND,eitherexpressorimplied.
  15. *SeetheLicenseforthespecificlanguagegoverningpermissionsand
  16. *limitationsundertheLicense.
  17. */
  18. packageorg.apache.hadoop.hdfs.server.balancer;
  20. importstaticorg.apache.hadoop.hdfs.protocolPB.PBHelper.vintPrefixed;
  22. importjava.io.BufferedInputStream;
  23. importjava.io.BufferedOutputStream;
  24. importjava.io.DataInputStream;
  25. importjava.io.DataOutputStream;
  26. importjava.io.IOException;
  27. importjava.io.InputStream;
  28. importjava.io.OutputStream;
  29. importjava.net.Socket;
  30. importjava.util.ArrayList;
  31. importjava.util.Arrays;
  32. importjava.util.Collection;
  33. importjava.util.EnumMap;
  34. importjava.util.HashMap;
  35. importjava.util.HashSet;
  36. importjava.util.Iterator;
  37. importjava.util.List;
  38. importjava.util.Map;
  39. importjava.util.Set;
  40. importjava.util.concurrent.ExecutionException;
  41. importjava.util.concurrent.ExecutorService;
  42. importjava.util.concurrent.Executors;
  43. importjava.util.concurrent.Future;
  44. importjava.util.concurrent.atomic.AtomicLong;
  46. importorg.apache.commons.logging.Log;
  47. importorg.apache.commons.logging.LogFactory;
  48. importorg.apache.hadoop.classification.InterfaceAudience;
  49. importorg.apache.hadoop.conf.Configuration;
  50. importorg.apache.hadoop.fs.CommonConfigurationKeys;
  51. importorg.apache.hadoop.hdfs.DFSUtil;
  52. importorg.apache.hadoop.hdfs.StorageType;
  53. importorg.apache.hadoop.hdfs.protocol.Block;
  54. importorg.apache.hadoop.hdfs.protocol.DatanodeInfo;
  55. importorg.apache.hadoop.hdfs.protocol.ExtendedBlock;
  56. importorg.apache.hadoop.hdfs.protocol.HdfsConstants;
  57. importorg.apache.hadoop.hdfs.protocol.datatransfer.IOStreamPair;
  58. importorg.apache.hadoop.hdfs.protocol.datatransfer.Sender;
  59. importorg.apache.hadoop.hdfs.protocol.datatransfer.TrustedChannelResolver;
  60. importorg.apache.hadoop.hdfs.protocol.datatransfer.sasl.DataTransferSaslUtil;
  61. importorg.apache.hadoop.hdfs.protocol.datatransfer.sasl.SaslDataTransferClient;
  62. importorg.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto;
  63. importorg.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
  64. importorg.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
  65. importorg.apache.hadoop.hdfs.server.balancer.Dispatcher.DDatanode.StorageGroup;
  66. importorg.apache.hadoop.hdfs.server.common.HdfsServerConstants;
  67. importorg.apache.hadoop.hdfs.server.protocol.BlocksWithLocations;
  68. importorg.apache.hadoop.hdfs.server.protocol.BlocksWithLocations.BlockWithLocations;
  69. importorg.apache.hadoop.hdfs.server.protocol.DatanodeStorageReport;
  70. importorg.apache.hadoop.io.IOUtils;
  71. importorg.apache.hadoop.net.NetUtils;
  72. importorg.apache.hadoop.net.NetworkTopology;
  73. importorg.apache.hadoop.security.token.Token;
  74. importorg.apache.hadoop.util.HostsFileReader;
  75. importorg.apache.hadoop.util.StringUtils;
  76. importorg.apache.hadoop.util.Time;
  78. importcom.google.common.base.Preconditions;
  80. /**Dispatchingblockreplicamovesbetweendatanodes.*/
  81. @InterfaceAudience.Private
  82. publicclassDispatcher{
  83. staticfinalLogLOG=LogFactory.getLog(Dispatcher.class);
  85. privatestaticfinallongGB=1L<<30;//1GB
  86. privatestaticfinallongMAX_BLOCKS_SIZE_TO_FETCH=2*GB;
  88. privatestaticfinalintMAX_NO_PENDING_MOVE_ITERATIONS=5;
  89. privatestaticfinallongDELAY_AFTER_ERROR=10*1000L;//10seconds
  91. privatefinalNameNodeConnectornnc;
  92. privatefinalSaslDataTransferClientsaslClient;
  94. /**Setofdatanodestobeexcluded.*/
  95. privatefinalSet<String>excludedNodes;
  96. /**Restricttothefollowingnodes.*/
  97. privatefinalSet<String>includedNodes;
  99. privatefinalCollection<Source>sources=newHashSet<Source>();
  100. privatefinalCollection<StorageGroup>targets=newHashSet<StorageGroup>();
  102. privatefinalGlobalBlockMapglobalBlocks=newGlobalBlockMap();
  103. privatefinalMovedBlocks<StorageGroup>movedBlocks;
  105. /**Map(datanodeUuid,storageType->StorageGroup)*/
  106. privatefinalStorageGroupMapstorageGroupMap=newStorageGroupMap();
  108. privateNetworkTopologycluster;
  110. privatefinalExecutorServicemoveExecutor;
  111. privatefinalExecutorServicedispatchExecutor;
  112. /**Themaximumnumberofconcurrentblocksmovesatadatanode*/
  113. privatefinalintmaxConcurrentMovesPerNode;
  115. privatefinalAtomicLongbytesMoved=newAtomicLong();
  117. privatestaticclassGlobalBlockMap{
  118. privatefinalMap<Block,DBlock>map=newHashMap<Block,DBlock>();
  120. /**
  121. *Gettheblockfromthemap;
  122. *iftheblockisnotfound,createanewblockandputitinthemap.
  123. */
  124. privateDBlockget(Blockb){
  125. DBlockblock=map.get(b);
  126. if(block==null){
  127. block=newDBlock(b);
  128. map.put(b,block);
  129. }
  130. returnblock;
  131. }
  133. /**Removeallblocksexceptforthemovedblocks.*/
  134. privatevoidremoveAllButRetain(MovedBlocks<StorageGroup>movedBlocks){
  135. for(Iterator<Block>i=map.keySet().iterator();i.hasNext();){
  136. if(!movedBlocks.contains(i.next())){
  137. i.remove();
  138. }
  139. }
  140. }
  141. }
  143. staticclassStorageGroupMap{
  144. privatestaticStringtoKey(StringdatanodeUuid,StorageTypestorageType){
  145. returndatanodeUuid+":"+storageType;
  146. }
  148. privatefinalMap<String,StorageGroup>map=newHashMap<String,StorageGroup>();
  150. StorageGroupget(StringdatanodeUuid,StorageTypestorageType){
  151. returnmap.get(toKey(datanodeUuid,storageType));
  152. }
  154. voidput(StorageGroupg){
  155. finalStringkey=toKey(g.getDatanodeInfo().getDatanodeUuid(),g.storageType);
  156. finalStorageGroupexisting=map.put(key,g);
  157. Preconditions.checkState(existing==null);
  158. }
  160. intsize(){
  161. returnmap.size();
  162. }
  164. voidclear(){
  165. map.clear();
  166. }
  167. }
  169. /**Thisclasskeepstrackofascheduledblockmove*/
  170. privateclassPendingMove{
  171. privateDBlockblock;
  172. privateSourcesource;
  173. privateDDatanodeproxySource;
  174. privateStorageGrouptarget;
  176. privatePendingMove(){
  177. }
  179. @Override
  180. publicStringtoString(){
  181. finalBlockb=block.getBlock();
  182. returnb+"withsize="+b.getNumBytes()+"from"
  183. +source.getDisplayName()+"to"+target.getDisplayName()
  184. +"through"+proxySource.datanode;
  185. }
  187. /**
  188. *Chooseablock&aproxysourceforthispendingMovewhosesource&
  189. *targethavealreadybeenchosen.
  190. *
  191. *@returntrueifablockanditsproxyarechosen;falseotherwise
  192. */
  193. privatebooleanchooseBlockAndProxy(){
  194. //iterateallsource'sblocksuntilfindagoodone
  195. for(Iterator<DBlock>i=source.getBlockIterator();i.hasNext();){
  196. if(markMovedIfGoodBlock(i.next())){
  197. i.remove();
  198. returntrue;
  199. }
  200. }
  201. returnfalse;
  202. }
  204. /**
  205. *@returntrueifthegivenblockisgoodforthetentativemove.
  206. */
  207. privatebooleanmarkMovedIfGoodBlock(DBlockblock){
  208. synchronized(block){
  209. synchronized(movedBlocks){
  210. if(isGoodBlockCandidate(source,target,block)){
  211. this.block=block;
  212. if(chooseProxySource()){
  213. movedBlocks.put(block);
  214. if(LOG.isDebugEnabled()){
  215. LOG.debug("Decidedtomove"+this);
  216. }
  217. returntrue;
  218. }
  219. }
  220. }
  221. }
  222. returnfalse;
  223. }
  225. /**
  226. *Chooseaproxysource.
  227. *
  228. *@returntrueifaproxyisfound;otherwisefalse
  229. */
  230. privatebooleanchooseProxySource(){
  231. finalDatanodeInfotargetDN=target.getDatanodeInfo();
  232. //ifnodegroupissupported,firsttryaddnodesinthesamenodegroup
  233. if(cluster.isNodeGroupAware()){
  234. for(StorageGrouploc:block.getLocations()){
  235. if(cluster.isOnSameNodeGroup(loc.getDatanodeInfo(),targetDN)
  236. &&addTo(loc)){
  237. returntrue;
  238. }
  239. }
  240. }
  241. //checkifthereisreplicawhichisonthesamerackwiththetarget
  242. for(StorageGrouploc:block.getLocations()){
  243. if(cluster.isOnSameRack(loc.getDatanodeInfo(),targetDN)&&addTo(loc)){
  244. returntrue;
  245. }
  246. }
  247. //findoutanon-busyreplica
  248. for(StorageGrouploc:block.getLocations()){
  249. if(addTo(loc)){
  250. returntrue;
  251. }
  252. }
  253. returnfalse;
  254. }
  256. /**addtoaproxysourceforspecificblockmovement*/
  257. privatebooleanaddTo(StorageGroupg){
  258. finalDDatanodedn=g.getDDatanode();
  259. if(dn.addPendingBlock(this)){
  260. proxySource=dn;
  261. returntrue;
  262. }
  263. returnfalse;
  264. }
  266. /**Dispatchthemovetotheproxysource&waitfortheresponse.*/
  267. privatevoiddispatch(){
  268. if(LOG.isDebugEnabled()){
  269. LOG.debug("Startmoving"+this);
  270. }
  272. Socketsock=newSocket();
  273. DataOutputStreamout=null;
  274. DataInputStreamin=null;
  275. try{
  276. sock.connect(
  277. NetUtils.createSocketAddr(target.getDatanodeInfo().getXferAddr()),
  278. HdfsServerConstants.READ_TIMEOUT);
  280. sock.setKeepAlive(true);
  282. OutputStreamunbufOut=sock.getOutputStream();
  283. InputStreamunbufIn=sock.getInputStream();
  284. ExtendedBlockeb=newExtendedBlock(nnc.getBlockpoolID(),
  285. block.getBlock());
  286. finalKeyManagerkm=nnc.getKeyManager();
  287. Token<BlockTokenIdentifier>accessToken=km.getAccessToken(eb);
  288. IOStreamPairsaslStreams=saslClient.socketSend(sock,unbufOut,
  289. unbufIn,km,accessToken,target.getDatanodeInfo());
  290. unbufOut=saslStreams.out;
  291. unbufIn=saslStreams.in;
  292. out=newDataOutputStream(newBufferedOutputStream(unbufOut,
  293. HdfsConstants.IO_FILE_BUFFER_SIZE));
  294. in=newDataInputStream(newBufferedInputStream(unbufIn,
  295. HdfsConstants.IO_FILE_BUFFER_SIZE));
  297. sendRequest(out,eb,accessToken);
  298. receiveResponse(in);
  299. bytesMoved.addAndGet(block.getNumBytes());
  300. LOG.info("Successfullymoved"+this);
  301. }catch(IOExceptione){
  302. LOG.warn("Failedtomove"+this+":"+e.getMessage());
  303. //Proxyortargetmayhavesomeissues,delaybeforeusingthesenodes
  304. //furtherinordertoavoidapotentialstormof"threadsquota
  305. //exceeded"warningswhenthedispatchergetsoutofsyncwithwork
  306. //goingonindatanodes.
  307. proxySource.activateDelay(DELAY_AFTER_ERROR);
  308. target.getDDatanode().activateDelay(DELAY_AFTER_ERROR);
  309. }finally{
  310. IOUtils.closeStream(out);
  311. IOUtils.closeStream(in);
  312. IOUtils.closeSocket(sock);
  314. proxySource.removePendingBlock(this);
  315. target.getDDatanode().removePendingBlock(this);
  317. synchronized(this){
  318. reset();
  319. }
  320. synchronized(Dispatcher.this){
  321. Dispatcher.this.notifyAll();
  322. }
  323. }
  324. }
  326. /**Sendablockreplacerequesttotheoutputstream*/
  327. privatevoidsendRequest(DataOutputStreamout,ExtendedBlockeb,
  328. Token<BlockTokenIdentifier>accessToken)throwsIOException{
  329. newSender(out).replaceBlock(eb,target.storageType,accessToken,
  330. source.getDatanodeInfo().getDatanodeUuid(),proxySource.datanode);
  331. }
  333. /**Receiveablockcopyresponsefromtheinputstream*/
  334. privatevoidreceiveResponse(DataInputStreamin)throwsIOException{
  335. BlockOpResponseProtoresponse=
  336. BlockOpResponseProto.parseFrom(vintPrefixed(in));
  337. while(response.getStatus()==Status.IN_PROGRESS){
  338. //readintermediateresponses
  339. response=BlockOpResponseProto.parseFrom(vintPrefixed(in));
  340. }
  341. if(response.getStatus()!=Status.SUCCESS){
  342. if(response.getStatus()==Status.ERROR_ACCESS_TOKEN){
  343. thrownewIOException("blockmovefailedduetoaccesstokenerror");
  344. }
  345. thrownewIOException("blockmoveisfailed:"+response.getMessage());
  346. }
  347. }
  349. /**resettheobject*/
  350. privatevoidreset(){
  351. block=null;
  352. source=null;
  353. proxySource=null;
  354. target=null;
  355. }
  356. }
  358. /**Aclassforkeepingtrackofblocklocationsinthedispatcher.*/
  359. privatestaticclassDBlockextendsMovedBlocks.Locations<StorageGroup>{
  360. DBlock(Blockblock){
  361. super(block);
  362. }
  363. }
  365. /**Theclassrepresentsadesiredmove.*/
  366. staticclassTask{
  367. privatefinalStorageGrouptarget;
  368. privatelongsize;//bytesscheduledtomove
  370. Task(StorageGrouptarget,longsize){
  371. this.target=target;
  372. this.size=size;
  373. }
  375. longgetSize(){
  376. returnsize;
  377. }
  378. }
  380. /**Aclassthatkeepstrackofadatanode.*/
  381. staticclassDDatanode{
  383. /**Agroupofstoragesinadatanodewiththesamestoragetype.*/
  384. classStorageGroup{
  385. finalStorageTypestorageType;
  386. finallongmaxSize2Move;
  387. privatelongscheduledSize=0L;
  389. privateStorageGroup(StorageTypestorageType,longmaxSize2Move){
  390. this.storageType=storageType;
  391. this.maxSize2Move=maxSize2Move;
  392. }
  394. privateDDatanodegetDDatanode(){
  395. returnDDatanode.this;
  396. }
  398. DatanodeInfogetDatanodeInfo(){
  399. returnDDatanode.this.datanode;
  400. }
  402. /**Decideifstillneedtomovemorebytes*/
  403. synchronizedbooleanhasSpaceForScheduling(){
  404. returnavailableSizeToMove()>0L;
  405. }
  407. /**@returnthetotalnumberofbytesthatneedtobemoved*/
  408. synchronizedlongavailableSizeToMove(){
  409. returnmaxSize2Move-scheduledSize;
  410. }
  412. /**incrementscheduledsize*/
  413. synchronizedvoidincScheduledSize(longsize){
  414. scheduledSize+=size;
  415. }
  417. /**@returnscheduledsize*/
  418. synchronizedlonggetScheduledSize(){
  419. returnscheduledSize;
  420. }
  422. /**Resetscheduledsizetozero.*/
  423. synchronizedvoidresetScheduledSize(){
  424. scheduledSize=0L;
  425. }
  427. /**@returnthenamefordisplay*/
  428. StringgetDisplayName(){
  429. returndatanode+":"+storageType;
  430. }
  432. @Override
  433. publicStringtoString(){
  434. returngetDisplayName();
  435. }
  436. }
  438. finalDatanodeInfodatanode;
  439. finalEnumMap<StorageType,StorageGroup>storageMap
  440. =newEnumMap<StorageType,StorageGroup>(StorageType.class);
  441. protectedlongdelayUntil=0L;
  442. /**blocksbeingmovedbutnotconfirmedyet*/
  443. privatefinalList<PendingMove>pendings;
  444. privatefinalintmaxConcurrentMoves;
  446. @Override
  447. publicStringtoString(){
  448. returngetClass().getSimpleName()+":"+datanode+":"+storageMap.values();
  449. }
  451. privateDDatanode(DatanodeStorageReportr,intmaxConcurrentMoves){
  452. this.datanode=r.getDatanodeInfo();
  453. this.maxConcurrentMoves=maxConcurrentMoves;
  454. this.pendings=newArrayList<PendingMove>(maxConcurrentMoves);
  455. }
  457. privatevoidput(StorageTypestorageType,StorageGroupg){
  458. finalStorageGroupexisting=storageMap.put(storageType,g);
  459. Preconditions.checkState(existing==null);
  460. }
  462. StorageGroupaddStorageGroup(StorageTypestorageType,longmaxSize2Move){
  463. finalStorageGroupg=newStorageGroup(storageType,maxSize2Move);
  464. put(storageType,g);
  465. returng;
  466. }
  468. SourceaddSource(StorageTypestorageType,longmaxSize2Move,Dispatcherd){
  469. finalSources=d.newSource(storageType,maxSize2Move,this);
  470. put(storageType,s);
  471. returns;
  472. }
  474. synchronizedprivatevoidactivateDelay(longdelta){
  475. delayUntil=Time.monotonicNow()+delta;
  476. }
  478. synchronizedprivatebooleanisDelayActive(){
  479. if(delayUntil==0||Time.monotonicNow()>delayUntil){
  480. delayUntil=0;
  481. returnfalse;
  482. }
  483. returntrue;
  484. }
  486. /**Checkifthenodecanschedulemoreblockstomove*/
  487. synchronizedbooleanisPendingQNotFull(){
  488. returnpendings.size()<maxConcurrentMoves;
  489. }
  491. /**Checkifallthedispatchedmovesaredone*/
  492. synchronizedbooleanisPendingQEmpty(){
  493. returnpendings.isEmpty();
  494. }
  496. /**Addascheduledblockmovetothenode*/
  497. synchronizedbooleanaddPendingBlock(PendingMovependingBlock){
  498. if(!isDelayActive()&&isPendingQNotFull()){
  499. returnpendings.add(pendingBlock);
  500. }
  501. returnfalse;
  502. }
  504. /**Removeascheduledblockmovefromthenode*/
  505. synchronizedbooleanremovePendingBlock(PendingMovependingBlock){
  506. returnpendings.remove(pendingBlock);
  507. }
  508. }
  510. /**Anodethatcanbethesourcesofablockmove*/
  511. classSourceextendsDDatanode.StorageGroup{
  513. privatefinalList<Task>tasks=newArrayList<Task>(2);
  514. privatelongblocksToReceive=0L;
  515. /**
  516. *Sourceblockspointtotheobjectsin{@linkDispatcher#globalBlocks}
  517. *becausewewanttokeeponecopyofablockandbeawarethatthe
  518. *locationsarechangingovertime.
  519. */
  520. privatefinalList<DBlock>srcBlocks=newArrayList<DBlock>();
  522. privateSource(StorageTypestorageType,longmaxSize2Move,DDatanodedn){
  523. dn.super(storageType,maxSize2Move);
  524. }
  526. /**Addatask*/
  527. voidaddTask(Tasktask){
  528. Preconditions.checkState(task.target!=this,
  529. "Sourceandtargetarethesamestoragegroup"+getDisplayName());
  530. incScheduledSize(task.size);
  531. tasks.add(task);
  532. }
  534. /**@returnaniteratortothissource'sblocks*/
  535. Iterator<DBlock>getBlockIterator(){
  536. returnsrcBlocks.iterator();
  537. }
  539. /**
  540. *Fetchnewblocksofthissourcefromnamenodeandupdatethissource's
  541. *blocklist&{@linkDispatcher#globalBlocks}.
  542. *
  543. *@returnthetotalsizeofthereceivedblocksinthenumberofbytes.
  544. */
  545. privatelonggetBlockList()throwsIOException{
  546. finallongsize=Math.min(MAX_BLOCKS_SIZE_TO_FETCH,blocksToReceive);
  547. finalBlocksWithLocationsnewBlocks=nnc.getBlocks(getDatanodeInfo(),size);
  549. longbytesReceived=0;
  550. for(BlockWithLocationsblk:newBlocks.getBlocks()){
  551. bytesReceived+=blk.getBlock().getNumBytes();
  552. synchronized(globalBlocks){
  553. finalDBlockblock=globalBlocks.get(blk.getBlock());
  554. synchronized(block){
  555. block.clearLocations();
  557. //updatelocations
  558. finalString[]datanodeUuids=blk.getDatanodeUuids();
  559. finalStorageType[]storageTypes=blk.getStorageTypes();
  560. for(inti=0;i<datanodeUuids.length;i++){
  561. finalStorageGroupg=storageGroupMap.get(
  562. datanodeUuids[i],storageTypes[i]);
  563. if(g!=null){//notunknown
  564. block.addLocation(g);
  565. }
  566. }
  567. }
  568. if(!srcBlocks.contains(block)&&isGoodBlockCandidate(block)){
  569. //filterbadcandidates
  570. srcBlocks.add(block);
  571. }
  572. }
  573. }
  574. returnbytesReceived;
  575. }
  577. /**Decideifthegivenblockisagoodcandidatetomoveornot*/
  578. privatebooleanisGoodBlockCandidate(DBlockblock){
  579. for(Taskt:tasks){
  580. if(Dispatcher.this.isGoodBlockCandidate(this,t.target,block)){
  581. returntrue;
  582. }
  583. }
  584. returnfalse;
  585. }
  587. /**
  588. *Chooseamoveforthesource.Theblock'ssource,target,andproxy
  589. *aredeterminedtoo.Whenchoosingproxyandtarget,source&
  590. *targetthrottlinghasbeenconsidered.Theyarechosenonlywhenthey
  591. *havethecapacitytosupportthisblockmove.Theblockshouldbe
  592. *dispatchedimmediatelyafterthismethodisreturned.
  593. *
  594. *@returnamovethat'sgoodforthesourcetodispatchimmediately.
  595. */
  596. privatePendingMovechooseNextMove(){
  597. for(Iterator<Task>i=tasks.iterator();i.hasNext();){
  598. finalTasktask=i.next();
  599. finalDDatanodetarget=task.target.getDDatanode();
  600. PendingMovependingBlock=newPendingMove();
  601. if(target.addPendingBlock(pendingBlock)){
  602. //targetisnotbusy,sodoatentativeblockallocation
  603. pendingBlock.source=this;
  604. pendingBlock.target=task.target;
  605. if(pendingBlock.chooseBlockAndProxy()){
  606. longblockSize=pendingBlock.block.getNumBytes();
  607. incScheduledSize(-blockSize);
  608. task.size-=blockSize;
  609. if(task.size==0){
  610. i.remove();
  611. }
  612. returnpendingBlock;
  613. }else{
  614. //cancelthetentativemove
  615. target.removePendingBlock(pendingBlock);
  616. }
  617. }
  618. }
  619. returnnull;
  620. }
  622. /**Iterateallsource'sblockstoremovemovedones*/
  623. privatevoidremoveMovedBlocks(){
  624. for(Iterator<DBlock>i=getBlockIterator();i.hasNext();){
  625. if(movedBlocks.contains(i.next().getBlock())){
  626. i.remove();
  627. }
  628. }
  629. }
  631. privatestaticfinalintSOURCE_BLOCKS_MIN_SIZE=5;
  633. /**@returnifshouldfetchmoreblocksfromnamenode*/
  634. privatebooleanshouldFetchMoreBlocks(){
  635. returnsrcBlocks.size()<SOURCE_BLOCKS_MIN_SIZE&&blocksToReceive>0;
  636. }
  638. privatestaticfinallongMAX_ITERATION_TIME=20*60*1000L;//20mins
  640. /**
  641. *Thismethoditerativelydoesthefollowing:itfirstselectsablockto
  642. *move,thensendsarequesttotheproxysourcetostarttheblockmove
  643. *whenthesource'sblocklistfallsbelowathreshold,itasksthe
  644. *namenodeformoreblocks.Itterminateswhenithasdispatchenoughblock
  645. *movetasksorithasreceivedenoughblocksfromthenamenode,orthe
  646. *elapsedtimeoftheiterationhasexceededthemaxtimelimit.
  647. */
  648. privatevoiddispatchBlocks(){
  649. finallongstartTime=Time.monotonicNow();
  650. this.blocksToReceive=2*getScheduledSize();
  651. booleanisTimeUp=false;
  652. intnoPendingMoveIteration=0;
  653. while(!isTimeUp&&getScheduledSize()>0
  654. &&(!srcBlocks.isEmpty()||blocksToReceive>0)){
  655. finalPendingMovep=chooseNextMove();
  656. if(p!=null){
  657. //movetheblock
  658. moveExecutor.execute(newRunnable(){
  659. @Override
  660. publicvoidrun(){
  661. p.dispatch();
  662. }
  663. });
  664. continue;
  665. }
  667. //Sincewecannotscheduleanyblocktomove,
  668. //removeanymovedblocksfromthesourceblocklistand
  669. removeMovedBlocks();//filteralreadymovedblocks
  670. //checkifweshouldfetchmoreblocksfromthenamenode
  671. if(shouldFetchMoreBlocks()){
  672. //fetchnewblocks
  673. try{
  674. blocksToReceive-=getBlockList();
  675. continue;
  676. }catch(IOExceptione){
  677. LOG.warn("Exceptionwhilegettingblocklist",e);
  678. return;
  679. }
  680. }else{
  681. //sourcenodecannotfindapendingblocktomove,iteration+1
  682. noPendingMoveIteration++;
  683. //incasenoblockscanbemovedforsourcenode'stask,
  684. //jumpoutofwhile-loopafter5iterations.
  685. if(noPendingMoveIteration>=MAX_NO_PENDING_MOVE_ITERATIONS){
  686. resetScheduledSize();
  687. }
  688. }
  690. //checkiftimeisupornot
  691. if(Time.monotonicNow()-startTime>MAX_ITERATION_TIME){
  692. isTimeUp=true;
  693. continue;
  694. }
  696. //Nowwecannotscheduleanyblocktomoveandthereare
  697. //nonewblocksaddedtothesourceblocklist,sowewait.
  698. try{
  699. synchronized(Dispatcher.this){
  700. Dispatcher.this.wait(1000);//waitfortargets/sourcestobeidle
  701. }
  702. }catch(InterruptedExceptionignored){
  703. }
  704. }
  705. }
  706. }
  708. publicDispatcher(NameNodeConnectornnc,Set<String>includedNodes,
  709. Set<String>excludedNodes,longmovedWinWidth,intmoverThreads,
  710. intdispatcherThreads,intmaxConcurrentMovesPerNode,Configurationconf){
  711. this.nnc=nnc;
  712. this.excludedNodes=excludedNodes;
  713. this.includedNodes=includedNodes;
  714. this.movedBlocks=newMovedBlocks<StorageGroup>(movedWinWidth);
  716. this.cluster=NetworkTopology.getInstance(conf);
  718. this.moveExecutor=Executors.newFixedThreadPool(moverThreads);
  719. this.dispatchExecutor=Executors.newFixedThreadPool(dispatcherThreads);
  720. this.maxConcurrentMovesPerNode=maxConcurrentMovesPerNode;
  722. finalbooleanfallbackToSimpleAuthAllowed=conf.getBoolean(
  723. CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_KEY,
  724. CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_DEFAULT);
  725. this.saslClient=newSaslDataTransferClient(
  726. DataTransferSaslUtil.getSaslPropertiesResolver(conf),
  727. TrustedChannelResolver.getInstance(conf),fallbackToSimpleAuthAllowed);
  728. }
  730. StorageGroupMapgetStorageGroupMap(){
  731. returnstorageGroupMap;
  732. }
  734. NetworkTopologygetCluster(){
  735. returncluster;
  736. }
  738. longgetBytesMoved(){
  739. returnbytesMoved.get();
  740. }
  742. longbytesToMove(){
  743. Preconditions.checkState(
  744. storageGroupMap.size()>=sources.size()+targets.size(),
  745. "Mismatchednumberofstoragegroups("+storageGroupMap.size()
  746. +"<"+sources.size()+"sources+"+targets.size()
  747. +"targets)");
  749. longb=0L;
  750. for(Sourcesrc:sources){
  751. b+=src.getScheduledSize();
  752. }
  753. returnb;
  754. }
  756. voidadd(Sourcesource,StorageGrouptarget){
  757. sources.add(source);
  758. targets.add(target);
  759. }
  761. privatebooleanshouldIgnore(DatanodeInfodn){
  762. //ignoredecommissionednodes
  763. finalbooleandecommissioned=dn.isDecommissioned();
  764. //ignoredecommissioningnodes
  765. finalbooleandecommissioning=dn.isDecommissionInProgress();
  766. //ignorenodesinexcludelist
  767. finalbooleanexcluded=Util.isExcluded(excludedNodes,dn);
  768. //ignorenodesnotintheincludelist(ifincludelistisnotempty)
  769. finalbooleannotIncluded=!Util.isIncluded(includedNodes,dn);
  771. if(decommissioned||decommissioning||excluded||notIncluded){
  772. if(LOG.isTraceEnabled()){
  773. LOG.trace("Excludingdatanode"+dn+":"+decommissioned+","
  774. +decommissioning+","+excluded+","+notIncluded);
  775. }
  776. returntrue;
  777. }
  778. returnfalse;
  779. }
  781. /**Getlivedatanodestoragereportsandthenbuildthenetworktopology.*/
  782. List<DatanodeStorageReport>init()throwsIOException{
  783. finalDatanodeStorageReport[]reports=nnc.getLiveDatanodeStorageReport();
  784. finalList<DatanodeStorageReport>trimmed=newArrayList<DatanodeStorageReport>();
  785. //createnetworktopologyandclassifyutilizationcollections:
  786. //over-utilized,above-average,below-averageandunder-utilized.
  787. for(DatanodeStorageReportr:DFSUtil.shuffle(reports)){
  788. finalDatanodeInfodatanode=r.getDatanodeInfo();
  789. if(shouldIgnore(datanode)){
  790. continue;
  791. }
  792. trimmed.add(r);
  793. cluster.add(datanode);
  794. }
  795. returntrimmed;
  796. }
  798. publicDDatanodenewDatanode(DatanodeStorageReportr){
  799. returnnewDDatanode(r,maxConcurrentMovesPerNode);
  800. }
  802. publicbooleandispatchAndCheckContinue()throwsInterruptedException{
  803. returnnnc.shouldContinue(dispatchBlockMoves());
  804. }
  806. /**
  807. *Dispatchblockmovesforeachsource.Thethreadselectsblockstomove&
  808. *sendsrequesttoproxysourcetoinitiateblockmove.Theprocessisflow
  809. *controlled.Blockselectionisblockediftherearetoomanyun-confirmed
  810. *blockmoves.
  811. *
  812. *@returnthetotalnumberofbytessuccessfullymovedinthisiteration.
  813. */
  814. privatelongdispatchBlockMoves()throwsInterruptedException{
  815. finallongbytesLastMoved=bytesMoved.get();
  816. finalFuture<?>[]futures=newFuture<?>[sources.size()];
  818. finalIterator<Source>i=sources.iterator();
  819. for(intj=0;j<futures.length;j++){
  820. finalSources=i.next();
  821. futures[j]=dispatchExecutor.submit(newRunnable(){
  822. @Override
  823. publicvoidrun(){
  824. s.dispatchBlocks();
  825. }
  826. });
  827. }
  829. //waitforalldispatcherthreadstofinish
  830. for(Future<?>future:futures){
  831. try{
  832. future.get();
  833. }catch(ExecutionExceptione){
  834. LOG.warn("Dispatcherthreadfailed",e.getCause());
  835. }
  836. }
  838. //waitforallblockmovingtobedone
  839. waitForMoveCompletion();
  841. returnbytesMoved.get()-bytesLastMoved;
  842. }
  844. /**Thesleepingperiodbeforecheckingifblockmoveiscompletedagain*/
  845. staticprivatelongblockMoveWaitTime=30000L;
  847. /**setthesleepingperiodforblockmovecompletioncheck*/
  848. staticvoidsetBlockMoveWaitTime(longtime){
  849. blockMoveWaitTime=time;
  850. }
  852. /**Waitforallblockmoveconfirmations.*/
  853. privatevoidwaitForMoveCompletion(){
  854. for(;;){
  855. booleanempty=true;
  856. for(StorageGroupt:targets){
  857. if(!t.getDDatanode().isPendingQEmpty()){
  858. empty=false;
  859. break;
  860. }
  861. }
  862. if(empty){
  863. return;//allpendingqueuesareempty
  864. }
  865. try{
  866. Thread.sleep(blockMoveWaitTime);
  867. }catch(InterruptedExceptionignored){
  868. }
  869. }
  870. }
  872. /**
  873. *Decideiftheblockisagoodcandidatetobemovedfromsourcetotarget.
  874. *Ablockisagoodcandidateif
  875. *1.theblockisnotintheprocessofbeingmoved/hasnotbeenmoved;
  876. *移动的块不是正在被移动的块
  877. *2.theblockdoesnothaveareplicaonthetarget;
  878. *在目标节点上没有移动的block块
  879. *3.doingthemovedoesnotreducethenumberofracksthattheblockhas
  880. *移动之后,不同机架上的block块的数量应该是不变的.
  881. */
  882. privatebooleanisGoodBlockCandidate(Sourcesource,StorageGrouptarget,
  883. DBlockblock){
  884. if(source.storageType!=target.storageType){
  885. returnfalse;
  886. }
  887. //checkiftheblockismovedornot
  888. //如果所要移动的块是存在于正在被移动的块列表,则返回false
  889. if(movedBlocks.contains(block.getBlock())){
  890. returnfalse;
  891. }
  892. //如果移动的块已经存在于目标节点上,则返回false,将不会予以移动
  893. if(block.isLocatedOn(target)){
  894. returnfalse;
  895. }
  897. //如果开启了机架感知的配置,则目标节点不应该有相同的block
  898. if(cluster.isNodeGroupAware()
  899. &&isOnSameNodeGroupWithReplicas(target,block,source)){
  900. returnfalse;
  901. }
  903. //需要维持机架上的block块数量不变
  904. if(reduceNumOfRacks(source,target,block)){
  905. returnfalse;
  906. }
  907. returntrue;
  908. }
  910. /**
  911. *Determinewhethermovingthegivenblockreplicafromsourcetotarget
  912. *wouldreducethenumberofracksoftheblockreplicas.
  913. */
  914. privatebooleanreduceNumOfRacks(Sourcesource,StorageGrouptarget,
  915. DBlockblock){
  916. finalDatanodeInfosourceDn=source.getDatanodeInfo();
  917. if(cluster.isOnSameRack(sourceDn,target.getDatanodeInfo())){
  918. //sourceandtargetareonthesamerack
  919. returnfalse;
  920. }
  921. booleannotOnSameRack=true;
  922. synchronized(block){
  923. for(StorageGrouploc:block.getLocations()){
  924. if(cluster.isOnSameRack(loc.getDatanodeInfo(),target.getDatanodeInfo())){
  925. notOnSameRack=false;
  926. break;
  927. }
  928. }
  929. }
  930. if(notOnSameRack){
  931. //targetisnotonthesamerackasanyreplica
  932. returnfalse;
  933. }
  934. for(StorageGroupg:block.getLocations()){
  935. if(g!=source&&cluster.isOnSameRack(g.getDatanodeInfo(),sourceDn)){
  936. //sourceisonthesamerackofanotherreplica
  937. returnfalse;
  938. }
  939. }
  940. returntrue;
  941. }
  943. /**
  944. *Checkifthereareanyreplica(otherthansource)onthesamenodegroup
  945. *withtarget.Iftrue,thentargetisnotagoodcandidateforplacing
  946. *specificreplicaaswedon'twant2replicasunderthesamenodegroup.
  947. *
  948. *@returntrueifthereareanyreplica(otherthansource)onthesamenode
  949. *groupwithtarget
  950. */
  951. privatebooleanisOnSameNodeGroupWithReplicas(
  952. StorageGrouptarget,DBlockblock,Sourcesource){
  953. finalDatanodeInfotargetDn=target.getDatanodeInfo();
  954. for(StorageGroupg:block.getLocations()){
  955. if(g!=source&&cluster.isOnSameNodeGroup(g.getDatanodeInfo(),targetDn)){
  956. returntrue;
  957. }
  958. }
  959. returnfalse;
  960. }
  962. /**Resetallfieldsinordertoprepareforthenextiteration*/
  963. voidreset(Configurationconf){
  964. cluster=NetworkTopology.getInstance(conf);
  965. storageGroupMap.clear();
  966. sources.clear();
  967. targets.clear();
  968. globalBlocks.removeAllButRetain(movedBlocks);
  969. movedBlocks.cleanup();
  970. }
  972. /**shutdownthreadpools*/
  973. voidshutdownNow(){
  974. dispatchExecutor.shutdownNow();
  975. moveExecutor.shutdownNow();
  976. }
  978. staticclassUtil{
  979. /**@returntrueifdatanodeispartoftheexcludedNodes.*/
  980. staticbooleanisExcluded(Set<String>excludedNodes,DatanodeInfodn){
  981. returnisIn(excludedNodes,dn);
  982. }
  984. /**
  985. *@returntrueifincludedNodesisemptyordatanodeispartofthe
  986. *includedNodes.
  987. */
  988. staticbooleanisIncluded(Set<String>includedNodes,DatanodeInfodn){
  989. return(includedNodes.isEmpty()||isIn(includedNodes,dn));
  990. }
  992. /**
  993. *Matchischeckedusinghostname,ipaddresswithandwithoutport
  994. *number.
  995. *
  996. *@returntrueifthedatanode'stransferaddressmatchesthesetofnodes.
  997. */
  998. privatestaticbooleanisIn(Set<String>datanodes,DatanodeInfodn){
  999. returnisIn(datanodes,dn.getPeerHostName(),dn.getXferPort())
  1000. ||isIn(datanodes,dn.getIpAddr(),dn.getXferPort())
  1001. ||isIn(datanodes,dn.getHostName(),dn.getXferPort());
  1002. }
  1004. /**@returntrueifnodescontainshostorhost:port*/
  1005. privatestaticbooleanisIn(Set<String>nodes,Stringhost,intport){
  1006. if(host==null){
  1007. returnfalse;
  1008. }
  1009. return(nodes.contains(host)||nodes.contains(host+":"+port));
  1010. }
  1012. /**
  1013. *Parseacommaseparatedstringtoobtainsetofhostnames
  1014. *
  1015. *@returnsetofhostnames
  1016. */
  1017. staticSet<String>parseHostList(Stringstring){
  1018. String[]addrs=StringUtils.getTrimmedStrings(string);
  1019. returnnewHashSet<String>(Arrays.asList(addrs));
  1020. }
  1022. /**
  1023. *Readsetofhostnamesfromafile
  1024. *
  1025. *@returnsetofhostnames
  1026. */
  1027. staticSet<String>getHostListFromFile(StringfileName,Stringtype){
  1028. Set<String>nodes=newHashSet<String>();
  1029. try{
  1030. HostsFileReader.readFileToSet(type,fileName,nodes);
  1031. returnStringUtils.getTrimmedStrings(nodes);
  1032. }catch(IOExceptione){
  1033. thrownewIllegalArgumentException(
  1034. "Failedtoreadhostlistfromfile:"+fileName);
  1035. }
  1036. }
  1037. }
  1038. }

相关推荐