Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ueshin wants to merge 8 commits into from
Closed

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

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
b9d99bd
Add a test to check thread safety of ScalaRefection.mirror method.
ueshin Mar 3, 2016
389d644
Synchronize ScalaReflection.mirror method.
ueshin Mar 3, 2016
549b8f8
Modify test to check other methods using reflection.
ueshin Mar 3, 2016
cee5896
Synchronize ScalaReflection.localTypeOf method.
ueshin Mar 3, 2016
36cfa08
Add comments.
ueshin Mar 7, 2016
0055fd1
Modify test.
ueshin Mar 7, 2016
7ac9648
Merge branch 'master' into issues/SPARK-13640
ueshin Mar 7, 2016
63b1797
Re-modify test.
ueshin Mar 7, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,8 +32,9 @@ 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 =
override def mirror: universe.Mirror = ScalaReflectionLock.synchronized {
universe.runtimeMirror(Thread.currentThread().getContextClassLoader)
}

import universe._

Expand Down Expand Up @@ -665,7 +666,7 @@ trait ScalaReflection {
*
* @see SPARK-5281
*/
def localTypeOf[T: TypeTag]: `Type` = {
def localTypeOf[T: TypeTag]: `Type` = ScalaReflectionLock.synchronized {
val tag = implicitly[TypeTag[T]]
tag.in(mirror).tpe.normalize
}
Expand Down
45 changes: 45 additions & 0 deletions sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/ScalaReflectionSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,12 @@ package org.apache.spark.sql.catalyst
import java.math.BigInteger
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,
Expand Down Expand Up @@ -237,4 +241,45 @@ class ScalaReflectionSuite extends SparkFunSuite {
assert(anyTypes.forall(!_.isPrimitive))
assert(anyTypes === Seq(classOf[java.lang.Object], classOf[java.lang.Object]))
}

private def testThreadSafetyFor(name: String)(exec: () => Any) = {
test(s"thread safety of ${name}") {
for (_ <- 0 until 100) {
Copy link
Member

@srowen srowen Mar 6, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(0 until 100).foreach? probably just a matter of taste but it's symmetrical with your inner loop then.
You can import java.net.URLClassLoader.
It doesn't really seem like you need a method here; it took a moment to see there was a test in here.

Maybe it's obvious to you but why do all these classes/methods need to be tested separately?

And is this locking still safe in 2.11?

Copy link
Member Author

@ueshin ueshin Mar 7, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@srowen Thank you for your comment.

(0 until 100).foreach?

I repeated the test 100 times here because it is for thread-safety. Thread safety problem sometimes happens but sometimes doesn't.

You can import java.net.URLClassLoader.

I'll modify to use import.

It doesn't really seem like you need a method here; it took a moment to see there was a test in here.

I'll modify to move out of the method.

Maybe it's obvious to you but why do all these classes/methods need to be tested separately?

The methods are public, i.e. can be called by multi-thread, so I thought these also need to be tested.
But I'm wondering some of them could be removed?

And is this locking still safe in 2.11?

Yes, reflection in Scala 2.11 is thread-safe.
If we don't support Scala 2.10, these lockings in ScalaReflection would not be needed.

Copy link
Member

@srowen srowen Mar 7, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes I understand repeating it 100 times, but your idiom for iterating n times is different on two nearby lines.

I actually meant the opposite -- remove the helper method but run tests in a loop still -- but leave it.

My question was more like, does the extra locking cause trouble in 2.11? I know you said 2.11 doesn't need this change per se.

val loader = new java.net.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)
}
}
}
}
}

private val dataTypeForComplexData = dataTypeFor[ComplexData]
private val typeOfComplexData = typeOf[ComplexData]

for {
(name, exec) <- 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)))
} {
testThreadSafetyFor(name)(exec)
}
}