You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by jl...@apache.org on 2016/02/29 17:12:21 UTC
[3/3] tez git commit: TEZ-3147. Intermediate mem-to-mem: Fix early
exit when only one segment can fit into memory (rbalamohan) (cherry picked
from commit 3e409ae0ee7233b4cf631cac1bc366679a08b7d1)
TEZ-3147. Intermediate mem-to-mem: Fix early exit when only one segment can fit into memory (rbalamohan)
(cherry picked from commit 3e409ae0ee7233b4cf631cac1bc366679a08b7d1)
Conflicts:
CHANGES.txt
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/0c4d0cbf
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/0c4d0cbf
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/0c4d0cbf
Branch: refs/heads/branch-0.7
Commit: 0c4d0cbfc13c5a57e0e4c974480ca848ed46c7ec
Parents: ef61eb2
Author: Jason Lowe <jl...@apache.org>
Authored: Mon Feb 29 16:11:48 2016 +0000
Committer: Jason Lowe <jl...@apache.org>
Committed: Mon Feb 29 16:11:48 2016 +0000
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../shuffle/orderedgrouped/MergeManager.java | 8 ++-
.../orderedgrouped/TestMergeManager.java | 59 ++++++++++++++++++++
3 files changed, 67 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/0c4d0cbf/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 92d9de8..df8cf59 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -9,6 +9,7 @@ INCOMPATIBLE CHANGES
TEZ-2972. Avoid task rescheduling when a node turns unhealthy
ALL CHANGES:
+ TEZ-3147. Intermediate mem-to-mem: Fix early exit when only one segment can fit into memory
TEZ-1911. MergeManager's unconditionalReserve() should check for memory limits before allocating.
TEZ-3141. mapreduce.task.timeout is not translated to container heartbeat timeout
TEZ-3128. Avoid stopping containers on the AM shutdown thread.
http://git-wip-us.apache.org/repos/asf/tez/blob/0c4d0cbf/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java
index b716c01..2d519d2 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java
@@ -596,6 +596,7 @@ public class MergeManager {
synchronized (manager) {
Iterator<MapOutput> it = inputs.iterator();
+ MapOutput lastAddedMapOutput = null;
while(it.hasNext() && !Thread.currentThread().isInterrupted()) {
MapOutput mo = it.next();
if ((mergeOutputSize + mo.getSize() + manager.getUsedMemory()) > memoryLimit) {
@@ -614,6 +615,7 @@ public class MergeManager {
mo.getAttemptIdentifier(), mo.getMemory(), 0, mo.getMemory().length);
inMemorySegments.add(new Segment(reader, true,
(mo.isPrimaryMapOutput() ? mergedMapOutputsCounter : null)));
+ lastAddedMapOutput = mo;
it.remove();
LOG.debug("Added segment for merging. mergeOutputSize=" + mergeOutputSize);
}
@@ -622,8 +624,12 @@ public class MergeManager {
//Add any unused MapOutput back
inMemoryMapOutputs.addAll(inputs);
+ //Exit early, if 0 or 1 segment is available
if (inMemorySegments.size() <= 1) {
- return; //no need to proceed further.
+ if (lastAddedMapOutput != null) {
+ inMemoryMapOutputs.add(lastAddedMapOutput);
+ }
+ return;
}
mergedMapOutputs = unconditionalReserve(dummyMapId, mergeOutputSize, false);
http://git-wip-us.apache.org/repos/asf/tez/blob/0c4d0cbf/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestMergeManager.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestMergeManager.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestMergeManager.java
index 1a8cb7a..398e94e 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestMergeManager.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestMergeManager.java
@@ -420,6 +420,65 @@ public class TestMergeManager {
mergeManager.close();
+ /**
+ * Test #4
+ * - Set number of segments for merging to 4.
+ * - Have 4 in-memory segments of size {490000,490000,490000,230000}
+ * - Committing 4 segments would trigger mem-to-mem
+ * - But only 300000 can fit into memory. This should not be
+ * merged as there is no point in merging single segment. It should be
+ * added back to the inMemorySegments
+ */
+ conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMTOMEM_SEGMENTS, 4);
+ mergeManager =
+ new MergeManager(conf, localFs, localDirAllocator, inputContext, null, null, null, null,
+ exceptionReporter, 2000000, null, false, -1);
+ mergeManager.configureAndStart();
+
+ //Single shuffle limit is 25% of 2000000
+ data1 = generateDataBySize(conf, 490000);
+ data2 = generateDataBySize(conf, 490000);
+ data3 = generateDataBySize(conf, 490000);
+ data4 = generateDataBySize(conf, 230000);
+
+ mo1 = mergeManager.reserve(new InputAttemptIdentifier(0,0), data1.length, data1.length, 0);
+ mo2 = mergeManager.reserve(new InputAttemptIdentifier(1,0), data2.length, data2.length, 0);
+ mo3 = mergeManager.reserve(new InputAttemptIdentifier(2,0), data3.length, data3.length, 0);
+ mo4 = mergeManager.reserve(new InputAttemptIdentifier(3,0), data4.length, data4.length, 0);
+
+ assertTrue(mergeManager.getUsedMemory() >= (490000 + 490000 + 490000 + 23000));
+
+ assertEquals(MapOutput.Type.MEMORY, mo1.getType());
+ assertEquals(MapOutput.Type.MEMORY, mo2.getType());
+ assertEquals(MapOutput.Type.MEMORY, mo3.getType());
+ assertEquals(MapOutput.Type.MEMORY, mo4.getType());
+ assertEquals(0, mergeManager.getCommitMemory());
+
+ assertEquals(data1.length + data2.length + data3.length + data4.length,
+ mergeManager.getUsedMemory());
+
+ System.arraycopy(data1, 0, mo1.getMemory(), 0, data1.length);
+ System.arraycopy(data2, 0, mo2.getMemory(), 0, data2.length);
+ System.arraycopy(data3, 0, mo3.getMemory(), 0, data3.length);
+ System.arraycopy(data4, 0, mo4.getMemory(), 0, data4.length);
+
+ //Committing 4 segments should trigger mem-to-mem merge
+ mo1.commit();
+ mo2.commit();
+ mo3.commit();
+ mo4.commit();
+
+ //4 segments were there originally in inMemoryMapOutput.
+ int numberOfMapOutputs = 4;
+
+ //Wait for mem-to-mem to complete. Since only 1 segment (230000) can fit
+ //into memory, it should return early
+ mergeManager.waitForMemToMemMerge();
+
+ //Check if inMemorySegment has got the MapOutput back for merging later
+ assertEquals(numberOfMapOutputs, mergeManager.inMemoryMapOutputs.size());
+
+ mergeManager.close();
}
private byte[] generateDataBySize(Configuration conf, int rawLen) throws IOException {