Merge branch 'develop' of github.com:twitter/chill into Kryo3Upgrade

Author: ianoc

.travis.yml

@@ -7,3 +7,5 @@ jdk:
   - openjdk6
   - oraclejdk7
   - oraclejdk8
+script: ./sbt ++$TRAVIS_SCALA_VERSION clean test doc
+

CHANGES.md

@@ -1,5 +1,8 @@
 # chill #
 
+### 0.7.4 ###
+* If the inner Instantiator is cached we should just cache the kayo and…: https://github.com/twitter/chill/pull/243
+
 ### 0.7.3 ###
 * Update the build: https://github.com/twitter/chill/pull/248
 * Use currentThread classloader: https://github.com/twitter/chill/pull/247

README.md

@@ -159,7 +159,7 @@ Discussion occurs primarily on the [Chill mailing list](https://groups.google.co
 
 ## Maven
 
-Chill modules are available on Maven Central. The current groupid and version for all modules is, respectively, `"com.twitter"` and  `0.7.3` and each scala project is published for `2.10` and `2.11`. Search [search.maven.org](http://search.maven.org/#search%7Cga%7C1%7Cchill) when in doubt.
+Chill modules are available on Maven Central. The current groupid and version for all modules is, respectively, `"com.twitter"` and  `0.7.4` and each scala project is published for `2.10` and `2.11`. Search [search.maven.org](http://search.maven.org/#search%7Cga%7C1%7Cchill) when in doubt.
 
 ## Authors
 

chill-hadoop/src/main/java/com/twitter/chill/hadoop/KryoSerialization.java

@@ -37,7 +37,24 @@
 public class KryoSerialization extends Configured implements Serialization<Object> {
     // can't be final because we need to set them in setConf (for Configured)
     KryoPool kryoPool;
-    Kryo testKryo;
+
+    private static KryoPool cachedPool = null;
+    private static KryoInstantiator cachedKryoInst = null;
+
+    /**
+     * Hadoop will re-initialize the KryoSerialization on every spill
+     * This gets very expensive if you output a lot from a mapper to initialize the chill/kryo stack
+     * The KryoInstantiator's already do some caching, and figuring out if its safe to cache,
+     * so here we piggy back on that to avoid generating new kryo's or kryo pools
+     */
+    public static synchronized void resetOrUpdateFromCache(KryoSerialization instance, KryoInstantiator kryoInst){
+      if(kryoInst != cachedKryoInst) {
+        cachedPool = KryoPool.withByteArrayOutputStream(MAX_CACHED_KRYO, kryoInst);
+        cachedKryoInst = kryoInst;
+      }
+      instance.kryoPool = cachedPool;
+    }
+
     /**
      * Since each thread only needs 1 Kryo, the pool doesn't need more
      * space than the number of threads. We guess that there are 4 hyperthreads /
@@ -69,8 +86,7 @@ public void setConf(Configuration conf) {
 	if (conf != null) {
 	    try {
 		KryoInstantiator kryoInst = new ConfiguredInstantiator(new HadoopConfig(conf));
-		testKryo = kryoInst.newKryo();
-		kryoPool = KryoPool.withByteArrayOutputStream(MAX_CACHED_KRYO, kryoInst);
+        resetOrUpdateFromCache(this, kryoInst);
 	    }
 	    catch(ConfigurationException cx) {
 		// This interface can't throw
@@ -86,7 +102,7 @@ public void setConf(Configuration conf) {
      */
     public boolean accept(Class<?> aClass) {
         try {
-            return (testKryo.getRegistration(aClass) != null);
+            return kryoPool.hasRegistration(aClass);
         } catch (IllegalArgumentException e) {
             return false;
         }

chill-java/src/main/java/com/twitter/chill/KryoPool.java

@@ -131,4 +131,14 @@ public byte[] toBytesWithoutClass(Object obj) {
       release(serde);
     }
   }
+
+  public boolean hasRegistration(Class obj) {
+    SerDeState serde = borrow();
+    try {
+      return serde.hasRegistration(obj);
+    }
+    finally {
+      release(serde);
+    }
+  }
 }

chill-java/src/main/java/com/twitter/chill/SerDeState.java

@@ -68,4 +68,8 @@ public Object readClassAndObject() {
   public void writeOutputTo(OutputStream os) throws IOException {
     os.write(outputToBytes());
   }
+
+  public boolean hasRegistration(Class obj) {
+    return kryo.getRegistration(obj) != null;
+  }
 }

chill-java/src/main/java/com/twitter/chill/config/ConfiguredInstantiator.java

@@ -40,7 +40,7 @@ public class ConfiguredInstantiator extends KryoInstantiator {
   protected final KryoInstantiator delegate;
 
   /** Key we use to configure this class.
-   * Format: <class of KryoInstantiator>(:<base64 serialized instantiator>)
+   * Format: {@literal<class of KryoInstantiator>(:<base64 serialized instantiator>)}
    * if there is no serialized instantiator, we use the reflected instance
    * as the delegate
    */

version.sbt

@@ -1 +1 @@
-version in ThisBuild := "0.8.0-SNAPSHOT"
+version in ThisBuild := "0.8.0-SNAPSHOT"