TEZ-3634. reduce the default buffer sizes in PipelinedSorter by a small

amount. (sseth)

CHANGES.txt

@@ -7,6 +7,7 @@ INCOMPATIBLE CHANGES
 
 ALL CHANGES:
 
+  TEZ-3634. reduce the default buffer sizes in PipelinedSorter by a small amount.
   TEZ-3627. Use queue name available in RegisterApplicationMasterResponse for publishing to ATS.
   TEZ-3610. TEZ UI 0.7 0.9 compatibility for url query params and tez-app sub-routes
   TEZ-3625. Dag.getVertex should obtain a readlock.

tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/PipelinedSorter.java

@@ -132,7 +132,8 @@ public PipelinedSorter(OutputContext outputContext, Configuration conf, int numO
        * When lazy-allocation is enabled, framework takes care of auto
        * allocating memory on need basis. Desirable block size is set to 256MB
        */
-      MIN_BLOCK_SIZE = 256 << 20; //256 MB
+      //256 MB - 64 bytes. See comment for the 32MB allocation.
+      MIN_BLOCK_SIZE = ((256 << 20) - 64);
     } else {
       int minBlockSize = conf.getInt(TezRuntimeConfiguration
               .TEZ_RUNTIME_PIPELINED_SORTER_MIN_BLOCK_SIZE_IN_MB,
@@ -267,7 +268,11 @@ int computeBlockSize(long availableMem, long maxAllocatedMemory) {
      */
     if (lazyAllocateMem) {
       if (buffers == null || buffers.isEmpty()) {
-        return 32 << 20; //32 MB
+        //32 MB - 64 bytes
+        // These buffers end up occupying 33554456 (32M + 24) bytes.
+        // On large JVMs (64G+), with G1Gc - the region size maxes out at
+        // 32M. Without the -64, this structure would end up using 2 regions.
+        return ((32 << 20) - 64);
       }
     }
 

tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/sort/impl/TestPipelinedSorter.java

@@ -601,13 +601,13 @@ public void test_with_lazyMemAllocation() throws IOException {
         numOutputs, (128l << 20));
     assertTrue("Expected 1 sort buffers. current len=" + sorter.buffers.size(),
         sorter.buffers.size() == 1);
-    assertTrue(sorter.buffers.get(0).capacity() == 32 * 1024 * 1024);
+    assertTrue(sorter.buffers.get(0).capacity() == 32 * 1024 * 1024 - 64);
     writeData(sorter, 100, 1024*1024, false); //100 1 MB KV. Will spill
 
     //Now it should have created 2 buffers, 32 & 96 MB buffers.
     assertTrue(sorter.buffers.size() == 2);
-    assertTrue(sorter.buffers.get(0).capacity() == 32 * 1024 * 1024);
-    assertTrue(sorter.buffers.get(1).capacity() == 96 * 1024 * 1024);
+    assertTrue(sorter.buffers.get(0).capacity() == 32 * 1024 * 1024 - 64);
+    assertTrue(sorter.buffers.get(1).capacity() == 96 * 1024 * 1024 + 64);
     closeSorter(sorter);
     verifyCounters(sorter, outputContext);
 
@@ -619,12 +619,12 @@ public void test_with_lazyMemAllocation() throws IOException {
         .TEZ_RUNTIME_PIPELINED_SORTER_LAZY_ALLOCATE_MEMORY, true);
     sorter = new PipelinedSorter(this.outputContext, conf, numOutputs, (300l << 20));
     assertTrue(sorter.buffers.size() == 1);
-    assertTrue(sorter.buffers.get(0).capacity() == 32 * 1024 * 1024);
+    assertTrue(sorter.buffers.get(0).capacity() == 32 * 1024 * 1024 - 64);
 
     writeData(sorter, 50, 1024*1024, false); //50 1 MB KV to allocate 2nd buf
     assertTrue(sorter.buffers.size() == 2);
-    assertTrue(sorter.buffers.get(0).capacity() == 32 * 1024 * 1024);
-    assertTrue(sorter.buffers.get(1).capacity() == 268 * 1024 * 1024);
+    assertTrue(sorter.buffers.get(0).capacity() == 32 * 1024 * 1024 - 64);
+    assertTrue(sorter.buffers.get(1).capacity() == 268 * 1024 * 1024 + 64);
 
     //48 MB. Do not pre-allocate.
     // Get 32 MB buffer first invariably and proceed with the rest.
@@ -635,13 +635,13 @@ public void test_with_lazyMemAllocation() throws IOException {
         numOutputs, (48l << 20));
     assertTrue("Expected 1 sort buffers. current len=" + sorter.buffers.size(),
         sorter.buffers.size() == 1);
-    assertTrue(sorter.buffers.get(0).capacity() == 32 * 1024 * 1024);
+    assertTrue(sorter.buffers.get(0).capacity() == 32 * 1024 * 1024 - 64);
     writeData(sorter, 20, 1024*1024, false); //100 1 MB KV. Will spill
 
     //Now it should have created 2 buffers, 32 & 96 MB buffers.
     assertTrue(sorter.buffers.size() == 2);
-    assertTrue(sorter.buffers.get(0).capacity() == 32 * 1024 * 1024);
-    assertTrue(sorter.buffers.get(1).capacity() == 16 * 1024 * 1024);
+    assertTrue(sorter.buffers.get(0).capacity() == 32 * 1024 * 1024 - 64);
+    assertTrue(sorter.buffers.get(1).capacity() == 16 * 1024 * 1024 + 64);
     closeSorter(sorter);
   }