Merge branch 'develop'

Author: ianoc

.gitignore

@@ -1,3 +1,4 @@
+sonatype.sbt
 # see also test/files/.gitignore
 /test/files/.gitignore
 

CHANGES.md

@@ -1,5 +1,12 @@
 # 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
+
 ### 0.7.2 ###
 * Setting the class loader by default: https://github.com/twitter/chill/pull/242
 

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.2` 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
 

build.sbt

@@ -0,0 +1,231 @@
+import ReleaseTransformations._
+import com.typesafe.sbt.SbtScalariform._
+import com.typesafe.tools.mima.plugin.MimaKeys._
+import com.typesafe.tools.mima.plugin.MimaPlugin.mimaDefaultSettings
+import scala.collection.JavaConverters._
+import scalariform.formatter.preferences._
+
+val kryoVersion = "2.21"
+val bijectionVersion = "0.9.0"
+val algebirdVersion = "0.12.0"
+
+def isScala210x(scalaVersion: String) = scalaVersion match {
+    case version if version startsWith "2.10" => true
+    case _ => false
+}
+
+val sharedSettings = Project.defaultSettings ++ mimaDefaultSettings ++ scalariformSettings ++ Seq(
+  organization := "com.twitter",
+  scalaVersion := "2.10.5",
+  crossScalaVersions := Seq("2.10.5", "2.11.7"),
+  scalacOptions ++= Seq("-unchecked", "-deprecation"),
+  ScalariformKeys.preferences := formattingPreferences,
+
+  // Twitter Hadoop needs this, sorry 1.7 fans
+  javacOptions ++= Seq("-target", "1.6", "-source", "1.6", "-Xlint:-options"),
+  javacOptions in doc := Seq("-source", "1.6"),
+
+  resolvers ++= Seq(
+    Opts.resolver.sonatypeSnapshots,
+    Opts.resolver.sonatypeReleases
+  ),
+  libraryDependencies ++= Seq(
+    "org.scalacheck" %% "scalacheck" % "1.11.5" % "test",
+    "org.scalatest" %% "scalatest" % "2.2.2" % "test",
+    "com.esotericsoftware.kryo" % "kryo" % kryoVersion
+  ),
+
+  parallelExecution in Test := true,
+
+  // Publishing options:
+  releaseCrossBuild := true,
+  releasePublishArtifactsAction := PgpKeys.publishSigned.value,
+  publishMavenStyle := true,
+  publishArtifact in Test := false,
+  pomIncludeRepository := { x => false },
+
+  publishTo <<= version { v =>
+    Some(
+      if (v.trim.toUpperCase.endsWith("SNAPSHOT"))
+        Opts.resolver.sonatypeSnapshots
+      else
+        Opts.resolver.sonatypeStaging
+        //"twttr" at "http://artifactory.local.twitter.com/libs-releases-local"
+    )
+  },
+  pomExtra := (
+    <url>https://github.com/twitter/chill</url>
+        <licenses>
+      <license>
+      <name>Apache 2</name>
+      <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+        <distribution>repo</distribution>
+      <comments>A business-friendly OSS license</comments>
+      </license>
+      </licenses>
+      <scm>
+      <url>git@github.com:twitter/chill.git</url>
+      <connection>scm:git:git@github.com:twitter/chill.git</connection>
+      </scm>
+      <developers>
+      <developer>
+      <id>oscar</id>
+      <name>Oscar Boykin</name>
+      <url>http://twitter.com/posco</url>
+        </developer>
+      <developer>
+      <id>sritchie</id>
+      <name>Sam Ritchie</name>
+      <url>http://twitter.com/sritchie</url>
+        </developer>
+      </developers>)
+)
+
+// Aggregated project
+lazy val chillAll = Project(
+  id = "chill-all",
+  base = file("."),
+  settings = sharedSettings
+).settings(
+  test := { },
+  publish := { },
+  publishLocal := { }
+).aggregate(
+  chill,
+  chillBijection,
+  chillScrooge,
+  chillStorm,
+  chillJava,
+  chillHadoop,
+  chillThrift,
+  chillProtobuf,
+  chillAkka,
+  chillAvro,
+  chillAlgebird
+  )
+
+lazy val formattingPreferences = {
+ import scalariform.formatter.preferences._
+ FormattingPreferences().
+   setPreference(AlignParameters, false).
+   setPreference(PreserveSpaceBeforeArguments, true)
+}
+
+/**
+  * This returns the youngest jar we released that is compatible
+  * with the current.
+  */
+val unreleasedModules = Set[String]("akka")
+val javaOnly = Set[String]("storm", "java", "hadoop", "thrift", "protobuf")
+
+def youngestForwardCompatible(subProj: String) =
+  Some(subProj)
+    .filterNot(unreleasedModules.contains(_))
+    .map { s =>
+    val suffix = if (javaOnly.contains(s)) "" else "_2.10"
+    "com.twitter" % ("chill-" + s + suffix) % "0.7.2"
+  }
+
+def module(name: String) = {
+  val id = "chill-%s".format(name)
+  Project(id = id, base = file(id), settings = sharedSettings ++ Seq(
+    Keys.name := id,
+    previousArtifact := youngestForwardCompatible(name),
+    // Disable cross publishing for java artifacts
+    publishArtifact <<= (scalaVersion) { scalaVersion =>
+      if(javaOnly.contains(name) && scalaVersion.startsWith("2.11")) false else true
+    }
+    )
+  )
+}
+
+// We usually do the pattern of having a core module, but we don't want to cause
+// pain for legacy deploys. With this, they can stay the same.
+lazy val chill = Project(
+  id = "chill",
+  base = file("chill-scala"),
+  settings = sharedSettings
+).settings(
+  name := "chill",
+  previousArtifact := Some("com.twitter" % "chill_2.10" % "0.7.2")
+).dependsOn(chillJava)
+
+lazy val chillAkka = module("akka").settings(
+  resolvers += "Typesafe Repository" at "http://repo.typesafe.com/typesafe/releases/",
+  libraryDependencies ++= Seq(
+    "com.typesafe" % "config" % "1.2.1",
+    "com.typesafe.akka" %% "akka-actor" % "2.3.6" % "provided"
+  )
+).dependsOn(chill % "test->test;compile->compile")
+
+lazy val chillBijection = module("bijection").settings(
+  libraryDependencies ++= Seq(
+    "com.twitter" %% "bijection-core" % bijectionVersion
+  )
+).dependsOn(chill % "test->test;compile->compile")
+
+// This can only have java deps!
+lazy val chillJava = module("java").settings(
+  crossPaths := false,
+  autoScalaLibrary := false
+)
+
+// This can only have java deps!
+lazy val chillStorm = module("storm").settings(
+  crossPaths := false,
+  autoScalaLibrary := false,
+  resolvers ++= Seq(
+    "Clojars Repository" at "http://clojars.org/repo",
+    "Conjars Repository" at "http://conjars.org/repo"
+  ),
+  libraryDependencies += "storm" % "storm" % "0.9.0-wip9" % "provided"
+).dependsOn(chillJava)
+
+// This can only have java deps!
+lazy val chillHadoop = module("hadoop").settings(
+  crossPaths := false,
+  autoScalaLibrary := false,
+  libraryDependencies ++= Seq(
+    "org.apache.hadoop" % "hadoop-core" % "0.20.2" % "provided",
+    "org.slf4j" % "slf4j-api" % "1.6.6",
+    "org.slf4j" % "slf4j-log4j12" % "1.6.6" % "provided"
+  )
+).dependsOn(chillJava)
+
+// This can only have java deps!
+lazy val chillThrift = module("thrift").settings(
+  crossPaths := false,
+  autoScalaLibrary := false,
+  libraryDependencies ++= Seq(
+    "org.apache.thrift" % "libthrift" % "0.6.1" % "provided"
+  )
+)
+
+lazy val chillScrooge = module("scrooge").settings(
+  libraryDependencies ++= Seq(
+    "org.apache.thrift" % "libthrift" % "0.6.1" exclude("junit", "junit"),
+    "com.twitter" %% "scrooge-serializer" % "3.20.0"
+  )
+).dependsOn(chill % "test->test;compile->compile")
+
+// This can only have java deps!
+lazy val chillProtobuf = module("protobuf").settings(
+  crossPaths := false,
+  autoScalaLibrary := false,
+  libraryDependencies ++= Seq(
+    "com.google.protobuf" % "protobuf-java" % "2.3.0" % "provided"
+  )
+).dependsOn(chillJava)
+
+lazy val chillAvro = module("avro").settings(
+  libraryDependencies ++= Seq(
+    "com.twitter" %% "bijection-avro" % bijectionVersion,
+    "junit" % "junit" % "4.5" % "test"
+  )
+).dependsOn(chill,chillJava, chillBijection)
+
+lazy val chillAlgebird = module("algebird").settings(
+  libraryDependencies ++= Seq(
+    "com.twitter" %% "algebird-core" % algebirdVersion
+  )
+).dependsOn(chill)

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/KryoInstantiator.java

@@ -27,6 +27,17 @@
 public class KryoInstantiator implements Serializable {
   public Kryo newKryo() { return new Kryo(); }
 
+  /** Use this to set a specific classloader
+   */
+  public KryoInstantiator setClassLoader(final ClassLoader cl) {
+    return new KryoInstantiator() {
+      public Kryo newKryo() {
+        Kryo k = KryoInstantiator.this.newKryo();
+        k.setClassLoader(cl);
+        return k;
+      }
+    };
+  }
   /** If true, Kryo will error if it sees a class that has not been registered
    */
   public KryoInstantiator setInstantiatorStrategy(final InstantiatorStrategy inst) {
@@ -72,6 +83,17 @@ public Kryo newKryo() {
       }
     };
   }
+  /** Use Thread.currentThread().getContextClassLoader() as the ClassLoader where ther newKryo is called
+   */
+  public KryoInstantiator setThreadContextClassLoader() {
+    return new KryoInstantiator() {
+      public Kryo newKryo() {
+        Kryo k = KryoInstantiator.this.newKryo();
+        k.setClassLoader(Thread.currentThread().getContextClassLoader());
+        return k;
+      }
+    };
+  }
 
   public KryoInstantiator withRegistrar(final IKryoRegistrar r) {
     return new KryoInstantiator() {

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

@@ -57,7 +57,7 @@ public ConfiguredInstantiator(Config conf) throws ConfigurationException {
         throw new ConfigurationException("Invalid Config Key: " + conf.get(KEY));
       }
       KryoInstantiator reflected = null;
-      try { reflected = reflect((Class<? extends KryoInstantiator>)Class.forName(parts[0]), conf); }
+      try { reflected = reflect((Class<? extends KryoInstantiator>)Class.forName(parts[0], true, Thread.currentThread().getContextClassLoader()), conf); }
       catch(ClassNotFoundException x) {
         throw new ConfigurationException("Could not find class for: " + parts[0], x);
       }