提交 2c5af7d4 编写于 作者: T Takuya UESHIN 提交者: Sean Owen

[SPARK-13640][SQL] Synchronize ScalaReflection.mirror method.

## What changes were proposed in this pull request?

`ScalaReflection.mirror` method should be synchronized when scala version is `2.10` because `universe.runtimeMirror` is not thread safe.

## How was this patch tested?

I added a test to check thread safety of `ScalaRefection.mirror` method in `ScalaReflectionSuite`, which will throw the following Exception in Scala `2.10` without this patch:

```
[info] - thread safety of mirror *** FAILED *** (49 milliseconds)
[info]   java.lang.UnsupportedOperationException: tail of empty list
[info]   at scala.collection.immutable.Nil$.tail(List.scala:339)
[info]   at scala.collection.immutable.Nil$.tail(List.scala:334)
[info]   at scala.reflect.internal.SymbolTable.popPhase(SymbolTable.scala:172)
[info]   at scala.reflect.internal.Symbols$Symbol.unsafeTypeParams(Symbols.scala:1477)
[info]   at scala.reflect.internal.Symbols$TypeSymbol.tpe(Symbols.scala:2777)
[info]   at scala.reflect.internal.Mirrors$RootsBase.init(Mirrors.scala:235)
[info]   at scala.reflect.runtime.JavaMirrors$class.createMirror(JavaMirrors.scala:34)
[info]   at scala.reflect.runtime.JavaMirrors$class.runtimeMirror(JavaMirrors.scala:61)
[info]   at scala.reflect.runtime.JavaUniverse.runtimeMirror(JavaUniverse.scala:12)
[info]   at scala.reflect.runtime.JavaUniverse.runtimeMirror(JavaUniverse.scala:12)
[info]   at org.apache.spark.sql.catalyst.ScalaReflection$.mirror(ScalaReflection.scala:36)
[info]   at org.apache.spark.sql.catalyst.ScalaReflectionSuite$$anonfun$12$$anonfun$apply$mcV$sp$1$$anonfun$apply$1$$anonfun$apply$2.apply(ScalaReflectionSuite.scala:256)
[info]   at org.apache.spark.sql.catalyst.ScalaReflectionSuite$$anonfun$12$$anonfun$apply$mcV$sp$1$$anonfun$apply$1$$anonfun$apply$2.apply(ScalaReflectionSuite.scala:252)
[info]   at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
[info]   at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
[info]   at scala.concurrent.impl.ExecutionContextImpl$$anon$3.exec(ExecutionContextImpl.scala:107)
[info]   at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
[info]   at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
[info]   at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
[info]   at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
```

Notice that the test will pass when Scala version is `2.11`.

Author: Takuya UESHIN <ueshin@happy-camper.st>

Closes #11487 from ueshin/issues/SPARK-13640.
上级 f3201aee
......@@ -32,8 +32,10 @@ object ScalaReflection extends ScalaReflection {
// Since we are creating a runtime mirror usign the class loader of current thread,
// we need to use def at here. So, every time we call mirror, it is using the
// class loader of the current thread.
override def mirror: universe.Mirror =
// SPARK-13640: Synchronize this because universe.runtimeMirror is not thread-safe in Scala 2.10.
override def mirror: universe.Mirror = ScalaReflectionLock.synchronized {
universe.runtimeMirror(Thread.currentThread().getContextClassLoader)
}
import universe._
......@@ -665,7 +667,8 @@ trait ScalaReflection {
*
* @see SPARK-5281
*/
def localTypeOf[T: TypeTag]: `Type` = {
// SPARK-13640: Synchronize this because TypeTag.tpe is not thread-safe in Scala 2.10.
def localTypeOf[T: TypeTag]: `Type` = ScalaReflectionLock.synchronized {
val tag = implicitly[TypeTag[T]]
tag.in(mirror).tpe.normalize
}
......
......@@ -17,10 +17,15 @@
package org.apache.spark.sql.catalyst
import java.net.URLClassLoader
import java.sql.{Date, Timestamp}
import scala.reflect.runtime.universe.typeOf
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.catalyst.expressions.BoundReference
import org.apache.spark.sql.types._
import org.apache.spark.util.Utils
case class PrimitiveData(
intField: Int,
......@@ -236,4 +241,40 @@ class ScalaReflectionSuite extends SparkFunSuite {
assert(anyTypes.forall(!_.isPrimitive))
assert(anyTypes === Seq(classOf[java.lang.Object], classOf[java.lang.Object]))
}
private val dataTypeForComplexData = dataTypeFor[ComplexData]
private val typeOfComplexData = typeOf[ComplexData]
Seq(
("mirror", () => mirror),
("dataTypeFor", () => dataTypeFor[ComplexData]),
("constructorFor", () => constructorFor[ComplexData]),
("extractorsFor", {
val inputObject = BoundReference(0, dataTypeForComplexData, nullable = false)
() => extractorsFor[ComplexData](inputObject)
}),
("getConstructorParameters(cls)", () => getConstructorParameters(classOf[ComplexData])),
("getConstructorParameterNames", () => getConstructorParameterNames(classOf[ComplexData])),
("getClassFromType", () => getClassFromType(typeOfComplexData)),
("schemaFor", () => schemaFor[ComplexData]),
("localTypeOf", () => localTypeOf[ComplexData]),
("getClassNameFromType", () => getClassNameFromType(typeOfComplexData)),
("getParameterTypes", () => getParameterTypes(() => ())),
("getConstructorParameters(tpe)", () => getClassNameFromType(typeOfComplexData))).foreach {
case (name, exec) =>
test(s"SPARK-13640: thread safety of ${name}") {
(0 until 100).foreach { _ =>
val loader = new URLClassLoader(Array.empty, Utils.getContextOrSparkClassLoader)
(0 until 10).par.foreach { _ =>
val cl = Thread.currentThread.getContextClassLoader
try {
Thread.currentThread.setContextClassLoader(loader)
exec()
} finally {
Thread.currentThread.setContextClassLoader(cl)
}
}
}
}
}
}
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册