hadoop之BlockPoolManager源码分析

在HDFS Federation架构中, 一个HDFS集群可以创建多个命名空间，每一个DataNode都可以存储多个BlockPool的的数据块，所以在

DataNode定义了一个BlockPoolManager用于管理DataNode上所有的块池。

DataNode 其他模块要对BlockPool操作必须通过BlockPool Manager来执行，每一个DataNode都有一个BlockManager的实例

 

一 BPServiceActor分析

BPServiceActor负责与一个NameNode进行通信，每一个BPServiceActor都是一个独立的线程，主要功能：

>>与NameNode进行第一次握手，获取命名空间的信息

>>向NameNode注册当前DataNode

>>定期向NameNode发送心跳，增量块汇报，全量块汇报，缓存块汇报等

>>执行NameNode传回的指令

 

static
final Log
LOG = DataNode.LOG;
//NameNode
地址
finalInetSocketAddress
nnAddr;
//NameNode
状态
HAServiceStatestate;
//所持有的BPOfferService对象
final BPOfferService
bpos;

//当前的工作线程
ThreadbpThread;
//向Name Node发送RPC请求的代理
DatanodeProtocolClientSideTranslatorPBbpNamenode;
//当前BPServiceActor的运行状态，初始状态时CONNECTING
static
enum RunningState {
   
CONNECTING, INIT_FAILED,
RUNNING, EXITED,
FAILED;
}
private
volatile RunningState
runningState = RunningState.CONNECTING; 

//用于保存2次块汇报之间Data Node存储数据块的变化
private
final Map<DatanodeStorage, PerStoragePendingIncrementalBR>
     
pendingIncrementalBRperStorage = Maps.newHashMap();

//DataNode对象的引用
private
final DataNode
dn;

//用于记录Data Node的注册信息
private DatanodeRegistration
bpRegistration;

 

//初始化
try {
    //与Name Node握手并进行Data Node注册
    connectToNNAndHandshake();
    break;
}
catch (IOException ioe) {
    //初始化握手出现失败，运行状态置为INIT_FAILED
    runningState = RunningState.INIT_FAILED;
    if (shouldRetryInit()) {
        // Retry until all namenode's of BPOSfailed initialization
        sleepAndLogInterrupts(5000,
"initializing");
    }
else {
        runningState = RunningState.FAILED;
       
return;
    }
}
}
//初始化成功，状态置为RUNNING
runningState = RunningState.RUNNING;
//循环调用offerService方法向NameNode发送心跳，块汇报，增量汇报以及缓存快汇报等
while (shouldRun()) {
   
try {
     
offerService();
    }
catch (Exception ex) {
 
//收到异常也不会处理直到BPServiceActor停止或者Data Node停止
 
sleepAndLogInterrupts(5000,
"offeringservice");
    }
  }
 
//BPServiceActor停止以后，状态置为EXITED
 
runningState = RunningState.EXITED;

 

 
private void
connectToNNAndHandshake() throws IOException {
   
//获取Name Node 的PRC 代理
   
bpNamenode = dn.connectToNN(nnAddr);
   
//第一次握手去获取namespace 信息
    NamespaceInfo
nsInfo =
retrieveNamespaceInfo();
   
bpos.verifyAndSetNamespaceInfo(nsInfo);
   
//第二次握手则是向Name Node注册这个Data Node
   
register(nsInfo);
  }

 

private
void offerService()
throwsException {
   
long fullBlockReportLeaseId =
0;
   
while (shouldRun()) {
     
try {
       
final long
startTime =
scheduler.monotonicNow();
       
//判断是否发送心跳信息
       
final boolean
sendHeartbeat =
scheduler.isHeartbeatDue(startTime);
        HeartbeatResponse
resp = null;
       
//如果要发送心跳
       
if (sendHeartbeat) {
         
boolean requestBlockReportLease = (fullBlockReportLeaseId ==
0) &&
                 
scheduler.isBlockReportDue(startTime);
         
scheduler.scheduleNextHeartbeat();
         
if (!dn.areHeartbeatsDisabledForTests()) {
           
//发送心跳信息
           
resp = sendHeartBeat(requestBlockReportLease);
           
assert
resp != null;
           
if (resp.getFullBlockReportLeaseId() !=
0) {
             
fullBlockReportLeaseId =
resp.getFullBlockReportLeaseId();
            }
           
dn.getMetrics().addHeartbeat(scheduler.monotonicNow() -
startTime);
           
//对心跳响应中携带的NameNode的HA状态进行处理
           
bpos.updateActorStatesFromHeartbeat(
               
this,
resp.getNameNodeHaState());
           
state = resp.getNameNodeHaState().getState();
 
           
if (state == HAServiceState.ACTIVE) {
             
handleRollingUpgradeStatus(resp);
            }
 
           
long startProcessCommands =
monotonicNow();
           
//处理响应中带回的Name Node指令
           
if (!processCommand(resp.getCommands()))
             
continue;
           
long endProcessCommands =
monotonicNow();
          }
        }
       
if (sendImmediateIBR ||
sendHeartbeat) {
         
reportReceivedDeletedBlocks();
        }
 
       
List<DatanodeCommand>
cmds = null;
       
boolean forceFullBr =
           
scheduler.forceFullBlockReport.getAndSet(false);
       
if (forceFullBr) {
         
LOG.info("Forcinga full block report to " +
nnAddr);
        }
       
if ((fullBlockReportLeaseId !=
0) || forceFullBr) {
         
cmds = blockReport(fullBlockReportLeaseId);
         
fullBlockReportLeaseId =
0;
        }
       
processCommand(cmds ==
null ? null :
cmds.toArray(new
DatanodeCommand[cmds.size()]));
 
       
if (!dn.areCacheReportsDisabledForTests()) {
         
DatanodeCommand cmd =
cacheReport();
         
processCommand(new
DatanodeCommand[]{
cmd });
        }
 
       
//
       
// There is no work to do;  sleep until hearbeat timer elapses,

       
// or work arrives, and then iterate again.
       
//
       
long waitTime =
scheduler.getHeartbeatWaitTime();
       
synchronized(pendingIncrementalBRperStorage) {
         
if (waitTime >
0 && !sendImmediateIBR) {
           
try {
             
pendingIncrementalBRperStorage.wait(waitTime);
            }
catch (InterruptedException
ie) {
             
LOG.warn("BPOfferServicefor " +
this +
" interrupted");
            }
          }
        }
// synchronized
      }
catch(RemoteException
re) {
        String
reClass =
re.getClassName();
       
if (UnregisteredNodeException.class.getName().equals(reClass)
||
            DisallowedDatanodeException.class.getName().equals(reClass)
||
            IncorrectVersionException.class.getName().equals(reClass))
{
         
LOG.warn(this +
" is shutting down",
re);
         
shouldServiceRun = false;
         
return;
        }
       
LOG.warn("RemoteExceptionin offerService",
re);
       
try {
         
long sleepTime = Math.min(1000,
dnConf.heartBeatInterval);
          Thread.sleep(sleepTime);
        }
catch (InterruptedException
ie) {
          Thread.currentThread().interrupt();
        }
      }
catch (IOException e) {
       
LOG.warn("IOExceptionin offerService",
e);
      }
     
processQueueMessages();
    }
// while (shouldRun())
  }
// offerService

 

void
updateActorStatesFromHeartbeat(BPServiceActor
actor,
      NNHAStatusHeartbeat
nnHaState) {
   
writeLock();
   
try {
     
//取到Name Node的txid
     
final long
txid = nnHaState.getTxId();
     
//当前这个NameNode是否声明自己为Active NameNode
     
final boolean
nnClaimsActive =
         
nnHaState.getState() == HAServiceState.ACTIVE;
     
//BPOfferService是否认为当前Name Node为Active NameNode
     
final boolean
bposThinksActive =
bpServiceToActive == actor;
     
//当前Name Node携带的txid是否大于原Active NameNode 的txid
     
final boolean
isMoreRecentClaim =
txid > lastActiveClaimTxId;
     
//当前这个是Acitve，但是BPOfferService所记录的不是Active,说明Standby已经切换成Active
     
if (nnClaimsActive && !bposThinksActive) {
       
//如果有两个namenode声明为active,当前的请求过时
       
if (!isMoreRecentClaim) {
          
//直接忽略
         
return;
        }
else {//当前请求是最新的请求
         
if (bpServiceToActive ==
null) {
           
//BPOfferService上还没有保存active name node
          }
else {
          }
         
//将bpServiceToActive指向当前的NameNode对应的 BPOfferService
         
bpServiceToActive =
actor;
        }
      }
else if (!nnClaimsActive &&
bposThinksActive) {
       
//原来Active Name Node现在声明为StandbyName Node
       
bpServiceToActive = null;
      }
     
//更新lastActiveClaimTxId
     
if (bpServiceToActive ==
actor) {
       
assert
txid >= lastActiveClaimTxId;
       
lastActiveClaimTxId =
txid;
      }
    }
finally {
     
writeUnlock();
    }
  }
 

二 BPOfferService

BPOfferService就是对DataNode每一个BlockPool进行管理的类。

重要的字段：

//握手之后获取到的Namespace信息
NamespaceInfobpNSInfo;
//当前Block Pool在Name Node上的注册信息，这个信息是在Data Node注册阶段获取的
volatileDatanodeRegistration
bpRegistration;
//当前DataNode的引用
private
final DataNode
dn;
//BPServiceActor的引用，这个代表着是Active的Name Node对应的对象
privateBPServiceActor
bpServiceToActive = null;
//当前命名空间中所有NameNode对应的BPServiceActor对象
private
final List<BPServiceActor>
bpServices = new
CopyOnWriteArrayList<BPServiceActor>();
//每当收到一个NameNode的时候，就记录最近的txid
private
long lastActiveClaimTxId = -1;

主要的方法分类：

1)触发汇报

trySendErrorReport(),reportRemoteBadBlock,reportBadBlock()实现了向NameNode发送错误汇报，汇报远程坏块以及本地坏块的操作，会直接调用BPServiceActor对应操作

void
trySendErrorReport(int
errCode, String errMsg) {
   
for (BPServiceActor actor :
bpServices) {
      ErrorReportAction
errorReportAction = new
ErrorReportAction
          (errCode,
errMsg);
     
actor.bpThreadEnqueue(errorReportAction);
    }
}

2)添加与删除数据块操作

当DataNode接收一个新的数据块时，比如客户端通过数据流管道写入一个数据块，或者通过DataTransferProtocal流式接口复制一个数据块时候，都会调用BPOfferService.notifyNameNodeReceiveBlock()。

当DataNdoe删除一个已有的数据块的时候，会调用BPOfferService

.notifyNamenodeDeletedBlock()方法通知命名空间。

 

3)响应NameNode的指令

boolean
processCommandFromActor(DatanodeCommand
cmd,BPServiceActor
actor) throws IOException {
   
if (cmd ==
null) {
     
return true;
    }
   
//如果Name Node返回的指令要求Data Node重新注册的，则调用BPServiceActor.register方法
   
if (DatanodeProtocol.DNA_REGISTER ==
cmd.getAction()) {
     
actor.reRegister();
     
return false;
    }
   
writeLock();
   
try {
     
//对于Active Name Node返回的指令，调用processCommandFromActive
     
if (actor ==
bpServiceToActive) {
       
return
processCommandFromActive(cmd,
actor);
      }
else {
     
//对于Standbu Name Node返回的指令，调用processCommandFromStandby
       
return
processCommandFromStandby(cmd,
actor);
      }
    }
finally {
     
writeUnlock();
    }
}

processCommandFromStandby处理来自StandbyName Node的指令，直接忽略即可。防止在HA部署下出现脑裂的情况，也就是ActiveNameNode和StandbyNameNode同时向DataNode下指令。所以BPOfferService对象并不执行Standby返回的字节指令

 

三BlockPoolManager

BlockPoolManager类负责管理所有的BPOfferService实例，对外提供添加、删除、启动关闭BPOfferService类的接口。所有BPOfferService的操作，都必须通过BlockPoolManager类提供的方法来执行

 

DataNode启动的时候，会初始化BlockPoolManager对象，然后调用BlockPoolManager.refreshNamenodes()完成对BlockPoolManager的构造

//<namespaceId，BPOfferService>命名空间id与BPOfferService
private
final Map<String, BPOfferService>
bpByNameserviceId = Maps.newHashMap();
//<blockId,BPOfferService>块池id与BPOfferService映射
private
final Map<String, BPOfferService>
bpByBlockPoolId = Maps.newHashMap();
//持有一个BPOfferService List
private
final List<BPOfferService>
offerServices =Lists.newArrayList();
//持有一个Data Node的引用
private
final DataNode
dn;