Fix a thread-safety issue in "local" mode.

The issue is that SparkConf is not thread-safe; so it was possible
for the executor thread to try to read the configuration while the
context thread was modifying it. In my tests this caused the executor
to consistently miss the "spark.driver.port" config and fail tests.

Long term, it would probably be better to investigate using a concurrent
map implementation in SparkConf (instead of a HashMap).

Author: Marcelo Vanzin

core/src/main/scala/org/apache/spark/executor/Executor.scala

@@ -43,18 +43,25 @@ private[spark] class Executor(
     executorId: String,
     slaveHostname: String,
     env: SparkEnv,
+    conf: SparkConf,
     isLocal: Boolean = false)
   extends Logging
 {
+
+  def this(executorId: String,
+      slaveHostname: String,
+      env: SparkEnv,
+      isLocal: Boolean = false) = {
+    this(executorId, slaveHostname, env, env.conf, isLocal)
+  }
+
   // Application dependencies (added through SparkContext) that we've fetched so far on this node.
   // Each map holds the master's timestamp for the version of that file or JAR we got.
   private val currentFiles: HashMap[String, Long] = new HashMap[String, Long]()
   private val currentJars: HashMap[String, Long] = new HashMap[String, Long]()
 
   private val EMPTY_BYTE_BUFFER = ByteBuffer.wrap(new Array[Byte](0))
 
-  private val conf = env.conf
-
   @volatile private var isStopped = false
 
   // No ip or host:port - just hostname

core/src/main/scala/org/apache/spark/scheduler/local/LocalBackend.scala

@@ -21,7 +21,7 @@ import java.nio.ByteBuffer
 
 import akka.actor.{Actor, ActorRef, Props}
 
-import org.apache.spark.{Logging, SparkContext, SparkEnv, TaskState}
+import org.apache.spark.{Logging, SparkConf, SparkContext, SparkEnv, TaskState}
 import org.apache.spark.TaskState.TaskState
 import org.apache.spark.executor.{Executor, ExecutorBackend}
 import org.apache.spark.scheduler.{SchedulerBackend, TaskSchedulerImpl, WorkerOffer}
@@ -43,6 +43,7 @@ private case class StopExecutor()
 private[spark] class LocalActor(
     scheduler: TaskSchedulerImpl,
     executorBackend: LocalBackend,
+    conf: SparkConf,
     private val totalCores: Int)
   extends Actor with ActorLogReceive with Logging {
 
@@ -52,7 +53,7 @@ private[spark] class LocalActor(
   private val localExecutorHostname = "localhost"
 
   private val executor = new Executor(
-    localExecutorId, localExecutorHostname, SparkEnv.get, isLocal = true)
+    localExecutorId, localExecutorHostname, SparkEnv.get, conf, isLocal = true)
 
   override def receiveWithLogging = {
     case ReviveOffers =>
@@ -90,11 +91,12 @@ private[spark] class LocalBackend(scheduler: TaskSchedulerImpl, val totalCores:
   extends SchedulerBackend with ExecutorBackend {
 
   private val appId = "local-" + System.currentTimeMillis
+  private val conf = SparkEnv.get.conf.clone()
   var localActor: ActorRef = null
 
   override def start() {
     localActor = SparkEnv.get.actorSystem.actorOf(
-      Props(new LocalActor(scheduler, this, totalCores)),
+      Props(new LocalActor(scheduler, this, conf, totalCores)),
       "LocalBackendActor")
   }