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 we...@apache.org on 2019/10/25 20:19:31 UTC
[hadoop] branch branch-3.2 updated: HDFS-14308.
DFSStripedInputStream curStripeBuf is not freed by unbuffer() (#1667)
This is an automated email from the ASF dual-hosted git repository.
weichiu pushed a commit to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/branch-3.2 by this push:
new 9316ca1 HDFS-14308. DFSStripedInputStream curStripeBuf is not freed by unbuffer() (#1667)
9316ca1 is described below
commit 9316ca149f5e29d62af78808a7f897c81264cce8
Author: zhaoyim <zh...@asiainfo.com>
AuthorDate: Sat Oct 26 04:09:14 2019 +0800
HDFS-14308. DFSStripedInputStream curStripeBuf is not freed by unbuffer() (#1667)
Reviewed-by: Aravindan Vijayan <av...@cloudera.com>
Reviewed-by: Wei-Chiu Chuang <we...@apache.org>
(cherry picked from commit 30db895b59d250788d029cb2013bb4712ef9b546)
---
.../apache/hadoop/hdfs/DFSStripedInputStream.java | 18 +++++++++--
.../hadoop/hdfs/TestDFSStripedInputStream.java | 37 ++++++++++++++++++++++
2 files changed, 53 insertions(+), 2 deletions(-)
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSStripedInputStream.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSStripedInputStream.java
index 3f688d4..cf29791 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSStripedInputStream.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSStripedInputStream.java
@@ -70,7 +70,8 @@ public class DFSStripedInputStream extends DFSInputStream {
private final int groupSize;
/** the buffer for a complete stripe. */
private ByteBuffer curStripeBuf;
- private ByteBuffer parityBuf;
+ @VisibleForTesting
+ protected ByteBuffer parityBuf;
private final ErasureCodingPolicy ecPolicy;
private RawErasureDecoder decoder;
@@ -129,7 +130,7 @@ public class DFSStripedInputStream extends DFSInputStream {
curStripeRange = new StripeRange(0, 0);
}
- protected ByteBuffer getParityBuffer() {
+ protected synchronized ByteBuffer getParityBuffer() {
if (parityBuf == null) {
parityBuf = BUFFER_POOL.getBuffer(useDirectBuffer(),
cellSize * parityBlkNum);
@@ -554,4 +555,17 @@ public class DFSStripedInputStream extends DFSInputStream {
throw new UnsupportedOperationException(
"Not support enhanced byte buffer access.");
}
+
+ @Override
+ public synchronized void unbuffer() {
+ super.unbuffer();
+ if (curStripeBuf != null) {
+ BUFFER_POOL.putBuffer(curStripeBuf);
+ curStripeBuf = null;
+ }
+ if (parityBuf != null) {
+ BUFFER_POOL.putBuffer(parityBuf);
+ parityBuf = null;
+ }
+ }
}
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSStripedInputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSStripedInputStream.java
index 3355691..7973cfa 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSStripedInputStream.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSStripedInputStream.java
@@ -627,4 +627,41 @@ public class TestDFSStripedInputStream {
}
}
}
+
+ @Test
+ public void testUnbuffer() throws Exception {
+ final int numBlocks = 2;
+ final int fileSize = numBlocks * blockGroupSize;
+ DFSTestUtil.createStripedFile(cluster, filePath, null, numBlocks,
+ stripesPerBlock, false, ecPolicy);
+ LocatedBlocks lbs = fs.getClient().namenode.
+ getBlockLocations(filePath.toString(), 0, fileSize);
+
+ for (LocatedBlock lb : lbs.getLocatedBlocks()) {
+ assert lb instanceof LocatedStripedBlock;
+ LocatedStripedBlock bg = (LocatedStripedBlock)(lb);
+ for (int i = 0; i < dataBlocks; i++) {
+ Block blk = new Block(bg.getBlock().getBlockId() + i,
+ stripesPerBlock * cellSize,
+ bg.getBlock().getGenerationStamp());
+ blk.setGenerationStamp(bg.getBlock().getGenerationStamp());
+ cluster.injectBlocks(i, Arrays.asList(blk),
+ bg.getBlock().getBlockPoolId());
+ }
+ }
+ DFSStripedInputStream in = new DFSStripedInputStream(fs.getClient(),
+ filePath.toString(), false, ecPolicy, null);
+ ByteBuffer readBuffer = ByteBuffer.allocate(fileSize);
+ int done = 0;
+ while (done < fileSize) {
+ int ret = in.read(readBuffer);
+ assertTrue(ret > 0);
+ done += ret;
+ }
+ in.unbuffer();
+ ByteBuffer curStripeBuf = (in.getCurStripeBuf());
+ assertNull(curStripeBuf);
+ assertNull(in.parityBuf);
+ in.close();
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org