diff options
| -rw-r--r-- | core/src/main/scala/org/apache/spark/SparkEnv.scala | 19 |
1 files changed, 7 insertions, 12 deletions
diff --git a/core/src/main/scala/org/apache/spark/SparkEnv.scala b/core/src/main/scala/org/apache/spark/SparkEnv.scala index 224b5c1744..b581c7b074 100644 --- a/core/src/main/scala/org/apache/spark/SparkEnv.scala +++ b/core/src/main/scala/org/apache/spark/SparkEnv.scala @@ -17,8 +17,9 @@ package org.apache.spark -import collection.mutable -import serializer.Serializer +import java.util.concurrent.atomic.AtomicInteger + +import scala.collection.mutable import akka.actor._ import akka.remote.RemoteActorRefProvider @@ -60,7 +61,7 @@ class SparkEnv private[spark] ( private val pythonWorkers = mutable.HashMap[(String, Map[String, String]), PythonWorkerFactory]() // Number of tasks currently running across all threads - @volatile private var _numRunningTasks = 0 + private val _numRunningTasks = new AtomicInteger(0) // A general, soft-reference map for metadata needed during HadoopRDD split computation // (e.g., HadoopFileRDD uses this to cache JobConfs and InputFormats). @@ -93,15 +94,9 @@ class SparkEnv private[spark] ( /** * Return the number of tasks currently running across all threads */ - def numRunningTasks: Int = _numRunningTasks - - def incrementNumRunningTasks() = synchronized { - _numRunningTasks += 1 - } - - def decrementNumRunningTasks() = synchronized { - _numRunningTasks -= 1 - } + def numRunningTasks: Int = _numRunningTasks.intValue() + def incrementNumRunningTasks(): Int = _numRunningTasks.incrementAndGet() + def decrementNumRunningTasks(): Int = _numRunningTasks.decrementAndGet() } object SparkEnv extends Logging { |