Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HDFS-7398. Reset cached thread-local FSEditLogOp's on every FSEditLog#logEdit. Contributed by Gera Shegalov.

This commit is contained in:
cnauroth 2014-11-18 16:56:44 -08:00
parent 90a968d675
commit 9e81be0114
4 changed files with 293 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
View File

@ -370,6 +370,9 @@ Release 2.7.0 - UNRELEASED
HDFS-7404. Remove o.a.h.hdfs.server.datanode.web.resources.
(Li Lu via wheat9)
HDFS-7398. Reset cached thread-local FSEditLogOp's on every
FSEditLog#logEdit. (Gera Shegalov via cnauroth)
OPTIMIZATIONS
BUG FIXES

4
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLog.java
View File

@ -422,6 +422,8 @@ public class FSEditLog implements LogsPurgeable {
editLogStream.write(op);
} catch (IOException ex) {
// All journals failed, it is handled in logSync.
} finally {
op.reset();
}
endTransaction(start);
@ -708,7 +710,6 @@ public class FSEditLog implements LogsPurgeable {
Preconditions.checkArgument(newNode.isUnderConstruction());
PermissionStatus permissions = newNode.getPermissionStatus();
AddOp op = AddOp.getInstance(cache.get())
.reset()
.setInodeId(newNode.getId())
.setPath(path)
.setReplication(newNode.getFileReplication())
@ -779,7 +780,6 @@ public class FSEditLog implements LogsPurgeable {
public void logMkDir(String path, INode newNode) {
PermissionStatus permissions = newNode.getPermissionStatus();
MkdirOp op = MkdirOp.getInstance(cache.get())
.reset()
.setInodeId(newNode.getId())
.setPath(path)
.setTimestamp(newNode.getModificationTime())

297
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLogOp.java
View File

@ -138,9 +138,18 @@ import com.google.common.collect.Lists;
@InterfaceStability.Unstable
public abstract class FSEditLogOp {
public final FSEditLogOpCodes opCode;
long txid = HdfsConstants.INVALID_TXID;
byte[] rpcClientId = RpcConstants.DUMMY_CLIENT_ID;
int rpcCallId = RpcConstants.INVALID_CALL_ID;
long txid;
byte[] rpcClientId;
int rpcCallId;
final void reset() {
txid = HdfsConstants.INVALID_TXID;
rpcClientId = RpcConstants.DUMMY_CLIENT_ID;
rpcCallId = RpcConstants.INVALID_CALL_ID;
resetSubFields();
}
abstract void resetSubFields();
final public static class OpInstanceCache {
private final EnumMap<FSEditLogOpCodes, FSEditLogOp> inst =
@ -220,6 +229,7 @@ public abstract class FSEditLogOp {
@VisibleForTesting
protected FSEditLogOp(FSEditLogOpCodes opCode) {
this.opCode = opCode;
reset();
}
public long getTransactionId() {
@ -418,11 +428,24 @@ public abstract class FSEditLogOp {
storagePolicyId = BlockStoragePolicySuite.ID_UNSPECIFIED;
assert(opCode == OP_ADD || opCode == OP_CLOSE);
}
<T extends AddCloseOp> T reset() {
this.aclEntries = null;
this.xAttrs = null;
return (T)this;
@Override
void resetSubFields() {
length = 0;
inodeId = 0L;
path = null;
replication = 0;
mtime = 0L;
atime = 0L;
blockSize = 0L;
blocks = null;
permissions = null;
aclEntries = null;
xAttrs = null;
clientName = null;
clientMachine = null;
overwrite = false;
storagePolicyId = 0;
}
<T extends AddCloseOp> T setInodeId(long inodeId) {
@ -802,6 +825,13 @@ public abstract class FSEditLogOp {
static AddBlockOp getInstance(OpInstanceCache cache) {
return (AddBlockOp) cache.get(OP_ADD_BLOCK);
}
@Override
void resetSubFields() {
path = null;
penultimateBlock = null;
lastBlock = null;
}
AddBlockOp setPath(String path) {
this.path = path;
@ -843,7 +873,7 @@ public abstract class FSEditLogOp {
// clientId and callId
writeRpcIds(rpcClientId, rpcCallId, out);
}
@Override
void readFields(DataInputStream in, int logVersion) throws IOException {
path = FSImageSerialization.readString(in);
@ -907,6 +937,12 @@ public abstract class FSEditLogOp {
static UpdateBlocksOp getInstance(OpInstanceCache cache) {
return (UpdateBlocksOp)cache.get(OP_UPDATE_BLOCKS);
}
@Override
void resetSubFields() {
path = null;
blocks = null;
}
UpdateBlocksOp setPath(String path) {
this.path = path;
@ -995,6 +1031,12 @@ public abstract class FSEditLogOp {
return (SetReplicationOp)cache.get(OP_SET_REPLICATION);
}
@Override
void resetSubFields() {
path = null;
replication = 0;
}
SetReplicationOp setPath(String path) {
this.path = path;
return this;
@ -1068,6 +1110,14 @@ public abstract class FSEditLogOp {
return (ConcatDeleteOp)cache.get(OP_CONCAT_DELETE);
}
@Override
void resetSubFields() {
length = 0;
trg = null;
srcs = null;
timestamp = 0L;
}
ConcatDeleteOp setTarget(String trg) {
this.trg = trg;
return this;
@ -1218,6 +1268,14 @@ public abstract class FSEditLogOp {
return (RenameOldOp)cache.get(OP_RENAME_OLD);
}
@Override
void resetSubFields() {
length = 0;
src = null;
dst = null;
timestamp = 0L;
}
RenameOldOp setSource(String src) {
this.src = src;
return this;
@ -1322,6 +1380,13 @@ public abstract class FSEditLogOp {
return (DeleteOp)cache.get(OP_DELETE);
}
@Override
void resetSubFields() {
length = 0;
path = null;
timestamp = 0L;
}
DeleteOp setPath(String path) {
this.path = path;
return this;
@ -1416,10 +1481,15 @@ public abstract class FSEditLogOp {
return (MkdirOp)cache.get(OP_MKDIR);
}
MkdirOp reset() {
this.aclEntries = null;
this.xAttrs = null;
return this;
@Override
void resetSubFields() {
length = 0;
inodeId = 0L;
path = null;
timestamp = 0L;
permissions = null;
aclEntries = null;
xAttrs = null;
}
MkdirOp setInodeId(long inodeId) {
@ -1584,6 +1654,11 @@ public abstract class FSEditLogOp {
return (SetGenstampV1Op)cache.get(OP_SET_GENSTAMP_V1);
}
@Override
void resetSubFields() {
genStampV1 = 0L;
}
SetGenstampV1Op setGenerationStamp(long genStamp) {
this.genStampV1 = genStamp;
return this;
@ -1637,6 +1712,11 @@ public abstract class FSEditLogOp {
return (SetGenstampV2Op)cache.get(OP_SET_GENSTAMP_V2);
}
@Override
void resetSubFields() {
genStampV2 = 0L;
}
SetGenstampV2Op setGenerationStamp(long genStamp) {
this.genStampV2 = genStamp;
return this;
@ -1690,6 +1770,11 @@ public abstract class FSEditLogOp {
return (AllocateBlockIdOp)cache.get(OP_ALLOCATE_BLOCK_ID);
}
@Override
void resetSubFields() {
blockId = 0L;
}
AllocateBlockIdOp setBlockId(long blockId) {
this.blockId = blockId;
return this;
@ -1744,6 +1829,12 @@ public abstract class FSEditLogOp {
return (SetPermissionsOp)cache.get(OP_SET_PERMISSIONS);
}
@Override
void resetSubFields() {
src = null;
permissions = null;
}
SetPermissionsOp setSource(String src) {
this.src = src;
return this;
@ -1811,6 +1902,13 @@ public abstract class FSEditLogOp {
return (SetOwnerOp)cache.get(OP_SET_OWNER);
}
@Override
void resetSubFields() {
src = null;
username = null;
groupname = null;
}
SetOwnerOp setSource(String src) {
this.src = src;
return this;
@ -1891,6 +1989,12 @@ public abstract class FSEditLogOp {
return (SetNSQuotaOp)cache.get(OP_SET_NS_QUOTA);
}
@Override
void resetSubFields() {
src = null;
nsQuota = 0L;
}
@Override
public
void writeFields(DataOutputStream out) throws IOException {
@ -1943,6 +2047,11 @@ public abstract class FSEditLogOp {
return (ClearNSQuotaOp)cache.get(OP_CLEAR_NS_QUOTA);
}
@Override
void resetSubFields() {
src = null;
}
@Override
public
void writeFields(DataOutputStream out) throws IOException {
@ -1992,6 +2101,13 @@ public abstract class FSEditLogOp {
return (SetQuotaOp)cache.get(OP_SET_QUOTA);
}
@Override
void resetSubFields() {
src = null;
nsQuota = 0L;
dsQuota = 0L;
}
SetQuotaOp setSource(String src) {
this.src = src;
return this;
@ -2071,6 +2187,14 @@ public abstract class FSEditLogOp {
return (TimesOp)cache.get(OP_TIMES);
}
@Override
void resetSubFields() {
length = 0;
path = null;
mtime = 0L;
atime = 0L;
}
TimesOp setPath(String path) {
this.path = path;
return this;
@ -2172,6 +2296,17 @@ public abstract class FSEditLogOp {
return (SymlinkOp)cache.get(OP_SYMLINK);
}
@Override
void resetSubFields() {
length = 0;
inodeId = 0L;
path = null;
value = null;
mtime = 0L;
atime = 0L;
permissionStatus = null;
}
SymlinkOp setId(long inodeId) {
this.inodeId = inodeId;
return this;
@ -2320,6 +2455,15 @@ public abstract class FSEditLogOp {
return (RenameOp)cache.get(OP_RENAME);
}
@Override
void resetSubFields() {
length = 0;
src = null;
dst = null;
timestamp = 0L;
options = null;
}
RenameOp setSource(String src) {
this.src = src;
return this;
@ -2477,6 +2621,13 @@ public abstract class FSEditLogOp {
return (ReassignLeaseOp)cache.get(OP_REASSIGN_LEASE);
}
@Override
void resetSubFields() {
leaseHolder = null;
path = null;
newHolder = null;
}
ReassignLeaseOp setLeaseHolder(String leaseHolder) {
this.leaseHolder = leaseHolder;
return this;
@ -2552,6 +2703,12 @@ public abstract class FSEditLogOp {
return (GetDelegationTokenOp)cache.get(OP_GET_DELEGATION_TOKEN);
}
@Override
void resetSubFields() {
token = null;
expiryTime = 0L;
}
GetDelegationTokenOp setDelegationTokenIdentifier(
DelegationTokenIdentifier token) {
this.token = token;
@ -2625,6 +2782,12 @@ public abstract class FSEditLogOp {
return (RenewDelegationTokenOp)cache.get(OP_RENEW_DELEGATION_TOKEN);
}
@Override
void resetSubFields() {
token = null;
expiryTime = 0L;
}
RenewDelegationTokenOp setDelegationTokenIdentifier(
DelegationTokenIdentifier token) {
this.token = token;
@ -2697,6 +2860,11 @@ public abstract class FSEditLogOp {
return (CancelDelegationTokenOp)cache.get(OP_CANCEL_DELEGATION_TOKEN);
}
@Override
void resetSubFields() {
token = null;
}
CancelDelegationTokenOp setDelegationTokenIdentifier(
DelegationTokenIdentifier token) {
this.token = token;
@ -2751,6 +2919,11 @@ public abstract class FSEditLogOp {
return (UpdateMasterKeyOp)cache.get(OP_UPDATE_MASTER_KEY);
}
@Override
void resetSubFields() {
key = null;
}
UpdateMasterKeyOp setDelegationKey(DelegationKey key) {
this.key = key;
return this;
@ -2805,6 +2978,11 @@ public abstract class FSEditLogOp {
return (LogSegmentOp)cache.get(code);
}
@Override
void resetSubFields() {
// no data stored in these ops yet
}
@Override
public void readFields(DataInputStream in, int logVersion)
throws IOException {
@ -2847,6 +3025,10 @@ public abstract class FSEditLogOp {
return (InvalidOp)cache.get(OP_INVALID);
}
@Override
void resetSubFields() {
}
@Override
public
void writeFields(DataOutputStream out) throws IOException {
@ -2893,7 +3075,13 @@ public abstract class FSEditLogOp {
static CreateSnapshotOp getInstance(OpInstanceCache cache) {
return (CreateSnapshotOp)cache.get(OP_CREATE_SNAPSHOT);
}
@Override
void resetSubFields() {
snapshotRoot = null;
snapshotName = null;
}
CreateSnapshotOp setSnapshotName(String snapName) {
this.snapshotName = snapName;
return this;
@ -2963,6 +3151,12 @@ public abstract class FSEditLogOp {
static DeleteSnapshotOp getInstance(OpInstanceCache cache) {
return (DeleteSnapshotOp)cache.get(OP_DELETE_SNAPSHOT);
}
@Override
void resetSubFields() {
snapshotRoot = null;
snapshotName = null;
}
DeleteSnapshotOp setSnapshotName(String snapName) {
this.snapshotName = snapName;
@ -3034,6 +3228,13 @@ public abstract class FSEditLogOp {
static RenameSnapshotOp getInstance(OpInstanceCache cache) {
return (RenameSnapshotOp) cache.get(OP_RENAME_SNAPSHOT);
}
@Override
void resetSubFields() {
snapshotRoot = null;
snapshotOldName = null;
snapshotNewName = null;
}
RenameSnapshotOp setSnapshotOldName(String snapshotOldName) {
this.snapshotOldName = snapshotOldName;
@ -3120,6 +3321,11 @@ public abstract class FSEditLogOp {
return (AllowSnapshotOp) cache.get(OP_ALLOW_SNAPSHOT);
}
@Override
void resetSubFields() {
snapshotRoot = null;
}
public AllowSnapshotOp setSnapshotRoot(String snapRoot) {
snapshotRoot = snapRoot;
return this;
@ -3174,6 +3380,10 @@ public abstract class FSEditLogOp {
return (DisallowSnapshotOp) cache.get(OP_DISALLOW_SNAPSHOT);
}
void resetSubFields() {
snapshotRoot = null;
}
public DisallowSnapshotOp setSnapshotRoot(String snapRoot) {
snapshotRoot = snapRoot;
return this;
@ -3225,6 +3435,11 @@ public abstract class FSEditLogOp {
.get(OP_ADD_CACHE_DIRECTIVE);
}
@Override
void resetSubFields() {
directive = null;
}
public AddCacheDirectiveInfoOp setDirective(
CacheDirectiveInfo directive) {
this.directive = directive;
@ -3291,6 +3506,11 @@ public abstract class FSEditLogOp {
.get(OP_MODIFY_CACHE_DIRECTIVE);
}
@Override
void resetSubFields() {
directive = null;
}
public ModifyCacheDirectiveInfoOp setDirective(
CacheDirectiveInfo directive) {
this.directive = directive;
@ -3363,6 +3583,11 @@ public abstract class FSEditLogOp {
.get(OP_REMOVE_CACHE_DIRECTIVE);
}
@Override
void resetSubFields() {
id = 0L;
}
public RemoveCacheDirectiveInfoOp setId(long id) {
this.id = id;
return this;
@ -3415,6 +3640,11 @@ public abstract class FSEditLogOp {
return (AddCachePoolOp) cache.get(OP_ADD_CACHE_POOL);
}
@Override
void resetSubFields() {
info = null;
}
public AddCachePoolOp setPool(CachePoolInfo info) {
this.info = info;
assert(info.getPoolName() != null);
@ -3476,6 +3706,11 @@ public abstract class FSEditLogOp {
return (ModifyCachePoolOp) cache.get(OP_MODIFY_CACHE_POOL);
}
@Override
void resetSubFields() {
info = null;
}
public ModifyCachePoolOp setInfo(CachePoolInfo info) {
this.info = info;
return this;
@ -3544,6 +3779,11 @@ public abstract class FSEditLogOp {
return (RemoveCachePoolOp) cache.get(OP_REMOVE_CACHE_POOL);
}
@Override
void resetSubFields() {
poolName = null;
}
public RemoveCachePoolOp setPoolName(String poolName) {
this.poolName = poolName;
return this;
@ -3596,6 +3836,12 @@ public abstract class FSEditLogOp {
return new RemoveXAttrOp();
}
@Override
void resetSubFields() {
xAttrs = null;
src = null;
}
@Override
void readFields(DataInputStream in, int logVersion) throws IOException {
XAttrEditLogProto p = XAttrEditLogProto.parseDelimitedFrom(in);
@ -3643,6 +3889,12 @@ public abstract class FSEditLogOp {
return new SetXAttrOp();
}
@Override
void resetSubFields() {
xAttrs = null;
src = null;
}
@Override
void readFields(DataInputStream in, int logVersion) throws IOException {
XAttrEditLogProto p = XAttrEditLogProto.parseDelimitedFrom(in);
@ -3690,6 +3942,12 @@ public abstract class FSEditLogOp {
return new SetAclOp();
}
@Override
void resetSubFields() {
aclEntries = null;
src = null;
}
@Override
void readFields(DataInputStream in, int logVersion) throws IOException {
AclEditLogProto p = AclEditLogProto.parseDelimitedFrom(in);
@ -3790,6 +4048,11 @@ public abstract class FSEditLogOp {
return (RollingUpgradeOp) cache.get(OP_ROLLING_UPGRADE_FINALIZE);
}
@Override
void resetSubFields() {
time = 0L;
}
long getTime() {
return time;
}
@ -3843,6 +4106,12 @@ public abstract class FSEditLogOp {
return (SetStoragePolicyOp) cache.get(OP_SET_STORAGE_POLICY);
}
@Override
void resetSubFields() {
path = null;
policyId = 0;
}
SetStoragePolicyOp setPath(String path) {
this.path = path;
return this;

5
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestEditLog.java
View File

@ -109,6 +109,11 @@ public class TestEditLog {
super(FSEditLogOpCodes.OP_MKDIR);
}
@Override
void resetSubFields() {
// nop
}
@Override
void readFields(DataInputStream in, int logVersion) throws IOException {
throw new IOException("cannot decode GarbageMkdirOp");