LuciferYang commented on code in PR #57710:
URL: https://github.com/apache/spark/pull/57710#discussion_r3764427775
##########
common/utils/src/main/scala/org/apache/spark/util/ClosureCleaner.scala:
##########
@@ -1075,26 +1116,19 @@ private[spark] object IndylambdaScalaClosures extends
Logging {
private[spark] class ReturnStatementInClosureException
extends SparkException("Return statements aren't allowed in Spark closures")
-private class ReturnStatementFinder(targetMethodName: Option[String] = None)
- extends ClassVisitor(Opcodes.ASM9) {
+/** Collects the names of all closure methods that contain a non-local return.
*/
+private class ReturnStatementCollector extends ClassVisitor(Opcodes.ASM9) {
+ val found = Set.empty[String]
+
override def visitMethod(access: Int, name: String, desc: String,
sig: String, exceptions: Array[String]): MethodVisitor = {
// $anonfun$ covers indylambda closures
if (name.contains("apply") || name.contains("$anonfun$")) {
- // A method with suffix "$adapted" will be generated in cases like
- // { _:Int => return; Seq()} but not { _:Int => return; true}
- // closure passed is $anonfun$t$1$adapted while actual code resides in
$anonfun$s$1
- // visitor will see only $anonfun$s$1$adapted, so we remove the suffix,
see
- // https://github.com/scala/scala-dev/issues/109
- val isTargetMethod = targetMethodName.isEmpty ||
- name == targetMethodName.get || name ==
targetMethodName.get.stripSuffix("$adapted")
-
new MethodVisitor(Opcodes.ASM9) {
override def visitTypeInsn(op: Int, tp: String): Unit = {
- if (op == Opcodes.NEW &&
tp.contains("scala/runtime/NonLocalReturnControl") &&
- isTargetMethod) {
- throw new ReturnStatementInClosureException
+ if (op == Opcodes.NEW &&
tp.contains("scala/runtime/NonLocalReturnControl")) {
Review Comment:
Line 1136 returns an empty `MethodVisitor` rather than `null` for
non-`apply`/`$anonfun$` methods, and only a `null` return makes ASM skip a
method's Code attribute, so those methods are still fully decoded and the
callbacks land in a no-op. That makes the description's "decode instructions
only for `apply`/`$anonfun$` methods" inaccurate, and a reader underestimates
the per-parse cost. Rewording it to "callbacks fire only for
`apply`/`$anonfun$` methods" would be enough; actually skipping the work needs
a `return null` here, as `FieldAccessFinder:1172` does, which for a 204 KB
`SparkContext.class` is most of the bytes. The deleted code had the same shape,
so this is carried forward rather than introduced here.
##########
core/src/test/scala/org/apache/spark/util/ClosureCleanerSuite.scala:
##########
@@ -199,6 +228,40 @@ object TestObjectWithBogusReturns {
}
}
+object TestObjectWithBogusReturnsAndNullCapture {
+ def run(): Int = {
+ withSpark(new SparkContext("local", "test")) { sc =>
+ val nums = sc.parallelize(Array(1, 2, 3, 4).toImmutableArraySeq)
+ val s: String = null
+ // The closure's first captured argument may be the (null) `s` rather
than the non-local
+ // return's key: the cleaner must still detect the invalid return rather
than bail out on
+ // the null capture.
+ nums.map { x => if (s != null) return 1; x * 2 }
Review Comment:
The test at `ClosureCleanerSuite.scala:63` exists to pin that the
capture-count hoist must not be extended to the null-`getCapturedArg(0)`
bail-out at `ClosureCleaner.scala:284`. That only holds while the closure's
first captured argument is the null `s`. Capture order follows first reference
inside the lambda body, and no assertion guards it. After a harmless-looking
rewrite of the closure body the test degenerates into a duplicate of
`TestObjectWithBogusReturns` and still passes. Consider adding an `Int => Int`
`lastClosure` var to `TestObjectWithBogusReturnsAndNullCapture`, assigned
before `nums.map`, then after `intercept` taking the proxy as :79 does and
asserting `getCapturedArgCount == 2` and `getCapturedArg(0) == null` on it.
##########
core/src/test/scala/org/apache/spark/util/ClosureCleanerSuite.scala:
##########
@@ -199,6 +228,40 @@ object TestObjectWithBogusReturns {
}
}
+object TestObjectWithBogusReturnsAndNullCapture {
+ def run(): Int = {
+ withSpark(new SparkContext("local", "test")) { sc =>
+ val nums = sc.parallelize(Array(1, 2, 3, 4).toImmutableArraySeq)
+ val s: String = null
+ // The closure's first captured argument may be the (null) `s` rather
than the non-local
+ // return's key: the cleaner must still detect the invalid return rather
than bail out on
+ // the null capture.
+ nums.map { x => if (s != null) return 1; x * 2 }
+ 1
+ }
+ }
+}
+
+object TestObjectWithReturnInClosure {
+ // The non-local `return` forces the enclosing method to have type Int, so
it cannot return the
+ // closure to the caller directly; stash it for the test to inspect instead.
+ var lastClosure: Int => Int = null
+ def run(): Int = {
+ val f = (x: Int) => { if (x < 0) return -1; x }
+ lastClosure = f
+ f(1)
+ }
+}
+
+object TestObjectWithoutReturnInClosure {
+ var lastClosure: Int => Int = null
Review Comment:
The `lastClosure` var at :257 is assigned at :260 and read by no test
(:93-94 only use `run()` and `getClass`); it was copied from the twin object,
so a reader assumes some test depends on it. Dropping the var is enough. The
closure itself has to stay, since the assertion at :94 needs it, and the
`run()` call can go too, because referencing the object already forces class
loading.
##########
core/src/test/scala/org/apache/spark/util/ClosureCleanerSuite.scala:
##########
@@ -60,11 +60,40 @@ class ClosureCleanerSuite extends SparkFunSuite {
}
}
+ test("return statements in closures capturing a null value are identified at
cleaning time") {
+ intercept[ReturnStatementInClosureException] {
+ TestObjectWithBogusReturnsAndNullCapture.run()
+ }
+ }
+
test("return statements from named functions nested in closures don't raise
exceptions") {
val result = TestObjectWithNestedReturns.run()
assert(result === 1)
}
+ test("hasReturnStatement identifies non-local returns per method") {
+ TestObjectWithReturnInClosure.run()
+ val cls = TestObjectWithReturnInClosure.getClass
+ val implMethodName = {
+ val proxy =
+
IndylambdaScalaClosures.getSerializationProxy(TestObjectWithReturnInClosure.lastClosure)
+ assert(proxy.isDefined)
+ proxy.get.getImplMethodName
+ }
+ // Any-method query.
+ assert(ClosureCleaner.hasReturnStatement(cls, None))
+ // Targeted query with the exact impl method name.
+ assert(ClosureCleaner.hasReturnStatement(cls, Some(implMethodName)))
+ // An "$adapted" wrapper name resolves to the underlying method that holds
the closure body
+ // (see https://github.com/scala/scala-dev/issues/109).
+ assert(ClosureCleaner.hasReturnStatement(cls, Some(implMethodName +
"$adapted")))
Review Comment:
The name asserted at :89 is synthesized as `implMethodName + "$adapted"`,
but that closure is a fully specialized `Int => Int` with no `$adapted` method
on the class, so the assertion only checks that `stripSuffix` removes the
suffix it just appended. The real bridge scenario goes uncovered, and if scalac
renames the bridge this assertion still passes. A closure that really gets one
would close the gap, e.g. `(x: Int) => { if (x < 0) return Seq.empty[Int];
Seq(x) }` in a new test object whose `run()` returns `Seq[Int]`, since the
`return` exits the enclosing method and will not compile inside `run(): Int`;
stash it in a `lastClosure` var, then take its proxy's `implMethodName` and
assert it `endsWith("$adapted")`.
##########
common/utils/src/main/scala/org/apache/spark/util/ClosureCleaner.scala:
##########
@@ -34,6 +35,40 @@ import org.apache.spark.internal.Logging
* A cleaner that renders closures serializable if they can be done so safely.
*/
private[spark] object ClosureCleaner extends Logging {
+ /**
+ * Per-class memo of which closure methods contain a non-local return, i.e.
allocate a
+ * `scala/runtime/NonLocalReturnControl`. The verdict is a pure function of
the class's
+ * immutable bytecode, so one ASM parse per class answers for every
`clean()` call.
+ */
+ private val methodsWithNonLocalReturn = new
ClassValue[immutable.Set[String]] {
+ override def computeValue(cls: Class[_]): immutable.Set[String] = {
+ val collector = new ReturnStatementCollector
+ val reader = getClassReader(cls)
+ if (reader != null) {
+ reader.accept(collector, 0)
+ } else {
+ logDebug(s"Cannot get class bytes for ${cls.getName}; skipping
return-statement check")
Review Comment:
Lines 47-52 return an empty set when `getClassReader` yields null, and
`ClassValue` caches it, so "unreadable this once" becomes "this class is
treated as return-free from now on". The old code NPE'd here, so skipping is an
improvement, but the memo makes the skip permanent. For a class whose bytecode
the current classloader cannot reach (dynamic proxies, hidden classes) the
fail-fast is off for the JVM's lifetime, and the user gets an obscure runtime
error instead. In practice the capturing class's bytecode is reachable, and the
fail-fast is best-effort anyway, so no result is computed wrongly. A comment is
enough, with no retry needed: null means the resource is absent, so a retry
gets nothing, while a genuinely transient read failure throws from
`copyStream`, and `ClassValue` records nothing when `computeValue` throws, so
the next `get` recomputes anyway.
##########
common/utils/src/main/scala/org/apache/spark/util/ClosureCleaner.scala:
##########
@@ -312,7 +351,9 @@ private[spark] object ClosureCleaner extends Logging {
}
// Fail fast if we detect return statements in closures
- getClassReader(func.getClass).accept(new ReturnStatementFinder(), 0)
+ if (hasReturnStatement(func.getClass, None)) {
Review Comment:
"Removes a latent NPE" holds only for the indylambda path. On the legacy
path, `false` from :354 falls through to :370, which dereferences
`getClassReader(cls)` for the same `func.getClass` without a guard, so for the
starting closure the NPE just moves to a different stack. A later NPE triage
would then read the description as closing this class of bug, while :370, :938
and :1203 still dereference unconditionally. Scoping the claim to the
indylambda path's return-statement check would make it accurate. The missing
guard at :370 is better left to a follow-up: skipping the scan there leaves
`accessedFields` without this class's fields, so :161 copies none and the
cleaned closure silently loses captured state, which is not the same trade as
the best-effort return-statement check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
