[SPARK-6843][core]Add volatile for the "state"

[zhichao-li]

Fix potential visibility problem for the "state" of Executor

The field of "state" is shared and modified by multiple threads. i.e:

Within ExecutorRunner.scala

(1) workerThread = new Thread("ExecutorRunner for " + fullId) {
  override def run() { fetchAndRunExecutor() }
}
 workerThread.start()
// Shutdown hook that kills actors on shutdown.

(2)shutdownHook = new Thread() {
  override def run() {
    killProcess(Some("Worker shutting down"))
  }
}

(3)and also the "Actor thread" for worker. 

I think we should at lease add volatile to ensure the visibility among threads otherwise the worker might send an out-of-date status to the master.

https://issues.apache.org/jira/browse/SPARK-6843

[SparkQA]

Test build #30020 has started for PR 5448 at commit a2386e7.

[zhichao-li]

From the the code of worker , we can see it pass through the state without any guard, but the state of executors might be changed by the other threads.

case RequestWorkerState =>
  sender ! WorkerStateResponse(host, port, workerId, **executors.values.toList**,
    finishedExecutors.values.toList, drivers.values.toList,
    finishedDrivers.values.toList, activeMasterUrl, cores, memory,
    coresUsed, memoryUsed, activeMasterWebUiUrl)

[SparkQA]

Test build #30020 has finished for PR 5448 at commit a2386e7.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.
• This patch does not change any dependencies.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/30020/
Test PASSed.

[srowen]

I think that's technically right, although I find there are probably hundreds more fields in Spark that should be @volatile (or else designed so as to not be shared state across threads). It's OK to fix but rather than peck at the problem, can you identify any logically similar changes that need to happen? in this class, or, in similar 'state' variables? I don't mean to evaluate every field in the code base, but just to see if we can make this about more than one particular instance.