You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by wa...@apache.org on 2015/08/25 19:12:30 UTC
[19/42] hadoop git commit: HDFS-8934. Move ShortCircuitShm to
hdfs-client. Contributed by Mingliang Liu.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
index 8e81fdc..beaa903 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
@@ -695,7 +695,7 @@ public class ClientNamenodeProtocolServerSideTranslatorPB implements
RpcController controller, GetDatanodeReportRequestProto req)
throws ServiceException {
try {
- List<? extends DatanodeInfoProto> result = PBHelper.convert(server
+ List<? extends DatanodeInfoProto> result = PBHelperClient.convert(server
.getDatanodeReport(PBHelper.convert(req.getType())));
return GetDatanodeReportResponseProto.newBuilder()
.addAllDi(result).build();
@@ -890,7 +890,7 @@ public class ClientNamenodeProtocolServerSideTranslatorPB implements
server.setQuota(req.getPath(), req.getNamespaceQuota(),
req.getStoragespaceQuota(),
req.hasStorageType() ?
- PBHelper.convertStorageType(req.getStorageType()): null);
+ PBHelperClient.convertStorageType(req.getStorageType()): null);
return VOID_SETQUOTA_RESPONSE;
} catch (IOException e) {
throw new ServiceException(e);
@@ -990,7 +990,7 @@ public class ClientNamenodeProtocolServerSideTranslatorPB implements
GetDelegationTokenResponseProto.Builder rspBuilder =
GetDelegationTokenResponseProto.newBuilder();
if (token != null) {
- rspBuilder.setToken(PBHelper.convert(token));
+ rspBuilder.setToken(PBHelperClient.convert(token));
}
return rspBuilder.build();
} catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
index d6afa6e..d30982a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
@@ -390,7 +390,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
String holder) throws AccessControlException, FileNotFoundException,
UnresolvedLinkException, IOException {
AbandonBlockRequestProto req = AbandonBlockRequestProto.newBuilder()
- .setB(PBHelper.convert(b)).setSrc(src).setHolder(holder)
+ .setB(PBHelperClient.convert(b)).setSrc(src).setHolder(holder)
.setFileId(fileId).build();
try {
rpcProxy.abandonBlock(null, req);
@@ -409,9 +409,9 @@ public class ClientNamenodeProtocolTranslatorPB implements
AddBlockRequestProto.Builder req = AddBlockRequestProto.newBuilder()
.setSrc(src).setClientName(clientName).setFileId(fileId);
if (previous != null)
- req.setPrevious(PBHelper.convert(previous));
- if (excludeNodes != null)
- req.addAllExcludeNodes(PBHelper.convert(excludeNodes));
+ req.setPrevious(PBHelperClient.convert(previous));
+ if (excludeNodes != null)
+ req.addAllExcludeNodes(PBHelperClient.convert(excludeNodes));
if (favoredNodes != null) {
req.addAllFavoredNodes(Arrays.asList(favoredNodes));
}
@@ -433,10 +433,10 @@ public class ClientNamenodeProtocolTranslatorPB implements
.newBuilder()
.setSrc(src)
.setFileId(fileId)
- .setBlk(PBHelper.convert(blk))
- .addAllExistings(PBHelper.convert(existings))
+ .setBlk(PBHelperClient.convert(blk))
+ .addAllExistings(PBHelperClient.convert(existings))
.addAllExistingStorageUuids(Arrays.asList(existingStorageIDs))
- .addAllExcludes(PBHelper.convert(excludes))
+ .addAllExcludes(PBHelperClient.convert(excludes))
.setNumAdditionalNodes(numAdditionalNodes)
.setClientName(clientName)
.build();
@@ -458,7 +458,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
.setClientName(clientName)
.setFileId(fileId);
if (last != null)
- req.setLast(PBHelper.convert(last));
+ req.setLast(PBHelperClient.convert(last));
try {
return rpcProxy.complete(null, req.build()).getResult();
} catch (ServiceException e) {
@@ -819,7 +819,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
.setNamespaceQuota(namespaceQuota)
.setStoragespaceQuota(storagespaceQuota);
if (type != null) {
- builder.setStorageType(PBHelper.convertStorageType(type));
+ builder.setStorageType(PBHelperClient.convertStorageType(type));
}
final SetQuotaRequestProto req = builder.build();
try {
@@ -897,7 +897,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
String clientName) throws IOException {
UpdateBlockForPipelineRequestProto req = UpdateBlockForPipelineRequestProto
.newBuilder()
- .setBlock(PBHelper.convert(block))
+ .setBlock(PBHelperClient.convert(block))
.setClientName(clientName)
.build();
try {
@@ -913,8 +913,8 @@ public class ClientNamenodeProtocolTranslatorPB implements
ExtendedBlock newBlock, DatanodeID[] newNodes, String[] storageIDs) throws IOException {
UpdatePipelineRequestProto req = UpdatePipelineRequestProto.newBuilder()
.setClientName(clientName)
- .setOldBlock(PBHelper.convert(oldBlock))
- .setNewBlock(PBHelper.convert(newBlock))
+ .setOldBlock(PBHelperClient.convert(oldBlock))
+ .setNewBlock(PBHelperClient.convert(newBlock))
.addAllNewNodes(Arrays.asList(PBHelper.convert(newNodes)))
.addAllStorageIDs(storageIDs == null ? null : Arrays.asList(storageIDs))
.build();
@@ -945,7 +945,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
public long renewDelegationToken(Token<DelegationTokenIdentifier> token)
throws IOException {
RenewDelegationTokenRequestProto req = RenewDelegationTokenRequestProto.newBuilder().
- setToken(PBHelper.convert(token)).
+ setToken(PBHelperClient.convert(token)).
build();
try {
return rpcProxy.renewDelegationToken(null, req).getNewExpiryTime();
@@ -959,7 +959,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
throws IOException {
CancelDelegationTokenRequestProto req = CancelDelegationTokenRequestProto
.newBuilder()
- .setToken(PBHelper.convert(token))
+ .setToken(PBHelperClient.convert(token))
.build();
try {
rpcProxy.cancelDelegationToken(null, req);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/DatanodeProtocolClientSideTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/DatanodeProtocolClientSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/DatanodeProtocolClientSideTranslatorPB.java
index 94028a2..0b46927 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/DatanodeProtocolClientSideTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/DatanodeProtocolClientSideTranslatorPB.java
@@ -298,11 +298,11 @@ public class DatanodeProtocolClientSideTranslatorPB implements
) throws IOException {
CommitBlockSynchronizationRequestProto.Builder builder =
CommitBlockSynchronizationRequestProto.newBuilder()
- .setBlock(PBHelper.convert(block)).setNewGenStamp(newgenerationstamp)
+ .setBlock(PBHelperClient.convert(block)).setNewGenStamp(newgenerationstamp)
.setNewLength(newlength).setCloseFile(closeFile)
.setDeleteBlock(deleteblock);
for (int i = 0; i < newtargets.length; i++) {
- builder.addNewTaragets(PBHelper.convert(newtargets[i]));
+ builder.addNewTaragets(PBHelperClient.convert(newtargets[i]));
builder.addNewTargetStorages(newtargetstorages[i]);
}
CommitBlockSynchronizationRequestProto req = builder.build();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/InterDatanodeProtocolTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/InterDatanodeProtocolTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/InterDatanodeProtocolTranslatorPB.java
index fee62a4..17ba196 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/InterDatanodeProtocolTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/InterDatanodeProtocolTranslatorPB.java
@@ -105,7 +105,7 @@ public class InterDatanodeProtocolTranslatorPB implements
long recoveryId, long newBlockId, long newLength) throws IOException {
UpdateReplicaUnderRecoveryRequestProto req =
UpdateReplicaUnderRecoveryRequestProto.newBuilder()
- .setBlock(PBHelper.convert(oldBlock))
+ .setBlock(PBHelperClient.convert(oldBlock))
.setNewLength(newLength).setNewBlockId(newBlockId)
.setRecoveryId(recoveryId).build();
try {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/NamenodeProtocolTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/NamenodeProtocolTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/NamenodeProtocolTranslatorPB.java
index 82c5c4c..bcb96ba 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/NamenodeProtocolTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/NamenodeProtocolTranslatorPB.java
@@ -101,7 +101,7 @@ public class NamenodeProtocolTranslatorPB implements NamenodeProtocol,
public BlocksWithLocations getBlocks(DatanodeInfo datanode, long size)
throws IOException {
GetBlocksRequestProto req = GetBlocksRequestProto.newBuilder()
- .setDatanode(PBHelper.convert((DatanodeID)datanode)).setSize(size)
+ .setDatanode(PBHelperClient.convert((DatanodeID)datanode)).setSize(size)
.build();
try {
return PBHelper.convert(rpcProxy.getBlocks(NULL_CONTROLLER, req)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
index 4ca5b26..887accf 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
@@ -346,7 +346,7 @@ public class PBHelper {
if (types == null || types.length == 0) {
return null;
}
- List<StorageTypeProto> list = convertStorageTypes(types);
+ List<StorageTypeProto> list = PBHelperClient.convertStorageTypes(types);
return StorageTypesProto.newBuilder().addAllStorageTypes(list).build();
}
@@ -381,20 +381,6 @@ public class PBHelper {
.getInfoSecurePort() : 0, dn.getIpcPort());
}
- public static DatanodeIDProto convert(DatanodeID dn) {
- // For wire compatibility with older versions we transmit the StorageID
- // which is the same as the DatanodeUuid. Since StorageID is a required
- // field we pass the empty string if the DatanodeUuid is not yet known.
- return DatanodeIDProto.newBuilder()
- .setIpAddr(dn.getIpAddr())
- .setHostName(dn.getHostName())
- .setXferPort(dn.getXferPort())
- .setDatanodeUuid(dn.getDatanodeUuid() != null ? dn.getDatanodeUuid() : "")
- .setInfoPort(dn.getInfoPort())
- .setInfoSecurePort(dn.getInfoSecurePort())
- .setIpcPort(dn.getIpcPort()).build();
- }
-
// Arrays of DatanodeId
public static DatanodeIDProto[] convert(DatanodeID[] did) {
if (did == null)
@@ -402,7 +388,7 @@ public class PBHelper {
final int len = did.length;
DatanodeIDProto[] result = new DatanodeIDProto[len];
for (int i = 0; i < len; ++i) {
- result[i] = convert(did[i]);
+ result[i] = PBHelperClient.convert(did[i]);
}
return result;
}
@@ -433,7 +419,7 @@ public class PBHelper {
.setBlock(convert(blk.getBlock()))
.addAllDatanodeUuids(Arrays.asList(blk.getDatanodeUuids()))
.addAllStorageUuids(Arrays.asList(blk.getStorageIDs()))
- .addAllStorageTypes(convertStorageTypes(blk.getStorageTypes()))
+ .addAllStorageTypes(PBHelperClient.convertStorageTypes(blk.getStorageTypes()))
.build();
}
@@ -595,16 +581,6 @@ public class PBHelper {
eb.getGenerationStamp());
}
- public static ExtendedBlockProto convert(final ExtendedBlock b) {
- if (b == null) return null;
- return ExtendedBlockProto.newBuilder().
- setPoolId(b.getBlockPoolId()).
- setBlockId(b.getBlockId()).
- setNumBytes(b.getNumBytes()).
- setGenerationStamp(b.getGenerationStamp()).
- build();
- }
-
public static RecoveringBlockProto convert(RecoveringBlock b) {
if (b == null) {
return null;
@@ -625,17 +601,6 @@ public class PBHelper {
new RecoveringBlock(block, locs, b.getNewGenStamp());
}
- public static DatanodeInfoProto.AdminState convert(
- final DatanodeInfo.AdminStates inAs) {
- switch (inAs) {
- case NORMAL: return DatanodeInfoProto.AdminState.NORMAL;
- case DECOMMISSION_INPROGRESS:
- return DatanodeInfoProto.AdminState.DECOMMISSION_INPROGRESS;
- case DECOMMISSIONED: return DatanodeInfoProto.AdminState.DECOMMISSIONED;
- default: return DatanodeInfoProto.AdminState.NORMAL;
- }
- }
-
static public DatanodeInfo convert(DatanodeInfoProto di) {
if (di == null) return null;
return new DatanodeInfo(
@@ -647,12 +612,6 @@ public class PBHelper {
di.getXceiverCount(), PBHelper.convert(di.getAdminState()));
}
- static public DatanodeInfoProto convertDatanodeInfo(DatanodeInfo di) {
- if (di == null) return null;
- return convert(di);
- }
-
-
static public DatanodeInfo[] convert(DatanodeInfoProto di[]) {
if (di == null) return null;
DatanodeInfo[] result = new DatanodeInfo[di.length];
@@ -662,27 +621,6 @@ public class PBHelper {
return result;
}
- public static List<? extends HdfsProtos.DatanodeInfoProto> convert(
- DatanodeInfo[] dnInfos) {
- return convert(dnInfos, 0);
- }
-
- /**
- * Copy from {@code dnInfos} to a target of list of same size starting at
- * {@code startIdx}.
- */
- public static List<? extends HdfsProtos.DatanodeInfoProto> convert(
- DatanodeInfo[] dnInfos, int startIdx) {
- if (dnInfos == null)
- return null;
- ArrayList<HdfsProtos.DatanodeInfoProto> protos = Lists
- .newArrayListWithCapacity(dnInfos.length);
- for (int i = startIdx; i < dnInfos.length; i++) {
- protos.add(convert(dnInfos[i]));
- }
- return protos;
- }
-
public static DatanodeInfo[] convert(List<DatanodeInfoProto> list) {
DatanodeInfo[] info = new DatanodeInfo[list.size()];
for (int i = 0; i < info.length; i++) {
@@ -690,32 +628,11 @@ public class PBHelper {
}
return info;
}
-
- public static DatanodeInfoProto convert(DatanodeInfo info) {
- DatanodeInfoProto.Builder builder = DatanodeInfoProto.newBuilder();
- if (info.getNetworkLocation() != null) {
- builder.setLocation(info.getNetworkLocation());
- }
- builder
- .setId(PBHelper.convert((DatanodeID)info))
- .setCapacity(info.getCapacity())
- .setDfsUsed(info.getDfsUsed())
- .setRemaining(info.getRemaining())
- .setBlockPoolUsed(info.getBlockPoolUsed())
- .setCacheCapacity(info.getCacheCapacity())
- .setCacheUsed(info.getCacheUsed())
- .setLastUpdate(info.getLastUpdate())
- .setLastUpdateMonotonic(info.getLastUpdateMonotonic())
- .setXceiverCount(info.getXceiverCount())
- .setAdminState(PBHelper.convert(info.getAdminState()))
- .build();
- return builder.build();
- }
public static DatanodeStorageReportProto convertDatanodeStorageReport(
DatanodeStorageReport report) {
return DatanodeStorageReportProto.newBuilder()
- .setDatanodeInfo(convert(report.getDatanodeInfo()))
+ .setDatanodeInfo(PBHelperClient.convert(report.getDatanodeInfo()))
.addAllStorageReports(convertStorageReports(report.getStorageReports()))
.build();
}
@@ -767,7 +684,7 @@ public class PBHelper {
Lists.newLinkedList(Arrays.asList(b.getCachedLocations()));
for (int i = 0; i < locs.length; i++) {
DatanodeInfo loc = locs[i];
- builder.addLocs(i, PBHelper.convert(loc));
+ builder.addLocs(i, PBHelperClient.convert(loc));
boolean locIsCached = cachedLocs.contains(loc);
builder.addIsCached(locIsCached);
if (locIsCached) {
@@ -781,7 +698,7 @@ public class PBHelper {
StorageType[] storageTypes = b.getStorageTypes();
if (storageTypes != null) {
for (int i = 0; i < storageTypes.length; ++i) {
- builder.addStorageTypes(PBHelper.convertStorageType(storageTypes[i]));
+ builder.addStorageTypes(PBHelperClient.convertStorageType(storageTypes[i]));
}
}
final String[] storageIDs = b.getStorageIDs();
@@ -789,8 +706,8 @@ public class PBHelper {
builder.addAllStorageIDs(Arrays.asList(storageIDs));
}
- return builder.setB(PBHelper.convert(b.getBlock()))
- .setBlockToken(PBHelper.convert(b.getBlockToken()))
+ return builder.setB(PBHelperClient.convert(b.getBlock()))
+ .setBlockToken(PBHelperClient.convert(b.getBlockToken()))
.setCorrupt(b.isCorrupt()).setOffset(b.getStartOffset()).build();
}
@@ -831,14 +748,6 @@ public class PBHelper {
return lb;
}
- public static TokenProto convert(Token<?> tok) {
- return TokenProto.newBuilder().
- setIdentifier(ByteString.copyFrom(tok.getIdentifier())).
- setPassword(ByteString.copyFrom(tok.getPassword())).
- setKind(tok.getKind().toString()).
- setService(tok.getService().toString()).build();
- }
-
public static Token<BlockTokenIdentifier> convert(
TokenProto blockToken) {
return new Token<BlockTokenIdentifier>(blockToken.getIdentifier()
@@ -890,7 +799,7 @@ public class PBHelper {
DatanodeRegistration registration) {
DatanodeRegistrationProto.Builder builder = DatanodeRegistrationProto
.newBuilder();
- return builder.setDatanodeID(PBHelper.convert((DatanodeID) registration))
+ return builder.setDatanodeID(PBHelperClient.convert((DatanodeID) registration))
.setStorageInfo(PBHelper.convert(registration.getStorageInfo()))
.setKeys(PBHelper.convert(registration.getExportedKeys()))
.setSoftwareVersion(registration.getSoftwareVersion()).build();
@@ -982,7 +891,7 @@ public class PBHelper {
if (types != null) {
for (StorageType[] ts : types) {
StorageTypesProto.Builder builder = StorageTypesProto.newBuilder();
- builder.addAllStorageTypes(convertStorageTypes(ts));
+ builder.addAllStorageTypes(PBHelperClient.convertStorageTypes(ts));
list.add(builder.build());
}
}
@@ -1013,7 +922,7 @@ public class PBHelper {
DatanodeInfosProto[] ret = new DatanodeInfosProto[targets.length];
for (int i = 0; i < targets.length; i++) {
ret[i] = DatanodeInfosProto.newBuilder()
- .addAllDatanodes(PBHelper.convert(targets[i])).build();
+ .addAllDatanodes(PBHelperClient.convert(targets[i])).build();
}
return Arrays.asList(ret);
}
@@ -1337,7 +1246,7 @@ public class PBHelper {
fs.getFileBufferSize(),
fs.getEncryptDataTransfer(),
fs.getTrashInterval(),
- PBHelper.convert(fs.getChecksumType()));
+ PBHelperClient.convert(fs.getChecksumType()));
}
public static FsServerDefaultsProto convert(FsServerDefaults fs) {
@@ -1350,7 +1259,7 @@ public class PBHelper {
.setFileBufferSize(fs.getFileBufferSize())
.setEncryptDataTransfer(fs.getEncryptDataTransfer())
.setTrashInterval(fs.getTrashInterval())
- .setChecksumType(PBHelper.convert(fs.getChecksumType()))
+ .setChecksumType(PBHelperClient.convert(fs.getChecksumType()))
.build();
}
@@ -1738,7 +1647,7 @@ public class PBHelper {
if (cs.hasTypeQuotaInfos()) {
for (HdfsProtos.StorageTypeQuotaInfoProto info :
cs.getTypeQuotaInfos().getTypeQuotaInfoList()) {
- StorageType type = PBHelper.convertStorageType(info.getType());
+ StorageType type = PBHelperClient.convertStorageType(info.getType());
builder.typeConsumed(type, info.getConsumed());
builder.typeQuota(type, info.getQuota());
}
@@ -1762,7 +1671,7 @@ public class PBHelper {
for (StorageType t: StorageType.getTypesSupportingQuota()) {
HdfsProtos.StorageTypeQuotaInfoProto info =
HdfsProtos.StorageTypeQuotaInfoProto.newBuilder().
- setType(convertStorageType(t)).
+ setType(PBHelperClient.convertStorageType(t)).
setConsumed(cs.getTypeConsumed(t)).
setQuota(cs.getTypeQuota(t)).
build();
@@ -1807,7 +1716,7 @@ public class PBHelper {
public static DatanodeStorageProto convert(DatanodeStorage s) {
return DatanodeStorageProto.newBuilder()
.setState(PBHelper.convertState(s.getState()))
- .setStorageType(PBHelper.convertStorageType(s.getStorageType()))
+ .setStorageType(PBHelperClient.convertStorageType(s.getStorageType()))
.setStorageUuid(s.getStorageID()).build();
}
@@ -1821,44 +1730,10 @@ public class PBHelper {
}
}
- public static List<StorageTypeProto> convertStorageTypes(
- StorageType[] types) {
- return convertStorageTypes(types, 0);
- }
-
- public static List<StorageTypeProto> convertStorageTypes(
- StorageType[] types, int startIdx) {
- if (types == null) {
- return null;
- }
- final List<StorageTypeProto> protos = new ArrayList<StorageTypeProto>(
- types.length);
- for (int i = startIdx; i < types.length; ++i) {
- protos.add(convertStorageType(types[i]));
- }
- return protos;
- }
-
- public static StorageTypeProto convertStorageType(StorageType type) {
- switch(type) {
- case DISK:
- return StorageTypeProto.DISK;
- case SSD:
- return StorageTypeProto.SSD;
- case ARCHIVE:
- return StorageTypeProto.ARCHIVE;
- case RAM_DISK:
- return StorageTypeProto.RAM_DISK;
- default:
- throw new IllegalStateException(
- "BUG: StorageType not found, type=" + type);
- }
- }
-
public static DatanodeStorage convert(DatanodeStorageProto s) {
return new DatanodeStorage(s.getStorageUuid(),
PBHelper.convertState(s.getState()),
- PBHelper.convertStorageType(s.getStorageType()));
+ PBHelperClient.convertStorageType(s.getStorageType()));
}
private static State convertState(StorageState state) {
@@ -1871,22 +1746,6 @@ public class PBHelper {
}
}
- public static StorageType convertStorageType(StorageTypeProto type) {
- switch(type) {
- case DISK:
- return StorageType.DISK;
- case SSD:
- return StorageType.SSD;
- case ARCHIVE:
- return StorageType.ARCHIVE;
- case RAM_DISK:
- return StorageType.RAM_DISK;
- default:
- throw new IllegalStateException(
- "BUG: StorageTypeProto not found, type=" + type);
- }
- }
-
public static StorageType[] convertStorageTypes(
List<StorageTypeProto> storageTypesList, int expectedSize) {
final StorageType[] storageTypes = new StorageType[expectedSize];
@@ -1895,7 +1754,7 @@ public class PBHelper {
Arrays.fill(storageTypes, StorageType.DEFAULT);
} else {
for (int i = 0; i < storageTypes.length; ++i) {
- storageTypes[i] = convertStorageType(storageTypesList.get(i));
+ storageTypes[i] = PBHelperClient.convertStorageType(storageTypesList.get(i));
}
}
return storageTypes;
@@ -2079,10 +1938,6 @@ public class PBHelper {
return reportProto;
}
- public static DataChecksum.Type convert(HdfsProtos.ChecksumTypeProto type) {
- return DataChecksum.Type.valueOf(type.getNumber());
- }
-
public static CacheDirectiveInfoProto convert
(CacheDirectiveInfo info) {
CacheDirectiveInfoProto.Builder builder =
@@ -2255,9 +2110,6 @@ public class PBHelper {
return new CachePoolEntry(info, stats);
}
- public static HdfsProtos.ChecksumTypeProto convert(DataChecksum.Type type) {
- return HdfsProtos.ChecksumTypeProto.valueOf(type.id);
- }
public static DatanodeLocalInfoProto convert(DatanodeLocalInfo info) {
DatanodeLocalInfoProto.Builder builder = DatanodeLocalInfoProto.newBuilder();
@@ -2272,17 +2124,6 @@ public class PBHelper {
proto.getConfigVersion(), proto.getUptime());
}
- public static InputStream vintPrefixed(final InputStream input)
- throws IOException {
- final int firstByte = input.read();
- if (firstByte == -1) {
- throw new EOFException("Premature EOF: no length prefix available");
- }
-
- int size = CodedInputStream.readRawVarint32(firstByte, input);
- assert size >= 0;
- return new ExactSizeInputStream(input, size);
- }
private static AclEntryScopeProto convert(AclEntryScope v) {
return AclEntryScopeProto.valueOf(v.ordinal());
@@ -2506,30 +2347,11 @@ public class PBHelper {
proto.getKeyName());
}
- public static ShortCircuitShmSlotProto convert(SlotId slotId) {
- return ShortCircuitShmSlotProto.newBuilder().
- setShmId(convert(slotId.getShmId())).
- setSlotIdx(slotId.getSlotIdx()).
- build();
- }
-
- public static ShortCircuitShmIdProto convert(ShmId shmId) {
- return ShortCircuitShmIdProto.newBuilder().
- setHi(shmId.getHi()).
- setLo(shmId.getLo()).
- build();
-
- }
-
public static SlotId convert(ShortCircuitShmSlotProto slotId) {
- return new SlotId(PBHelper.convert(slotId.getShmId()),
+ return new SlotId(PBHelperClient.convert(slotId.getShmId()),
slotId.getSlotIdx());
}
- public static ShmId convert(ShortCircuitShmIdProto shmId) {
- return new ShmId(shmId.getHi(), shmId.getLo());
- }
-
private static Event.CreateEvent.INodeType createTypeConvert(InotifyProtos.INodeType
type) {
switch (type) {
@@ -3036,18 +2858,6 @@ public class PBHelper {
ezKeyVersionName);
}
- public static List<Boolean> convert(boolean[] targetPinnings, int idx) {
- List<Boolean> pinnings = new ArrayList<Boolean>();
- if (targetPinnings == null) {
- pinnings.add(Boolean.FALSE);
- } else {
- for (; idx < targetPinnings.length; ++idx) {
- pinnings.add(Boolean.valueOf(targetPinnings[idx]));
- }
- }
- return pinnings;
- }
-
public static boolean[] convertBooleanList(
List<Boolean> targetPinningsList) {
final boolean[] targetPinnings = new boolean[targetPinningsList.size()];
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/InvalidBlockTokenException.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/InvalidBlockTokenException.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/InvalidBlockTokenException.java
deleted file mode 100644
index 2fa86fa..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/InvalidBlockTokenException.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.hadoop.hdfs.security.token.block;
-
-import java.io.IOException;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-
-/**
- * Access token verification failed.
- */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-public class InvalidBlockTokenException extends IOException {
- private static final long serialVersionUID = 168L;
-
- public InvalidBlockTokenException() {
- super();
- }
-
- public InvalidBlockTokenException(String msg) {
- super(msg);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
index a5e22ec..be1a9ef 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
@@ -17,7 +17,7 @@
*/
package org.apache.hadoop.hdfs.server.balancer;
-import static org.apache.hadoop.hdfs.protocolPB.PBHelper.vintPrefixed;
+import static org.apache.hadoop.hdfs.protocolPB.PBHelperClient.vintPrefixed;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/CachingStrategy.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/CachingStrategy.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/CachingStrategy.java
deleted file mode 100644
index 215df13..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/CachingStrategy.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hdfs.server.datanode;
-
-/**
- * The caching strategy we should use for an HDFS read or write operation.
- */
-public class CachingStrategy {
- private final Boolean dropBehind; // null = use server defaults
- private final Long readahead; // null = use server defaults
-
- public static CachingStrategy newDefaultStrategy() {
- return new CachingStrategy(null, null);
- }
-
- public static CachingStrategy newDropBehind() {
- return new CachingStrategy(true, null);
- }
-
- public static class Builder {
- private Boolean dropBehind;
- private Long readahead;
-
- public Builder(CachingStrategy prev) {
- this.dropBehind = prev.dropBehind;
- this.readahead = prev.readahead;
- }
-
- public Builder setDropBehind(Boolean dropBehind) {
- this.dropBehind = dropBehind;
- return this;
- }
-
- public Builder setReadahead(Long readahead) {
- this.readahead = readahead;
- return this;
- }
-
- public CachingStrategy build() {
- return new CachingStrategy(dropBehind, readahead);
- }
- }
-
- public CachingStrategy(Boolean dropBehind, Long readahead) {
- this.dropBehind = dropBehind;
- this.readahead = readahead;
- }
-
- public Boolean getDropBehind() {
- return dropBehind;
- }
-
- public Long getReadahead() {
- return readahead;
- }
-
- public String toString() {
- return "CachingStrategy(dropBehind=" + dropBehind +
- ", readahead=" + readahead + ")";
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
index ecf139c..5bc50b0 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
@@ -135,7 +135,7 @@ import org.apache.hadoop.hdfs.protocolPB.DatanodeProtocolClientSideTranslatorPB;
import org.apache.hadoop.hdfs.protocolPB.InterDatanodeProtocolPB;
import org.apache.hadoop.hdfs.protocolPB.InterDatanodeProtocolServerSideTranslatorPB;
import org.apache.hadoop.hdfs.protocolPB.InterDatanodeProtocolTranslatorPB;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
import org.apache.hadoop.hdfs.security.token.block.BlockPoolTokenSecretManager;
import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier.AccessMode;
@@ -2142,7 +2142,7 @@ public class DataNode extends ReconfigurableBase
// read ack
if (isClient) {
DNTransferAckProto closeAck = DNTransferAckProto.parseFrom(
- PBHelper.vintPrefixed(in));
+ PBHelperClient.vintPrefixed(in));
if (LOG.isDebugEnabled()) {
LOG.debug(getClass().getSimpleName() + ": close-ack=" + closeAck);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
index e9cf436..dfaa525 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
@@ -70,7 +70,7 @@ import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ReadOpChecksumIn
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ReleaseShortCircuitAccessResponseProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ShortCircuitShmResponseProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
import org.apache.hadoop.hdfs.server.datanode.DataNode.ShortCircuitFdsUnsupportedException;
import org.apache.hadoop.hdfs.server.datanode.DataNode.ShortCircuitFdsVersionException;
@@ -427,7 +427,7 @@ class DataXceiver extends Receiver implements Runnable {
throws IOException {
DataNodeFaultInjector.get().sendShortCircuitShmResponse();
ShortCircuitShmResponseProto.newBuilder().setStatus(SUCCESS).
- setId(PBHelper.convert(shmInfo.shmId)).build().
+ setId(PBHelperClient.convert(shmInfo.shmId)).build().
writeDelimitedTo(socketOut);
// Send the file descriptor for the shared memory segment.
byte buf[] = new byte[] { (byte)0 };
@@ -559,7 +559,7 @@ class DataXceiver extends Receiver implements Runnable {
// to respond with a Status enum.
try {
ClientReadStatusProto stat = ClientReadStatusProto.parseFrom(
- PBHelper.vintPrefixed(in));
+ PBHelperClient.vintPrefixed(in));
if (!stat.hasStatus()) {
LOG.warn("Client " + peer.getRemoteAddressString() +
" did not send a valid status code after reading. " +
@@ -745,7 +745,7 @@ class DataXceiver extends Receiver implements Runnable {
// read connect ack (only for clients, not for replication req)
if (isClient) {
BlockOpResponseProto connectAck =
- BlockOpResponseProto.parseFrom(PBHelper.vintPrefixed(mirrorIn));
+ BlockOpResponseProto.parseFrom(PBHelperClient.vintPrefixed(mirrorIn));
mirrorInStatus = connectAck.getStatus();
firstBadLink = connectAck.getFirstBadLink();
if (LOG.isDebugEnabled() || mirrorInStatus != SUCCESS) {
@@ -962,7 +962,7 @@ class DataXceiver extends Receiver implements Runnable {
.setBytesPerCrc(bytesPerCRC)
.setCrcPerBlock(crcPerBlock)
.setMd5(ByteString.copyFrom(md5.getDigest()))
- .setCrcType(PBHelper.convert(checksum.getChecksumType())))
+ .setCrcType(PBHelperClient.convert(checksum.getChecksumType())))
.build()
.writeDelimitedTo(out);
out.flush();
@@ -1147,8 +1147,8 @@ class DataXceiver extends Receiver implements Runnable {
// receive the response from the proxy
BlockOpResponseProto copyResponse = BlockOpResponseProto.parseFrom(
- PBHelper.vintPrefixed(proxyReply));
-
+ PBHelperClient.vintPrefixed(proxyReply));
+
String logInfo = "copy block " + block + " from "
+ proxySock.getRemoteSocketAddress();
DataTransferProtoUtil.checkBlockOpStatus(copyResponse, logInfo);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormatPBINode.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormatPBINode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormatPBINode.java
index 25fd99d..3a9c64e 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormatPBINode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormatPBINode.java
@@ -42,6 +42,7 @@ import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.protocol.proto.HdfsProtos.BlockProto;
import org.apache.hadoop.hdfs.protocolPB.PBHelper;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfo;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfoContiguous;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager;
@@ -155,7 +156,7 @@ public final class FSImageFormatPBINode {
QuotaByStorageTypeFeatureProto proto) {
ImmutableList.Builder<QuotaByStorageTypeEntry> b = ImmutableList.builder();
for (QuotaByStorageTypeEntryProto quotaEntry : proto.getQuotasList()) {
- StorageType type = PBHelper.convertStorageType(quotaEntry.getStorageType());
+ StorageType type = PBHelperClient.convertStorageType(quotaEntry.getStorageType());
long quota = quotaEntry.getQuota();
b.add(new QuotaByStorageTypeEntry.Builder().setStorageType(type)
.setQuota(quota).build());
@@ -462,7 +463,7 @@ public final class FSImageFormatPBINode {
if (q.getTypeSpace(t) >= 0) {
QuotaByStorageTypeEntryProto.Builder eb =
QuotaByStorageTypeEntryProto.newBuilder().
- setStorageType(PBHelper.convertStorageType(t)).
+ setStorageType(PBHelperClient.convertStorageType(t)).
setQuota(q.getTypeSpace(t));
b.addQuotas(eb);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShm.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShm.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShm.java
deleted file mode 100644
index 81cc68d..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShm.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hdfs.shortcircuit;
-
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.util.Iterator;
-
-import org.apache.hadoop.hdfs.net.DomainPeer;
-import org.apache.hadoop.hdfs.shortcircuit.DfsClientShmManager.EndpointShmManager;
-import org.apache.hadoop.net.unix.DomainSocket;
-import org.apache.hadoop.net.unix.DomainSocketWatcher;
-
-import com.google.common.base.Preconditions;
-
-/**
- * DfsClientShm is a subclass of ShortCircuitShm which is used by the
- * DfsClient.
- * When the UNIX domain socket associated with this shared memory segment
- * closes unexpectedly, we mark the slots inside this segment as disconnected.
- * ShortCircuitReplica objects that contain disconnected slots are stale,
- * and will not be used to service new reads or mmap operations.
- * However, in-progress read or mmap operations will continue to proceed.
- * Once the last slot is deallocated, the segment can be safely munmapped.
- *
- * Slots may also become stale because the associated replica has been deleted
- * on the DataNode. In this case, the DataNode will clear the 'valid' bit.
- * The client will then see these slots as stale (see
- * #{ShortCircuitReplica#isStale}).
- */
-public class DfsClientShm extends ShortCircuitShm
- implements DomainSocketWatcher.Handler {
- /**
- * The EndpointShmManager associated with this shared memory segment.
- */
- private final EndpointShmManager manager;
-
- /**
- * The UNIX domain socket associated with this DfsClientShm.
- * We rely on the DomainSocketWatcher to close the socket associated with
- * this DomainPeer when necessary.
- */
- private final DomainPeer peer;
-
- /**
- * True if this shared memory segment has lost its connection to the
- * DataNode.
- *
- * {@link DfsClientShm#handle} sets this to true.
- */
- private boolean disconnected = false;
-
- DfsClientShm(ShmId shmId, FileInputStream stream, EndpointShmManager manager,
- DomainPeer peer) throws IOException {
- super(shmId, stream);
- this.manager = manager;
- this.peer = peer;
- }
-
- public EndpointShmManager getEndpointShmManager() {
- return manager;
- }
-
- public DomainPeer getPeer() {
- return peer;
- }
-
- /**
- * Determine if the shared memory segment is disconnected from the DataNode.
- *
- * This must be called with the DfsClientShmManager lock held.
- *
- * @return True if the shared memory segment is stale.
- */
- public synchronized boolean isDisconnected() {
- return disconnected;
- }
-
- /**
- * Handle the closure of the UNIX domain socket associated with this shared
- * memory segment by marking this segment as stale.
- *
- * If there are no slots associated with this shared memory segment, it will
- * be freed immediately in this function.
- */
- @Override
- public boolean handle(DomainSocket sock) {
- manager.unregisterShm(getShmId());
- synchronized (this) {
- Preconditions.checkState(!disconnected);
- disconnected = true;
- boolean hadSlots = false;
- for (Iterator<Slot> iter = slotIterator(); iter.hasNext(); ) {
- Slot slot = iter.next();
- slot.makeInvalid();
- hadSlots = true;
- }
- if (!hadSlots) {
- free();
- }
- }
- return true;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShmManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShmManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShmManager.java
deleted file mode 100644
index 062539a..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShmManager.java
+++ /dev/null
@@ -1,514 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hdfs.shortcircuit;
-
-import java.io.BufferedOutputStream;
-import java.io.Closeable;
-import java.io.DataOutputStream;
-import java.io.EOFException;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map.Entry;
-import java.util.TreeMap;
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.ReentrantLock;
-
-import org.apache.commons.lang.mutable.MutableBoolean;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.hdfs.ExtendedBlockId;
-import org.apache.hadoop.hdfs.net.DomainPeer;
-import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
-import org.apache.hadoop.hdfs.protocol.datatransfer.DataTransferProtocol;
-import org.apache.hadoop.hdfs.protocol.datatransfer.Sender;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ShortCircuitShmResponseProto;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
-import org.apache.hadoop.hdfs.server.datanode.ShortCircuitRegistry;
-import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.ShmId;
-import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.Slot;
-import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.net.unix.DomainSocket;
-import org.apache.hadoop.net.unix.DomainSocketWatcher;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-
-/**
- * Manages short-circuit memory segments for an HDFS client.
- *
- * Clients are responsible for requesting and releasing shared memory segments used
- * for communicating with the DataNode. The client will try to allocate new slots
- * in the set of existing segments, falling back to getting a new segment from the
- * DataNode via {@link DataTransferProtocol#requestShortCircuitFds}.
- *
- * The counterpart to this class on the DataNode is {@link ShortCircuitRegistry}.
- * See {@link ShortCircuitRegistry} for more information on the communication protocol.
- */
-@InterfaceAudience.Private
-public class DfsClientShmManager implements Closeable {
- private static final Log LOG = LogFactory.getLog(DfsClientShmManager.class);
-
- /**
- * Manages short-circuit memory segments that pertain to a given DataNode.
- */
- class EndpointShmManager {
- /**
- * The datanode we're managing.
- */
- private final DatanodeInfo datanode;
-
- /**
- * Shared memory segments which have no empty slots.
- *
- * Protected by the manager lock.
- */
- private final TreeMap<ShmId, DfsClientShm> full =
- new TreeMap<ShmId, DfsClientShm>();
-
- /**
- * Shared memory segments which have at least one empty slot.
- *
- * Protected by the manager lock.
- */
- private final TreeMap<ShmId, DfsClientShm> notFull =
- new TreeMap<ShmId, DfsClientShm>();
-
- /**
- * True if this datanode doesn't support short-circuit shared memory
- * segments.
- *
- * Protected by the manager lock.
- */
- private boolean disabled = false;
-
- /**
- * True if we're in the process of loading a shared memory segment from
- * this DataNode.
- *
- * Protected by the manager lock.
- */
- private boolean loading = false;
-
- EndpointShmManager (DatanodeInfo datanode) {
- this.datanode = datanode;
- }
-
- /**
- * Pull a slot out of a preexisting shared memory segment.
- *
- * Must be called with the manager lock held.
- *
- * @param blockId The blockId to put inside the Slot object.
- *
- * @return null if none of our shared memory segments contain a
- * free slot; the slot object otherwise.
- */
- private Slot allocSlotFromExistingShm(ExtendedBlockId blockId) {
- if (notFull.isEmpty()) {
- return null;
- }
- Entry<ShmId, DfsClientShm> entry = notFull.firstEntry();
- DfsClientShm shm = entry.getValue();
- ShmId shmId = shm.getShmId();
- Slot slot = shm.allocAndRegisterSlot(blockId);
- if (shm.isFull()) {
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": pulled the last slot " + slot.getSlotIdx() +
- " out of " + shm);
- }
- DfsClientShm removedShm = notFull.remove(shmId);
- Preconditions.checkState(removedShm == shm);
- full.put(shmId, shm);
- } else {
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": pulled slot " + slot.getSlotIdx() +
- " out of " + shm);
- }
- }
- return slot;
- }
-
- /**
- * Ask the DataNode for a new shared memory segment. This function must be
- * called with the manager lock held. We will release the lock while
- * communicating with the DataNode.
- *
- * @param clientName The current client name.
- * @param peer The peer to use to talk to the DataNode.
- *
- * @return Null if the DataNode does not support shared memory
- * segments, or experienced an error creating the
- * shm. The shared memory segment itself on success.
- * @throws IOException If there was an error communicating over the socket.
- * We will not throw an IOException unless the socket
- * itself (or the network) is the problem.
- */
- private DfsClientShm requestNewShm(String clientName, DomainPeer peer)
- throws IOException {
- final DataOutputStream out =
- new DataOutputStream(
- new BufferedOutputStream(peer.getOutputStream()));
- new Sender(out).requestShortCircuitShm(clientName);
- ShortCircuitShmResponseProto resp =
- ShortCircuitShmResponseProto.parseFrom(
- PBHelper.vintPrefixed(peer.getInputStream()));
- String error = resp.hasError() ? resp.getError() : "(unknown)";
- switch (resp.getStatus()) {
- case SUCCESS:
- DomainSocket sock = peer.getDomainSocket();
- byte buf[] = new byte[1];
- FileInputStream fis[] = new FileInputStream[1];
- if (sock.recvFileInputStreams(fis, buf, 0, buf.length) < 0) {
- throw new EOFException("got EOF while trying to transfer the " +
- "file descriptor for the shared memory segment.");
- }
- if (fis[0] == null) {
- throw new IOException("the datanode " + datanode + " failed to " +
- "pass a file descriptor for the shared memory segment.");
- }
- try {
- DfsClientShm shm =
- new DfsClientShm(PBHelper.convert(resp.getId()),
- fis[0], this, peer);
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": createNewShm: created " + shm);
- }
- return shm;
- } finally {
- IOUtils.cleanup(LOG, fis[0]);
- }
- case ERROR_UNSUPPORTED:
- // The DataNode just does not support short-circuit shared memory
- // access, and we should stop asking.
- LOG.info(this + ": datanode does not support short-circuit " +
- "shared memory access: " + error);
- disabled = true;
- return null;
- default:
- // The datanode experienced some kind of unexpected error when trying to
- // create the short-circuit shared memory segment.
- LOG.warn(this + ": error requesting short-circuit shared memory " +
- "access: " + error);
- return null;
- }
- }
-
- /**
- * Allocate a new shared memory slot connected to this datanode.
- *
- * Must be called with the EndpointShmManager lock held.
- *
- * @param peer The peer to use to talk to the DataNode.
- * @param usedPeer (out param) Will be set to true if we used the peer.
- * When a peer is used
- *
- * @param clientName The client name.
- * @param blockId The block ID to use.
- * @return null if the DataNode does not support shared memory
- * segments, or experienced an error creating the
- * shm. The shared memory segment itself on success.
- * @throws IOException If there was an error communicating over the socket.
- */
- Slot allocSlot(DomainPeer peer, MutableBoolean usedPeer,
- String clientName, ExtendedBlockId blockId) throws IOException {
- while (true) {
- if (closed) {
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": the DfsClientShmManager has been closed.");
- }
- return null;
- }
- if (disabled) {
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": shared memory segment access is disabled.");
- }
- return null;
- }
- // Try to use an existing slot.
- Slot slot = allocSlotFromExistingShm(blockId);
- if (slot != null) {
- return slot;
- }
- // There are no free slots. If someone is loading more slots, wait
- // for that to finish.
- if (loading) {
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": waiting for loading to finish...");
- }
- finishedLoading.awaitUninterruptibly();
- } else {
- // Otherwise, load the slot ourselves.
- loading = true;
- lock.unlock();
- DfsClientShm shm;
- try {
- shm = requestNewShm(clientName, peer);
- if (shm == null) continue;
- // See #{DfsClientShmManager#domainSocketWatcher} for details
- // about why we do this before retaking the manager lock.
- domainSocketWatcher.add(peer.getDomainSocket(), shm);
- // The DomainPeer is now our responsibility, and should not be
- // closed by the caller.
- usedPeer.setValue(true);
- } finally {
- lock.lock();
- loading = false;
- finishedLoading.signalAll();
- }
- if (shm.isDisconnected()) {
- // If the peer closed immediately after the shared memory segment
- // was created, the DomainSocketWatcher callback might already have
- // fired and marked the shm as disconnected. In this case, we
- // obviously don't want to add the SharedMemorySegment to our list
- // of valid not-full segments.
- if (LOG.isDebugEnabled()) {
- LOG.debug(this + ": the UNIX domain socket associated with " +
- "this short-circuit memory closed before we could make " +
- "use of the shm.");
- }
- } else {
- notFull.put(shm.getShmId(), shm);
- }
- }
- }
- }
-
- /**
- * Stop tracking a slot.
- *
- * Must be called with the EndpointShmManager lock held.
- *
- * @param slot The slot to release.
- */
- void freeSlot(Slot slot) {
- DfsClientShm shm = (DfsClientShm)slot.getShm();
- shm.unregisterSlot(slot.getSlotIdx());
- if (shm.isDisconnected()) {
- // Stale shared memory segments should not be tracked here.
- Preconditions.checkState(!full.containsKey(shm.getShmId()));
- Preconditions.checkState(!notFull.containsKey(shm.getShmId()));
- if (shm.isEmpty()) {
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": freeing empty stale " + shm);
- }
- shm.free();
- }
- } else {
- ShmId shmId = shm.getShmId();
- full.remove(shmId); // The shm can't be full if we just freed a slot.
- if (shm.isEmpty()) {
- notFull.remove(shmId);
-
- // If the shared memory segment is now empty, we call shutdown(2) on
- // the UNIX domain socket associated with it. The DomainSocketWatcher,
- // which is watching this socket, will call DfsClientShm#handle,
- // cleaning up this shared memory segment.
- //
- // See #{DfsClientShmManager#domainSocketWatcher} for details about why
- // we don't want to call DomainSocketWatcher#remove directly here.
- //
- // Note that we could experience 'fragmentation' here, where the
- // DFSClient allocates a bunch of slots in different shared memory
- // segments, and then frees most of them, but never fully empties out
- // any segment. We make some attempt to avoid this fragmentation by
- // always allocating new slots out of the shared memory segment with the
- // lowest ID, but it could still occur. In most workloads,
- // fragmentation should not be a major concern, since it doesn't impact
- // peak file descriptor usage or the speed of allocation.
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": shutting down UNIX domain socket for " +
- "empty " + shm);
- }
- shutdown(shm);
- } else {
- notFull.put(shmId, shm);
- }
- }
- }
-
- /**
- * Unregister a shared memory segment.
- *
- * Once a segment is unregistered, we will not allocate any more slots
- * inside that segment.
- *
- * The DomainSocketWatcher calls this while holding the DomainSocketWatcher
- * lock.
- *
- * @param shmId The ID of the shared memory segment to unregister.
- */
- void unregisterShm(ShmId shmId) {
- lock.lock();
- try {
- full.remove(shmId);
- notFull.remove(shmId);
- } finally {
- lock.unlock();
- }
- }
-
- @Override
- public String toString() {
- return String.format("EndpointShmManager(%s, parent=%s)",
- datanode, DfsClientShmManager.this);
- }
-
- PerDatanodeVisitorInfo getVisitorInfo() {
- return new PerDatanodeVisitorInfo(full, notFull, disabled);
- }
-
- final void shutdown(DfsClientShm shm) {
- try {
- shm.getPeer().getDomainSocket().shutdown();
- } catch (IOException e) {
- LOG.warn(this + ": error shutting down shm: got IOException calling " +
- "shutdown(SHUT_RDWR)", e);
- }
- }
- }
-
- private boolean closed = false;
-
- private final ReentrantLock lock = new ReentrantLock();
-
- /**
- * A condition variable which is signalled when we finish loading a segment
- * from the Datanode.
- */
- private final Condition finishedLoading = lock.newCondition();
-
- /**
- * Information about each Datanode.
- */
- private final HashMap<DatanodeInfo, EndpointShmManager> datanodes =
- new HashMap<DatanodeInfo, EndpointShmManager>(1);
-
- /**
- * The DomainSocketWatcher which keeps track of the UNIX domain socket
- * associated with each shared memory segment.
- *
- * Note: because the DomainSocketWatcher makes callbacks into this
- * DfsClientShmManager object, you must MUST NOT attempt to take the
- * DomainSocketWatcher lock while holding the DfsClientShmManager lock,
- * or else deadlock might result. This means that most DomainSocketWatcher
- * methods are off-limits unless you release the manager lock first.
- */
- private final DomainSocketWatcher domainSocketWatcher;
-
- DfsClientShmManager(int interruptCheckPeriodMs) throws IOException {
- this.domainSocketWatcher = new DomainSocketWatcher(interruptCheckPeriodMs,
- "client");
- }
-
- public Slot allocSlot(DatanodeInfo datanode, DomainPeer peer,
- MutableBoolean usedPeer, ExtendedBlockId blockId,
- String clientName) throws IOException {
- lock.lock();
- try {
- if (closed) {
- LOG.trace(this + ": the DfsClientShmManager isclosed.");
- return null;
- }
- EndpointShmManager shmManager = datanodes.get(datanode);
- if (shmManager == null) {
- shmManager = new EndpointShmManager(datanode);
- datanodes.put(datanode, shmManager);
- }
- return shmManager.allocSlot(peer, usedPeer, clientName, blockId);
- } finally {
- lock.unlock();
- }
- }
-
- public void freeSlot(Slot slot) {
- lock.lock();
- try {
- DfsClientShm shm = (DfsClientShm)slot.getShm();
- shm.getEndpointShmManager().freeSlot(slot);
- } finally {
- lock.unlock();
- }
- }
-
- @VisibleForTesting
- public static class PerDatanodeVisitorInfo {
- public final TreeMap<ShmId, DfsClientShm> full;
- public final TreeMap<ShmId, DfsClientShm> notFull;
- public final boolean disabled;
-
- PerDatanodeVisitorInfo(TreeMap<ShmId, DfsClientShm> full,
- TreeMap<ShmId, DfsClientShm> notFull, boolean disabled) {
- this.full = full;
- this.notFull = notFull;
- this.disabled = disabled;
- }
- }
-
- @VisibleForTesting
- public interface Visitor {
- void visit(HashMap<DatanodeInfo, PerDatanodeVisitorInfo> info)
- throws IOException;
- }
-
- @VisibleForTesting
- public void visit(Visitor visitor) throws IOException {
- lock.lock();
- try {
- HashMap<DatanodeInfo, PerDatanodeVisitorInfo> info =
- new HashMap<DatanodeInfo, PerDatanodeVisitorInfo>();
- for (Entry<DatanodeInfo, EndpointShmManager> entry :
- datanodes.entrySet()) {
- info.put(entry.getKey(), entry.getValue().getVisitorInfo());
- }
- visitor.visit(info);
- } finally {
- lock.unlock();
- }
- }
-
- /**
- * Close the DfsClientShmManager.
- */
- @Override
- public void close() throws IOException {
- lock.lock();
- try {
- if (closed) return;
- closed = true;
- } finally {
- lock.unlock();
- }
- // When closed, the domainSocketWatcher will issue callbacks that mark
- // all the outstanding DfsClientShm segments as stale.
- IOUtils.cleanup(LOG, domainSocketWatcher);
- }
-
-
- @Override
- public String toString() {
- return String.format("ShortCircuitShmManager(%08x)",
- System.identityHashCode(this));
- }
-
- @VisibleForTesting
- public DomainSocketWatcher getDomainSocketWatcher() {
- return domainSocketWatcher;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitCache.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitCache.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitCache.java
index db4cbe2..15b8dea 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitCache.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitCache.java
@@ -44,7 +44,7 @@ import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.datatransfer.Sender;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ReleaseShortCircuitAccessResponseProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.Slot;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.ipc.RetriableException;
@@ -201,7 +201,7 @@ public class ShortCircuitCache implements Closeable {
DataInputStream in = new DataInputStream(sock.getInputStream());
ReleaseShortCircuitAccessResponseProto resp =
ReleaseShortCircuitAccessResponseProto.parseFrom(
- PBHelper.vintPrefixed(in));
+ PBHelperClient.vintPrefixed(in));
if (resp.getStatus() != Status.SUCCESS) {
String error = resp.hasError() ? resp.getError() : "(unknown)";
throw new IOException(resp.getStatus().toString() + ": " + error);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/490bb5eb/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitShm.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitShm.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitShm.java
deleted file mode 100644
index 7b89d0a..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitShm.java
+++ /dev/null
@@ -1,646 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hdfs.shortcircuit;
-
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.lang.reflect.Field;
-import java.util.BitSet;
-import java.util.Iterator;
-import java.util.NoSuchElementException;
-import java.util.Random;
-
-import org.apache.commons.lang.builder.EqualsBuilder;
-import org.apache.commons.lang.builder.HashCodeBuilder;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.fs.InvalidRequestException;
-import org.apache.hadoop.hdfs.ExtendedBlockId;
-import org.apache.hadoop.io.nativeio.NativeIO;
-import org.apache.hadoop.io.nativeio.NativeIO.POSIX;
-import org.apache.hadoop.util.Shell;
-import org.apache.hadoop.util.StringUtils;
-
-import sun.misc.Unsafe;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.ComparisonChain;
-import com.google.common.primitives.Ints;
-
-/**
- * A shared memory segment used to implement short-circuit reads.
- */
-public class ShortCircuitShm {
- private static final Log LOG = LogFactory.getLog(ShortCircuitShm.class);
-
- protected static final int BYTES_PER_SLOT = 64;
-
- private static final Unsafe unsafe = safetyDance();
-
- private static Unsafe safetyDance() {
- try {
- Field f = Unsafe.class.getDeclaredField("theUnsafe");
- f.setAccessible(true);
- return (Unsafe)f.get(null);
- } catch (Throwable e) {
- LOG.error("failed to load misc.Unsafe", e);
- }
- return null;
- }
-
- /**
- * Calculate the usable size of a shared memory segment.
- * We round down to a multiple of the slot size and do some validation.
- *
- * @param stream The stream we're using.
- * @return The usable size of the shared memory segment.
- */
- private static int getUsableLength(FileInputStream stream)
- throws IOException {
- int intSize = Ints.checkedCast(stream.getChannel().size());
- int slots = intSize / BYTES_PER_SLOT;
- if (slots == 0) {
- throw new IOException("size of shared memory segment was " +
- intSize + ", but that is not enough to hold even one slot.");
- }
- return slots * BYTES_PER_SLOT;
- }
-
- /**
- * Identifies a DfsClientShm.
- */
- public static class ShmId implements Comparable<ShmId> {
- private static final Random random = new Random();
- private final long hi;
- private final long lo;
-
- /**
- * Generate a random ShmId.
- *
- * We generate ShmIds randomly to prevent a malicious client from
- * successfully guessing one and using that to interfere with another
- * client.
- */
- public static ShmId createRandom() {
- return new ShmId(random.nextLong(), random.nextLong());
- }
-
- public ShmId(long hi, long lo) {
- this.hi = hi;
- this.lo = lo;
- }
-
- public long getHi() {
- return hi;
- }
-
- public long getLo() {
- return lo;
- }
-
- @Override
- public boolean equals(Object o) {
- if ((o == null) || (o.getClass() != this.getClass())) {
- return false;
- }
- ShmId other = (ShmId)o;
- return new EqualsBuilder().
- append(hi, other.hi).
- append(lo, other.lo).
- isEquals();
- }
-
- @Override
- public int hashCode() {
- return new HashCodeBuilder().
- append(this.hi).
- append(this.lo).
- toHashCode();
- }
-
- @Override
- public String toString() {
- return String.format("%016x%016x", hi, lo);
- }
-
- @Override
- public int compareTo(ShmId other) {
- return ComparisonChain.start().
- compare(hi, other.hi).
- compare(lo, other.lo).
- result();
- }
- };
-
- /**
- * Uniquely identifies a slot.
- */
- public static class SlotId {
- private final ShmId shmId;
- private final int slotIdx;
-
- public SlotId(ShmId shmId, int slotIdx) {
- this.shmId = shmId;
- this.slotIdx = slotIdx;
- }
-
- public ShmId getShmId() {
- return shmId;
- }
-
- public int getSlotIdx() {
- return slotIdx;
- }
-
- @Override
- public boolean equals(Object o) {
- if ((o == null) || (o.getClass() != this.getClass())) {
- return false;
- }
- SlotId other = (SlotId)o;
- return new EqualsBuilder().
- append(shmId, other.shmId).
- append(slotIdx, other.slotIdx).
- isEquals();
- }
-
- @Override
- public int hashCode() {
- return new HashCodeBuilder().
- append(this.shmId).
- append(this.slotIdx).
- toHashCode();
- }
-
- @Override
- public String toString() {
- return String.format("SlotId(%s:%d)", shmId.toString(), slotIdx);
- }
- }
-
- public class SlotIterator implements Iterator<Slot> {
- int slotIdx = -1;
-
- @Override
- public boolean hasNext() {
- synchronized (ShortCircuitShm.this) {
- return allocatedSlots.nextSetBit(slotIdx + 1) != -1;
- }
- }
-
- @Override
- public Slot next() {
- synchronized (ShortCircuitShm.this) {
- int nextSlotIdx = allocatedSlots.nextSetBit(slotIdx + 1);
- if (nextSlotIdx == -1) {
- throw new NoSuchElementException();
- }
- slotIdx = nextSlotIdx;
- return slots[nextSlotIdx];
- }
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException("SlotIterator " +
- "doesn't support removal");
- }
- }
-
- /**
- * A slot containing information about a replica.
- *
- * The format is:
- * word 0
- * bit 0:32 Slot flags (see below).
- * bit 33:63 Anchor count.
- * word 1:7
- * Reserved for future use, such as statistics.
- * Padding is also useful for avoiding false sharing.
- *
- * Little-endian versus big-endian is not relevant here since both the client
- * and the server reside on the same computer and use the same orientation.
- */
- public class Slot {
- /**
- * Flag indicating that the slot is valid.
- *
- * The DFSClient sets this flag when it allocates a new slot within one of
- * its shared memory regions.
- *
- * The DataNode clears this flag when the replica associated with this slot
- * is no longer valid. The client itself also clears this flag when it
- * believes that the DataNode is no longer using this slot to communicate.
- */
- private static final long VALID_FLAG = 1L<<63;
-
- /**
- * Flag indicating that the slot can be anchored.
- */
- private static final long ANCHORABLE_FLAG = 1L<<62;
-
- /**
- * The slot address in memory.
- */
- private final long slotAddress;
-
- /**
- * BlockId of the block this slot is used for.
- */
- private final ExtendedBlockId blockId;
-
- Slot(long slotAddress, ExtendedBlockId blockId) {
- this.slotAddress = slotAddress;
- this.blockId = blockId;
- }
-
- /**
- * Get the short-circuit memory segment associated with this Slot.
- *
- * @return The enclosing short-circuit memory segment.
- */
- public ShortCircuitShm getShm() {
- return ShortCircuitShm.this;
- }
-
- /**
- * Get the ExtendedBlockId associated with this slot.
- *
- * @return The ExtendedBlockId of this slot.
- */
- public ExtendedBlockId getBlockId() {
- return blockId;
- }
-
- /**
- * Get the SlotId of this slot, containing both shmId and slotIdx.
- *
- * @return The SlotId of this slot.
- */
- public SlotId getSlotId() {
- return new SlotId(getShmId(), getSlotIdx());
- }
-
- /**
- * Get the Slot index.
- *
- * @return The index of this slot.
- */
- public int getSlotIdx() {
- return Ints.checkedCast(
- (slotAddress - baseAddress) / BYTES_PER_SLOT);
- }
-
- /**
- * Clear the slot.
- */
- void clear() {
- unsafe.putLongVolatile(null, this.slotAddress, 0);
- }
-
- private boolean isSet(long flag) {
- long prev = unsafe.getLongVolatile(null, this.slotAddress);
- return (prev & flag) != 0;
- }
-
- private void setFlag(long flag) {
- long prev;
- do {
- prev = unsafe.getLongVolatile(null, this.slotAddress);
- if ((prev & flag) != 0) {
- return;
- }
- } while (!unsafe.compareAndSwapLong(null, this.slotAddress,
- prev, prev | flag));
- }
-
- private void clearFlag(long flag) {
- long prev;
- do {
- prev = unsafe.getLongVolatile(null, this.slotAddress);
- if ((prev & flag) == 0) {
- return;
- }
- } while (!unsafe.compareAndSwapLong(null, this.slotAddress,
- prev, prev & (~flag)));
- }
-
- public boolean isValid() {
- return isSet(VALID_FLAG);
- }
-
- public void makeValid() {
- setFlag(VALID_FLAG);
- }
-
- public void makeInvalid() {
- clearFlag(VALID_FLAG);
- }
-
- public boolean isAnchorable() {
- return isSet(ANCHORABLE_FLAG);
- }
-
- public void makeAnchorable() {
- setFlag(ANCHORABLE_FLAG);
- }
-
- public void makeUnanchorable() {
- clearFlag(ANCHORABLE_FLAG);
- }
-
- public boolean isAnchored() {
- long prev = unsafe.getLongVolatile(null, this.slotAddress);
- if ((prev & VALID_FLAG) == 0) {
- // Slot is no longer valid.
- return false;
- }
- return ((prev & 0x7fffffff) != 0);
- }
-
- /**
- * Try to add an anchor for a given slot.
- *
- * When a slot is anchored, we know that the block it refers to is resident
- * in memory.
- *
- * @return True if the slot is anchored.
- */
- public boolean addAnchor() {
- long prev;
- do {
- prev = unsafe.getLongVolatile(null, this.slotAddress);
- if ((prev & VALID_FLAG) == 0) {
- // Slot is no longer valid.
- return false;
- }
- if ((prev & ANCHORABLE_FLAG) == 0) {
- // Slot can't be anchored right now.
- return false;
- }
- if ((prev & 0x7fffffff) == 0x7fffffff) {
- // Too many other threads have anchored the slot (2 billion?)
- return false;
- }
- } while (!unsafe.compareAndSwapLong(null, this.slotAddress,
- prev, prev + 1));
- return true;
- }
-
- /**
- * Remove an anchor for a given slot.
- */
- public void removeAnchor() {
- long prev;
- do {
- prev = unsafe.getLongVolatile(null, this.slotAddress);
- Preconditions.checkState((prev & 0x7fffffff) != 0,
- "Tried to remove anchor for slot " + slotAddress +", which was " +
- "not anchored.");
- } while (!unsafe.compareAndSwapLong(null, this.slotAddress,
- prev, prev - 1));
- }
-
- @Override
- public String toString() {
- return "Slot(slotIdx=" + getSlotIdx() + ", shm=" + getShm() + ")";
- }
- }
-
- /**
- * ID for this SharedMemorySegment.
- */
- private final ShmId shmId;
-
- /**
- * The base address of the memory-mapped file.
- */
- private final long baseAddress;
-
- /**
- * The mmapped length of the shared memory segment
- */
- private final int mmappedLength;
-
- /**
- * The slots associated with this shared memory segment.
- * slot[i] contains the slot at offset i * BYTES_PER_SLOT,
- * or null if that slot is not allocated.
- */
- private final Slot slots[];
-
- /**
- * A bitset where each bit represents a slot which is in use.
- */
- private final BitSet allocatedSlots;
-
- /**
- * Create the ShortCircuitShm.
- *
- * @param shmId The ID to use.
- * @param stream The stream that we're going to use to create this
- * shared memory segment.
- *
- * Although this is a FileInputStream, we are going to
- * assume that the underlying file descriptor is writable
- * as well as readable. It would be more appropriate to use
- * a RandomAccessFile here, but that class does not have
- * any public accessor which returns a FileDescriptor,
- * unlike FileInputStream.
- */
- public ShortCircuitShm(ShmId shmId, FileInputStream stream)
- throws IOException {
- if (!NativeIO.isAvailable()) {
- throw new UnsupportedOperationException("NativeIO is not available.");
- }
- if (Shell.WINDOWS) {
- throw new UnsupportedOperationException(
- "DfsClientShm is not yet implemented for Windows.");
- }
- if (unsafe == null) {
- throw new UnsupportedOperationException(
- "can't use DfsClientShm because we failed to " +
- "load misc.Unsafe.");
- }
- this.shmId = shmId;
- this.mmappedLength = getUsableLength(stream);
- this.baseAddress = POSIX.mmap(stream.getFD(),
- POSIX.MMAP_PROT_READ | POSIX.MMAP_PROT_WRITE, true, mmappedLength);
- this.slots = new Slot[mmappedLength / BYTES_PER_SLOT];
- this.allocatedSlots = new BitSet(slots.length);
- if (LOG.isTraceEnabled()) {
- LOG.trace("creating " + this.getClass().getSimpleName() +
- "(shmId=" + shmId +
- ", mmappedLength=" + mmappedLength +
- ", baseAddress=" + String.format("%x", baseAddress) +
- ", slots.length=" + slots.length + ")");
- }
- }
-
- public final ShmId getShmId() {
- return shmId;
- }
-
- /**
- * Determine if this shared memory object is empty.
- *
- * @return True if the shared memory object is empty.
- */
- synchronized final public boolean isEmpty() {
- return allocatedSlots.nextSetBit(0) == -1;
- }
-
- /**
- * Determine if this shared memory object is full.
- *
- * @return True if the shared memory object is full.
- */
- synchronized final public boolean isFull() {
- return allocatedSlots.nextClearBit(0) >= slots.length;
- }
-
- /**
- * Calculate the base address of a slot.
- *
- * @param slotIdx Index of the slot.
- * @return The base address of the slot.
- */
- private final long calculateSlotAddress(int slotIdx) {
- long offset = slotIdx;
- offset *= BYTES_PER_SLOT;
- return this.baseAddress + offset;
- }
-
- /**
- * Allocate a new slot and register it.
- *
- * This function chooses an empty slot, initializes it, and then returns
- * the relevant Slot object.
- *
- * @return The new slot.
- */
- synchronized public final Slot allocAndRegisterSlot(
- ExtendedBlockId blockId) {
- int idx = allocatedSlots.nextClearBit(0);
- if (idx >= slots.length) {
- throw new RuntimeException(this + ": no more slots are available.");
- }
- allocatedSlots.set(idx, true);
- Slot slot = new Slot(calculateSlotAddress(idx), blockId);
- slot.clear();
- slot.makeValid();
- slots[idx] = slot;
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": allocAndRegisterSlot " + idx + ": allocatedSlots=" + allocatedSlots +
- StringUtils.getStackTrace(Thread.currentThread()));
- }
- return slot;
- }
-
- synchronized public final Slot getSlot(int slotIdx)
- throws InvalidRequestException {
- if (!allocatedSlots.get(slotIdx)) {
- throw new InvalidRequestException(this + ": slot " + slotIdx +
- " does not exist.");
- }
- return slots[slotIdx];
- }
-
- /**
- * Register a slot.
- *
- * This function looks at a slot which has already been initialized (by
- * another process), and registers it with us. Then, it returns the
- * relevant Slot object.
- *
- * @return The slot.
- *
- * @throws InvalidRequestException
- * If the slot index we're trying to allocate has not been
- * initialized, or is already in use.
- */
- synchronized public final Slot registerSlot(int slotIdx,
- ExtendedBlockId blockId) throws InvalidRequestException {
- if (slotIdx < 0) {
- throw new InvalidRequestException(this + ": invalid negative slot " +
- "index " + slotIdx);
- }
- if (slotIdx >= slots.length) {
- throw new InvalidRequestException(this + ": invalid slot " +
- "index " + slotIdx);
- }
- if (allocatedSlots.get(slotIdx)) {
- throw new InvalidRequestException(this + ": slot " + slotIdx +
- " is already in use.");
- }
- Slot slot = new Slot(calculateSlotAddress(slotIdx), blockId);
- if (!slot.isValid()) {
- throw new InvalidRequestException(this + ": slot " + slotIdx +
- " is not marked as valid.");
- }
- slots[slotIdx] = slot;
- allocatedSlots.set(slotIdx, true);
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": registerSlot " + slotIdx + ": allocatedSlots=" + allocatedSlots +
- StringUtils.getStackTrace(Thread.currentThread()));
- }
- return slot;
- }
-
- /**
- * Unregisters a slot.
- *
- * This doesn't alter the contents of the slot. It just means
- *
- * @param slotIdx Index of the slot to unregister.
- */
- synchronized public final void unregisterSlot(int slotIdx) {
- Preconditions.checkState(allocatedSlots.get(slotIdx),
- "tried to unregister slot " + slotIdx + ", which was not registered.");
- allocatedSlots.set(slotIdx, false);
- slots[slotIdx] = null;
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": unregisterSlot " + slotIdx);
- }
- }
-
- /**
- * Iterate over all allocated slots.
- *
- * Note that this method isn't safe if
- *
- * @return The slot iterator.
- */
- public SlotIterator slotIterator() {
- return new SlotIterator();
- }
-
- public void free() {
- try {
- POSIX.munmap(baseAddress, mmappedLength);
- } catch (IOException e) {
- LOG.warn(this + ": failed to munmap", e);
- }
- LOG.trace(this + ": freed");
- }
-
- @Override
- public String toString() {
- return this.getClass().getSimpleName() + "(" + shmId + ")";
- }
-}