diff options
author | Mridul Muralidharan <mridul@gmail.com> | 2013-04-16 17:05:10 +0530 |
---|---|---|
committer | Mridul Muralidharan <mridul@gmail.com> | 2013-04-16 17:05:10 +0530 |
commit | 323ab8ff3b822af28276e1460db0f9c73d6d6409 (patch) | |
tree | 6103c042413b6abc3cf1416c340361ae89b61130 /core/src | |
parent | 59c380d69a3831f0239b434a0fa1cf26a481d222 (diff) | |
download | spark-323ab8ff3b822af28276e1460db0f9c73d6d6409.tar.gz spark-323ab8ff3b822af28276e1460db0f9c73d6d6409.tar.bz2 spark-323ab8ff3b822af28276e1460db0f9c73d6d6409.zip |
Scala does not prevent variable shadowing ! Sick error due to it ...
Diffstat (limited to 'core/src')
-rw-r--r-- | core/src/main/scala/spark/MapOutputTracker.scala | 1 | ||||
-rw-r--r-- | core/src/main/scala/spark/storage/BlockManager.scala | 4 |
2 files changed, 2 insertions, 3 deletions
diff --git a/core/src/main/scala/spark/MapOutputTracker.scala b/core/src/main/scala/spark/MapOutputTracker.scala index 6e9da02893..fde597ffd1 100644 --- a/core/src/main/scala/spark/MapOutputTracker.scala +++ b/core/src/main/scala/spark/MapOutputTracker.scala @@ -148,7 +148,6 @@ private[spark] class MapOutputTracker extends Logging { logInfo("Doing the fetch; tracker actor = " + trackerActor) val hostPort = Utils.localHostPort() // This try-finally prevents hangs due to timeouts: - var fetchedStatuses: Array[MapStatus] = null try { val fetchedBytes = askTracker(GetMapOutputStatuses(shuffleId, hostPort)).asInstanceOf[Array[Byte]] diff --git a/core/src/main/scala/spark/storage/BlockManager.scala b/core/src/main/scala/spark/storage/BlockManager.scala index c98ee5a0e7..6e861ac734 100644 --- a/core/src/main/scala/spark/storage/BlockManager.scala +++ b/core/src/main/scala/spark/storage/BlockManager.scala @@ -1092,7 +1092,7 @@ class BlockFetcherIterator( logDebug("Got local blocks in " + Utils.getUsedTimeMs(startTime) + " ms") //an iterator that will read fetched blocks off the queue as they arrive. - var resultsGotten = 0 + @volatile private var resultsGotten = 0 def hasNext: Boolean = resultsGotten < totalBlocks @@ -1102,7 +1102,7 @@ class BlockFetcherIterator( val result = results.take() val stopFetchWait = System.currentTimeMillis() _fetchWaitTime += (stopFetchWait - startFetchWait) - bytesInFlight -= result.size + if (! result.failed) bytesInFlight -= result.size while (!fetchRequests.isEmpty && (bytesInFlight == 0 || bytesInFlight + fetchRequests.front.size <= maxBytesInFlight)) { sendRequest(fetchRequests.dequeue()) |