ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

track remoteFetchTime

Browse source
This commit is contained in:
Imran Rashid 2013-02-05 14:29:16 -08:00
parent b29f9cc978
commit 696e4b2167
1 changed files with 4 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
core/src/main/scala/spark/storage/BlockManager.scala
View file

@ -852,6 +852,7 @@ class BlockFetcherIterator(
import blockManager._
private var remoteBytesRead = 0l
private var remoteFetchTime = 0l
if (blocksByAddress == null) {
throw new IllegalArgumentException("BlocksByAddress is null")
@ -893,9 +894,12 @@ class BlockFetcherIterator(
})
bytesInFlight += req.size
val sizeMap = req.blocks.toMap // so we can look up the size of each blockID
val fetchStart = System.currentTimeMillis()
val future = connectionManager.sendMessageReliably(cmId, blockMessageArray.toBufferMessage)
future.onSuccess {
case Some(message) => {
val fetchDone = System.currentTimeMillis()
remoteFetchTime += fetchDone - fetchStart
val bufferMessage = message.asInstanceOf[BufferMessage]
val blockMessageArray = BlockMessageArray.fromBufferMessage(bufferMessage)
for (blockMessage <- blockMessageArray) {