tez-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From hit...@apache.org
Subject git commit: TEZ-874. Fetcher inputstream is not buffered. (Rajesh Balamohan via hitesh)
Date Sun, 23 Feb 2014 18:01:07 GMT
Repository: incubator-tez
Updated Branches:
  refs/heads/master 1c112c103 -> 52cfa784d


TEZ-874. Fetcher inputstream is not buffered. (Rajesh Balamohan via hitesh)


Project: http://git-wip-us.apache.org/repos/asf/incubator-tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tez/commit/52cfa784
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tez/tree/52cfa784
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tez/diff/52cfa784

Branch: refs/heads/master
Commit: 52cfa784ddf5db0e696f4f145b3537eddfaca079
Parents: 1c112c1
Author: Hitesh Shah <hitesh@apache.org>
Authored: Sun Feb 23 09:59:28 2014 -0800
Committer: Hitesh Shah <hitesh@apache.org>
Committed: Sun Feb 23 09:59:28 2014 -0800

----------------------------------------------------------------------
 tez-api/src/main/java/org/apache/tez/common/TezJobConfig.java | 7 +++++++
 .../tez/runtime/library/common/shuffle/impl/Fetcher.java      | 7 ++++++-
 2 files changed, 13 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/52cfa784/tez-api/src/main/java/org/apache/tez/common/TezJobConfig.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/TezJobConfig.java b/tez-api/src/main/java/org/apache/tez/common/TezJobConfig.java
index 444a830..a3d6561 100644
--- a/tez-api/src/main/java/org/apache/tez/common/TezJobConfig.java
+++ b/tez-api/src/main/java/org/apache/tez/common/TezJobConfig.java
@@ -217,6 +217,13 @@ public class TezJobConfig {
   public static final String TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT = "tez.runtime.shuffle.read.timeout";
   public final static int DEFAULT_TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT = 
       3 * 60 * 1000;
+  
+  /**
+   * 
+   */
+  public static final String TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE = "tez.runtime.shuffle.buffersize";
+  public final static int DEFAULT_TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE = 
+      8 * 1024;
 
   /**
    * 

http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/52cfa784/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/Fetcher.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/Fetcher.java
b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/Fetcher.java
index e8038a4..042e8f5 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/Fetcher.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/Fetcher.java
@@ -17,6 +17,7 @@
  */
 package org.apache.tez.runtime.library.common.shuffle.impl;
 
+import java.io.BufferedInputStream;
 import java.io.DataInputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -78,6 +79,7 @@ class Fetcher extends Thread {
   
   private final int connectionTimeout;
   private final int readTimeout;
+  private final int bufferSize;
   
   // Decompression of map-outputs
   private final CompressionCodec codec;
@@ -137,6 +139,9 @@ class Fetcher extends Thread {
     this.readTimeout = 
         job.getInt(TezJobConfig.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT, 
             TezJobConfig.DEFAULT_TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT);
+    
+    this.bufferSize = job.getInt(TezJobConfig.TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE, 
+            TezJobConfig.DEFAULT_TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE);
 
     setName("fetcher#" + id);
     setDaemon(true);
@@ -262,7 +267,7 @@ class Fetcher extends Thread {
           ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION);
       connect(connection, connectionTimeout);
       connectSucceeded = true;
-      input = new DataInputStream(connection.getInputStream());
+      input = new DataInputStream(new BufferedInputStream(connection.getInputStream(), bufferSize));
 
       // Validate response code
       int rc = connection.getResponseCode();


Mime
View raw message