+ lz4, - lzo and zlib, mark 0.1.3

Author: ztellman

README.md

@@ -27,14 +27,16 @@ Available methods can be found via `available-hash-functions`, `available-compre
 
 ```clj
 byte-transforms> (available-hash-functions)
-(:sha384 :md2 :crc32 :sha512 :sha1 :murmur32 :murmur128 :adler32 :sha256 :md5 :murmur64)
+(:sha384 :md2 :crc32 :crc64 :sha512 :sha1 :murmur32 :murmur128 :adler32 :sha256 :md5 :murmur64)
 byte-transforms> (available-compressors)
-(:lzo :bzip2 :snappy :gzip :zlib)
+(:lz4 :bzip2 :snappy :gzip)
 byte-transforms> (available-encoders)
 (:base64)
 ```
 
-When choosing a compression algorithm, `snappy` is typically the fastest, and `bzip2` yields the highest compression.  Full stats on all methods can be found by cloning the project and running `lein test :benchmark`.
+When choosing a compression algorithm, `snappy` is typically the fastest, `bzip2` yields the highest compression, and `lz4` provides a good balance between higher compression rate and fast decompression.  All the compression algorithms except `lz4` are concat-able; multiple compressed segments can be concatenated and decompressed as a single stream.
+
+Full stats on all methods can be found by cloning the project and running `lein test :benchmark`.
 
 ## License
 

project.clj

@@ -1,14 +1,14 @@
-(defproject byte-transforms "0.1.3-SNAPSHOT"
+(defproject byte-transforms "0.1.3"
   :description "Methods for hashing, compressing, and encoding bytes."
   :license {:name "Apache License 2.0"
             :url "http://www.apache.org/licenses/LICENSE-2.0.html"}
-  :dependencies [[byte-streams "0.1.11-SNAPSHOT"]
+  :dependencies [[byte-streams "0.1.11"]
                  [org.xerial.snappy/snappy-java "1.1.0.1"]
                  [commons-codec/commons-codec "1.9"]
-                 [org.anarres.lzo/lzo-core "1.0.0"]
+                 [net.jpountz.lz4/lz4 "1.2.0"]
                  [org.apache.commons/commons-compress "1.8"
                   :exclusions [org.tukaani/xz]]]
-  :profiles {:dev {:dependencies [[org.clojure/clojure "1.5.1"]
+  :profiles {:dev {:dependencies [[org.clojure/clojure "1.6.0"]
                                   [criterium "0.4.3"]
                                   [reiddraper/simple-check "0.5.6"]
                                   [codox-md "0.2.0" :exclusions [org.clojure/clojure]]]}}

src/byte_transforms.clj

@@ -5,22 +5,26 @@
     [primitive-math :as p])
   (:import
     [byte_transforms
-     CassandraMurmurHash]
+     CassandraMurmurHash
+     CRC64]
     [java.util
      UUID]
     [java.lang.reflect
      Array]
     [java.util.zip
      CRC32
-     Adler32
-     DeflaterInputStream
-     InflaterInputStream]
+     Adler32]
     [java.io
      OutputStream
      InputStream
      PipedInputStream
      PipedOutputStream
      ByteArrayOutputStream]
+    [net.jpountz.lz4
+     LZ4BlockOutputStream
+     LZ4BlockInputStream
+     LZ4Factory
+     LZ4Compressor]
     [java.security
      MessageDigest]
     [java.nio
@@ -40,12 +44,7 @@
     [org.xerial.snappy
      Snappy
      SnappyInputStream
-     SnappyOutputStream]
-    [org.anarres.lzo
-     LzoInputStream
-     LzoOutputStream
-     LzoDecompressor1x
-     LzoLibrary]))
+     SnappyOutputStream]))
 
 ;;;
 
@@ -142,7 +141,17 @@
 (def-hash crc32
   [x options]
   (let [crc (CRC32.)]
-    (when-let [seed (:seed options)]
+    (when-let [seed (get options :seed)]
+      (.update crc (byte seed)))
+    (doseq [^bytes ary (bytes/to-byte-arrays x options)]
+      (.update crc ary))
+    (.getValue crc)))
+
+;; CRC64 hash
+(def-hash crc64
+  [x options]
+  (let [crc (CRC64.)]
+    (when-let [seed (get options :seed)]
       (.update crc (byte seed)))
     (doseq [^bytes ary (bytes/to-byte-arrays x options)]
       (.update crc ary))
@@ -152,14 +161,14 @@
 (def-hash adler32
   [x options]
   (let [adler (Adler32.)]
-    (when-let [seed (:seed options)]
+    (when-let [seed (get options :seed)]
       (.update adler (byte seed)))
     (doseq [^bytes ary (bytes/to-byte-arrays x options)]
       (.update adler ary))
     (.getValue adler)))
 
 (defn- hash-digest [^MessageDigest digest bufs options]
-  (when-let [seed (:seed options)]
+  (when-let [seed (get options :seed)]
     (.update digest (byte seed)))
   (doseq [^ByteBuffer buf bufs]
     (.update digest buf))
@@ -222,7 +231,8 @@
 
 (let [murmur32 (get @hash-functions :murmur32)
       murmur64 (get @hash-functions :murmur64)
-      murmur128 (get @hash-functions :murmur128)]
+      murmur128 (get @hash-functions :murmur128)
+      crc64 (get @hash-functions :crc64)]
   (defn hash
     "Takes a byte stream, and returns a value representing its hash, which will be an integer if
    the hash is 32 or 64-bit, or a byte array otherwise.  By default, this will use the murmur64
@@ -236,6 +246,7 @@
          :murmur32 (murmur32 bytes options)
          :murmur64 (murmur64 bytes options)
          :murmur128 (murmur128 bytes options)
+         :crc64 (crc64 bytes options)
          (if-let [f (@hash-functions (keyword function))]
            (f bytes options)
            (throw
@@ -273,14 +284,6 @@
          (IllegalArgumentException.
            (str "Don't recognize decompressor '" (name algorithm) "'"))))))
 
-(def-compressor zlib
-  [x options]
-  (DeflaterInputStream. (bytes/to-input-stream x options)))
-
-(def-decompressor zlib
-  [x options]
-  (InflaterInputStream. (bytes/to-input-stream x options)))
-
 (defn- in->wrapped-out->in
   [^InputStream stream output-wrapper options]
   (let [chunk-size (get options :chunk-size 65536)
@@ -374,19 +377,27 @@
   [x options]
   (BZip2CompressorInputStream. (bytes/to-input-stream x options) true))
 
-(def-decompressor lzo
-  [x options]
-  (LzoInputStream. (bytes/to-input-stream x options) (LzoDecompressor1x.)))
-
-(def-compressor lzo
-  [x options]
+(def-compressor lz4
+  [x {:keys [safe? fastest? chunk-size]
+      :or {safe? false, fastest? false, chunk-size 1e5}
+      :as options}]
   (bytes->wrapped-out->bytes
     x
-    #(LzoOutputStream. %
-       (-> (LzoLibrary/getInstance) (.newCompressor nil nil))
-       (* 25 1024))
+    #(LZ4BlockOutputStream. %
+       chunk-size
+       (let [^LZ4Factory factory (if safe?
+                                   (LZ4Factory/safeInstance)
+                                   (LZ4Factory/fastestInstance))]
+         (if fastest?
+           (.fastCompressor factory)
+           (.highCompressor factory))))
     options))
 
+(def-decompressor lz4
+  [x options]
+  (LZ4BlockInputStream. (bytes/to-input-stream x options)))
+
+
 ;;;
 
 (defn available-encoders

src/byte_transforms/CRC64.java

@@ -0,0 +1,56 @@
+/*
+ * CRC64
+ *
+ * Author: Lasse Collin <[email protected]>
+ *
+ * This file has been put into the public domain.
+ * You can do whatever you want with this file.
+ */
+
+// this file has been slightly modified, but is otherwise
+// as originally written by Lasse Collin
+package byte_transforms;
+
+public class CRC64 {
+
+    private static final long poly = 0xC96C5795D7870F42L;
+    private static final long crcTable[] = new long[256];
+
+    private long crc = -1;
+
+    static {
+        for (int b = 0; b < crcTable.length; ++b) {
+                long r = b;
+                for (int i = 0; i < 8; ++i) {
+                        if ((r & 1) == 1)
+                                r = (r >>> 1) ^ poly;
+                        else
+                                r >>>= 1;
+                }
+
+                crcTable[b] = r;
+        }
+    }
+
+    public CRC64() {
+    }
+
+    public void update(byte b) {
+        crc = crcTable[(b ^ (int)crc) & 0xFF] ^ (crc >>> 8);
+    }
+
+    public void update(byte[] buf) {
+        update(buf, 0, buf.length);
+    }
+
+    public void update(byte[] buf, int off, int len) {
+        int end = off + len;
+
+        while (off < end)
+            crc = crcTable[(buf[off++] ^ (int)crc) & 0xFF] ^ (crc >>> 8);
+    }
+
+    public long getValue() {
+        return ~crc;
+    }
+}

test/byte_transforms_simple_check.clj

@@ -10,14 +10,14 @@
 
 (def compression-type (gen/elements (bt/available-compressors)))
 
-(def concat-compression-type (gen/elements [:gzip :bzip2 :lzo :snappy]))
+(def concat-compression-type (gen/elements [:gzip :bzip2 :snappy]))
 
 (def not-empty-byte-array (gen/such-that not-empty gen/bytes))
 
 (defn roundtrip-equiv
   [b comp-type]
   (java.util.Arrays/equals
-    b
+    ^bytes b
     (-> b
       (bt/compress comp-type)
       (bt/decompress comp-type)
@@ -26,7 +26,7 @@
 (defn concat-roundtrip-equiv
   [b chunk-size comp-type]
   (java.util.Arrays/equals
-    b
+    ^bytes b
     (->> (bs/to-byte-buffers b {:chunk-size chunk-size})
       (map #(bt/compress % comp-type))
       (#(bt/decompress % comp-type))

test/byte_transforms_test.clj

@@ -54,21 +54,20 @@
 (defn benchmark-compression-fn [algorithm ^bytes data]
   (let [len (alength data)
         now #(System/currentTimeMillis)
-        start (now)
-        iterations (loop [cnt 0]
-                     (if (< (- (now) start) 2000)
-                       (do
-                         (-> data
-                           (bt/compress algorithm)
-                           (bt/decompress algorithm)
-                           bs/to-byte-array)
-                         (recur (inc cnt)))
-                       cnt))
-        end (now)]
-    (float
-      (/
-        (* len iterations 1000 (Math/pow 2 -20))
-        (- end start)))))
+        measure (fn [f]
+                  (let [start (now)]
+                    (loop [cnt 0]
+                      (if (< (- (now) start) 2000)
+                        (do (f) (recur (inc cnt)))
+                        (let [end (now)]
+                          (float
+                            (/
+                              (* len cnt 1000 (Math/pow 2 -20))
+                              (- end start))))))))
+        compressed-data (-> data (bt/compress algorithm) bs/to-byte-array)]
+    [(measure #(-> data (bt/compress algorithm) (bt/decompress algorithm) bs/to-byte-array))
+     (measure #(-> data (bt/compress algorithm) bs/to-byte-array))
+     (measure #(-> compressed-data (bt/decompress algorithm) bs/to-byte-array))]))
 
 (defn measure-compression-fn [algorithm ^bytes data]
   (float
@@ -77,12 +76,16 @@
       (-> data (bt/compress algorithm) bs/to-byte-array alength))))
 
 (deftest ^:benchmark benchmark-compression-algorithms
-  (println "\ncompression roundtrip throughput:")
+  (println "\ncompression throughput:")
   (doseq [c (sort (bt/available-compressors))]
     ;; warmup
     (bt/compress warmup-data c)
-    (let [throughput (benchmark-compression-fn c world-facts)]
-      (println (format "%12s: %.2f MB/s" c throughput))))
+    (let [[roundtrip compress decompress] (benchmark-compression-fn c world-facts)]
+      (println (format "%12s: roundtrip %.2f MB/s, compress %.2f MB/s, decompress %.2f MB/s"
+                 c
+                 roundtrip
+                 compress
+                 decompress))))
   (println "\ncompression factor:")
     (doseq [c (sort (bt/available-compressors))]
       (println (format "%12s: %.2fx" c (measure-compression-fn c world-facts)))))