HADOOP-9319. Update bundled LZ4 source to r99. (Binglin Chang via llu)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1510734 13f79535-47bb-0310-9956-ffa450edef68

Author: vicaya

hadoop-common-project/hadoop-common/CHANGES.txt

@@ -283,6 +283,8 @@ Release 2.3.0 - UNRELEASED
 
   IMPROVEMENTS
 
+    HADOOP-9319. Update bundled LZ4 source to r99. (Binglin Chang via llu)
+
     HADOOP-9241. DU refresh interval is not configurable (harsh)
 
     HADOOP-9417.  Support for symlink resolution in LocalFileSystem /

hadoop-common-project/hadoop-common/LICENSE.txt

@@ -252,24 +252,26 @@ in src/main/native/src/org/apache/hadoop/util:
  *   BSD-style license that can be found in the LICENSE file.
  */
 
- For src/main/native/src/org/apache/hadoop/io/compress/lz4/lz4.c:
+For src/main/native/src/org/apache/hadoop/io/compress/lz4/{lz4.h,lz4.c,
+lz4_encoder.h,lz4hc.h,lz4hc.c,lz4hc_encoder.h},
 
 /*
    LZ4 - Fast LZ compression algorithm
-   Copyright (C) 2011, Yann Collet.
-   BSD License
+   Header File
+   Copyright (C) 2011-2013, Yann Collet.
+   BSD 2-Clause License (http://www.opensource.org/licenses/bsd-license.php)
 
    Redistribution and use in source and binary forms, with or without
    modification, are permitted provided that the following conditions are
    met:
-  
+
        * Redistributions of source code must retain the above copyright
    notice, this list of conditions and the following disclaimer.
        * Redistributions in binary form must reproduce the above
    copyright notice, this list of conditions and the following disclaimer
    in the documentation and/or other materials provided with the
    distribution.
-  
+
    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
    "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
    LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
@@ -281,4 +283,8 @@ in src/main/native/src/org/apache/hadoop/util:
    THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
    (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
    OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+   You can contact the author at :
+   - LZ4 homepage : http://fastcompression.blogspot.com/p/lz4.html
+   - LZ4 source repository : http://code.google.com/p/lz4/
 */

hadoop-common-project/hadoop-common/pom.xml

@@ -456,7 +456,12 @@
             <exclude>src/test/empty-file</exclude>
             <exclude>src/test/all-tests</exclude>
             <exclude>src/test/resources/kdc/ldif/users.ldif</exclude>
+            <exclude>src/main/native/src/org/apache/hadoop/io/compress/lz4/lz4.h</exclude>
             <exclude>src/main/native/src/org/apache/hadoop/io/compress/lz4/lz4.c</exclude>
+            <exclude>src/main/native/src/org/apache/hadoop/io/compress/lz4/lz4_encoder.h</exclude>
+            <exclude>src/main/native/src/org/apache/hadoop/io/compress/lz4/lz4hc.h</exclude>
+            <exclude>src/main/native/src/org/apache/hadoop/io/compress/lz4/lz4hc.c</exclude>
+            <exclude>src/main/native/src/org/apache/hadoop/io/compress/lz4/lz4hc_encoder.h</exclude>
             <exclude>src/test/java/org/apache/hadoop/fs/test-untar.tgz</exclude>
           </excludes>
         </configuration>

hadoop-common-project/hadoop-common/src/CMakeLists.txt

@@ -170,6 +170,7 @@ add_dual_library(hadoop
     ${D}/io/compress/lz4/Lz4Compressor.c
     ${D}/io/compress/lz4/Lz4Decompressor.c
     ${D}/io/compress/lz4/lz4.c
+    ${D}/io/compress/lz4/lz4hc.c
     ${SNAPPY_SOURCE_FILES}
     ${D}/io/compress/zlib/ZlibCompressor.c
     ${D}/io/compress/zlib/ZlibDecompressor.c

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java

@@ -96,14 +96,22 @@ public class CommonConfigurationKeys extends CommonConfigurationKeysPublic {
   public static final int IO_COMPRESSION_CODEC_SNAPPY_BUFFERSIZE_DEFAULT =
       256 * 1024;
 
-  /** Internal buffer size for Snappy compressor/decompressors */
+  /** Internal buffer size for Lz4 compressor/decompressors */
   public static final String IO_COMPRESSION_CODEC_LZ4_BUFFERSIZE_KEY =
       "io.compression.codec.lz4.buffersize";
 
   /** Default value for IO_COMPRESSION_CODEC_SNAPPY_BUFFERSIZE_KEY */
   public static final int IO_COMPRESSION_CODEC_LZ4_BUFFERSIZE_DEFAULT =
       256 * 1024;
 
+  /** Use lz4hc(slow but with high compression ratio) for lz4 compression */
+  public static final String IO_COMPRESSION_CODEC_LZ4_USELZ4HC_KEY =
+      "io.compression.codec.lz4.use.lz4hc";
+
+  /** Default value for IO_COMPRESSION_CODEC_USELZ4HC_KEY */
+  public static final boolean IO_COMPRESSION_CODEC_LZ4_USELZ4HC_DEFAULT =
+      false;
+
   /**
    * Service Authorization
    */

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/Lz4Codec.java

@@ -107,7 +107,7 @@ public CompressionOutputStream createOutputStream(OutputStream out,
         CommonConfigurationKeys.IO_COMPRESSION_CODEC_LZ4_BUFFERSIZE_KEY,
         CommonConfigurationKeys.IO_COMPRESSION_CODEC_LZ4_BUFFERSIZE_DEFAULT);
 
-    int compressionOverhead = Math.max((int)(bufferSize * 0.01), 10);
+    int compressionOverhead = bufferSize/255 + 16;
 
     return new BlockCompressorStream(out, compressor, bufferSize,
         compressionOverhead);
@@ -140,7 +140,10 @@ public Compressor createCompressor() {
     int bufferSize = conf.getInt(
         CommonConfigurationKeys.IO_COMPRESSION_CODEC_LZ4_BUFFERSIZE_KEY,
         CommonConfigurationKeys.IO_COMPRESSION_CODEC_LZ4_BUFFERSIZE_DEFAULT);
-    return new Lz4Compressor(bufferSize);
+    boolean useLz4HC = conf.getBoolean(
+        CommonConfigurationKeys.IO_COMPRESSION_CODEC_LZ4_USELZ4HC_KEY,
+        CommonConfigurationKeys.IO_COMPRESSION_CODEC_LZ4_USELZ4HC_DEFAULT);
+    return new Lz4Compressor(bufferSize, useLz4HC);
   }
 
   /**

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/lz4/Lz4Compressor.java

@@ -52,6 +52,7 @@ public class Lz4Compressor implements Compressor {
   private long bytesRead = 0L;
   private long bytesWritten = 0L;
 
+  private final boolean useLz4HC;
 
   static {
     if (NativeCodeLoader.isNativeCodeLoaded()) {
@@ -72,15 +73,27 @@ public class Lz4Compressor implements Compressor {
    * Creates a new compressor.
    *
    * @param directBufferSize size of the direct buffer to be used.
+   * @param useLz4HC use high compression ratio version of lz4, 
+   *                 which trades CPU for compression ratio.
    */
-  public Lz4Compressor(int directBufferSize) {
+  public Lz4Compressor(int directBufferSize, boolean useLz4HC) {
+    this.useLz4HC = useLz4HC;
     this.directBufferSize = directBufferSize;
 
     uncompressedDirectBuf = ByteBuffer.allocateDirect(directBufferSize);
     compressedDirectBuf = ByteBuffer.allocateDirect(directBufferSize);
     compressedDirectBuf.position(directBufferSize);
   }
 
+  /**
+   * Creates a new compressor.
+   *
+   * @param directBufferSize size of the direct buffer to be used.
+   */
+  public Lz4Compressor(int directBufferSize) {
+    this(directBufferSize, false);
+  }
+
   /**
    * Creates a new compressor with the default buffer size.
    */
@@ -227,7 +240,7 @@ public synchronized int compress(byte[] b, int off, int len)
     }
 
     // Compress data
-    n = compressBytesDirect();
+    n = useLz4HC ? compressBytesDirectHC() : compressBytesDirect();
     compressedDirectBuf.limit(n);
     uncompressedDirectBuf.clear(); // lz4 consumes all buffer input
 
@@ -297,5 +310,7 @@ public synchronized void end() {
 
   private native int compressBytesDirect();
 
+  private native int compressBytesDirectHC();
+
   public native static String getLibraryName();
 }

hadoop-common-project/hadoop-common/src/main/native/native.vcxproj

@@ -72,6 +72,7 @@
   </ItemDefinitionGroup>
   <ItemGroup>
     <ClCompile Include="src\org\apache\hadoop\io\compress\lz4\lz4.c" />
+    <ClCompile Include="src\org\apache\hadoop\io\compress\lz4\lz4hc.c" />
     <ClCompile Include="src\org\apache\hadoop\io\compress\lz4\Lz4Compressor.c" />
     <ClCompile Include="src\org\apache\hadoop\io\compress\lz4\Lz4Decompressor.c" />
     <ClCompile Include="src\org\apache\hadoop\io\nativeio\file_descriptor.c" />

hadoop-common-project/hadoop-common/src/main/native/native.vcxproj.filters

@@ -51,6 +51,9 @@
     <ClCompile Include="src\org\apache\hadoop\io\compress\lz4\lz4.c">
       <Filter>Source Files</Filter>
     </ClCompile>
+    <ClCompile Include="src\org\apache\hadoop\io\compress\lz4\lz4hc.c">
+      <Filter>Source Files</Filter>
+    </ClCompile>
     <ClCompile Include="src\org\apache\hadoop\io\compress\lz4\Lz4Compressor.c">
       <Filter>Source Files</Filter>
     </ClCompile>

hadoop-common-project/hadoop-common/src/main/native/src/org/apache/hadoop/io/compress/lz4/Lz4Compressor.c

@@ -23,21 +23,9 @@
 #ifdef UNIX
 #include "config.h"
 #endif // UNIX
+#include "lz4.h"
+#include "lz4hc.h"
 
-//****************************
-// Simple Functions
-//****************************
-
-extern int LZ4_compress   (const char* source, char* dest, int isize);
-
-/*
-LZ4_compress() :
- return : the number of bytes in compressed buffer dest
- note : destination buffer must be already allocated.
-  To avoid any problem, size it to handle worst cases situations (input data not compressible)
-  Worst case size is : "inputsize + 0.4%", with "0.4%" being at least 8 bytes.
-
-*/
 
 static jfieldID Lz4Compressor_clazz;
 static jfieldID Lz4Compressor_uncompressedDirectBuf;
@@ -107,5 +95,45 @@ JNIEXPORT jstring JNICALL
 Java_org_apache_hadoop_io_compress_lz4_Lz4Compressor_getLibraryName(
  JNIEnv *env, jclass class
  ) {
-  return (*env)->NewStringUTF(env, "revision:43");
+  return (*env)->NewStringUTF(env, "revision:99");
+}
+
+JNIEXPORT jint JNICALL Java_org_apache_hadoop_io_compress_lz4_Lz4Compressor_compressBytesDirectHC
+(JNIEnv *env, jobject thisj){
+  const char* uncompressed_bytes = NULL;
+  char* compressed_bytes = NULL;
+
+  // Get members of Lz4Compressor
+  jobject clazz = (*env)->GetStaticObjectField(env, thisj, Lz4Compressor_clazz);
+  jobject uncompressed_direct_buf = (*env)->GetObjectField(env, thisj, Lz4Compressor_uncompressedDirectBuf);
+  jint uncompressed_direct_buf_len = (*env)->GetIntField(env, thisj, Lz4Compressor_uncompressedDirectBufLen);
+  jobject compressed_direct_buf = (*env)->GetObjectField(env, thisj, Lz4Compressor_compressedDirectBuf);
+  jint compressed_direct_buf_len = (*env)->GetIntField(env, thisj, Lz4Compressor_directBufferSize);
+
+  // Get the input direct buffer
+  LOCK_CLASS(env, clazz, "Lz4Compressor");
+  uncompressed_bytes = (const char*)(*env)->GetDirectBufferAddress(env, uncompressed_direct_buf);
+  UNLOCK_CLASS(env, clazz, "Lz4Compressor");
+
+  if (uncompressed_bytes == 0) {
+    return (jint)0;
+  }
+
+  // Get the output direct buffer
+  LOCK_CLASS(env, clazz, "Lz4Compressor");
+  compressed_bytes = (char *)(*env)->GetDirectBufferAddress(env, compressed_direct_buf);
+  UNLOCK_CLASS(env, clazz, "Lz4Compressor");
+
+  if (compressed_bytes == 0) {
+    return (jint)0;
+  }
+
+  compressed_direct_buf_len = LZ4_compressHC(uncompressed_bytes, compressed_bytes, uncompressed_direct_buf_len);
+  if (compressed_direct_buf_len < 0){
+    THROW(env, "java/lang/InternalError", "LZ4_compressHC failed");
+  }
+
+  (*env)->SetIntField(env, thisj, Lz4Compressor_uncompressedDirectBufLen, 0);
+
+  return (jint)compressed_direct_buf_len;
 }