Revert "HDFS-16898. Make write lock fine-grain in method processCommandFromActor (#5330). Contributed by ZhangHB."
This reverts commit eb04ecd29d
.
This commit is contained in:
parent
eb04ecd29d
commit
7e919212c4
@ -679,20 +679,15 @@ boolean processCommandFromActor(DatanodeCommand cmd,
|
|||||||
actor.reRegister();
|
actor.reRegister();
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
boolean isActiveActor;
|
writeLock();
|
||||||
InetSocketAddress nnSocketAddress;
|
|
||||||
readLock();
|
|
||||||
try {
|
try {
|
||||||
isActiveActor = (actor == bpServiceToActive);
|
if (actor == bpServiceToActive) {
|
||||||
nnSocketAddress = actor.getNNSocketAddress();
|
return processCommandFromActive(cmd, actor);
|
||||||
|
} else {
|
||||||
|
return processCommandFromStandby(cmd, actor);
|
||||||
|
}
|
||||||
} finally {
|
} finally {
|
||||||
readUnlock();
|
writeUnlock();
|
||||||
}
|
|
||||||
|
|
||||||
if (isActiveActor) {
|
|
||||||
return processCommandFromActive(cmd, nnSocketAddress);
|
|
||||||
} else {
|
|
||||||
return processCommandFromStandby(cmd, nnSocketAddress);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -720,7 +715,7 @@ private String blockIdArrayToString(long ids[]) {
|
|||||||
* @throws IOException
|
* @throws IOException
|
||||||
*/
|
*/
|
||||||
private boolean processCommandFromActive(DatanodeCommand cmd,
|
private boolean processCommandFromActive(DatanodeCommand cmd,
|
||||||
InetSocketAddress nnSocketAddress) throws IOException {
|
BPServiceActor actor) throws IOException {
|
||||||
final BlockCommand bcmd =
|
final BlockCommand bcmd =
|
||||||
cmd instanceof BlockCommand? (BlockCommand)cmd: null;
|
cmd instanceof BlockCommand? (BlockCommand)cmd: null;
|
||||||
final BlockIdCommand blockIdCmd =
|
final BlockIdCommand blockIdCmd =
|
||||||
@ -773,7 +768,7 @@ assert getBlockPoolId().equals(bp) :
|
|||||||
dn.finalizeUpgradeForPool(bp);
|
dn.finalizeUpgradeForPool(bp);
|
||||||
break;
|
break;
|
||||||
case DatanodeProtocol.DNA_RECOVERBLOCK:
|
case DatanodeProtocol.DNA_RECOVERBLOCK:
|
||||||
String who = "NameNode at " + nnSocketAddress;
|
String who = "NameNode at " + actor.getNNSocketAddress();
|
||||||
dn.getBlockRecoveryWorker().recoverBlocks(who,
|
dn.getBlockRecoveryWorker().recoverBlocks(who,
|
||||||
((BlockRecoveryCommand)cmd).getRecoveringBlocks());
|
((BlockRecoveryCommand)cmd).getRecoveringBlocks());
|
||||||
break;
|
break;
|
||||||
@ -815,11 +810,11 @@ assert getBlockPoolId().equals(bp) :
|
|||||||
* DNA_REGISTER which should be handled earlier itself.
|
* DNA_REGISTER which should be handled earlier itself.
|
||||||
*/
|
*/
|
||||||
private boolean processCommandFromStandby(DatanodeCommand cmd,
|
private boolean processCommandFromStandby(DatanodeCommand cmd,
|
||||||
InetSocketAddress nnSocketAddress) throws IOException {
|
BPServiceActor actor) throws IOException {
|
||||||
switch(cmd.getAction()) {
|
switch(cmd.getAction()) {
|
||||||
case DatanodeProtocol.DNA_ACCESSKEYUPDATE:
|
case DatanodeProtocol.DNA_ACCESSKEYUPDATE:
|
||||||
LOG.info("DatanodeCommand action from standby NN {}: DNA_ACCESSKEYUPDATE",
|
LOG.info("DatanodeCommand action from standby NN {}: DNA_ACCESSKEYUPDATE",
|
||||||
nnSocketAddress);
|
actor.getNNSocketAddress());
|
||||||
if (dn.isBlockTokenEnabled) {
|
if (dn.isBlockTokenEnabled) {
|
||||||
dn.blockPoolTokenSecretManager.addKeys(
|
dn.blockPoolTokenSecretManager.addKeys(
|
||||||
getBlockPoolId(),
|
getBlockPoolId(),
|
||||||
@ -836,11 +831,11 @@ private boolean processCommandFromStandby(DatanodeCommand cmd,
|
|||||||
case DatanodeProtocol.DNA_UNCACHE:
|
case DatanodeProtocol.DNA_UNCACHE:
|
||||||
case DatanodeProtocol.DNA_ERASURE_CODING_RECONSTRUCTION:
|
case DatanodeProtocol.DNA_ERASURE_CODING_RECONSTRUCTION:
|
||||||
LOG.warn("Got a command from standby NN {} - ignoring command: {}",
|
LOG.warn("Got a command from standby NN {} - ignoring command: {}",
|
||||||
nnSocketAddress, cmd.getAction());
|
actor.getNNSocketAddress(), cmd.getAction());
|
||||||
break;
|
break;
|
||||||
default:
|
default:
|
||||||
LOG.warn("Unknown DatanodeCommand action: {} from standby NN {}",
|
LOG.warn("Unknown DatanodeCommand action: {} from standby NN {}",
|
||||||
cmd.getAction(), nnSocketAddress);
|
cmd.getAction(), actor.getNNSocketAddress());
|
||||||
}
|
}
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
@ -1499,7 +1499,7 @@ private boolean processCommand(DatanodeCommand[] cmds) {
|
|||||||
dn.getMetrics().addNumProcessedCommands(processCommandsMs);
|
dn.getMetrics().addNumProcessedCommands(processCommandsMs);
|
||||||
}
|
}
|
||||||
if (processCommandsMs > dnConf.getProcessCommandsThresholdMs()) {
|
if (processCommandsMs > dnConf.getProcessCommandsThresholdMs()) {
|
||||||
LOG.warn("Took {} ms to process {} commands from NN",
|
LOG.info("Took {} ms to process {} commands from NN",
|
||||||
processCommandsMs, cmds.length);
|
processCommandsMs, cmds.length);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user