diff options
Diffstat (limited to 'core')
| -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()) |