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

HDFS-15709. Socket file descriptor leak in StripedBlockChecksumReconstructor. (#2518) 

(cherry picked from commit 40f7543a6d)
This commit is contained in:
crossfire 2020-12-08 08:49:45 +09:00 committed by Wei-Chiu Chuang
parent 6a5864ee4a
commit edd9b659ca
2 changed files with 37 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BlockChecksumHelper.java
View File

@ -700,24 +700,25 @@ final class BlockChecksumHelper {
blockGroup, ecPolicy, blockIndices, datanodes, errIndices); blockGroup, ecPolicy, blockIndices, datanodes, errIndices);
BlockChecksumType groupChecksumType = BlockChecksumType groupChecksumType =
getBlockChecksumOptions().getBlockChecksumType(); getBlockChecksumOptions().getBlockChecksumType();
final StripedBlockChecksumReconstructor checksumRecon = try (StripedBlockChecksumReconstructor checksumRecon =
groupChecksumType == BlockChecksumType.COMPOSITE_CRC ? groupChecksumType == BlockChecksumType.COMPOSITE_CRC ?
new StripedBlockChecksumCompositeCrcReconstructor( new StripedBlockChecksumCompositeCrcReconstructor(
getDatanode().getErasureCodingWorker(), stripedReconInfo, getDatanode().getErasureCodingWorker(), stripedReconInfo,
blockChecksumBuf, blockLength) : blockChecksumBuf, blockLength) :
new StripedBlockChecksumMd5CrcReconstructor( new StripedBlockChecksumMd5CrcReconstructor(
getDatanode().getErasureCodingWorker(), stripedReconInfo, getDatanode().getErasureCodingWorker(), stripedReconInfo,
blockChecksumBuf, blockLength); blockChecksumBuf, blockLength)) {
checksumRecon.reconstruct(); checksumRecon.reconstruct();
DataChecksum checksum = checksumRecon.getChecksum(); DataChecksum checksum = checksumRecon.getChecksum();
long crcPerBlock = checksum.getChecksumSize() <= 0 ? 0 long crcPerBlock = checksum.getChecksumSize() <= 0 ? 0
: checksumRecon.getChecksumDataLen() / checksum.getChecksumSize(); : checksumRecon.getChecksumDataLen() / checksum.getChecksumSize();
setOrVerifyChecksumProperties(errBlkIndex, setOrVerifyChecksumProperties(errBlkIndex,
checksum.getBytesPerChecksum(), crcPerBlock, checksum.getBytesPerChecksum(), crcPerBlock,
checksum.getChecksumType()); checksum.getChecksumType());
LOG.debug("Recalculated checksum for the block index:{}, checksum={}", LOG.debug("Recalculated checksum for the block index:{}, checksum={}",
errBlkIndex, checksumRecon.getDigestObject()); errBlkIndex, checksumRecon.getDigestObject());
}
} }
private void setOrVerifyChecksumProperties(int blockIdx, int bpc, private void setOrVerifyChecksumProperties(int blockIdx, int bpc,

47
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/erasurecode/StripedBlockChecksumReconstructor.java
View File

@ -17,6 +17,7 @@
*/ */
package org.apache.hadoop.hdfs.server.datanode.erasurecode; package org.apache.hadoop.hdfs.server.datanode.erasurecode;
import java.io.Closeable;
import java.io.IOException; import java.io.IOException;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.util.Arrays; import java.util.Arrays;
@ -32,7 +33,7 @@ import org.apache.hadoop.io.DataOutputBuffer;
*/ */
@InterfaceAudience.Private @InterfaceAudience.Private
public abstract class StripedBlockChecksumReconstructor public abstract class StripedBlockChecksumReconstructor
extends StripedReconstructor { extends StripedReconstructor implements Closeable {
private ByteBuffer targetBuffer; private ByteBuffer targetBuffer;
private final byte[] targetIndices; private final byte[] targetIndices;
@ -73,31 +74,27 @@ public abstract class StripedBlockChecksumReconstructor
public void reconstruct() throws IOException { public void reconstruct() throws IOException {
prepareDigester(); prepareDigester();
long maxTargetLength = getMaxTargetLength(); long maxTargetLength = getMaxTargetLength();
try { while (requestedLen > 0 && getPositionInBlock() < maxTargetLength) {
while (requestedLen > 0 && getPositionInBlock() < maxTargetLength) { long remaining = maxTargetLength - getPositionInBlock();
long remaining = maxTargetLength - getPositionInBlock(); final int toReconstructLen = (int) Math
final int toReconstructLen = (int) Math .min(getStripedReader().getBufferSize(), remaining);
.min(getStripedReader().getBufferSize(), remaining); // step1: read from minimum source DNs required for reconstruction.
// step1: read from minimum source DNs required for reconstruction. // The returned success list is the source DNs we do real read from
// The returned success list is the source DNs we do real read from getStripedReader().readMinimumSources(toReconstructLen);
getStripedReader().readMinimumSources(toReconstructLen);
// step2: decode to reconstruct targets // step2: decode to reconstruct targets
reconstructTargets(toReconstructLen); reconstructTargets(toReconstructLen);
// step3: calculate checksum // step3: calculate checksum
checksumDataLen += checksumWithTargetOutput( checksumDataLen += checksumWithTargetOutput(
getBufferArray(targetBuffer), toReconstructLen); getBufferArray(targetBuffer), toReconstructLen);
updatePositionInBlock(toReconstructLen); updatePositionInBlock(toReconstructLen);
requestedLen -= toReconstructLen; requestedLen -= toReconstructLen;
clearBuffers(); clearBuffers();
}
commitDigest();
} finally {
cleanup();
} }
commitDigest();
} }
/** /**
@ -222,4 +219,10 @@ public abstract class StripedBlockChecksumReconstructor
} }
return buff; return buff;
} }
@Override
public void close() throws IOException {
getStripedReader().close();
cleanup();
}
} }