Stylistic changes and cleanups

andrewor14 · Apr 5, 2014 · f4cafa0 · f4cafa0
1 parent b3d9c86
commit f4cafa0
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 15 deletions.
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -1028,21 +1028,18 @@ class SparkContext(
   def cancelAllJobs() {
     dagScheduler.cancelAllJobs()
   }
-
-  /**
-   * Clean a closure to make it ready to serialized and send to tasks
-   * (removes unreferenced variables in $outer's, updates REPL variables)
-   */
-  private[spark] def clean[F <: AnyRef : ClassTag](f: F): F = {
-    clean(f, true)
-  }
-
+
   /**
    * Clean a closure to make it ready to serialized and send to tasks
    * (removes unreferenced variables in $outer's, updates REPL variables)
+   *
+   * @param f closure to be cleaned and optionally serialized
+   * @param captureNow whether or not to serialize this closure and capture any free 
+   * variables immediately; defaults to true.  If this is set and f is not serializable, 
+   * it will raise an exception.
    */
-  private[spark] def clean[F <: AnyRef : ClassTag](f: F, checkSerializable: Boolean): F = {
-    ClosureCleaner.clean(f, checkSerializable)
+  private[spark] def clean[F <: AnyRef : ClassTag](f: F, captureNow: Boolean = true): F = {
+    ClosureCleaner.clean(f, captureNow)
   }
 
   /**

diff --git a/core/src/main/scala/org/apache/spark/util/ClosureCleaner.scala b/core/src/main/scala/org/apache/spark/util/ClosureCleaner.scala
@@ -105,7 +105,7 @@ private[spark] object ClosureCleaner extends Logging {
     }
   }
 
-  def clean[F <: AnyRef : ClassTag](func: F, checkSerializable: Boolean = true): F = {
+  def clean[F <: AnyRef : ClassTag](func: F, captureNow: Boolean = true): F = {
     // TODO: cache outerClasses / innerClasses / accessedFields
     val outerClasses = getOuterClasses(func)
     val innerClasses = getInnerClasses(func)
@@ -155,14 +155,14 @@ private[spark] object ClosureCleaner extends Logging {
       field.set(func, outer)
     }
 
-    if (checkSerializable) {
-      ensureSerializable(func)
+    if (captureNow) {
+      cloneViaSerializing(func)
     } else {
       func
     }
   }
 
-  private def ensureSerializable[T: ClassTag](func: T) = {
+  private def cloneViaSerializing[T: ClassTag](func: T): T = {
     try {
       val serializer = SparkEnv.get.closureSerializer.newInstance()
       serializer.deserialize[T](serializer.serialize[T](func))